Qwen-LiveTranslate-Realtime Python SDK

​

前提条件

1. SDK をインストールします。DashScope SDK のバージョンが 1.25.6 以降であることを確認してください。
2. API キーを取得します。

​

リクエストパラメーター

• OmniRealtimeConversation クラスのコンストラクターで、以下のパラメーターを設定します。
  サンプルコードの表示

  Copy

  from dashscope.audio.qwen_omni import (
      OmniRealtimeConversation,
      OmniRealtimeCallback,
      MultiModality,
  )
  from dashscope.audio.qwen_omni.omni_realtime import TranslationParams
  
  class MyCallback(OmniRealtimeCallback):
      """リアルタイム音声翻訳のコールバックハンドラー"""
      def __init__(self, conversation=None):
          self.conversation = conversation
          self.handlers = {
              'session.created': self._handle_session_created,
              'response.audio_transcript.done': self._handle_translation_done,
              'response.audio.delta': self._handle_audio_delta,
              'response.done': lambda r: print('======Response Done======'),
              'input_audio_buffer.speech_started': lambda r: print('======Speech Start======'),
              'input_audio_buffer.speech_stopped': lambda r: print('======Speech Stop======'),
          }
  
      def on_open(self):
          print('Connection opened')
  
      def on_close(self, code, msg):
          print(f'Connection closed, code: {code}, msg: {msg}')
  
      def on_event(self, response):
          try:
              handler = self.handlers.get(response['type'])
              if handler:
                  handler(response)
          except Exception as e:
              print(f'[Error] {e}')
  
      def _handle_session_created(self, response):
          print(f"Session created: {response['session']['id']}")
  
      def _handle_translation_done(self, response):
          print(f"Translation result: {response['transcript']}")
  
      def _handle_audio_delta(self, response):
          # インクリメンタルな音声データを処理します。
          audio_b64 = response.get('delta', '')
          # 音声データをデコードして再生または保存します。
  
  conversation = OmniRealtimeConversation(
      model='qwen3.5-livetranslate-flash-realtime',
      # 中国 (北京) リージョン。{WorkspaceId} をお使いの Bailian ワークスペース ID に置き換えてください。URL はリージョンによって異なります。
      url='wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime',
      callback=MyCallback(conversation=None)  # 一時的に None を渡します。後でインジェクトされます。
  )
  # コールバックに self をインジェクトします。
  conversation.callback.conversation = conversation
  

  パラメーター	タイプ	必須	説明
  model	str	はい	使用するモデル名。 qwen3.5-livetranslate-flash-realtime (推奨) に設定します。
  callback	OmniRealtimeCallback	はい	サーバーサイドのイベントを処理するためのコールバックインスタンス。
  url	str	はい	リアルタイム音声翻訳のサービスエンドポイント： 中国 (北京)： wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime シンガポール： wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime {WorkspaceId} を実際の ワークスペース ID に置き換えてください。

• OmniRealtimeConversation クラスの update_session メソッドを使用して、以下のパラメーターを設定します。
  サンプルコードの表示

  Copy

  # 翻訳パラメーターの設定
  translation_params = TranslationParams(
      language='en',  # ターゲット言語
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )
  
  # セッション設定の更新
  conversation.update_session(
      output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
      voice='Tina',
      translation_params=translation_params,
  )
  

  パラメーター	タイプ	必須	説明
  output_modalities	List[MultiModality]	いいえ	翻訳結果の出力フォーマット。 デフォルト： [MultiModality.TEXT, MultiModality.AUDIO] 有効値： [MultiModality.TEXT]：テキストのみを出力 [MultiModality.TEXT, MultiModality.AUDIO]：テキストと音声を出力
  voice	str	いいえ	音声出力のボイス。 デフォルト： Qwen3.5-LiveTranslate-Flash-Realtime： Tina 有効値： サポートされているボイス
  input_audio_transcription_model	str	いいえ	このパラメーターを qwen3-asr-flash-realtime に設定すると、ソース言語の音声認識結果を受信できます。
  translation_params	TranslationParams	いいえ	翻訳設定 (ターゲット言語とホットワード)。
  enable_turn_detection	bool	いいえ	音声アクティビティ検出 (VAD) を有効にするかどうか。 デフォルト値： True。 サーバーが音声の開始/終了を自動検出し、翻訳をトリガーする VAD モードを有効にします。 False に設定すると手動モードに切り替わり、クライアントは commit メソッドを使って手動で音声を送信します。パラメーターの詳細な説明については、クライアントイベントドキュメントの turn_detection の説明をご参照ください。

