即時音視頻翻譯（Qwen-Livetranslate-Realtime）
即時音視頻翻譯（Qwen-LiveTranslate）Python SDK-API參考
本文檔介紹如何使用 DashScope Python SDK 調用即時音視頻翻譯（Qwen-LiveTranslate）模型。
前提條件
- 安裝SDK，確保DashScope SDK版本不低於1.25.6。
- 擷取API Key。
阿里雲百鍊為華北2（北京）、新加坡地區推出了業務空間專屬網域名稱，能夠為推理請求提供卓越的效能和更高的穩定性，建議遷移至新網域名稱：
- 華北2（北京）地區：從
wss://dashscope.aliyuncs.com 遷移至 wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com
- 新加坡地區：從
wss://dashscope-intl.aliyuncs.com 遷移至 wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com
其中 {WorkspaceId} 為您的業務空間 ID，可在阿里雲百鍊控制台的業務空間詳情頁面查看。現有網域名稱仍可正常使用。 請求參數
-
以下參數通過
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',
# 以下為華北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。
qwen3-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方法設定。
# 設定翻譯參數
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,
)
-
以下參數通過
TranslationParams的構造方法設定。
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類
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 commit(self) -> None:
| input_audio_buffer.committed
輸入音頻緩衝區已提交
| Manual 模式下，提交之前通過append_audio方法追加到雲端緩衝區的音頻，服務端收到後自動開始產生翻譯響應。VAD 模式下無需調用此方法，服務端會自動認可。 |
def clear_appended_audio(self) -> None:
| input_audio_buffer.cleared
輸入音頻緩衝區已清空
| 清空當前雲端緩衝區中尚未提交的音頻資料。 |
| 無 | 終止任務，並關閉串連。 |
def get_session_id(self) -> str:
| 無 | 擷取當前任務的session_id。 |
def get_last_response_id(self) -> str:
| 無 | 擷取最近一次response的response_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串連關閉時的日誌資訊 | WebSocket串連關閉時觸發。 |
完整樣本
以下樣本展示如何從麥克風即時錄音並進行翻譯。
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",
# 以下為新加坡地區URL，調用時請將{WorkspaceId}替換為真實的業務空間ID，各地區的URL不同。
url="wss://{WorkspaceId}.ap-southeast-1.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()
