Qwen 发布 Qwen3.8-LiveTranslate 实时同传模型，LAAL 降至 2.3 秒

API Demo

同声传译不仅关乎翻译速度——它还必须听得清楚、译得准确。Qwen3.8-LiveTranslate 以 Interleave 架构重构实时同声传译，全面提升忠实度、流畅度与简洁度，同时平均延迟（LAAL）从 2.8 秒降至 2.3 秒。

我们希望同声传译不仅传递语言，也传递对话背后的人与语境。在支持 60 种语言的基础上，Qwen3.8-LiveTranslate 新增三项能力，将同声传译带入更广泛的真实场景：实时说话人分离，为每一句话提供清晰的归属，并实现更稳定的声音克隆；在同一个双语屏幕上同步输出原文与译文；以及长上下文消歧，结合前文理解当下，让姓名和术语更加精准。

核心亮点#

Interleave 架构：音频与文本交织融合，质量更高、延迟更低。这一代将同声传译重构为单一的音频-文本交织流；已经听到的音频和已经产出的译文都可以被缓存并复用，因此翻译质量较上一代显著提升，同时平均延迟（LAAL）从上一代的 2.8 秒降至 2.3 秒。

实时说话人分离：内容归属更清晰，声音克隆更稳定。当多人轮流发言时，模型能够区分不同的说话人以及各自所说的内容，并帮助翻译后的语音更准确、更稳定地保留每位说话人的音色。

原文与译文同步。同传过程中的双语对齐既支持即时理解，也支持原文核对，为字幕显示、内容整理和检索等后续功能奠定基础——拥有更广阔的应用空间。

长上下文消歧：跨轮次关联历史，翻译更精准。通过借助前文和历史上下文，模型缓解了复杂场景中专有名词、指代等的歧义，保持表达一致。

模型架构#

Qwen3.8-LiveTranslate 采用基于 Hybrid-MoE 的 Thinker–Talker 双模块设计，通过交错方式连接流式理解、文本输出和语音生成。其中，Thinker 将视频、音频、源文本和译文编排为单一因果序列——按时间顺序交错并以端到端方式生成——使理解和翻译在同一序列内完成；Talker 随后将译文与源音频结合，把译文合成为保留原说话人音色的语音。

性能#

多说话人长音频#

在Omnilingua-MSpeaker这一覆盖 14 个语言方向的多说话人长音频评测集上，Qwen3.8-LiveTranslate 在四个维度上均优于当今主流的实时同传系统：翻译忠实度、流畅度、简洁度，以及说话人分离错误率（DER）。

多语言性能#

我们在公开的 FLEURS 音频测试集上评估了 70 个语言方向的实时同传性能。Qwen3.8-LiveTranslate 在四个维度上均领先于上一代及当前主流的实时同传系统：翻译质量、平均延迟、语音识别准确率和语音合成质量。

🎬 实际效果演示#

以下演示使用了《西游记》中的两段对话以及一组同音词示例，展示了说话人归属与声音克隆、同步双语输出，以及上下文和视觉信息如何帮助翻译消歧。

视频 · 前往原文观看

视频 · 前往原文观看

视频 · 前往原文观看

支持的语言#

类别支持的语言

输入音频与输出文本60 种语言 南非荷兰语、阿拉伯语、阿斯图里亚斯语、阿塞拜疆语、白俄罗斯语、孟加拉语、波斯尼亚语、保加利亚语、粤语、加泰罗尼亚语、宿务语、中文、克罗地亚语、捷克语、丹麦语、荷兰语、英语、爱沙尼亚语、菲律宾语、芬兰语、法语、加利西亚语、古吉拉特语、德语、希腊语、希伯来语、印地语、匈牙利语、冰岛语、印度尼西亚语、意大利语、日语、爪哇语、卡纳达语、哈萨克语、韩语、吉尔吉斯语、拉脱维亚语、马其顿语、马来语、马拉雅拉姆语、马拉地语、挪威语、波斯语、波兰语、葡萄牙语、旁遮普语、罗马尼亚语、俄语、斯洛伐克语、斯洛文尼亚语、西班牙语、斯瓦希里语、瑞典语、塔吉克语、泰语、土耳其语、乌克兰语、乌尔都语、越南语

输出音频29 种语言 中文、英语、德语、意大利语、葡萄牙语、西班牙语、日语、韩语、法语、俄语、泰语、印度尼西亚语、阿拉伯语、越南语、土耳其语、芬兰语、波兰语、印地语、荷兰语、捷克语、乌尔都语、菲律宾语、瑞典语、丹麦语、希伯来语、冰岛语、马来语、挪威语、波斯语

通过 DashScope API 使用 Qwen3.8-LiveTranslate#

