Qwen-LiveTranslate Python SDK - API reference

​

Prerequisites

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

​

Request parameters

• Set the following parameters in the constructor of the OmniRealtimeConversation class.
  Click to view sample code

  Copy

  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.5-livetranslate-flash-realtime',
      # China (Beijing) region. Replace {WorkspaceId} with your Bailian workspace ID. URLs vary by region.
      url='wss://{WorkspaceId}.cn-beijing.maas.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.5-livetranslate-flash-realtime (recommended).
  callback	OmniRealtimeCallback	Yes	A callback instance to handle server-side events.
  url	str	Yes	The service endpoint for real-time translation: China (Beijing): wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime Singapore: wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime Replace {WorkspaceId} with your actual workspace ID.

• Set the following parameters using the update_session method of the OmniRealtimeConversation class.
  Click to view sample code

  Copy

  # 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='Tina',
      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: Qwen3.5-LiveTranslate-Flash-Realtime: Tina 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).
  enable_turn_detection	bool	No	Whether to enable VAD (Voice Activity Detection). Default value: True, which enables VAD mode where the server automatically detects speech start/end and triggers translation. Set to False to switch to Manual mode, where the client manually submits audio via the commit method. For detailed parameter descriptions, see the turn_detection description in the Client events document.

• Set the following parameters in the constructor of the TranslationParams class.
  Click to view sample code

  Copy

  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

def connect(self) -> None:

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

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

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

def commit(self) -> None:

def clear_appended_audio(self) -> None:

def close(self) -> None:

def get_session_id(self) -> str:

def get_last_response_id(self) -> str:

​

Callback interface (OmniRealtimeCallback)

def on_open(self) -> None:

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

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

​

Complete example

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.5-livetranslate-flash-realtime",
        # China (Beijing) region. Replace {WorkspaceId} with your Bailian workspace ID. URLs vary by region.
        url="wss://{WorkspaceId}.cn-beijing.maas.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="Tina",
        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.end_session()
        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()