跳转到主要内容
千问云 home page
开放平台文档
实时

LiveTranslate Python SDK

LiveTranslate Python SDK 参考文档

通过 DashScope Python SDK 调用 Qwen-LiveTranslate，实现实时语音翻译。 使用指南： 教程和完整示例请参见实时翻译。

请求参数

在 OmniRealtimeConversation 构造函数中设置以下参数： 
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',
  url='wss://dashscope.aliyuncs.com/api-ws/v1/realtime',
  callback=MyCallback(conversation=None)  # 临时传入 None，稍后注入
)
# 将 conversation 注入回调对象
conversation.callback.conversation = conversation
参数类型必选说明
modelstr是模型名称。推荐使用 qwen3.5-livetranslate-flash-realtime。qwen3-livetranslate-flash-realtime 为旧版模型。
callbackOmniRealtimeCallback是处理服务端事件的回调对象。
urlstr否服务端点：wss://dashscope.aliyuncs.com/api-ws/v1/realtime。默认为 DashScope 端点。
api_keystr否用于鉴权的 API Key。未提供时，SDK 自动读取 DASHSCOPE_API_KEY 环境变量。
通过 OmniRealtimeConversation.update_session 设置以下参数： 
# 设置翻译参数
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='Cherry',
  translation_params=translation_params,
)
参数类型必选说明
output_modalitiesList[MultiModality]否输出类型。默认值：[MultiModality.TEXT, MultiModality.AUDIO]。可选值：[MultiModality.TEXT]（仅文本）或 [MultiModality.TEXT, MultiModality.AUDIO]（文本和音频）。
voicestr否音频输出的音色。默认值：Cherry。参见支持的音色。
input_audio_transcription_modelstr否设为 qwen3-asr-flash-realtime 可获取源语言的语音识别结果。使用时须同时设置 translation_params，否则服务端返回参数无效错误。
translation_paramsTranslationParams否翻译设置。
在 TranslationParams 构造函数中设置以下参数： 
translation_params = TranslationParams(
  language='en',  # 目标语言代码
  corpus=TranslationParams.Corpus(
    phrases={
      'Inteligencia Artificial': 'Artificial Intelligence',  # 源语言词汇: 目标翻译
      'Aprendizaje Automático': 'Machine Learning'
    }
  )
)
参数类型必选说明
languagestr否目标语言代码。默认值：en。参见支持的语种。
corpusTranslationParams.Corpus否热词设置，用于提高特定术语的翻译准确度。
corpus.phrasesdict否热词映射（key：源语言术语，value：目标语言翻译）。示例：{'Inteligencia Artificial': 'Artificial Intelligence'}

核心接口

OmniRealtimeConversation 类

导入方式：from dashscope.audio.qwen_omni import OmniRealtimeConversation
方法签名服务端事件（通过回调返回）说明
def connect(self) -> None:服务端事件：会话已创建；服务端事件：会话配置已更新连接服务端。
def update_session(self, output_modalities: List[MultiModality], voice: str = None, translation_params: TranslationParams = None, **kwargs) -> None:服务端事件：会话配置已更新更新会话设置。连接后立即调用。未调用时使用默认值。参见 OmniRealtimeConversation.update_session 参数。
def end_session(self, timeout: int = 20) -> None:session.finished：服务端完成翻译并结束会话结束会话。服务端会完成剩余翻译后再关闭。
def append_audio(self, audio_b64: str) -> None:无向输入缓冲区发送 Base64 编码的音频数据。服务端自动检测语音边界并触发翻译。
def close(self) -> None:无停止任务并关闭连接。
def get_session_id(self) -> str:无返回当前会话 ID。
def get_last_response_id(self) -> str:无返回最近一次响应 ID。

回调接口 (OmniRealtimeCallback)

服务端通过回调向客户端发送事件。继承此类并实现相应方法来处理事件。 导入方式：from dashscope.audio.qwen_omni import OmniRealtimeCallback
方法签名参数说明
def on_open(self) -> None:无WebSocket 连接建立时调用。
def on_event(self, message: dict) -> None:message：服务端事件收到服务端事件时调用。
def on_close(self, close_status_code, close_msg) -> None:close_status_code：状态码。close_msg：日志信息。WebSocket 连接关闭时调用。