LiveTranslate Python SDK

Use the DashScope SDK for Python to call Qwen-LiveTranslate for real-time speech translation. User guide: For tutorials and complete examples, see Real-time translation.

Prerequisites

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

Request parameters

Set these in the OmniRealtimeConversation constructor:

Click to view sample code

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',
  url='wss://dashscope-intl.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). `qwen3-livetranslate-flash-realtime` is a legacy model.
`callback`	`OmniRealtimeCallback`	Yes	Callback object that handles server events.
`url`	`str`	No	Service endpoint: `wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime`. Defaults to the DashScope endpoint.

Set these with OmniRealtimeConversation.update_session:

Click to view sample code

# 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	Output types. Default: `[MultiModality.TEXT, MultiModality.AUDIO]`. Valid values: `[MultiModality.TEXT]` (text only) or `[MultiModality.TEXT, MultiModality.AUDIO]` (text and audio).
`voice`	`str`	No	Voice for audio output. Default: `Tina` for Qwen3.5-LiveTranslate-Flash-Realtime, or `Cherry` for Qwen3-LiveTranslate-Flash-Realtime. See Supported voices.
`input_audio_transcription_model`	`str`	No	Set to `qwen3-asr-flash-realtime` to get speech recognition results for the source language.
`translation_params`	`TranslationParams`	No	Translation settings.
`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 these in the TranslationParams constructor:

Click to view sample code

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	Target language code. Default: `en`. See Supported languages.
`corpus`	`TranslationParams.Corpus`	No	Hotword settings to improve accuracy for specific terms.
`corpus.phrases`	`dict`	No	Hotword map (key: source term, value: target translation). Example: `{'Inteligencia Artificial': 'Artificial Intelligence'}`

Key interfaces

OmniRealtimeConversation class

Import: from dashscope.audio.qwen_omni import OmniRealtimeConversation

Method signature	Server event (via callback)	Description
`def connect(self) -> None:`	Server event: Session created; Server event: Session config updated	Connects to the server.
`def update_session(self, output_modalities: List[MultiModality], voice: str = None, translation_params: TranslationParams = None, **kwargs) -> None:`	Server event: Session updated	Updates session settings. Call right after connecting. If not called, defaults apply. See the `OmniRealtimeConversation.update_session` parameters.
`def end_session(self, timeout: int = 20) -> None:`	session.finished: The server finishes translation and ends the session	Ends the session. The server finishes any remaining translation before closing.
`def append_audio(self, audio_b64: str) -> None:`	None	Sends Base64-encoded audio to the input buffer. The server auto-detects speech boundaries 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	Stops the task and closes the connection.
`def get_session_id(self) -> str:`	None	Returns the current session ID.
`def get_last_response_id(self) -> str:`	None	Returns the last response ID.

Callback interface (OmniRealtimeCallback)

The server sends events to the client through callbacks. Inherit this class and implement its methods to handle them. Import: from dashscope.audio.qwen_omni import OmniRealtimeCallback

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 event	Called when a server event arrives.
`def on_close(self, close_status_code, close_msg) -> None:`	close_status_code: Status code. close_msg: Log message.	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

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():
  if not os.environ.get("DASHSCOPE_API_KEY"):
    print("Set the DASHSCOPE_API_KEY environment variable.")
    sys.exit(1)

  pya = pyaudio.PyAudio()

  speaker = pya.open(
    format=pyaudio.paInt16,
    channels=1,
    rate=24000,
    output=True,
    frames_per_buffer=2400
  )

  mic = pya.open(
    format=pyaudio.paInt16,
    channels=1,
    rate=16000,
    input=True,
    frames_per_buffer=1600
  )

  callback = Callback(speaker=speaker)

  conversation = OmniRealtimeConversation(
    model="qwen3.5-livetranslate-flash-realtime",
    url="wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime",
    callback=callback
  )

  conversation.connect()

  translation_params = TranslationParams(
    language="en",
    corpus=TranslationParams.Corpus(
      phrases={
        "Source Term 1": "Target Translation 1",
        "Source Term 2": "Target Translation 2"
      }
    )
  )

  conversation.update_session(
    output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
    input_audio_transcription_model="qwen3-asr-flash-realtime",
    voice="Tina",
    translation_params=translation_params,
  )

  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.")

  while True:
    audio_data = mic.read(1600, exception_on_overflow=False)
    conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))


if __name__ == "__main__":
  main()

​Prerequisites

​Request parameters

​Key interfaces

​OmniRealtimeConversation class

​Callback interface (OmniRealtimeCallback)

​Complete example

Prerequisites

Request parameters

Key interfaces

OmniRealtimeConversation class

Callback interface (OmniRealtimeCallback)

Complete example