Alibaba Cloud Model Studio:リアルタイム音声・映像翻訳 - Qwen

特徴

• 多言語対応：中国語、英語、フランス語、ドイツ語、ロシア語、日本語、韓国語をはじめとする 18 言語と、北京語（普通話）、広東語、四川語を含む 6 種類の中国語方言をサポートします。

• 視覚情報の活用：口の動き、ジェスチャー、画面上のテキストなどの視覚的ヒントを分析し、特にノイズ環境下や曖昧な単語において翻訳精度を向上させます。

• 3 秒の遅延：遅延がわずか 3 秒の同時通訳を実現します。

• 高品質な同時通訳：意味単位予測により言語間の語順の違いを解消し、オフライン翻訳に匹敵するリアルタイム翻訳品質を提供します。

• 自然な音声：ソース音声のイントネーションや感情を自動的にマッチングし、自然な音声を生成します。

• ホットワード設定：特定の用語に対する翻訳精度を向上させるためにホットワードを設定できます。

操作手順

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

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-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
   
   or
   
   pip install pyaudio

   CentOS

   sudo yum install -y portaudio portaudio-devel && pip install pyaudio

   Windows

   pip install pyaudio

   次に、WebSocket の依存関係をインストールします。

   pip install websocket-client==1.8.0 websockets

2. クライアントの作成

   以下のコードを含む 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", voice: str | None = "Cherry", *, 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.voice = voice if audio_enabled else "Cherry"
           self.ws = None
           self.api_url = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-livetranslate-flash-realtime"
           
           # Audio input configuration (from microphone)
           self.input_rate = 16000
           self.input_chunk = 1600
           self.input_format = pyaudio.paInt16
           self.input_channels = 1
           
           # Audio output configuration (for playback)
           self.output_rate = 24000
           self.output_chunk = 2400
           self.output_format = pyaudio.paInt16
           self.output_channels = 1
           
           # State management
           self.is_connected = False
           self.audio_player_thread = None
           self.audio_playback_queue = queue.Queue()
           self.pyaudio_instance = pyaudio.PyAudio()
   
       async def connect(self):
           """Establish 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 the 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, setting the target language, voice, etc."""
           config = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "session.update",
               "session": {
                   # 'modalities' controls the output type.
                   # ["text", "audio"]: Returns both translated text and synthesized audio (recommended).
                   # ["text"]: Returns only the translated text.
                   "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                   **({"voice": self.voice} if self.audio_enabled and self.voice else {}),
                   "input_audio_format": "pcm",
                   "output_audio_format": "pcm",
                   # 'input_audio_transcription' configures source language recognition.
                   # Set 'model' to 'qwen3-asr-flash-realtime' to also output the source language recognition result.
                   # "input_audio_transcription": {
                   #     "model": "qwen3-asr-flash-realtime",
                   #     "language": "zh"  # source language, default 'en'
                   # },
                   "translation": {
                       "language": self.target_language,
                       # 'corpus' configures hotwords to improve the translation accuracy of specific terms.
                       # "corpus": {
                       #     "phrases": {
                       #         "人工智能": "Artificial Intelligence",
                       #         "机器学习": "Machine Learning"
                       #     }
                       # }
                   }
               }
           }
           print(f"Sending session configuration: {json.dumps(config, indent=2, ensure_ascii=False)}")
           await self.ws.send(json.dumps(config))
   
       async def send_audio_chunk(self, audio_data: bytes):
           """Encode and send an audio chunk 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.
           if not self.is_connected:
               return
   
           if not image_bytes:
               raise ValueError("image_bytes cannot be empty.")
   
           # Encode to 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: # Termination signal
                           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):
           """Start the audio player thread (only when audio output is enabled)."""
           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):
           """Handle incoming messages from the server in a loop."""
           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 round complete.")
                       usage = event.get("response", {}).get("usage", {})
                       if usage:
                           print(f"[INFO] token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                   # Process source language recognition results (requires enabling input_audio_transcription.model)
                   # elif event_type == "conversation.item.input_audio_transcription.text":
                   #     stash = event.get("stash", "")  # Pending recognition text
                   #     print(f"[Recognizing] {stash}")
                   # elif event_type == "conversation.item.input_audio_transcription.completed":
                   #     transcript = event.get("transcript", "")  # Complete recognition result
                   #     print(f"[Source language] {transcript}")
                   elif event_type == "response.audio_transcript.done":
                       print("\n[INFO] Translation complete.")
                       text = event.get("transcript", "")
                       if text:
                           print(f"[INFO] Translated text: {text}")
                   elif event_type == "response.text.done":
                       print("\n[INFO] Translation complete.")
                       text = event.get("text", "")
                       if text:
                           print(f"[INFO] Translated text: {text}")
   
           except websockets.exceptions.ConnectionClosed as e:
               print(f"[WARNING] Connection closed: {e}")
               self.is_connected = False
           except Exception as e:
               print(f"[ERROR] An unexpected error occurred while processing messages: {e}")
               traceback.print_exc()
               self.is_connected = False
   
       async def start_microphone_streaming(self):
           """Capture audio from the microphone and stream it to the server."""
           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 is on. Start 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 connection and release resources."""
           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) # Send termination signal
               self.audio_player_thread.join(timeout=1)
               print("Audio player thread stopped.")
               
           self.pyaudio_instance.terminate()
           print("PyAudio instance released.")

