Alibaba Cloud Model Studio:リアルタイム音声・動画翻訳 - Qwen

qwen3.5-livetranslate-flash-realtime は、60 言語 (音声+テキスト 29 言語、テキストのみ 31 言語) をサポートする視覚拡張リアルタイム翻訳モデルです。ビデオストリームやローカルファイルから音声と画像の入力を処理し、視覚的なコンテキストを使用して精度を向上させ、翻訳されたテキストと音声をリアルタイムで出力します。

Function Compute を使用したワンクリックデプロイでオンラインデモをお試しください。

特徴

• 多言語サポート：中国語、英語、フランス語、ドイツ語、ロシア語、日本語、韓国語、スペイン語、ポルトガル語、アラビア語など、60 言語間の翻訳に対応しています (29 言語は音声とテキスト出力、31 言語はテキストのみの出力)。
• 視覚拡張：唇の動き、ジェスチャー、画面上のテキストなどの視覚的な手がかりを分析して、特に騒がしい環境や曖昧な単語の翻訳精度を向上させます。
• 2.8 秒のレイテンシー：わずか 2.8 秒の低レイテンシーで同時通訳を実現します。
• ロスレス同時通訳：意味単位を予測して言語間の語順の違いを解決し、オフライン翻訳に匹敵する品質を実現します。
• 自然な音声：ソース音声のイントネーションや感情に自動的に合わせます。
• ホットワード構成：設定可能なホットワードにより、特定の用語の翻訳精度が向上します。
• 音声クローニング：話者の声をクローンして翻訳出力に使用します。サーバーサイドのリアルタイムクローニングと、事前にクローンされた音声プロファイルをサポートしています。
• 同一言語出力のスキップ：ソース言語とターゲット言語が同じ場合、モデルはテキスト出力、音声出力、またはその両方をスキップできます。この機能は、ターゲット言語が中国語 (zh) または英語 (en) の場合にのみ有効です。

操作手順

1. 接続の構成

このモデルは、WebSocket を介して以下のパラメーターで接続します：

WebSocket に加えて、このモデルは AOQ および WebRTC プロトコルもサポートしています。安定したレイテンシー、低速ネットワークでの回復力、組み込みの全二重ノイズ抑制およびエコーキャンセレーションを優先するクライアントサイドの統合には、AOQ を推奨します。プロトコルの比較については、「リアルタイム API の概要」をご参照ください。

接続コードのサンプル (Python)：

# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"サーバーに接続しました: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("受信したイベント:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("エラー:", error)

ws = websocket.WebSocketApp(
    API_URL,
    header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)

ws.run_forever()

2. 言語、モダリティ、音声の構成

以下のパラメーターを指定して session.update クライアントイベントを送信します：

• 言語

  • ソース言語：session.input_audio_transcription.language パラメーターを使用して構成します。

    指定しない場合、モデルはソース言語を自動的に検出します。

  • ターゲット言語：session.translation.language パラメーターを使用して構成します。

    デフォルト値は en (英語) です。

  詳細については、「サポート対象言語」をご参照ください。

• ソース言語の認識結果の出力

  session.input_audio_transcription.model を qwen3-asr-flash-realtime に設定します。これにより、サーバーは翻訳結果と入力音声の音声認識結果 (原文テキスト) の両方を返します。

  サーバーは以下のイベントを返します：

  • conversation.item.input_audio_transcription.text：認識結果をストリーム配信します。
  • conversation.item.input_audio_transcription.completed：認識完了後に最終結果を返します。
  • conversation.item.input_audio_transcription.failed：認識が失敗した場合にエラー情報を返します。
• 出力モダリティ

  session.modalities パラメーターを ["text"] (テキストのみ) または ["text","audio"] (テキストと音声) に設定します。

• 音声アクティビティ検出 (VAD) と手動モード

  session.turn_detection パラメーターを使用して、発話の境界をどのように検出するかを構成します：

  • VAD モード (デフォルト)：turn_detection を構成オブジェクトに設定します。サーバーは自動的に発話の境界を検出し、翻訳をトリガーします。クライアントが継続的に音声ストリームを送信するシナリオに適しています。
  • 手動モード：turn_detection を null に設定します。クライアントが発話の境界を決定し、各発話の後に input_audio_buffer.commit イベントを送信して音声を提出します。プッシュツートークのシナリオに適しています。

  両モードでの完全なインタラクション手順については、「3. 音声と画像の入力」をご参照ください。