下面是一个完整的实时同传客户端示例：它采集麦克风音频，将其流式传输到服务器，并接收和播放回传的译文。这一代新增的说话人分离和源语言与译文同步输出能力均可通过会话配置启用；服务器会在返回译文的同时返回说话人标识和源语言文本，无需再单独调用 ASR 接口。

import os import time import base64 import asyncio import json import websockets import pyaudio import queue import threading import traceback class LiveTranslateClient: """Client for the DashScope live-translation service: captures mic audio, sends it to the server, and plays back the translated speech.""" def __init__(self, api_key: str, workspace_id: str, target_language: str = "en", *, audio_enabled: bool = True): if not api_key: raise ValueError("API key cannot be empty.") if not workspace_id: raise ValueError("Workspace ID cannot be empty.") self.api_key = api_key self.workspace_id = workspace_id self.target_language = target_language self.audio_enabled = audio_enabled self.ws = None # URL for the China (Beijing) region; replace {workspace_id} with your Model Studio workspace ID. URLs differ by region. self.api_url = ( f"wss://{workspace_id}.cn-beijing.maas.aliyuncs.com" "/api-ws/v1/realtime?model=qwen3.8-livetranslate-flash-realtime" ) # Audio input parameters (microphone capture) self.input_rate = 16000 self.input_chunk = 1600 self.input_format = pyaudio.paInt16 self.input_channels = 1 # Audio output parameters (local playback) self.output_rate = 24000 self.output_chunk = 2400 self.output_format = pyaudio.paInt16 self.output_channels = 1 # Runtime state and playback resources self.is_connected = False self.audio_player_thread = None self.audio_playback_queue = queue.Queue() self.pyaudio_instance = pyaudio.PyAudio() # Set once the server returns session.finished, so close() can wait for a clean shutdown. self.session_finished_event = asyncio.Event() async def connect(self): """Open a WebSocket connection to the translation service.""" 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"Successfully connected to server: {self.api_url}") await self.configure_session() except Exception as e: print(f"Connection failed: {e}") self.is_connected = False raise async def configure_session(self): """Configure the translation session: target language, audio formats, and optional features.""" config = { "event_id": f"event_{int(time.time() * 1000)}", "type": "session.update", "session": { # `output_modalities` decides what the server returns: # ["text", "audio"] — both translated text and synthesized speech (recommended) # ["text"] — translated text only "output_modalities": ["text", "audio"] if self.audio_enabled else ["text"], "input_audio_format": "pcm", "output_audio_format": "pcm", # `input_audio_transcription`: enable source-language ASR. # Setting `model` to 'qwen3-asr-flash-realtime' also streams back the source transcript. # "input_audio_transcription": { # "model": "qwen3-asr-flash-realtime", # "language": "zh" # source language; defaults to 'en' # }, "translation": { "language": self.target_language, # `corpus`: register hotwords to boost accuracy on proper nouns and domain-specific terms. # "corpus": { # "phrases": { # "人工智能": "Artificial Intelligence", # "机器学习": "Machine Learning" # } # } } } } print(f"Sending session config: {json.dumps(config, indent=2, ensure_ascii=False)}") await self.ws.send(json.dumps(config)) async def send_audio_chunk(self, audio_data: bytes): """Base64-encode an audio chunk and send it to the server.""" 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): """Send an image frame to the server as visual context for translation.""" if not self.is_connected: return if not image_bytes: raise ValueError("image_bytes cannot be empty") 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): """Background thread task: drain PCM chunks from the playback queue and write them to the speaker output stream.""" 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: # sentinel: stop the playback loop 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): """Spin up the background audio playback thread (no-op when audio output is disabled).""" 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): """Continuously receive and dispatch event messages pushed by the server.""" 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] Response complete.") usage = event.get("response", {}).get("usage", {}) if usage: print(f"[INFO] Token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}") # Receive source-language ASR results (requires input_audio_transcription.model to be enabled) # elif event_type == "conversation.item.input_audio_transcription.delta": # delta = event.get("delta", "") # incremental transcript # print(f"[Recognizing] {delta}", end="", flush=True) # elif event_type == "conversation.item.input_audio_transcription.completed": # transcript = event.get("transcript", "") # final transcript for an utterance # print(f"[Source] {transcript}") # In voice + text mode, the incremental translation text arrives alongside synthesized audio elif event_type == "response.audio_transcript.delta": on_text_received(event.get("delta", "")) # In text-only mode, the incremental translation arrives via response.text.delta elif event_type == "response.text.delta": on_text_received(event.get("delta", "")) # The server acknowledges session.finish and completes the shutdown handshake elif event_type == "session.finished": print("\n[INFO] Session finished.") self.session_finished_event.set() except websockets.exceptions.ConnectionClosed as e: print(f"[WARNING] Connection closed: {e}") self.is_connected = False except Exception as e: print(f"[ERROR] Unknown error during message handling: {e}") traceback.print_exc() self.is_connected = False async def start_microphone_streaming(self): """Continuously capture microphone audio and stream it to the server in real time.""" 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("Microphone started, please begin speaking...") 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): """Gracefully close the WebSocket connection and release audio resources.""" # Ask the server to finish the session, then wait for its session.finished acknowledgement. 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("Sent session.finish, waiting for the server to finish processing...") try: await asyncio.wait_for(self.session_finished_event.wait(), timeout=15) print("Server finished processing.") except asyncio.TimeoutError: print("Timed out waiting for session.finished.") self.is_connected = False if self.ws: await self.ws.close() print("WebSocket connection closed.") if self.audio_player_thread: self.audio_playback_queue.put(None) # signal the playback thread to exit self.audio_player_thread.join(timeout=1) print("Audio playback thread stopped.") self.pyaudio_instance.terminate() print("PyAudio instance released.") def print_banner(): print("=" * 60) print(" Powered by Qwen qwen3.8-livetranslate-flash-realtime") print("=" * 60 + "\n") def get_user_config(): """Collect runtime parameters from the user via CLI: output mode and target language.""" print("Select mode:") print("1. Voice + Text [default] | 2. Text only") mode_choice = input("Enter option (press Enter for Voice + Text): ").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("Select target translation language (Voice + Text mode):") print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Korean | 10. Japanese | 11. Cantonese") 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("Select target translation language (Text only mode):") print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Indonesian | 10. Korean | 11. Japanese | 12. Vietnamese | 13. Thai | 14. Arabic | 15. Cantonese | 16. Hindi | 17. Greek | 18. Turkish") choice = input("Enter option (default is the first one): ").strip() target_language = lang_map.get(choice, next(iter(lang_map.values()))) return target_language, audio_enabled async def main(): """Program entry point: connect, configure the session, and drive the live-translation loop.""" print_banner() api_key = os.environ.get("DASHSCOPE_API_KEY") if not api_key: print("[ERROR] Please set the environment variable DASHSCOPE_API_KEY") print(" Example: export DASHSCOPE_API_KEY='your_api_key_here'") return workspace_id = os.environ.get("DASHSCOPE_WORKSPACE_ID") if not workspace_id: print("[ERROR] Please set the environment variable DASHSCOPE_WORKSPACE_ID (your Model Studio workspace ID)") print(" Example: export DASHSCOPE_WORKSPACE_ID='your_workspace_id_here'") return target_language, audio_enabled = get_user_config() print("\nConfiguration complete:") print(f" - Target language: {target_language}") if not audio_enabled: print(" - Output mode: Text only") client = LiveTranslateClient(api_key=api_key, workspace_id=workspace_id, target_language=target_language, audio_enabled=audio_enabled) # Callback fired as translated text arrives — stream it to stdout, character by character def on_translation_text(text): print(text, end="", flush=True) try: print("Connecting to the translation service...") await client.connect() # Launch the audio playback thread (only does real work when audio output is enabled) client.start_audio_player() print("\n" + "-" * 60) print("Connected! Please speak into the microphone.") print("The program will translate your speech in real time and play the results. Press Ctrl+C to exit.") print("-" * 60 + "\n") # Run two coroutines concurrently: server-message handling + microphone audio upload message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text)) tasks = [message_handler] # Microphone capture is the translation input source — required regardless of output mode microphone_streamer = asyncio.create_task(client.start_microphone_streaming()) tasks.append(microphone_streamer) await asyncio.gather(*tasks) except KeyboardInterrupt: print("\n\nUser interrupted, exiting...") except Exception as e: print(f"\nFatal error occurred: {e}") finally: print("\nCleaning up resources...") await client.close() print("Program exited.") if __name__ == "__main__": asyncio.run(main())

未来方向#

同声传译远未到达它的终点。在“听得更完整、传得更忠实、用得更广泛”的指引下，我们正朝着以下方向努力：

逼近同声传译的延迟极限：持续压缩端到端延迟，把“听到”与“译出”之间的间隔推向极限。

跨会话的长期记忆：将记忆跨会话带入同一项目、同一群人的下一次对话中，越用越好用。

覆盖更多语言：持续将覆盖范围扩展到更多长尾语言和地区方言，让实时传译能够服务世界上的每一个人。

引用#

如果你觉得 Qwen3.8-LiveTranslate 有帮助，欢迎引用以下文章：

@misc{qwen38livetranslateblog, title = {Qwen3.8-LiveTranslate: Names the speaker. Carries the meaning.}, url = {https://qwen.ai/blog?id=qwen3.8-livetranslate}, author = {Qwen Team}, month = {September}, year = {2026} }