Real-time audio and video translation - Qwen

Try an online demo with one-click deployment using Function Compute .

​

Features

• Multi-language support: Translates between 60 languages — 29 with audio and text output, 31 with text-only output — including Chinese, English, French, German, Russian, Japanese, Korean, Spanish, Portuguese, and Arabic.
• Visual enhancement: Analyzes visual cues, such as lip movements, gestures, and on-screen text, to improve translation accuracy, especially in noisy environments or for ambiguous words.
• 2.3-second latency: Delivers simultaneous interpretation with latency as low as 2.3 seconds.
• Real-time speaker diarization: Distinguishes different speakers and their speech when multiple people take turns speaking, so listeners can clearly understand who said what.
• Lossless simultaneous interpretation: Predicts semantic units to resolve cross-language word order differences, achieving quality comparable to offline translation.
• Natural voice: Matches the intonation and emotion of the source audio automatically.
• Hotword configuration: Configurable hotwords improve translation accuracy for specific terms.
• Voice cloning: Clones the speaker's voice for translated output. Supports real-time cloning and pre-cloned voice profiles.

​

Procedure

​

1. Configure the connection

# 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.8-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()

​

2. Configure language, modality, and voice

• Target language: Set session.translation.language, such as en for English. With qwen3.8-livetranslate-flash-realtime, the source language is detected automatically.
  With qwen3.5-livetranslate-flash-realtime, set session.input_audio_transcription.language to specify the source language, or omit it for automatic detection.
• Output modalities: With qwen3.8-livetranslate-flash-realtime, set session.output_modalities to ["text"] for text only or ["text", "audio"] for text and audio (default).
  With qwen3.5-livetranslate-flash-realtime, use session.modalities. The supported values and default are the same.
• Source transcription: With qwen3.8-livetranslate-flash-realtime, source transcription is always enabled and cannot be disabled. Receive increments through conversation.item.input_audio_transcription.delta and complete transcripts through conversation.item.input_audio_transcription.completed.
  With qwen3.5-livetranslate-flash-realtime, source transcription is enabled by default. Set session.input_audio_transcription.model to qwen3-asr-flash-realtime to enable it or null to disable it. Receive increments through conversation.item.input_audio_transcription.text and complete transcripts through conversation.item.input_audio_transcription.completed.
• Audio and voice: With qwen3.8-livetranslate-flash-realtime, defaults are 16000 Hz PCM input, 24000 Hz PCM output, and voice Tina. Check session.audio.output.voice in session.created for the voice. To preserve the speaker’s voice, see Voice cloning.
  With qwen3.5-livetranslate-flash-realtime, configure audio through session.sample_rate, session.input_audio_format, and session.output_audio_format, and the voice through session.voice. The default voice is also Tina.
• Hotwords: Set session.translation.corpus.phrases to specify preferred translations. See Configure hotwords.
  We recommend using no more than 1,000 hotwords.

​

3. Input audio and images

Images can be from a local file or captured in real time from a video stream.

​

4. Receive the model response

• Text only: Concatenate the delta values from response.text.delta to obtain the translation.
• Text and audio: Concatenate delta from response.audio_transcript.delta for text, and Base64-decode delta from response.audio.delta to obtain audio chunks.

​

5. End the session

​

Supported models

​

Recommended models

​

Getting started

​

Translate microphone input in real time

1. Use Python 3.10 or later and install pyaudio:
   • macOS
   • Debian/Ubuntu
   • CentOS
   • Windows

   Copy

   brew install portaudio && pip install pyaudio
   

   Copy

   sudo apt-get install python3-pyaudio
   

   Copy

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

   Copy

   pip install pyaudio
   
   Install the WebSocket dependency: pip install websocket-client.
2. Set these environment variables. The API key and workspace must belong to the region in the connection URL:
   • DASHSCOPE_API_KEY: An API key for the region and workspace. See Get an API key.
   • DASHSCOPE_WORKSPACE_ID: Your workspace ID. See Workspaces and regions.
3. Save the code below as livetranslate_client.py, run python livetranslate_client.py, and speak into the microphone.

import base64
import json
import os
import signal
import threading
import time

import pyaudio
import websocket

model = "qwen3.8-livetranslate-flash-realtime"
workspace_id = os.environ["DASHSCOPE_WORKSPACE_ID"]
api_key = os.environ["DASHSCOPE_API_KEY"]
url = (
    f"wss://{workspace_id}.ap-southeast-1.maas.aliyuncs.com"
    f"/api-ws/v1/realtime?model={model}"
)
stop_recording = False
send_errors = []
ws = None
audio = None
input_stream = None
output_stream = None
sender = None
previous_sigint_handler = None