• 音声

  session.voice パラメーターを使用して構成します。詳細については、「サポート対象の音声」をご参照ください。

• ホットワード

  session.translation.corpus.phrases パラメーターを使用してホットワードを構成します。ホットワードは、ソース用語をターゲットの翻訳にマッピングするキーと値のペアであり、特定の用語の翻訳精度を向上させます。

  例："artificial intelligence" を "Artificial Intelligence" にマッピングします。

• 音声クローニング

  session.enable_voice_clone、session.voice_clone_options.frequency、session.voice パラメーターを使用して構成します。事前にクローンされた音声プロファイル (frequency：never)、セッション開始時にサーバーサイドで一度クローン (once)、または各応答の前にリアルタイムでクローン (always) の 3 つのモードをサポートしています。詳細については、「音声クローニング」をご参照ください。

3. 音声と画像の入力

input_audio_buffer.append および input_image_buffer.append イベントを使用して、Base64 エンコードされた音声および画像データを送信します。音声入力は必須ですが、画像入力はオプションです。

画像はローカルファイルから、またはビデオストリームからリアルタイムでキャプチャできます。

モデルが発話の完了を判断する方法は、turn_detection パラメーターを介して構成された VAD モードまたは手動モードによって異なります：

• VAD モード (デフォルト)：クライアントは継続的に input_audio_buffer.append イベントを送信します。サーバーが発話の開始/終了を検出すると、それぞれ input_audio_buffer.speech_started および input_audio_buffer.speech_stopped イベントを返し、音声バッファーを自動的にコミットして翻訳をトリガーします。翻訳応答はストリーミング音声と同期して生成され、通常は発話が終了するのを待たずに音声入力中に開始されます。
• 手動モード：session.turn_detection を null に設定します。クライアントは完全な発話の送信を終えた後、input_audio_buffer.commit イベントを送信して音声バッファーをコミットします。サーバーが input_audio_buffer.committed イベントを返して確認した後、自動的に翻訳応答の生成を開始します。クライアントは応答をトリガーするために他のイベントを送信する必要はありません。コミットする前に未コミットの音声をクリアするには、input_audio_buffer.clear イベントを送信します。

4. モデルの応答の受信

翻訳応答はストリーミング音声と同期して生成され、通常は発話の終了を待つ必要はありません (前のセクションの VAD/手動モードの説明をご参照ください)。応答形式は出力モダリティによって異なります。

• テキストのみの出力

  サーバーは response.text.text イベントを介して、インクリメンタルな翻訳テキスト (確定テキストと仮の予測テキストを含む) をストリーム配信します。完了すると、完全な翻訳テキストが response.text.done イベントで返されます。

• テキストと音声の出力
  • テキスト

    サーバーは response.audio_transcript.text イベントを介して、インクリメンタルな翻訳テキストをストリーム配信します。完了すると、完全な翻訳テキストが response.audio_transcript.done イベントで返されます。

  • 音声

    サーバーは、response.audio.delta イベントで、インクリメンタルな Base64 エンコードされた音声データを返します。

5. セッションの終了

すべての音声を送信した後、クライアントイベントを送信し、サーバーが session.finished イベントを返すのを待ってから WebSocket 接続を閉じます。

session.finish を送信せずに WebSocket を閉じると、サーバーの VAD は最後の発話セグメントの終了を検出できません。これにより、そのセグメントの翻訳結果が完全に失われ、接続が無期限にハングする可能性があります。切断する前に必ずこのイベントを送信してください。

サポート対象モデル

推奨モデル

レガシーモデル

以下のモデルは引き続き利用可能ですが、推奨される選択肢ではありません。新しいユースケースでは、より良い翻訳品質とコスト効率のために、上記の新しいモデルを使用してください。

クイックスタート

1. 環境の準備

   Python 3.10 以降が必要です。

   まず、pyaudio をインストールします。

   macOS

   brew install portaudio && pip install pyaudio
   

   Debian/Ubuntu

   sudo apt-get install python3-pyaudio
   
   or
   
   pip install pyaudio
   

   CentOS

   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
   

   Windows

   pip install pyaudio
   

   次に、WebSocket の依存関係をインストールします：