3. モデルとの対話

   同じディレクトリに、以下のコードを含む main.py という名前のファイルを作成します。

   main.py

   import os
   import asyncio
   from livetranslate_client import LiveTranslateClient
   
   def print_banner():
       print("=" * 60)
       print("  Powered by Qwen qwen3-livetranslate-flash-realtime")
       print("=" * 60 + "\n")
   
   def get_user_config():
       """Get user configuration."""
       print("Select a mode:")
       print("1. Voice + Text [Default] | 2. Text Only")
       mode_choice = input("Enter your choice (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 the target 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 the target 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 your choice (defaults to the first option): ").strip()
       target_language = lang_map.get(choice, next(iter(lang_map.values())))
   
       voice = None
       if audio_enabled:
           print("\nSelect a speech synthesis voice:")
           voice_map = {"1": "Cherry", "2": "Nofish", "3": "Sunny", "4": "Jada", "5": "Dylan", "6": "Peter", "7": "Eric", "8": "Kiki"}
           print("1. Cherry (Female) [Default] | 2. Nofish (Male) | 3. Sunny (Sichuan Female) | 4. Jada (Shanghai Female) | 5. Dylan (Beijing Male) | 6. Peter (Tianjin Male) | 7. Eric (Sichuan Male) | 8. Kiki (Cantonese Female)")
           voice_choice = input("Enter your choice (press Enter for Cherry): ").strip()
           voice = voice_map.get(voice_choice, "Cherry")
       return target_language, voice, audio_enabled
   
   async def main():
       """Main program entry point."""
       print_banner()
       
       api_key = os.environ.get("DASHSCOPE_API_KEY")
       if not api_key:
           print("[ERROR] Please set the DASHSCOPE_API_KEY environment variable.")
           print("  For example: export DASHSCOPE_API_KEY='your_api_key_here'")
           return
           
       target_language, voice, audio_enabled = get_user_config()
       print("\nConfiguration complete:")
       print(f"  - Target language: {target_language}")
       if audio_enabled:
           print(f"  - Synthesized voice: {voice}")
       else:
           print("  - Output mode: Text Only")
       
       client = LiveTranslateClient(api_key=api_key, target_language=target_language, voice=voice, audio_enabled=audio_enabled)
       
       # Define the callback function.
       def on_translation_text(text):
           print(text, end="", flush=True)
   
       try:
           print("Connecting to the translation service...")
           await client.connect()
           
           # Start audio playback based on the mode.
           client.start_audio_player()
           
           print("\n" + "-" * 60)
           print("Connection successful! Speak into the microphone.")
           print("The program will translate your speech in real time and play the translated audio. Press Ctrl+C to exit.")
           print("-" * 60 + "\n")
   
           # Run message handling and microphone recording concurrently.
           message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
           tasks = [message_handler]
           # Capture audio from the microphone for translation, regardless of whether audio output is enabled.
           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"\nA critical error occurred: {e}")
       finally:
           print("\nCleaning up resources...")
           await client.close()
           print("Program exited.")
   
   if __name__ == "__main__":
       asyncio.run(main())

   main.py を実行し、マイクに向かって話しかけてください。モデルはリアルタイムで翻訳された音声およびテキストを出力します。システムは自動的に発話を検出し、サーバーに送信します。

