Qwen-LiveTranslate Python SDK - API reference

​

Pré-requisitos

1. Instale o SDK. Verifique se a versão do seu DashScope SDK é 1.25.6 ou superior.
2. Obtenha uma chave de API.

​

Parâmetros da solicitação

• Defina os seguintes parâmetros no construtor da classe OmniRealtimeConversation.
  Clique em visualizar o código de exemplo

  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
  

  Parâmetro	Tipo	Obrigatório	Descrição
  model	str	Sim	Nome do modelo a ser utilizado. Defina como qwen3.5-livetranslate-flash-realtime (recomendado). qwen3-livetranslate-flash-realtime é um modelo legado.
  callback	OmniRealtimeCallback	Sim	Instância de callback para tratar eventos do lado do servidor.
  url	str	Sim	Endpoint do service para tradução em tempo real: 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 Substitua {WorkspaceId} pelo seu workspace ID real.

• Configure os parâmetros abaixo por meio do método update_session da classe OmniRealtimeConversation.
  Clique em visualizar o código de exemplo

  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,
  )
  

  Parâmetro	Tipo	Obrigatório	Descrição
  output_modalities	List[MultiModality]	Não	Formato de saída dos resultados da tradução. Padrão: [MultiModality.TEXT, MultiModality.AUDIO]. Valores válidos: [MultiModality.TEXT]: Gera apenas texto [MultiModality.TEXT, MultiModality.AUDIO]: Gera texto e áudio
  voice	str	Não	Voz utilizada na saída de áudio. Padrão: Qwen3.5-LiveTranslate-Flash-Realtime: Tina Qwen3-LiveTranslate-Flash-Realtime: Cherry Valores válidos: Supported voices.
  input_audio_transcription_model	str	Não	Defina este parâmetro como qwen3-asr-flash-realtime para receber resultados de reconhecimento de fala no idioma de origem.
  translation_params	TranslationParams	Não	Configurações de tradução (idioma alvo e palavras-chave).
  enable_turn_detection	bool	Não	Indica se o VAD (Detecção de Atividade de Voz) deve ser ativado. Valor padrão: True, que habilita o modo VAD, onde o servidor detecta automaticamente o início e o fim da fala e aciona a tradução. Defina como False para alternar para o modo Manual, no qual o cliente envia o áudio manualmente através do método commit. Para descrições detalhadas dos parâmetros, consulte a descrição de turn_detection no documento Client events.

• Especifique os seguintes parâmetros no construtor da classe TranslationParams.
  Clique em visualizar o código de exemplo

  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'
          }
      )
  )
  

  Parâmetro	Tipo	Obrigatório	Descrição
  language	str	Não	Idioma alvo da tradução. Padrão: en. Valores válidos: Supported languages.
  corpus	TranslationParams.Corpus	Não	Mapeamento de palavras-chave para melhorar a precisão da tradução de termos específicos.
  corpus.phrases	dict	Não	Tabela de mapeamento de palavras-chave, onde as chaves são os termos de origem e os valores são as traduções alvo. Exemplo: {'Inteligencia Artificial': 'Artificial Intelligence'}

​

Interfaces principais

​

Classe OmniRealtimeConversation

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:

​

Interface de callback (OmniRealtimeCallback)

def on_open(self) -> None:

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

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

​

Exemplo completo

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",
        # The following is the Singapore region URL. Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        url="wss://{WorkspaceId}.ap-southeast-1.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()