实时语音/音视频翻译-千问

qwen3.5-livetranslate-flash-realtime 是视觉增强型实时翻译模型，支持 60 种语言互译（其中 29 种支持音频+文本输出、31 种仅支持文本输出），可同时处理音频与图像输入，适用于实时视频流或本地视频文件，利用视觉上下文信息提升翻译准确性，并实时输出高质量的翻译文本与音频。

在线体验参见通过函数计算一键部署。

功能特性

• 多语言支持：支持 60 种语言互译，其中 29 种支持音频+文本输出、31 种仅支持文本输出，覆盖中文、英语、法语、德语、俄语、日语、韩语、西班牙语、葡萄牙语、阿拉伯语等主流语种。
• 视觉增强：利用视觉内容提升翻译准确性。模型通过分析画面中的口型、动作和文字，改善在嘈杂环境下或一词多义场景中的翻译效果。
• 2.8 秒延迟：实现低至 2.8 秒的同传延迟。
• 无损同传：通过语义单元预测技术，解决跨语言语序问题。实时翻译质量接近离线翻译结果。
• 音色自然：生成音色自然的拟人语音。模型能根据源语音内容，自适应调节语气和情感。
• 配置热词：通过热词提升特定词汇的翻译准确性。
• 声音复刻：支持复刻发言人音色用于翻译播报，让输出听起来像本人说外语。支持服务端实时复刻和使用预先复刻的固定音色。
• 跳过同语种输出：当源语种与目标语种相同时，可分别跳过文本或音频输出。仅当目标语种为中文（zh）或英语（en）时生效。

如何使用

1. 配置连接

qwen3.5-livetranslate-flash-realtime 模型通过 WebSocket 协议接入，连接时需要以下配置项：

该模型除 WebSocket 外，还支持通过 AOQ 和 WebRTC 协议接入；如果是客户端对接，且更看重稳定的延迟、弱网下的交互能力、实时双工的降噪与回声消除，可优先考虑 AOQ，协议对比与选型请参见Realtime API 概述。

可通过以下 Python 示例代码建立连接。

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

API_KEY=os.getenv("DASHSCOPE_API_KEY")
# 以下为华北2（北京）地域的URL。请将 {WorkspaceId} 替换为您的百炼业务空间ID，各地域的URL不同。
API_URL = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"Connected to server: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("Error:", 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）与 Manual 模式

  通过session.turn_detection参数配置语音起止的检测方式：

  • VAD 模式（默认）：turn_detection设为配置对象。服务端自动检测语音起止并自动触发翻译，适合客户端持续发送音频流的场景。
  • Manual 模式：turn_detection设为null。由客户端判断语音起止，说完一段话后主动发送input_audio_buffer.commit事件提交音频，适合"按下说话、松开发送"的场景。

  两种模式下的完整交互步骤参见本文3. 输入音频与图片。

• 音色

  通过session.voice参数配置。参见支持的音色。

• 热词

  通过session.translation.corpus.phrases参数配置。热词用于提升特定词汇的翻译准确性，以 key-value 形式指定源语言词汇与目标语言翻译的映射关系。

  示例：将"人工智能"指定翻译为"Artificial Intelligence"。

• 声音复刻

  通过 session.enable_voice_clone、session.voice_clone_options.frequency 与 session.voice 参数配置。支持三种模式：使用预先复刻的音色（frequency 为 never）、服务端复刻一次（once）或每次复刻（always）。详见 声音复刻。

3. 输入音频与图片

客户端通过 input_audio_buffer.append 和 input_image_buffer.append 事件发送 Base64 编码的音频和图片数据。音频输入是必需的；图片输入是可选的。

图片可以来自本地文件，或从视频流中实时采集。

模型判断一段语音"说完了"的方式，取决于turn_detection参数配置的 VAD 模式或 Manual 模式：

• VAD 模式（默认）：客户端持续发送input_audio_buffer.append事件；服务端检测到语音开始/结束时，分别返回input_audio_buffer.speech_started、input_audio_buffer.speech_stopped事件，并自动提交音频缓冲区、触发翻译。翻译响应基于流式语音同步生成，通常在语音输入过程中就已开始，无需等待语音结束。
• Manual 模式：将session.turn_detection设为null。客户端发送完一段完整的语音后，主动发送input_audio_buffer.commit事件提交音频缓冲区；服务端返回input_audio_buffer.committed事件确认后，自动开始生成翻译响应，客户端无需再发送其他事件触发响应。提交前若需清空未提交的音频，可发送input_audio_buffer.clear事件。

4. 接收模型响应

翻译响应基于流式语音同步生成，通常无需等待语音结束（参见上一节 VAD/Manual 模式说明）。模型的响应格式取决于配置的输出模态。

• 仅输出文本

  服务端通过response.text.text事件流式返回增量翻译文本（含已确认文本和待确认的预测文本）；翻译完成后，通过response.text.done事件返回完整的翻译文本。

• 输出文本+音频
  • 文本

    通过response.audio_transcript.text事件流式返回增量翻译文本；翻译完成后，通过response.audio_transcript.done事件返回完整的翻译文本。

  • 音频

    通过response.audio.delta事件返回 Base64 编码的增量音频数据。

5. 结束会话

音频发送完毕后，客户端必须发送 session.finish 事件通知服务端，然后等待服务端返回 session.finished 事件后再关闭 WebSocket 连接。

支持的模型

推荐模型

旧版模型

以下模型仍在服务中，但不再作为首选推荐。新场景建议使用上方的新一代模型，以获得更优的翻译质量与性价比。