画像による翻訳精度の向上

qwen3-livetranslate-flash-realtime モデルは画像入力を活用して音声翻訳を改善し、同音異義語の曖昧さを解消したり、一般的でない固有名詞を認識したりします。1 秒あたり最大 2 枚の画像を送信してください。

以下のサンプル画像をダウンロードします：medical mask.pngmasquerade mask.png

以下のコードを livetranslate_client.py と同じディレクトリにダウンロードして実行します。マイクに向かって "What is mask?" と話しかけてください。モデルは提供された画像を使用して「mask」という単語の曖昧さを解消します。たとえば、medical mask.png ファイルを使用すると、「What is a medical mask?」と翻訳され、masquerade mask.png ファイルを使用すると、「What is a masquerade mask?」と翻訳されます。

import os
import time
import json
import asyncio
import contextlib
import functools

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "medical mask.png"
# IMAGE_PATH = "masquerade mask.png"

def print_banner():
    print("=" * 60)
    print("  Powered by Qwen qwen3-livetranslate-flash-realtime — single-turn interaction example (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] Recording started. Please speak...")
    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)

            # Append an image frame every 0.5 seconds
            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] Please set the DASHSCOPE_API_KEY environment variable.")
        return

    client = LiveTranslateClient(api_key=api_key, target_language="zh", voice="Cherry", 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 キーを入力して、[Create and Deploy Default Environment] をクリックしてアプリケーションをテストします。

2. 約 1 分待ちます。Environment Details > Environment Context でエンドポイントを取得し、プロトコルを http から https に変更します（例：https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/）。ブラウザで URL を開き、モデルと対話します。

   重要

   このエンドポイントは自己署名証明書を使用しており、一時的なテスト専用です。初回アクセス時にブラウザがセキュリティ警告を表示しますが、これは想定された動作です。本番環境ではこのエンドポイントを使用しないでください。警告画面で画面上の指示に従って進めてください（例：詳細設定 → （安全ではないサイトに）進む）。

Resource Access Management 権限の構成を求められた場合は、画面上の指示に従ってください。

プロジェクトのソースコードを確認するには、Resource Information > Function Resources に移動します。

Function Compute および Model Studio は新規ユーザー向けに無料クォータを提供しており、基本的なデバッグに十分です。無料クォータを使い切った後は、従量課金が適用されます。

対話フロー

リアルタイム音声翻訳は、イベント駆動型の WebSocket モデルに従います。サーバーは自動的に発話区間を検出し、応答します。

API

Qwen-Livetranslate-Realtime をご参照ください。

課金

• 音声：音声入力または出力の 1 秒あたり 12.5 トークンを消費します。

• 画像：28×28 ピクセルごとに 0.5 トークンを消費します。

• テキスト：ソース言語の音声認識を有効にすると、サービスは翻訳に加えて入力音声のトランスクリプトを返します。このトランスクリプトは出力テキストトークンとして課金されます。

料金については、モデル一覧をご参照ください。

サポートされる言語

ソース言語およびターゲット言語を指定するには、以下の言語コードを使用します。

一部のターゲット言語はテキストのみをサポートします。

サポートされる音声