pip install websocket-client==1.8.0 websockets

2. クライアントの作成

   livetranslate_client.py という名前のファイルを作成し、以下のコードを記述します：

   クライアントコード - livetranslate_client.py

   import os
   import time
   import base64
   import asyncio
   import json
   import websockets
   import pyaudio
   import queue
   import threading
   import traceback
   
   class LiveTranslateClient:
       def __init__(self, api_key: str, target_language: str = "en", *, audio_enabled: bool = True):
           if not api_key:
               raise ValueError("API キーは空にできません。")
   
           self.api_key = api_key
           self.target_language = target_language
           self.audio_enabled = audio_enabled
           self.ws = None
           self.api_url = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"
   
           # 音声入力構成 (マイクから)
           self.input_rate = 16000
           self.input_chunk = 1600
           self.input_format = pyaudio.paInt16
           self.input_channels = 1
   
           # 音声出力構成 (再生用)
           self.output_rate = 24000
           self.output_chunk = 2400
           self.output_format = pyaudio.paInt16
           self.output_channels = 1
   
           # 状態管理
           self.is_connected = False
           self.audio_player_thread = None
           self.audio_playback_queue = queue.Queue()
           self.pyaudio_instance = pyaudio.PyAudio()
           self.session_finished_event = asyncio.Event()
   
       async def connect(self):
           """翻訳サービスへの WebSocket 接続を確立します。"""
           headers = {"Authorization": f"Bearer {self.api_key}"}
           try:
               self.ws = await websockets.connect(self.api_url, additional_headers=headers)
               self.is_connected = True
               print(f"サーバーに正常に接続しました: {self.api_url}")
               await self.configure_session()
           except Exception as e:
               print(f"接続に失敗しました: {e}")
               self.is_connected = False
               raise
   
       async def configure_session(self):
           """翻訳セッションを構成し、ターゲット言語や音声などを設定します。"""
           config = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "session.update",
               "session": {
                   # 'modalities' は出力タイプを制御します。
                   # ["text", "audio"]: 翻訳されたテキストと合成音声の両方を返します (推奨)。
                   # ["text"]: 翻訳されたテキストのみを返します。
                   "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                   "input_audio_format": "pcm",
                   "output_audio_format": "pcm",
                   # 'input_audio_transcription' はソース言語の認識を構成します。
                   # 'model' を 'qwen3-asr-flash-realtime' に設定すると、ソース言語の認識結果も出力されます。
                   # "input_audio_transcription": {
                   #     "model": "qwen3-asr-flash-realtime",
                   #     "language": "zh"  # ソース言語、デフォルトは 'en'
                   # },
                   "translation": {
                       "language": self.target_language,
                       # 'corpus' はホットワードを構成し、特定の用語の翻訳精度を向上させます。
                       # "corpus": {
                       #     "phrases": {
                       #         "Artificial Intelligence": "Artificial Intelligence",
                       #         "Machine Learning": "Machine Learning"
                       #     }
                       # }
                   }
               }
           }
           print(f"セッション構成を送信中: {json.dumps(config, indent=2, ensure_ascii=False)}")
           await self.ws.send(json.dumps(config))
   
       async def send_audio_chunk(self, audio_data: bytes):
           """音声チャンクをエンコードしてサーバーに送信します。"""
           if not self.is_connected:
               return
   
           event = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "input_audio_buffer.append",
               "audio": base64.b64encode(audio_data).decode()
           }
           await self.ws.send(json.dumps(event))
   
       async def send_image_frame(self, image_bytes: bytes, *, event_id: str | None = None):
           # 画像フレームをサーバーに送信します。
           if not self.is_connected:
               return
   
           if not image_bytes:
               raise ValueError("image_bytes は空にできません。")
   
           # Base64 にエンコード
           image_b64 = base64.b64encode(image_bytes).decode()
   
           event = {
               "event_id": event_id or f"event_{int(time.time() * 1000)}",
               "type": "input_image_buffer.append",
               "image": image_b64,
           }
   
           await self.ws.send(json.dumps(event))
   
       def _audio_player_task(self):
           stream = self.pyaudio_instance.open(
               format=self.output_format,
               channels=self.output_channels,
               rate=self.output_rate,
               output=True,
               frames_per_buffer=self.output_chunk,
           )
           try:
               while self.is_connected or not self.audio_playback_queue.empty():
                   try:
                       audio_chunk = self.audio_playback_queue.get(timeout=0.1)
                       if audio_chunk is None: # 終了信号
                           break
                       stream.write(audio_chunk)
                       self.audio_playback_queue.task_done()
                   except queue.Empty:
                       continue
           finally:
               stream.stop_stream()
               stream.close()
   
       def start_audio_player(self):
           """音声プレーヤースレッドを開始します (音声出力が有効な場合のみ)。"""
           if not self.audio_enabled:
               return
           if self.audio_player_thread is None or not self.audio_player_thread.is_alive():
               self.audio_player_thread = threading.Thread(target=self._audio_player_task, daemon=True)
               self.audio_player_thread.start()
   
       async def handle_server_messages(self, on_text_received):
           """サーバーからの受信メッセージをループで処理します。"""
           try:
               async for message in self.ws:
                   event = json.loads(message)
                   event_type = event.get("type")
                   if event_type == "response.audio.delta" and self.audio_enabled:
                       audio_b64 = event.get("delta", "")
                       if audio_b64:
                           audio_data = base64.b64decode(audio_b64)
                           self.audio_playback_queue.put(audio_data)
   
                   elif event_type == "response.done":
                       print("\n[INFO] 応答ラウンドが完了しました。")
                       usage = event.get("response", {}).get("usage", {})
                       if usage:
                           print(f"[INFO] トークン使用量: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                   elif event_type == "session.finished":
                       print("[INFO] セッションが終了しました。")
                       self.session_finished_event.set()
                   # ソース言語の認識結果を処理します (input_audio_transcription.model の有効化が必要)
                   # elif event_type == "conversation.item.input_audio_transcription.text":
                   #     stash = event.get("stash", "")  # 認識待ちのテキスト
                   #     print(f"[認識中] {stash}")
                   # elif event_type == "conversation.item.input_audio_transcription.completed":
                   #     transcript = event.get("transcript", "")  # 完全な認識結果
                   #     print(f"[ソース言語] {transcript}")
                   elif event_type == "response.text.text":
                       # テキストのみのモダリティで翻訳テキストをストリーミング
                       text = event.get("text", "")
                       stash = event.get("stash", "")
                       print(f"\r[翻訳中] {text}{stash}", end="", flush=True)
                   elif event_type == "response.audio_transcript.done":
                       print("\n[INFO] 翻訳が完了しました。")
                       text = event.get("transcript", "")
                       if text:
                           print(f"[INFO] 翻訳されたテキスト: {text}")
                   elif event_type == "response.text.done":
                       print("\n[INFO] 翻訳が完了しました。")
                       text = event.get("text", "")
                       if text:
                           print(f"[INFO] 翻訳されたテキスト: {text}")
   
           except websockets.exceptions.ConnectionClosed as e:
               print(f"[WARNING] 接続が閉じられました: {e}")
               self.is_connected = False
           except Exception as e:
               print(f"[ERROR] メッセージ処理中に予期せぬエラーが発生しました: {e}")
               traceback.print_exc()
               self.is_connected = False
   
       async def start_microphone_streaming(self):
           """マイクから音声をキャプチャし、サーバーにストリーミングします。"""
           stream = self.pyaudio_instance.open(
               format=self.input_format,
               channels=self.input_channels,
               rate=self.input_rate,
               input=True,
               frames_per_buffer=self.input_chunk
           )
           print("マイクがオンです。話し始めてください...")
           try:
               while self.is_connected:
                   audio_chunk = await asyncio.get_event_loop().run_in_executor(
                       None, stream.read, self.input_chunk
                   )
                   await self.send_audio_chunk(audio_chunk)
           finally:
               stream.stop_stream()
               stream.close()
   
       async def close(self):
           """接続を正常に閉じ、リソースを解放します。"""
           # サーバーが最後の発話セグメントの翻訳を完了できるように session.finish を送信します
           if self.is_connected and self.ws:
               finish_event = {
                   "event_id": f"event_{int(time.time() * 1000)}",
                   "type": "session.finish",
               }
               await self.ws.send(json.dumps(finish_event))
               print("session.finish を送信しました。サーバーの処理完了を待っています...")
               try:
                   await asyncio.wait_for(self.session_finished_event.wait(), timeout=15)
                   print("サーバーの処理が完了しました。")
               except asyncio.TimeoutError:
                   print("session.finished の待機がタイムアウトしました。")
           self.is_connected = False
           if self.ws:
               await self.ws.close()
               print("WebSocket 接続が閉じられました。")
   
           if self.audio_player_thread:
               self.audio_playback_queue.put(None) # 終了信号を送信
               self.audio_player_thread.join(timeout=1)
               print("音声プレーヤースレッドが停止しました。")
   
           self.pyaudio_instance.terminate()
           print("PyAudio インスタンスが解放されました。")
   

3. モデルとの対話

   同じディレクトリに main.py という名前のファイルを作成し、以下のコードを記述します：

   main.py

   import os
   import asyncio
   from livetranslate_client import LiveTranslateClient
   
   def print_banner():
       print("=" * 60)
       print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime")
       print("=" * 60 + "\n")
   
   def get_user_config():
       """ユーザー構成を取得します。"""
       print("モードを選択してください:")
       print("1. 音声 + テキスト [デフォルト] | 2. テキストのみ")
       mode_choice = input("選択肢を入力してください (音声 + テキストの場合は Enter を押してください): ").strip()
       audio_enabled = (mode_choice != "2")
   
       if audio_enabled:
           lang_map = {
               "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt",
               "7": "es", "8": "it", "9": "ko", "10": "ja", "11": "yue"
           }
           print("ターゲット言語を選択してください (音声 + テキストモード):")
           print("1. 英語 | 2. 中国語 | 3. ロシア語 | 4. フランス語 | 5. ドイツ語 | 6. ポルトガル語 | 7. スペイン語 | 8. イタリア語 | 9. 韓国語 | 10. 日本語 | 11. 広東語")
       else:
           lang_map = {
               "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt", "7": "es", "8": "it",
               "9": "id", "10": "ko", "11": "ja", "12": "vi", "13": "th", "14": "ar",
               "15": "yue", "16": "hi", "17": "el", "18": "tr"
           }
           print("ターゲット言語を選択してください (テキストのみモード):")
           print("1. 英語 | 2. 中国語 | 3. ロシア語 | 4. フランス語 | 5. ドイツ語 | 6. ポルトガル語 | 7. スペイン語 | 8. イタリア語 | 9. インドネシア語 | 10. 韓国語 | 11. 日本語 | 12. ベトナム語 | 13. タイ語 | 14. アラビア語 | 15. 広東語 | 16. ヒンディー語 | 17. ギリシャ語 | 18. トルコ語")
   
       choice = input("選択肢を入力してください (デフォルトは最初のオプションです): ").strip()
       target_language = lang_map.get(choice, next(iter(lang_map.values())))
   
       return target_language, audio_enabled
   
   async def main():
       """メインプログラムのエントリーポイント。"""
       print_banner()
   
       api_key = os.environ.get("DASHSCOPE_API_KEY")
       if not api_key:
           print("[ERROR] DASHSCOPE_API_KEY 環境変数を設定してください。")
           print("  例: export DASHSCOPE_API_KEY='your_api_key_here'")
           return
   
       target_language, audio_enabled = get_user_config()
       print("\n構成が完了しました:")
       print(f"  - ターゲット言語: {target_language}")
       if not audio_enabled:
           print("  - 出力モード: テキストのみ")
   
       client = LiveTranslateClient(api_key=api_key, target_language=target_language, audio_enabled=audio_enabled)
   
       # コールバック関数を定義します。
       def on_translation_text(text):
           print(text, end="", flush=True)
   
       try:
           print("翻訳サービスに接続中...")
           await client.connect()
   
           # モードに基づいて音声再生を開始します。
           client.start_audio_player()
   
           print("\n" + "-" * 60)
           print("接続に成功しました！マイクに向かって話してください。")
           print("プログラムはあなたの発話をリアルタイムで翻訳し、翻訳された音声を再生します。終了するには Ctrl+C を押してください。")
           print("-" * 60 + "\n")
   
           # メッセージ処理とマイク録音を同時に実行します。
           message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
           tasks = [message_handler]
           # 音声出力が有効かどうかにかかわらず、翻訳のためにマイクから音声をキャプチャします。
           microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
           tasks.append(microphone_streamer)
   
           await asyncio.gather(*tasks)
   
       except KeyboardInterrupt:
           print("\n\nユーザーが中断しました。終了します...")
       except Exception as e:
           print(f"\n重大なエラーが発生しました: {e}")
       finally:
           print("\nリソースをクリーンアップ中...")
           await client.close()
           print("プログラムが終了しました。")
   
   if __name__ == "__main__":
       asyncio.run(main())
   

   main.py を実行し、マイクに向かって話してください。モデルはあなたの発話を翻訳し、音声とテキストをリアルタイムで出力します。

音声クローニング

モデルは入力音声から話者の声をクローンし、翻訳出力に使用します。事前にクローンされた音声プロファイルを使用するか、サーバーにリアルタイムでクローンさせることができます。会議の通訳、ライブストリーミング、ビデオの吹き替えなどに役立ちます。

音声クローニングを有効にするには、session.update で以下のパラメーターを設定します：

• session.enable_voice_clone：true に設定して音声クローニングを有効にします。

• session.voice_clone_options.frequency：音声クローニングが行われるタイミングを制御します。指定可能な値：

  • never：サーバー上ではクローンしません。代わりに事前にクローンされた音声プロファイルを使用します。session.voice をカスタムのクローン音声 ID に設定します。
  • once：セッション開始時に入力音声から一度だけ声をクローンし、その後のすべての出力で再利用します。単一話者のシナリオに最適です。session.voice を default に設定します。
  • always：各応答の前に声をクローンし、話者の変更に動的に適応します。複数話者の会話に最適です。session.voice を default に設定します。

• session.voice：出力する音声を指定します。値は frequency の設定によって異なります：

  • default に設定：frequency を once または always に設定して使用します。サーバーは入力音声から話者の声をクローンします。クローニングが完了するまではデフォルトの音声が使用されます。
  • カスタムのクローン音声 ID (例：qwen-translate-vc-xxx-yyy-zzz) に設定：frequency を never に設定して使用します。音声クローニング API を使用して、targetModel を qwen3.5-livetranslate-flash-realtime に設定し、事前に音声を準備しておく必要があります。

frequency が once または always に設定されている場合、voice パラメーターは default に設定する必要があります。他の値を設定すると、サーバーはエラーを返します。

音声クローニングの構成例

事前にクローンされた音声プロファイル (安定した品質。安定した音声 ID が必要な場合に推奨)：

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "qwen-translate-vc-xxx-yyy-zzz",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "never"
        }
    }
}