快速开始

1. 准备运行环境

   您的 Python 版本需要不低于 3.10。

   首先安装 pyaudio。

   macOS

   brew install portaudio && pip install pyaudio
   

   Debian/Ubuntu

   sudo apt-get install python3-pyaudio
   
   或者
   
   pip install pyaudio
   

   CentOS

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

   Windows

   pip install pyaudio
   

   安装完成后，通过 pip 安装 websocket 相关的依赖：

pip install websocket-client==1.8.0 websockets

2. 创建客户端

   在本地新建一个 Python 文件，命名为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 key cannot be empty.")
   
           self.api_key = api_key
           self.target_language = target_language
           self.audio_enabled = audio_enabled
           self.ws = None
           # 以下为华北2（北京）地域的URL。请将 {WorkspaceId} 替换为您的百炼业务空间ID，各地域的URL不同。
           self.api_url = "wss://{WorkspaceId}.cn-beijing.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",
                       #         "机器学习": "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] Token 使用情况: {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. 与模型互动

   在livetranslate_client.py的同级目录下新建另一个 Python 文件，命名为main.py，并将以下代码复制进文件中：

   main.py

   import os
   import asyncio
   from livetranslate_client import LiveTranslateClient
   
   def print_banner():
       print("=" * 60)
       print("  基于千问 qwen3.5-livetranslate-flash-realtime")
       print("=" * 60 + "\n")
   
   def get_user_config():
       """获取用户配置"""
       print("请选择模式:")
       print("1. 语音+文本 [默认] | 2. 仅文本")
       mode_choice = input("请输入选项 (直接回车选择语音+文本): ").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，不可设置为其他预设音色，否则服务端会返回错误。

声音复刻配置示例

使用预先复刻的音色（音质稳定，推荐需要固定音色的场景）：

{
    "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"
        }
    }
}

服务端复刻一次（适合单人演讲）：

{
    "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"
        }
    }
}

利用图像提升翻译准确率

qwen3.5-livetranslate-flash-realtime 模型可以接收图像输入，辅助音频翻译，适用于同音异义、低频专有名词识别场景。建议每秒发送不超过2张图片。

将以下示例图片下载到本地：口罩.png面具.png

将以下代码下载到livetranslate_client.py同级目录并运行，向麦克风说"What is mask?"，在输入口罩图片时，模型会翻译为“什么是口罩？”；输入面具图片时，模型会翻译为“什么是面具？”

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

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "口罩.png"
# IMAGE_PATH = "面具.png"

def print_banner():
    print("=" * 60)
    print("  基于千问 qwen3.5-livetranslate-flash-realtime —— 单轮交互示例 (mask)")
    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 中配置 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())

通过函数计算一键部署

控制台暂不支持体验。可通过以下方式一键部署：

1. 打开我们写好的函数计算模板，填入 API Key， 单击创建并部署默认环境即可在线体验。

2. 等待约一分钟，在 环境详情 > 环境信息 中获取访问域名，将访问域名的http改成https（例如https://qwen-livetranslate-flash-realtime.fcv3.xxx.cn-hangzhou.fc.devsapp.net/），通过该链接与模型交互。

   重要此链接使用自签名证书，仅用于临时测试。首次访问时，浏览器会显示安全警告，这是预期行为，请勿在生产环境使用。如需继续，请按浏览器提示操作（如点击“高级” → “继续前往（不安全）”）。

如需开通访问控制权限，请跟随页面指引操作。

通过资源信息-函数资源查看项目源代码。

函数计算与阿里云百炼均为新用户提供免费额度，可以覆盖简单调试所需成本，额度耗尽后按量计费。只有在访问的情况下会产生费用。

交互流程

实时语音翻译的交互流程遵循标准的 WebSocket 事件驱动模型。语音起止的判断方式取决于 VAD 模式或 Manual 模式（参见3. 输入音频与图片），下表以 VAD 模式（默认）为主线，并标注了 Manual 模式下不同的服务端事件。

API 参考

• 实时音视频翻译（Qwen-Livetranslate-Realtime）。
• AOQ客户端API
• Realtime API 概述（WebRTC 协议说明）

计费说明

• Qwen3.5-LiveTranslate-Flash-Realtime
  • 音频：输入每秒音频消耗 7 Token，输出每秒音频消耗 12.5 Token。
  • 图片：每输入 32*32 像素消耗 0.5 Token。
  • 文本：启用源语言语音识别功能后，服务除返回翻译结果外，还会返回输入音频的语音识别文本（即源语言原文），该识别文本将按输出文本的 Token 标准计费。
• Qwen3-LiveTranslate-Flash-Realtime
  • 音频：输入或输出每秒音频均消耗 12.5 Token。
  • 图片：每输入 28*28 像素消耗 0.5 Token。
  • 文本：启用源语言语音识别功能后，服务除返回翻译结果外，还会返回输入音频的语音识别文本（即源语言原文），该识别文本将按输出文本的 Token 标准计费。

Token 费用请参见选择模型。

限流说明

模型的限流规则请参见限流。

支持的语种

下表中的语种代码可用于指定源语种与目标语种。

部分目标语种仅支持输出文本，不支持输出音频。老模型 qwen3-livetranslate-flash-realtime 仅支持以下 18 种语种：en、zh、ru、fr、de、pt、es、it、id、ko、ja、vi、th、ar、yue、hi、el、tr。

支持的音色

实时翻译支持的音色与voice参数取值参见音色列表。