Qwen3.8-LiveTranslate：知其人，傳其義

來源原標題

Qwen3.8-LiveTranslate：知其人，傳其義

機構

Alibaba

原始連結

https://qwen.ai/blog?id=qwen3.8-livetranslate

摘要原文

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 雙模組設計，通過 Interleave 方式銜接流式理解、文本輸出與語音生成。其中，Thinker 把影片、音訊、原文與譯文編排進同一條因果序列，按時序交替排列、端到端產出，讓理解與翻譯在單一序列內完成；Talker 再結合譯文和源音訊，把譯文合成為保留原說話人音色的譯文語音。 模型效果 # 多說話人長音訊能力 # 在覆蓋 14 個語向的多說話人長音訊評測集 Omnilingua-MSpeaker 上，Qwen3.8-LiveTranslate 在翻譯的忠實度、流暢度、簡潔度以及說話人識別的錯誤率 DER（Diarization Error Rate，說話人分離錯誤率）四個維度上均優於目前行業主流即時同傳系統。 多語言效果 # 我們在公開 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 : """與 DashScope 即時翻譯服務互動的客戶端：採集麥克風音訊、傳送至服務端、接收並播放譯文。""" 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 # 華北 2（北京）地域接入地址；{workspace_id} 為百鍊業務空間 ID，其它地域 URL 不同 self . api_url = ( f "wss:// { workspace_id } .cn-beijing.maas.aliyuncs.com" "/api-ws/v1/realtime?model=qwen3.8-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" : { # output_modalities 決定服務端返回內容的形式： # ["text", "audio"] —— 同時返回譯文文本和合成語音（推薦） # ["text"] —— 僅返回譯文文本 "output_modalities" : [ "text" , "audio" ] if self . audio_enabled else [ "text" ], "input_audio_format" : "pcm" , "output_audio_format" : "pcm" , # input_audio_transcription：開啟源語言識別（ASR）。 # 將 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 ): """將一塊音訊資料進行 base64 編碼後傳送至服務端。""" 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 不能為空" ) 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 ): """後臺執行緒任務：從播放佇列讀取 PCM 資料，寫入揚聲器輸出流。""" 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 ) # 語音+文本模式下，譯文文本隨合成語音流式返回，通過 response.audio_transcript.delta 累加 elif event_type == "response.audio_transcript.delta" : on_text_received ( event . get ( "delta" , "" )) # 僅文本模式下，譯文文本通過 response.text.delta 流式返回 elif event_type == "response.text.delta" : on_text_received ( event . get ( "delta" , "" )) 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 ) } " ) # 收到 session.finished 表示服務端已處理完 session.finish 請求 elif event_type == "session.finished" : print ( " \n [INFO] 會話已結束。" ) self . session_finished_event . set () # 接收源語言識別結果（需先啟用 input_audio_transcription.model） # elif event_type == "conversation.item.input_audio_transcription.delta": # print(event.get("delta", ""), end="", flush=True) # 流式識別增量 # elif event_type == "conversation.item.input_audio_transcription.completed": # transcript = event.get("transcript", "") # 一段語音的最終識別文本 # print(f"[源語言] {transcript}") 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 .