サーバーサイドクローニング、セッションごとに 1 回 (単一話者のシナリオに最適)：

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "once"
        }
    }
}

サーバーサイドクローニング、応答ごと (複数話者の会話に最適)：

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "always"
        }
    }
}

画像による翻訳精度の向上

画像入力は、翻訳中の同音異義語の曖昧さを解消し、一般的でない固有名詞の認識に役立ちます。1 秒あたり 2 枚以下の画像を送信してください。

以下のサンプル画像をダウンロードしてください：medical mask.png、masquerade mask.png

以下のコードを livetranslate_client.py と同じディレクトリにダウンロードして実行してください。マイクに向かって "What is mask?" と話してください。モデルは画像を使用して曖昧さを解消します：medical mask.png は「医療用マスクとは何ですか？」を、masquerade mask.png は「仮面舞踏会のマスクとは何ですか？」を生成します。

import os
import time
import json
import asyncio
import contextlib
import functools

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "medical mask.png"
# IMAGE_PATH = "masquerade mask.png"

def print_banner():
    print("=" * 60)
    print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime — シングルターンインタラクションの例 (マスク)")
    print("=" * 60 + "\n")

async def stream_microphone_once(client: LiveTranslateClient, image_bytes: bytes):
    pa = client.pyaudio_instance
    stream = pa.open(
        format=client.input_format,
        channels=client.input_channels,
        rate=client.input_rate,
        input=True,
        frames_per_buffer=client.input_chunk,
    )
    print(f"[INFO] 録音を開始しました。話してください...")
    loop = asyncio.get_event_loop()
    last_img_time = 0.0
    frame_interval = 0.5  # 2 fps
    try:
        while client.is_connected:
            data = await loop.run_in_executor(None, stream.read, client.input_chunk)
            await client.send_audio_chunk(data)

            # 0.5 秒ごとに画像フレームを追加
            now = time.time()
            if now - last_img_time >= frame_interval:
                await client.send_image_frame(image_bytes)
                last_img_time = now
    finally:
        stream.stop_stream()
        stream.close()

