SDK de Python de Qwen-LiveTranslate-Realtime

​

Requisitos previos

1. Instale el SDK. Asegúrese de que su versión del SDK de DashScope sea 1.25.6 o posterior.
2. Obtener una clave de API.

​

Parámetros de solicitud

• Establezca los siguientes parámetros en el constructor de la clase OmniRealtimeConversation.
  Haga clic para ver código de ejemplo

  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	Obligatorio	Descripción
  model	str	Sí	El nombre del modelo a utilizar. Defina este valor como qwen3.5-livetranslate-flash-realtime (recomendado).
  callback	OmniRealtimeCallback	Sí	Una instancia de devolución de llamada para gestionar eventos del lado del servidor.
  url	str	Sí	El endpoint de servicio para traducción en tiempo real: China (Pekín): wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime Singapur: wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime Reemplace {WorkspaceId} con su ID de espacio de trabajo real.

• Establezca los siguientes parámetros utilizando el método update_session de la clase OmniRealtimeConversation.
  Haga clic para ver código de ejemplo

  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	Obligatorio	Descripción
  output_modalities	List[MultiModality]	No	El formato de salida para los resultados de la traducción. Predeterminado: [MultiModality.TEXT, MultiModality.AUDIO]. Valores válidos: [MultiModality.TEXT]: Genera salida solo de texto [MultiModality.TEXT, MultiModality.AUDIO]: Genera salida de texto y audio
  voice	str	No	La voz para la salida de audio. Predeterminado: Qwen3.5-LiveTranslate-Flash-Realtime: Tina Valores válidos: Voces compatibles.
  input_audio_transcription_model	str	No	Defina este parámetro como qwen3-asr-flash-realtime para recibir resultados de reconocimiento de voz en el idioma de origen.
  translation_params	TranslationParams	No	Configuración de traducción (idioma objetivo y palabras clave).
  enable_turn_detection	bool	No	Indica si se debe habilitar VAD (Detección de Actividad de Voz). Valor predeterminado: True, que habilita el modo VAD donde el servidor detecta automáticamente el inicio y fin del habla y activa la traducción. Configure como False para cambiar al modo Manual, donde el cliente envía audio manualmente mediante el método commit. Para descripciones detalladas de los parámetros, consulte la descripción de turn_detection en el documento de Eventos del cliente.

• Establezca los siguientes parámetros en el constructor de la clase TranslationParams.
  Haga clic para ver código de ejemplo

  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	Obligatorio	Descripción
  language	str	No	El idioma de destino para la traducción. Predeterminado: en. Valores válidos: Idiomas compatibles.
  corpus	TranslationParams.Corpus	No	Un mapeo de palabras clave para mejorar la precisión de la traducción de términos específicos.
  corpus.phrases	dict	No	La tabla de mapeo de palabras clave donde las claves son términos fuente y los valores son traducciones objetivo. Ejemplo: {'Inteligencia Artificial': 'Artificial Intelligence'}

​

Interfaces clave

​

Clase 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:

​

Interfaz 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:

​

dashscope.Client(): recepción de eventos después de un error

​

Ejemplo 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()