SDK do LiveTranslate para Python

SDK Python do LiveTranslate

Use o DashScope SDK para Python para chamar Qwen-LiveTranslate para tradução de fala em tempo real.

Guia do usuário: Para tutoriais e exemplos completos, consulte Tradução em tempo real.

Pré-requisitos

1. Instale o SDK. Certifique-se de que a versão do SDK do DashScope seja 1.25.6 ou posterior.
2. Obtenha uma chave API.

Parâmetros de solicitação

Defina-os no OmniRealtimeConversation construtor:

Clique para ver o código de amostra.

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://maas.qwencloudapi.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

Defina-os com OmniRealtimeConversation.update_session:

Clique para ver o código de amostra.

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

Defina-os no TranslationParams construtor:

Clique para ver o código de amostra.

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

Interfaces principais

Classe OmniRealtimeConversation

Importar: from dashscope.audio.qwen_omni import OmniRealtimeConversation

Interface de retorno de chamada (OmniRealtimeCallback)

O servidor envia eventos ao cliente por meio de retornos de chamada. Herde esta classe e implemente seus métodos para lidar com eles.

Importar: from dashscope.audio.qwen_omni import OmniRealtimeCallback

Exemplo completo

Este exemplo grava o áudio do microfone e o traduz em tempo real.

Exemplo de código para tradução em tempo real a partir do microfone

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://maas.qwencloudapi.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()