DocsICP FilingConsole
Documentation
Home Page Alibaba Cloud Model Studio Model API Realtime API Connect to models and applications WebSocket Real-time audio and video translation Qwen-Livetranslate-Realtime Python SDK

Qwen-LiveTranslate Python SDK - API reference

Updated at:
Copy as MD
Product
My Favorites

Use the DashScope SDK for Python to call Qwen-LiveTranslate for real-time speech translation.

Prerequisites

  1. Install the SDK. Make sure that your DashScope SDK version is 2.0.1 or later.
  2. Obtain an API key.

ImportantAlibaba Cloud Model Studio has released workspace-specific domains for the China (Beijing) and Singapore regions. The new dedicated domains deliver superior performance and higher stability for inference requests. We recommend migrating to the new domains:

  • China (Beijing): from wss://dashscope.aliyuncs.com to wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com
  • Singapore: from wss://dashscope-intl.aliyuncs.com to wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com

{WorkspaceId} is your workspace ID, which can be found on the Workspace Details page in the Alibaba Cloud Model Studio console. The existing domain remains fully functional.

Request parameters

Key interfaces

Session (RealtimeSession)

Real-time translation sessions are created via client.audio.realtime.connect(model=...). Entering the with context establishes the connection automatically (you can also call the session's connect() method to connect manually).

Method signatureServer-side response event (delivered via events() iteration)Description
def connect(self) -> "RealtimeSession"

Server-side event

Session created

Server-side event

Session configuration updated

Connect to the server. Automatically called when entering the with context.

def update_session(self,
    output_modalities: List[str],
    *,
    voice: str = None,
    input_audio_transcription_model: str = None,
    translation_params: Dict[str, Any] = None,
    enable_turn_detection: bool = None,
    **kwargs) -> None:

Server-side event

Session updated

Update session configuration. We recommend calling this method immediately after the connection is established. If you do not call this method, default configurations apply. You only need to focus on the parameters involved in the session's update_session method.

def end_session(self) -> None

session.finished

The server completes the speech translation and ends the session

Notify the server to end the session. Upon receiving the end-session notification, the server completes the final speech translation.

def append_audio(self, audio_b64: Union[str, bytes]) -> None

None

Append audio data chunks to the cloud input audio buffer. Base64 strings and raw bytes (encoded automatically) are both supported. The server automatically detects the start and end of speech and triggers translation.

def commit(self) -> None

input_audio_buffer.committed

Input audio buffer committed

In Manual mode, submit the audio previously appended to the server buffer via the append_audio method. The server automatically starts generating translation responses upon receipt. In VAD mode, this method does not need to be called as the server commits automatically.

def clear_appended_audio(self) -> None

input_audio_buffer.cleared

Input audio buffer cleared

Clear uncommitted audio data in the current server buffer.

def close(self) -> None

None

Stop the task and close the connection. Automatically called when exiting the with context.

session.session_id

None

Get the session ID of the current task (read from the session.created event).

session.last_response_id

None

Get the response ID of the last response (read from the response.created event).

Event iteration (events())

The server returns response events and data to the client as events. Iterate the session's events() in a receiver thread to process server-side events; iteration ends after the session.finished event.

The error event is raised as a DashScopeError exception, which you can catch with try/except in the receiver thread.

Method signatureParametersDescription
def events(self) -> Iterator[Dict[str, Any]]

event: Server-side event

Iterate over server-side events (dict); iteration ends after the session.finished event. Consume events from a single receiver thread; do not create multiple concurrent readers.

def recv_event(self) -> Optional[Dict[str, Any]]

event: Server-side event

Read one server-side event; returns None after a normal close.

Complete example

This example records microphone audio and translates it in real time.

Sample code for real-time translation from a microphone

import os
import sys
import base64
import signal
import pyaudio
from concurrent.futures import ThreadPoolExecutor

import dashscope
from dashscope.v2 import DashScopeError

def main():
    # Check for the API key.
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("Set the DASHSCOPE_API_KEY environment variable.")
        sys.exit(1)

    # China (Beijing) region. Replace "{WorkspaceId}" with your actual workspace ID. Configurations differ by region.
    client = dashscope.Client(
        # If you have not configured the API key as an environment variable, replace the following line with your API key: api_key="sk-xxx"
        api_key=os.environ["DASHSCOPE_API_KEY"],
        workspace="{WorkspaceId}",
        region="cn-beijing",
    )

    # Initialize PyAudio.
    pya = pyaudio.PyAudio()

    # Initialize the speaker for playing back the translated audio.
    speaker = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True,
        frames_per_buffer=2400
    )

    # Initialize the microphone for capturing speech input.
    mic = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        frames_per_buffer=1600
    )

    # Create a real-time translation session: entering the with context establishes the connection automatically
    with client.audio.realtime.connect(
            model="qwen3.5-livetranslate-flash-realtime",
    ) as session:
        # Update the session configuration (recommended immediately after the connection is established)
        session.update_session(
            output_modalities=["text", "audio"],
            input_audio_transcription_model="qwen3-asr-flash-realtime",
            voice="Tina",
            translation_params={
                "language": "en",  # Target language for translation: English
                "corpus": {
                    "phrases": {
                        "人工智能": "Artificial Intelligence",
                        "机器学习": "Machine Learning"
                    }
                }
            },
        )

        # Receiver thread: iterate server-side events
        def receive_events():
            try:
                for event in session.events():
                    event_type = event.get("type", "")
                    if event_type == "input_audio_buffer.speech_started":
                        print("====== Speech input detected ======")
                    elif event_type == "input_audio_buffer.speech_stopped":
                        print("====== Speech input ended ======")
                    elif event_type == "conversation.item.input_audio_transcription.completed":
                        print(f"[Original text] {event.get('transcript', '')}")
                    elif event_type == "response.audio_transcript.done":
                        print(f"[Translation result] {event.get('transcript', '')}")
                    elif event_type == "response.audio.delta":
                        audio_b64 = event.get("delta", "")
                        if audio_b64:
                            speaker.write(base64.b64decode(audio_b64))
            except DashScopeError as e:
                print(f"[Error] {e.message}")

        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events)

            # Register the exit signal handler.
            def on_exit(sig, frame):
                print("\n[Exiting...]")
                session.end_session()
                session.close()
                mic.stop_stream()
                mic.close()
                speaker.stop_stream()
                speaker.close()
                pya.terminate()
                sys.exit(0)

            signal.signal(signal.SIGINT, on_exit)

            print("[Starting real-time translation] Speak into the microphone. Press Ctrl+C to exit.")

            # Continuously capture and send microphone audio (append_audio also accepts raw bytes, encoded automatically)
            try:
                while True:
                    audio_data = mic.read(1600, exception_on_overflow=False)
                    session.append_audio(audio_data)
            finally:
                session.end_session()
                receiver.result()

if __name__ == "__main__":
    main()
Previous：Qwen-Livetranslate-RealtimeNext：Qwen-LiveTranslate Java SDK