Skip to main content

即時音視頻翻譯（Qwen-LiveTranslate）Python SDK-API參考

本文檔介紹如何使用 DashScope Python SDK 調用即時音視頻翻譯（Qwen-LiveTranslate）模型。

前提條件

  1. 安裝SDK，確保DashScope SDK版本不低於1.25.6。
  2. 擷取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類

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 close(self) -> None:
無終止任務，並關閉串連。
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()