time audio and video translation - Qwen-Livetranslate-Realtime-Qwen-LiveTranslate Python SDK - API reference

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 1.25.6 or later.

2. Create an API key.

Request parameters

• Set the following parameters in the constructor of the OmniRealtimeConversation class.

  Click to view sample code

  HELPCODEESCAPE-python
  from dashscope.audio.qwen_omni import (
      OmniRealtimeConversation,
      OmniRealtimeCallback,
      MultiModality,
  )
  from dashscope.audio.qwen_omni.omni_realtime import TranslationParams
  
  
  class MyCallback(OmniRealtimeCallback):
      """Callback handler for real-time translation"""
      def __init__(self, conversation=None):
          self.conversation = conversation
          self.handlers = {
              'session.created': self._handle_session_created,
              'response.audio_transcript.done': self._handle_translation_done,
              'response.audio.delta': self._handle_audio_delta,
              'response.done': lambda r: print('======Response Done======'),
              'input_audio_buffer.speech_started': lambda r: print('======Speech Start======'),
              'input_audio_buffer.speech_stopped': lambda r: print('======Speech Stop======'),
          }
  
      def on_open(self):
          print('Connection opened')
  
      def on_close(self, code, msg):
          print(f'Connection closed, code: {code}, msg: {msg}')
  
      def on_event(self, response):
          try:
              handler = self.handlers.get(response['type'])
              if handler:
                  handler(response)
          except Exception as e:
              print(f'[Error] {e}')
  
      def _handle_session_created(self, response):
          print(f"Session created: {response['session']['id']}")
  
      def _handle_translation_done(self, response):
          print(f"Translation result: {response['transcript']}")
  
      def _handle_audio_delta(self, response):
          # Process incremental audio data.
          audio_b64 = response.get('delta', '')
          # Decode the audio data for playback or to save it.
  
  conversation = OmniRealtimeConversation(
      model='qwen3-livetranslate-flash-realtime',
      # The following URL is for Chinese Mainland region. For international use, replace the URL with: wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime
      url='wss://dashscope.aliyuncs.com/api-ws/v1/realtime',
      callback=MyCallback(conversation=None)  # Temporarily pass None. It will be injected later.
  )
  # Inject self into the callback.
  conversation.callback.conversation = conversation

Parameter	Type	Required	Description
model	str	Yes	The name of the model to use. Set this to qwen3-livetranslate-flash-realtime.
callback	OmniRealtimeCallback	Yes	A callback instance to handle server-side events.
url	str	Yes	The service endpoint for real-time translation: - Chinese Mainland: wss://dashscope.aliyuncs.com/api-ws/v1/realtime - International: wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime

• Set the following parameters using the update_session method of the OmniRealtimeConversation class.

  Click to view sample code

  HELPCODEESCAPE-python
  # Set translation parameters
  translation_params = TranslationParams(
      language='en',  # Target language
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )
  
  # Update session configuration
  conversation.update_session(
      output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
      voice='Cherry',
      translation_params=translation_params,
  )

Parameter	Type	Required	Description
output_modalities	List\[MultiModality\]	No	The output format for translation results. Default: \[MultiModality.TEXT, MultiModality.AUDIO\]. Valid values: - \[MultiModality.TEXT\]: Outputs text only - \[MultiModality.TEXT, MultiModality.AUDIO\]: Outputs text and audio
voice	str	No	The voice for audio output. Default: Cherry. Valid values: Supported voices.
input_audio_transcription_model	str	No	Set this parameter to qwen3-asr-flash-realtime to receive speech recognition results for the source language.
translation_params	TranslationParams	No	Translation settings (target language and hotwords).

• Set the following parameters in the constructor of the TranslationParams class.

  Click to view sample code

  HELPCODEESCAPE-python
  translation_params = TranslationParams(
      language='en',  # Target language code
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',  # Source phrase: Target translation
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )

Parameter	Type	Required	Description
language	str	No	The target language for translation. Default: en. Valid values: Supported languages.
corpus	TranslationParams.Corpus	No	A hotword mapping to improve translation accuracy for specific terms.
corpus.phrases	dict	No	The hotword mapping table where keys are source terms and values are target translations. Example: {'Inteligencia Artificial': 'Artificial Intelligence'}

Key interfaces

OmniRealtimeConversation class

Import: from dashscope.audio.qwen_omni import OmniRealtimeConversationMethod signature Server-side response event (sent via callback) Description

def connect(self) -\> None:

Server-side event ** Session created Server-side event Session configuration updated Connect to the server.

def update_session(self,
 output_modalities: List\[MultiModality\],
 voice: str = None,
 translation_params: TranslationParams = None,
 **kwargs) -\> None:

Server-side event Session configuration updated Update session configuration. Call immediately after connecting. If you do not call this method, default configurations apply.

def end_session(self, timeout: int = 20) -\> None:

session.finished The server completes the speech translation and ends the session End the session. The server completes final translation processing.

def append_audio(self, audio_b64: str) -\> None:

None Send Base64-encoded audio chunks to the server. Speech detection and translation are automatic.

def close(self) -\> None:

None Stop the task and close the connection.

def get_session_id(self) -\> str:

None Get the session ID.

def get_last_response_id(self) -\> str:

None Get the last response ID.

Callback interface (OmniRealtimeCallback)

Receive server events and data through callbacks.

Inherit this class and implement its methods to handle events from the server.

Import: from dashscope.audio.qwen_omni import OmniRealtimeCallback <b>Method signature** Parameters Description

def on_open(self) -\> None:

None Called when the WebSocket connection opens.

def on_event(self, message: dict) -\> None:

message: Server-side event Called when the server sends an event.

def on_close(self, close_status_code, close_msg) -\> None:

close_status_code: The status code. close_msg: The log message when the WebSocket connection is closed. Called when the WebSocket connection closes.

Complete example

This example records microphone audio and translates it in real time. Sample code for real-time translation from a microphone

HELPCODEESCAPE-python
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):
    """Callback handler class for real-time translation"""

    def __init__(self, speaker):
        self.speaker = speaker

    def on_open(self):
        print("[Connection established]")

    def on_close(self, code, msg):
        print(f"[Connection closed] code: {code}, msg: {msg}")

    def on_event(self, response):
        event_type = response.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] {response.get('transcript', '')}")
        elif event_type == "response.audio_transcript.done":
            print(f"[Translation result] {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"[Error] {response.get('error', {}).get('message', '')}")


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)

    # 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 callback instance.
    callback = Callback(speaker=speaker)

    # Create a real-time session.
    conversation = OmniRealtimeConversation(
        model="qwen3-livetranslate-flash-realtime",
        # For Chinese Mainland region, use wss://dashscope.aliyuncs.com/api-ws/v1/realtime
        url="wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime",
        callback=callback
    )

    # Connect to the server.
    conversation.connect()

    # Configure translation parameters.
    translation_params = TranslationParams(
        language="en",  # Target language for translation: English
        corpus=TranslationParams.Corpus(
            phrases={
                "Source Term 1": "Target Translation 1",
                "Source Term 2": "Target Translation 2"
            }
        )
    )

    # Update the session configuration.
    conversation.update_session(
        output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
        input_audio_transcription_model="qwen3-asr-flash-realtime",
        voice="Cherry",
        translation_params=translation_params,
    )

    # Register the exit signal handler.
    def on_exit(sig, frame):
        print("\n[Exiting...]")
        mic.stop_stream()
        mic.close()
        speaker.stop_stream()
        speaker.close()
        pya.terminate()
        conversation.close()
        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.
    while True:
        audio_data = mic.read(1600, exception_on_overflow=False)
        conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))


if __name__ == "__main__":
    main()