实时音视频翻译（Qwen-LiveTranslate）Python SDK-API参考

​

前提条件

1. 安装SDK，确保DashScope SDK版本不低于1.25.6。
2. 获取与配置 API Key。

​

请求参数

• 以下参数通过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',
      # 以下为华北2（北京）地域的URL。请将 {WorkspaceId} 替换为您的百炼业务空间ID，各地域的URL不同。
      url='wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime',
      callback=MyCallback(conversation=None)  # 暂时传None，稍后注入
  )
  # 注入自身到回调
  conversation.callback.conversation = conversation
  

  参数	类型	是否必须	说明
  model	str	是	指定要使用的模型名称，推荐使用qwen3.5-livetranslate-flash-realtime。
  callback	回调接口（OmniRealtimeCallback）	是	用于处理服务端事件的回调对象实例。
  url	str	是	实时翻译服务地址： 华北2（北京）：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={
              '人工智能': 'Artificial Intelligence',
              '机器学习': '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	否	将input_audio_transcription_model设为qwen3-asr-flash-realtime，服务端将返回源语言语音识别结果。
  translation_params	TranslationParams	否	翻译相关配置。
  enable_turn_detection	bool	否	是否启用 VAD（语音活动检测）。 默认值：True，即启用 VAD 模式，服务端自动检测语音起止并自动触发翻译。 设为False可切换为 Manual 模式，由客户端通过commit方法手动提交音频。详细参数说明参见turn_detection object （可选） 语音活动检测（VAD，Voice Activity Detection）配置，用于控制语音起止的检测方式： 设为配置对象（默认值）：启用 VAD 模式。服务端自动检测语音起止，自动提交音频缓冲区并触发翻译响应，客户端无需发送input_audio_buffer.commit事件。 设为null：启用 Manual 模式。由客户端通过input_audio_buffer.commit事件手动提交音频缓冲区，服务端收到后自动开始生成翻译响应。 属性 type string （可选） VAD 类型，固定为server_vad。 threshold float （可选） VAD 检测灵敏度。值越低，越容易将微弱声音（包括背景噪音）识别为语音；值越高，需要更清晰、音量更大的语音才能触发。 取值范围：[-1.0, 1.0]，默认值为 0.2。 silence_duration_ms integer （可选） 语音结束后需保持静音的最短时长（毫秒）。超过该时长后判定语音结束，服务端自动提交音频缓冲区并触发翻译响应。 取值范围：[200, 6000]，默认值为 1000。。

• 以下参数通过TranslationParams的构造方法设置。
  点击查看示例代码

  Copy

  translation_params = TranslationParams(
      language='en',  # 目标语言代码
      corpus=TranslationParams.Corpus(
          phrases={
              '人工智能': 'Artificial Intelligence',  # 源语言词: 目标语言翻译
              '机器学习': 'Machine Learning'
          }
      )
  )
  

  参数	类型	是否必须	说明
  language	str	否	翻译目标语言代码。 默认值：en。 可选值：参见支持的语种。
  corpus	TranslationParams.Corpus	否	热词配置，用于提升特定词汇的翻译准确性。
  corpus.phrases	dict	否	热词映射表。key 为源语言词汇，value 为目标语言对应翻译。 示例：{'人工智能': '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:

​

完整示例

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("[连接已建立]")

    def on_close(self, code, msg):
        print(f"[连接已关闭] code: {code}, msg: {msg}")

    def on_event(self, response):
        event_type = response.get("type", "")
        if event_type == "input_audio_buffer.speech_started":
            print("====== 检测到语音输入 ======")
        elif event_type == "input_audio_buffer.speech_stopped":
            print("====== 语音输入结束 ======")
        elif event_type == "conversation.item.input_audio_transcription.completed":
            print(f"[原文] {response.get('transcript', '')}")
        elif event_type == "response.audio_transcript.done":
            print(f"[翻译结果] {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"[错误] {response.get('error', {}).get('message', '')}")

def main():
    # 检查 API Key
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("请设置环境变量 DASHSCOPE_API_KEY")
        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",
        # 以下为华北2（北京）地域的URL。请将 {WorkspaceId} 替换为您的百炼业务空间ID，各地域的URL不同。
        url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        callback=callback
    )

    # 连接服务端
    conversation.connect()

    # 配置翻译参数
    translation_params = TranslationParams(
        language="en",  # 翻译目标语言：英语
        corpus=TranslationParams.Corpus(
            phrases={
                "人工智能": "Artificial Intelligence",
                "机器学习": "Machine Learning"
            }
        )
    )

    # 更新会话配置
    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[正在退出...]")
        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("[开始实时翻译] 请对着麦克风说话，按 Ctrl+C 退出")

    # 持续采集麦克风音频并发送
    while True:
        audio_data = mic.read(1600, exception_on_overflow=False)
        conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))

if __name__ == "__main__":
    main()