• TranslationParams クラスのコンストラクターで、以下のパラメーターを設定します。
  サンプルコードの表示

  Copy

  translation_params = TranslationParams(
      language='en',  # ターゲット言語コード
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',  # ソースフレーズ：ターゲット翻訳
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )
  

  パラメーター	タイプ	必須	説明
  language	str	いいえ	翻訳のターゲット言語。 デフォルト： en 有効値： サポートされている言語
  corpus	TranslationParams.Corpus	いいえ	特定の用語の翻訳精度を向上させるためのホットワードマッピング。
  corpus.phrases	dict	いいえ	キーがソース用語で、値がターゲット翻訳であるホットワードマッピングテーブル。 例： {'Inteligencia Artificial': 'Artificial Intelligence'}

​

主要なインターフェイス

​

OmniRealtimeConversation クラス

def connect(self) -> None:

def update_session(self,
    output_modalities: List[MultiModality],
    voice: str = None,
    translation_params: TranslationParams = None,
    **kwargs) -> None:

def end_session(self, timeout: int = 20) -> None:

def append_audio(self, audio_b64: str) -> None:

def commit(self) -> None:

def clear_appended_audio(self) -> None:

def close(self) -> None:

def get_session_id(self) -> str:

def get_last_response_id(self) -> str:

​

コールバックインターフェイス (OmniRealtimeCallback)

def on_open(self) -> None:

def on_event(self, message: dict) -> None:

def on_close(self, close_status_code, close_msg) -> None:

​

dashscope.Client()：例外発生後のイベント受信

​

完全なサンプルコード

import os
import sys
import base64
import signal
import pyaudio
from dashscope.audio.qwen_omni import (
    OmniRealtimeConversation,
    OmniRealtimeCallback,
    MultiModality,
)
from dashscope.audio.qwen_omni.omni_realtime import TranslationParams

class Callback(OmniRealtimeCallback):
    """リアルタイム翻訳のコールバックハンドラークラス"""

    def __init__(self, speaker):
        self.speaker = speaker

    def on_open(self):
        print("[Connection established]")

    def on_close(self, code, msg):
        print(f"[Connection closed] code: {code}, msg: {msg}")

    def on_event(self, response):
        event_type = response.get("type", "")
        if event_type == "input_audio_buffer.speech_started":
            print("====== Speech input detected ======")
        elif event_type == "input_audio_buffer.speech_stopped":
            print("====== Speech input ended ======")
        elif event_type == "conversation.item.input_audio_transcription.completed":
            print(f"[Original text] {response.get('transcript', '')}")
        elif event_type == "response.audio_transcript.done":
            print(f"[Translation result] {response.get('transcript', '')}")
        elif event_type == "response.audio.delta":
            audio_b64 = response.get("delta", "")
            if audio_b64:
                self.speaker.write(base64.b64decode(audio_b64))
        elif event_type == "error":
            print(f"[Error] {response.get('error', {}).get('message', '')}")

def main():
    # API キーの確認。
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("Set the DASHSCOPE_API_KEY environment variable.")
        sys.exit(1)

    # PyAudio の初期化。
    pya = pyaudio.PyAudio()

    # 翻訳された音声を再生するためのスピーカーを初期化します。
    speaker = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True,
        frames_per_buffer=2400
    )

    # 音声入力をキャプチャするためのマイクを初期化します。
    mic = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        frames_per_buffer=1600
    )

    # コールバックインスタンスを作成します。
    callback = Callback(speaker=speaker)

    # リアルタイムセッションを作成します。
    conversation = OmniRealtimeConversation(
        model="qwen3.5-livetranslate-flash-realtime",
        # 以下はシンガポールリージョンの URL です。{WorkspaceId} を実際のワークスペース ID に置き換えてください。URL はリージョンによって異なります。
        url="wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime",
        callback=callback
    )

    # サーバーに接続します。
    conversation.connect()

    # 翻訳パラメーターを設定します。
    translation_params = TranslationParams(
        language="en",  # 翻訳のターゲット言語：英語
        corpus=TranslationParams.Corpus(
            phrases={
                "Source Term 1": "Target Translation 1",
                "Source Term 2": "Target Translation 2"
            }
        )
    )

    # セッション設定を更新します。
    conversation.update_session(
        output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
        input_audio_transcription_model="qwen3-asr-flash-realtime",
        voice="Tina",
        translation_params=translation_params,
    )

    # 終了シグナルハンドラーを登録します。
    def on_exit(sig, frame):
        print("\n[Exiting...]")
        mic.stop_stream()
        mic.close()
        speaker.stop_stream()
        speaker.close()
        pya.terminate()
        conversation.end_session()
        conversation.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, on_exit)

    print("[Starting real-time translation] Speak into the microphone. Press Ctrl+C to exit.")

    # マイク音声を継続的にキャプチャして送信します。
    while True:
        audio_data = mic.read(1600, exception_on_overflow=False)
        conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))

if __name__ == "__main__":
    main()