async def main():
    print_banner()
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("[ERROR] DASHSCOPE_API_KEY 環境変数を設定してください。")
        return

    client = LiveTranslateClient(api_key=api_key, target_language="zh", audio_enabled=True)

    def on_text(text: str):
        print(text, end="", flush=True)

    try:
        await client.connect()
        client.start_audio_player()
        message_task = asyncio.create_task(client.handle_server_messages(on_text))
        with open(IMAGE_PATH, "rb") as f:
            img_bytes = f.read()
        await stream_microphone_once(client, img_bytes)
        await asyncio.sleep(15)
    finally:
        await client.close()
        if not message_task.done():
            message_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await message_task

if __name__ == "__main__":
    asyncio.run(main())

ワンクリック Function Compute デプロイメント

アプリケーションをデプロイするには：

1. Function Compute テンプレートを開き、API キーを入力して、[デフォルト環境の作成とデプロイ] をクリックしてアプリケーションをテストします。

2. 約 1 分待ちます。[環境詳細] > [環境コンテキスト] でエンドポイントを取得し、プロトコルを http から https に変更し (例：https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/)、ブラウザで URL を開いてモデルと対話します。

   重要このエンドポイントは自己署名証明書を使用しており、一時的なテスト専用です。初めてアクセスすると、ブラウザにセキュリティ警告が表示されます。これは予期された動作です。このエンドポイントを本番環境で使用しないでください。続行するには、画面の指示に従ってください (例：[詳細設定] → [安全でないサイトに移動] をクリック)。