def receive():
    event = json.loads(ws.recv())
    if event.get("type") == "error" or event.get("code"):
        raise RuntimeError(event)
    return event


def stop_input(signum, frame):
    global stop_recording
    stop_recording = True


def send_audio():
    try:
        while not stop_recording:
            chunk = input_stream.read(1600)
            ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode("ascii"),
            }))
    except Exception as error:
        send_errors.append(error)
    finally:
        try:
            ws.send(json.dumps({"type": "session.finish"}))
        except Exception as error:
            send_errors.append(error)
            ws.close()


def receive_translation():
    while True:
        event = receive()
        event_type = event.get("type")
        if event_type in ("response.text.delta", "response.audio_transcript.delta"):
            print(event["delta"], end="", flush=True)
        elif event_type == "response.audio.delta":
            output_stream.write(base64.b64decode(event["delta"]))
        elif event_type == "response.done":
            if event["response"]["status"] != "completed":
                raise RuntimeError(event["response"])
            print()
        elif event_type == "session.finished":
            return


try:
    ws = websocket.create_connection(
        url, header={"Authorization": f"Bearer {api_key}"}, timeout=30
    )
    while receive().get("type") != "session.created":
        pass
    ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "output_modalities": ["text", "audio"],
            "translation": {"language": "en"},
        },
    }))
    while receive().get("type") != "session.updated":
        pass
    audio = pyaudio.PyAudio()
    input_stream = audio.open(
        format=pyaudio.paInt16, channels=1, rate=16000,
        input=True, frames_per_buffer=1600,
    )
    output_stream = audio.open(
        format=pyaudio.paInt16, channels=1, rate=24000, output=True,
    )
    print("Start speaking. Press Ctrl+C to stop recording and wait for the remaining translation.")
    previous_sigint_handler = signal.signal(signal.SIGINT, stop_input)
    sender = threading.Thread(target=send_audio, daemon=True)
    sender.start()
    receive_translation()
    stop_recording = True
    sender.join()
    if send_errors:
        raise send_errors[0]
except Exception:
    if send_errors:
        raise send_errors[0]
    raise
finally:
    stop_recording = True
    if ws is not None:
        ws.close()
    if sender is not None:
        sender.join()
    for stream in (input_stream, output_stream):
        if stream is not None:
            stream.stop_stream()
            stream.close()
    if audio is not None:
        audio.terminate()
    if previous_sigint_handler is not None:
        signal.signal(signal.SIGINT, previous_sigint_handler)

​

Configure hotwords

• Fixed wording: Alibaba Cloud's slogan, “计算，为了无法计算的价值。”, has the agreed English wording “Computing, for the value beyond computation.”. The wording and word order need to be preserved.
• Brand names that sound alike: Qoder sounds like the common word coder. When the speaker means the product Qoder, the translation needs to retain the brand's spelling.

​

Configuration

{
  "event_id": "configure-hotwords",
  "type": "session.update",
  "session": {
    "translation": {
      "language": "en",
      "corpus": {
        "phrases": {
          "计算，为了无法计算的价值。": "Computing, for the value beyond computation.",
          "Qoder": "Qoder"
        }
      }
    }
  }
}

​

Identify speakers

{"type": "session.update", "session": {"audio": {"input": {"turn_detection": {"type": "speaker_detection", "threshold": 0.5}}}}}

1. Record the speaker_id and source item_id from input_audio_buffer.speech_started.
2. Read the source text for that item_id from conversation.item.input_audio_transcription.delta.
3. In the translation's conversation.item.created, associate item.id with the source ID in the event's top-level previous_item_id.
4. Use each translation increment's item_id to find its source text and speaker, then update the subtitle.

speaker_by_source = {}
source_by_translation = {}
source_text = {}
translation_text = {}


def show_subtitle(translation_id):
    source_id = source_by_translation.get(translation_id)
    if source_id not in speaker_by_source or translation_id not in translation_text:
        return
    print({
        "speaker_id": speaker_by_source[source_id],
        "source": source_text.get(source_id, ""),
        "translation": translation_text[translation_id],
    })


