Alibaba Cloud Model Studio：即時語音/音視頻翻譯-千問

功能特性

• 多語言支援：支援 60 種語言互譯，其中 29 種支援音頻+文本輸出、31 種僅支援文本輸出，覆蓋中文、英語、法語、德語、俄語、日語、韓語、西班牙語、葡萄牙語、阿拉伯語等主流語種。

• 視覺增強：利用視覺內容提升翻譯準確性。模型通過分析畫面中的口型、動作和文字，改善在嘈雜環境下或一詞多義情境中的翻譯效果。

• 2.8 秒延遲：實現低至 2.8 秒的同傳延遲。

• 無損同傳：通過語義單元預測技術，解決跨語言語序問題。即時翻譯品質接近離線翻譯結果。

• 音色自然：產生音色自然的擬人語音。模型能根據源語音內容，自適應調節語氣和情感。

• 配置熱詞：通過熱詞提升特定詞彙的翻譯準確性。

• 聲音複刻：支援複刻發言人音色用於翻譯播報，讓輸出聽起來像本人說外語。支援服務端即時複刻和使用預先複刻的固定音色。

如何使用

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

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://{WorkspaceId}.ap-southeast-1.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()

支援的模型

快速開始

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
           self.api_url = "wss://{WorkspaceId}.ap-southeast-1.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())

通過Function Compute一鍵部署

控制台暫不支援體驗。可通過以下方式一鍵部署：

1. 開啟我們寫好的Function Compute模板，填入 API Key， 單擊建立並部署預設環境即可線上體驗。

2. 等待約一分鐘，在 環境詳情 > 環境資訊 中擷取訪問網域名稱，將訪問網域名稱的http改成https（例如https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/），通過該連結與模型互動。

   重要

   此連結使用自我簽署憑證，僅用於臨時測試。首次訪問時，瀏覽器會顯示安全警告，這是預期行為，請勿在生產環境使用。如需繼續，請按瀏覽器提示操作（如點擊“進階” → “繼續前往（不安全）”）。

如需開通存取控制許可權，請跟隨頁面指引操作。

通過資源資訊-函數資源查看專案原始碼。

Function Compute與阿里雲百鍊均為新使用者提供免費額度，可以覆蓋簡單調試所需成本，額度耗盡後隨用隨付。只有在訪問的情況下會產生費用。

互動流程

即時語音翻譯的互動流程遵循標準的 WebSocket 事件驅動模型，服務端自動檢測語音起止並進行響應。

API 參考

請參見即時音視頻翻譯（Qwen-Livetranslate-Realtime）。

計費說明

• 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參數取值參見音色列表。