Resource Access Management の権限を設定するよう求められた場合は、画面の指示に従ってください。

プロジェクトのソースコードを表示するには、[リソース情報] > [関数リソース] に移動します。

Function Compute と Model Studio の両方で、新規ユーザー向けに無料クォータが提供されており、基本的なデバッグには十分です。無料クォータを使い切った後は、従量課金が適用されます。

インタラクションフロー

翻訳はイベント駆動型の WebSocket モデルを使用します。発話の境界を決定する方法は、VAD モードまたは手動モードによって異なります (「3. 音声と画像の入力」をご参照ください)。以下の表は VAD モード (デフォルト) に基づいており、手動モードでの異なるサーバーイベントにアノテーションを付けています。

すべての音声を送信した後、session.finish イベントを送信し、WebSocket を閉じる前に session.finished を待ってください。session.finish を送信せずに接続を閉じると、サーバーは音声入力が終了したことを知ることができず、最後の発話セグメントの認識および翻訳結果が失われます。

API

• Qwen-Livetranslate-Realtime。
• AOQ クライアント SDK
• リアルタイム API の概要 (WebRTC プロトコルの説明)

課金

• 音声：入力音声 1 秒あたり 7 トークン。出力音声 1 秒あたり 12.5 トークン。
• 画像：32×32 ピクセルごとに 0.5 トークンを消費します。
• テキスト：ソース言語の音声認識が有効な場合、サービスは翻訳に加えて入力音声のトランスクリプトを返します。このトランスクリプトは出力テキストトークンとして課金されます。

• 音声：音声入力または出力の 1 秒ごとに 12.5 トークンを消費します。
• 画像：28×28 ピクセルごとに 0.5 トークンを消費します。
• テキスト：ソース言語の音声認識が有効な場合、サービスは翻訳に加えて入力音声のトランスクリプトを返します。このトランスクリプトは出力テキストトークンとして課金されます。

料金：モデルリスト。

レート制限

モデルのレート制限については、「レート制限」をご参照ください。

サポート対象言語

ソース言語とターゲット言語を指定するには、以下の言語コードを使用します。

一部のターゲット言語はテキストのみをサポートしています。レガシーモデル qwen3-livetranslate-flash-realtime は、en、zh、ru、fr、de、pt、es、it、id、ko、ja、vi、th、ar、yue、hi、el、tr の 18 言語のみをサポートしています。

サポート対象の音声

サポートされている音声と対応する voice パラメーターの値については、「音声リスト」をご参照ください。