def handle_speaker_event(event):
    event_type = event.get("type")
    if event_type in ("response.text.delta", "response.audio_transcript.delta"):
        item_id = event["item_id"]
        translation_text[item_id] = translation_text.get(item_id, "") + event["delta"]
        show_subtitle(item_id)
        return
    if event_type == "conversation.item.created":
        item = event["item"]
        if any(part.get("type") == "input_audio" for part in item.get("content", [])):
            return
        source_id = event.get("previous_item_id")
        if item.get("role") == "assistant" and source_id is not None:
            source_by_translation[item["id"]] = source_id
            show_subtitle(item["id"])
        return
    if event_type == "input_audio_buffer.speech_started" and "speaker_id" in event:
        source_id = event["item_id"]
        speaker_by_source[source_id] = event["speaker_id"]
    elif event_type == "conversation.item.input_audio_transcription.delta":
        source_id = event["item_id"]
        source_text[source_id] = source_text.get(source_id, "") + event["delta"]
    elif event_type == "conversation.item.input_audio_transcription.completed":
        source_id = event["item_id"]
        source_text[source_id] = event["transcript"]
    else:
        return
    for translation_id, mapped_source in source_by_translation.items():
        if mapped_source == source_id:
            show_subtitle(translation_id)

​

Improve translation with images

1. Install Pillow: pip install Pillow.
2. Set translation.language to zh for Chinese output, then say “What is mask?” into the microphone when you run the example.
3. Place the image-loading code below before def send_audio(): in the microphone example, and replace the original send_audio() with this version. It uses the existing input_stream, stop_recording, ws, send_errors, and imports.

import io
from PIL import Image

IMAGE_PATH = "medical-mask.png"
# IMAGE_PATH = "masquerade-mask.png"
with Image.open(IMAGE_PATH) as image:
    buffer = io.BytesIO()
    image.convert("RGB").save(buffer, format="JPEG")
image_bytes = buffer.getvalue()
if len(image_bytes) > 500_000:
    raise ValueError("JPEG > 500 KB")
image_b64 = base64.b64encode(image_bytes).decode("ascii")


def send_audio():
    try:
        last_image_time = None
        while not stop_recording:
            chunk = input_stream.read(1600)
            ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode("ascii"),
            }))
            now = time.monotonic()
            if last_image_time is None or now - last_image_time >= 0.5:
                ws.send(json.dumps({
                    "type": "input_image_buffer.append",
                    "image": image_b64,
                }))
                last_image_time = now
    except Exception as error:
        send_errors.append(error)
    finally:
        try:
            ws.send(json.dumps({"type": "session.finish"}))
        except Exception as error:
            send_errors.append(error)
            ws.close()

​

Voice cloning

• session.enable_voice_clone: Set to true to enable voice cloning.
• session.voice_clone_options.frequency: Controls when voice cloning occurs. Accepted values:
  • never: Use a previously created voice without extracting a new voice during this session. Set session.voice to your cloned voice ID.
  • once: Extract the voice once from input audio at the start of the session and reuse it for subsequent translations in this session. “Once” describes extraction, not a single audio output. Suitable for a single-speaker presentation; set session.voice to default.
  • always: Extract the voice before each translated-audio response, so output follows speaker changes. Suitable for multiple speakers; set session.voice to default.
• session.voice: Specifies the output voice. The value depends on the frequency setting:
  • Set to default: Use with frequency set to once or always. Clones the speaker's voice from the input audio. A default voice is used until cloning completes.
  • Set to a custom cloned voice ID (for example, qwen-translate-vc-xxx-yyy-zzz): Use with frequency set to never. You must prepare the voice in advance using the Voice Cloning API with target_model set to the translation model you use.

When frequency is set to once or always , the voice parameter must be set to default . Any other value causes the server to return an error.

​

Voice cloning configuration examples

{
    "type": "session.update",
    "session": {
        "output_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": {
        "output_modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "once"
        }
    }
}

{
    "type": "session.update",
    "session": {
        "output_modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "always"
        }
    }
}

​

One-click Function Compute deployment

1. Open the Function Compute template, enter your API key, and click Create and Deploy Default Environment to test the application.
2. Wait for about a minute. In Environment Details > Environment Context, retrieve the endpoint, change the protocol from http to https (for example, https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/), and open the URL in a browser to interact with the model.
   This endpoint uses a self-signed certificate and is for temporary testing only. Your browser will display a security warning on your first visit. This is expected behavior. Do not use this endpoint in a production environment. To proceed, follow the on-screen instructions (for example, click Advanced → Proceed to (unsafe)).

If you are prompted to configure Resource Access Management permissions, follow the on-screen instructions.

To view the project source code, go to Resource Information > Function Resources .

Both Function Compute and Model Studio provide a free quota for new users, sufficient for basic debugging. After the free quota is used up, pay-as-you-go billing applies.

​

Interaction flow

​

API

• Qwen-Livetranslate-Realtime.
• Realtime API overview (WebRTC protocol description)

​

Billing

• Audio: 7 tokens per second of input audio; 12.5 tokens per second of output audio.
• Image: Every 32×32 pixels consumes 0.5 tokens. For token prices for each model, see Model pricing and billing.

​

Rate limits

​

Supported languages

Some target languages only support text.

​

Supported voices