Referensi API SDK Python Qwen-LiveTranslate

​

Prasyarat

1. Instal SDK. Pastikan versi DashScope SDK Anda adalah 1.25.6 atau lebih baru.
2. Dapatkan Kunci API.

​

Parameter permintaan

• Tetapkan parameter berikut dalam konstruktor kelas OmniRealtimeConversation.
  Klik untuk melihat contoh kode

  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):
          # Proses data audio inkremental.
          audio_b64 = response.get('delta', '')
          # Dekode data audio untuk pemutaran atau penyimpanan.
  
  conversation = OmniRealtimeConversation(
      model='qwen3.5-livetranslate-flash-realtime',
      # Wilayah China (Beijing). Ganti {WorkspaceId} dengan ID ruang kerja Bailian Anda. URL bervariasi berdasarkan wilayah.
      url='wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime',
      callback=MyCallback(conversation=None)  # Sementara berikan None. Nilai ini akan disuntikkan nanti.
  )
  # Suntikkan self ke dalam callback.
  conversation.callback.conversation = conversation
  

  Parameter	Tipe	Wajib	Deskripsi
  model	str	Ya	Nama model yang digunakan. Tetapkan nilai ini ke qwen3.5-livetranslate-flash-realtime (direkomendasikan). qwen3-livetranslate-flash-realtime adalah model lawas.
  callback	OmniRealtimeCallback	Ya	Instans callback untuk menangani event sisi server.
  url	str	Ya	Titik akhir layanan untuk terjemahan real-time: China (Beijing): wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime Singapura: wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime Ganti {WorkspaceId} dengan ID ruang kerja Anda yang sebenarnya.

• Tetapkan parameter berikut menggunakan metode update_session dari kelas OmniRealtimeConversation.
  Klik untuk melihat contoh kode

  Copy

  # Tetapkan parameter terjemahan
  translation_params = TranslationParams(
      language='en',  # Bahasa target
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )
  
  # Perbarui konfigurasi sesi
  conversation.update_session(
      output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
      voice='Tina',
      translation_params=translation_params,
  )
  

  Parameter	Tipe	Wajib	Deskripsi
  output_modalities	List[MultiModality]	Tidak	Format output untuk hasil terjemahan. Default: [MultiModality.TEXT, MultiModality.AUDIO]. Nilai yang valid: [MultiModality.TEXT]: Hanya menghasilkan teks [MultiModality.TEXT, MultiModality.AUDIO]: Menghasilkan teks dan audio
  voice	str	Tidak	Suara untuk output audio. Default: Qwen3.5-LiveTranslate-Flash-Realtime: Tina Qwen3-LiveTranslate-Flash-Realtime: Cherry Nilai yang valid: Suara yang didukung.
  input_audio_transcription_model	str	Tidak	Tetapkan parameter ini ke qwen3-asr-flash-realtime untuk menerima hasil pengenalan ucapan dalam bahasa sumber.
  translation_params	TranslationParams	Tidak	Pengaturan terjemahan (bahasa target dan hotword).
  enable_turn_detection	bool	Tidak	Apakah akan mengaktifkan VAD (Voice Activity Detection). Nilai default: True, yang mengaktifkan mode VAD di mana server secara otomatis mendeteksi awal/akhir ucapan dan memicu terjemahan. Tetapkan ke False untuk beralih ke mode Manual, di mana klien secara manual mengirimkan audio melalui metode commit. Untuk deskripsi parameter lengkap, lihat penjelasan turn_detection dalam dokumen Client events.

• Tetapkan parameter berikut dalam konstruktor kelas TranslationParams.
  Klik untuk melihat contoh kode

  Copy

  translation_params = TranslationParams(
      language='en',  # Kode bahasa target
      corpus=TranslationParams.Corpus(
          phrases={
              'Inteligencia Artificial': 'Artificial Intelligence',  # Frasa sumber: Terjemahan target
              'Aprendizaje Automático': 'Machine Learning'
          }
      )
  )
  

  Parameter	Tipe	Wajib	Deskripsi
  language	str	Tidak	Bahasa target untuk terjemahan. Default: en. Nilai yang valid: Bahasa yang didukung.
  corpus	TranslationParams.Corpus	Tidak	Pemetaan hotword untuk meningkatkan akurasi terjemahan istilah tertentu.
  corpus.phrases	dict	Tidak	Tabel pemetaan hotword di mana kunci adalah istilah sumber dan nilai adalah terjemahan target. Contoh: {'Inteligencia Artificial': 'Artificial Intelligence'}

​

Antarmuka utama

​

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

​

Antarmuka callback (OmniRealtimeCallback)

def on_open(self) -> None:

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

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

​

Contoh lengkap

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():
    # Periksa keberadaan Kunci API.
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("Setel variabel lingkungan DASHSCOPE_API_KEY.")
        sys.exit(1)

    # Inisialisasi PyAudio.
    pya = pyaudio.PyAudio()

    # Inisialisasi speaker untuk memutar ulang audio terjemahan.
    speaker = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True,
        frames_per_buffer=2400
    )

    # Inisialisasi mikrofon untuk menangkap input ucapan.
    mic = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        frames_per_buffer=1600
    )

    # Buat instans callback.
    callback = Callback(speaker=speaker)

    # Buat sesi real-time.
    conversation = OmniRealtimeConversation(
        model="qwen3.5-livetranslate-flash-realtime",
        # Berikut adalah URL wilayah Singapura. Ganti {WorkspaceId} dengan ID ruang kerja Anda yang sebenarnya. URL bervariasi berdasarkan wilayah.
        url="wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime",
        callback=callback
    )

    # Terhubung ke server.
    conversation.connect()

    # Konfigurasikan parameter terjemahan.
    translation_params = TranslationParams(
        language="en",  # Bahasa target untuk terjemahan: Inggris
        corpus=TranslationParams.Corpus(
            phrases={
                "Source Term 1": "Target Translation 1",
                "Source Term 2": "Target Translation 2"
            }
        )
    )

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

    # Daftarkan penangan sinyal keluar.
    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.")

    # Secara terus-menerus tangkap dan kirim audio mikrofon.
    while True:
        audio_data = mic.read(1600, exception_on_overflow=False)
        conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))

if __name__ == "__main__":
    main()