Alibaba Cloud Model Studio:Qwen-LiveTranslate Python SDK - API reference

Utilisez le SDK DashScope pour Python afin d'appeler Qwen-LiveTranslate pour la traduction vocale en temps réel.

Prérequis

1. Installez le SDK. Vérifiez que votre version du SDK DashScope est 1.25.6 ou ultérieure.
2. Obtenez une clé API.

Paramètres de requête

• Définissez les paramètres suivants dans le constructeur de la classe OmniRealtimeConversation.

  Cliquez pour afficher l'exemple de 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',
      # 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
  

  Paramètre	Type	Obligatoire	Description
  model	str	Oui	Nom du modèle à utiliser. Définissez cette valeur sur qwen3.5-livetranslate-flash-realtime (recommandé).
  callback	OmniRealtimeCallback	Oui	Instance de rappel pour gérer les événements côté serveur.
  url	str	Oui	Endpoint du service pour la traduction en temps réel : Chine (Pékin) : wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime Singapour : wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime Remplacez {WorkspaceId} par votre véritable ID d'espace de travail.

• Configurez les paramètres ci-dessous via la méthode update_session de la classe OmniRealtimeConversation.

  Cliquez pour afficher l'exemple de 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,
  )
  

  Paramètre	Type	Obligatoire	Description
  output_modalities	List[MultiModality]	Non	Format de sortie des résultats de traduction. Valeur par défaut : [MultiModality.TEXT, MultiModality.AUDIO]. Valeurs valides : [MultiModality.TEXT] : Sortie texte uniquement [MultiModality.TEXT, MultiModality.AUDIO] : Sortie texte et audio
  voice	str	Non	Voix utilisée pour la sortie audio. Valeur par défaut : Qwen3.5-LiveTranslate-Flash-Realtime : Tina Valeurs valides : Voix prises en charge.
  input_audio_transcription_model	str	Non	Définissez ce paramètre sur qwen3-asr-flash-realtime pour recevoir les résultats de reconnaissance vocale dans la langue source.
  translation_params	TranslationParams	Non	Paramètres de traduction (langue cible et mots-clés).
  enable_turn_detection	bool	Non	Active ou désactive la détection d'activité vocale (VAD). Valeur par défaut : True, ce qui active le mode VAD où le serveur détecte automatiquement le début et la fin de la parole pour déclencher la traduction. Définissez cette valeur sur False pour passer en mode manuel, où le client soumet explicitement l'audio via la méthode commit. Pour une description détaillée des paramètres, consultez la section turn_detection dans le document Événements client.

• Spécifiez les paramètres suivants dans le constructeur de la classe TranslationParams.

  Cliquez pour afficher l'exemple de 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'
          }
      )
  )
  

  Paramètre	Type	Obligatoire	Description
  language	str	Non	Langue cible de la traduction. Valeur par défaut : en. Valeurs valides : Langues prises en charge.
  corpus	TranslationParams.Corpus	Non	Tableau de mots-clés permettant d'améliorer la précision de traduction pour des termes spécifiques.
  corpus.phrases	dict	Non	Dictionnaire de correspondance des mots-clés, où les clés sont les termes sources et les valeurs leurs traductions cibles. Exemple : {'Inteligencia Artificial': 'Artificial Intelligence'}

Interfaces principales

Classe OmniRealtimeConversation

Importation : from dashscope.audio.qwen_omni import 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 rappel (OmniRealtimeCallback)

Recevez les événements et les données du serveur par le biais de rappels.

Héritez de cette classe et implémentez ses méthodes pour traiter les événements provenant du serveur.

Importation : from dashscope.audio.qwen_omni import OmniRealtimeCallback

def on_open(self) -> None:

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

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

Exemple complet

Cet exemple capture l'audio du microphone et le traduit en temps réel.

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