Alibaba Cloud Model Studio:Terjemahan audio dan video real-time - Qwen

qwen3.5-livetranslate-flash-realtime adalah model terjemahan real-time yang ditingkatkan dengan visi dan mendukung 60 bahasa (29 dengan output audio dan teks, 31 hanya teks). Model ini memproses input audio dan gambar dari aliran video atau file lokal, memanfaatkan konteks visual untuk meningkatkan akurasi, serta menghasilkan teks dan audio terjemahan secara real-time.

Coba demo online dengan penerapan satu klik menggunakan Function Compute.

Fitur

• Dukungan multi-bahasa: Menerjemahkan antara 60 bahasa — 29 dengan output audio dan teks, 31 dengan output hanya teks — termasuk Tiongkok, Inggris, Prancis, Jerman, Rusia, Jepang, Korea, Spanyol, Portugis, dan Arab.
• Peningkatan visual: Menganalisis petunjuk visual, seperti gerakan bibir, isyarat tubuh, dan teks di layar, untuk meningkatkan akurasi terjemahan, terutama di lingkungan berisik atau untuk kata-kata ambigu.
• Latensi 2,8 detik: Memberikan interpretasi simultan dengan latensi serendah 2,8 detik.
• Interpretasi simultan tanpa kehilangan kualitas: Memprediksi unit semantik untuk menyelesaikan perbedaan urutan kata lintas bahasa, mencapai kualitas yang sebanding dengan terjemahan offline.
• Suara alami: Secara otomatis mencocokkan intonasi dan emosi dari audio sumber.
• Konfigurasi hotword: Hotword yang dapat dikonfigurasi meningkatkan akurasi terjemahan untuk istilah tertentu.
• Kloning suara: Mengkloning suara pembicara untuk output terjemahan. Mendukung kloning real-time di sisi server dan profil suara yang telah dikloning sebelumnya.
• Lewati output bahasa yang sama: Ketika bahasa sumber dan target sama, model dapat melewatkan output teks, output audio, atau keduanya. Fitur ini hanya berlaku ketika bahasa target adalah Tiongkok (zh) atau Inggris (en).

Prosedur

1. Konfigurasikan koneksi

Model terhubung melalui WebSocket dengan parameter berikut:

Selain WebSocket, model ini juga mendukung protokol AOQ dan WebRTC. Untuk integrasi di sisi klien yang memprioritaskan latensi stabil, ketahanan pada jaringan lemah, serta penekanan kebisingan dan pembatalan gema full-duplex bawaan, disarankan menggunakan AOQ. Untuk perbandingan protokol, lihat Ikhtisar API Realtime.

Kode contoh koneksi (Python):

# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"Connected to server: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("Error:", error)

ws = websocket.WebSocketApp(
    API_URL,
    header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)

ws.run_forever()

2. Konfigurasikan bahasa, modalitas, dan suara

Kirim event klien session.update dengan parameter berikut:

• Bahasa

  • Bahasa sumber: Konfigurasikan menggunakan parameter session.input_audio_transcription.language.

    Jika tidak ditentukan, model akan secara otomatis mendeteksi bahasa sumber.

  • Bahasa target: Konfigurasikan menggunakan parameter session.translation.language.

    Nilai default-nya adalah en (Inggris).

  Lihat Bahasa yang didukung.

• Output hasil pengenalan bahasa sumber

  Atur session.input_audio_transcription.model ke qwen3-asr-flash-realtime. Server kemudian akan mengembalikan hasil terjemahan dan hasil pengenalan ucapan (teks asli) untuk audio input.

  Server mengembalikan event-event berikut:

  • conversation.item.input_audio_transcription.text: Mengalirkan hasil pengenalan.
  • conversation.item.input_audio_transcription.completed: Mengembalikan hasil akhir setelah pengenalan selesai.
  • conversation.item.input_audio_transcription.failed: Mengembalikan informasi error saat pengenalan gagal.
• Modalitas output

  Atur parameter session.modalities ke ["text"] (hanya teks) atau ["text","audio"] (teks dan audio).

• Voice Activity Detection (VAD) dan mode Manual

  Konfigurasikan cara mendeteksi batas ucapan menggunakan parameter session.turn_detection:

  • Mode VAD (default): Atur turn_detection ke objek konfigurasi. Server secara otomatis mendeteksi batas ucapan dan memicu terjemahan, cocok untuk skenario di mana klien terus-menerus mengirim aliran audio.
  • Mode Manual: Atur turn_detection ke null. Klien menentukan batas ucapan dan mengirim event input_audio_buffer.commit untuk mengirimkan audio setelah setiap tuturan, cocok untuk skenario push-to-talk.

  Untuk langkah interaksi lengkap dalam kedua mode, lihat 3. Input audio dan gambar.

• Suara

  Konfigurasikan menggunakan parameter session.voice. Lihat Suara yang didukung.

• Hotword

  Konfigurasikan hotword menggunakan parameter session.translation.corpus.phrases. Hotword adalah pasangan kunci-nilai yang memetakan istilah sumber ke terjemahan target, meningkatkan akurasi untuk istilah tertentu.

  Contoh: Petakan "artificial intelligence" ke "Artificial Intelligence".

• Kloning suara

  Konfigurasikan menggunakan parameter session.enable_voice_clone, session.voice_clone_options.frequency, dan session.voice. Mendukung tiga mode: profil suara yang telah dikloning sebelumnya (frequency: never), kloning sekali di sisi server saat sesi dimulai (once), atau kloning real-time sebelum setiap respons (always). Lihat Kloning suara.

3. Input audio dan gambar

Kirim data audio dan gambar yang di-encode Base64 menggunakan event input_audio_buffer.append dan input_image_buffer.append. Input audio wajib; input gambar opsional.

Gambar dapat berasal dari file lokal atau diambil secara real-time dari aliran video.

Cara model menentukan bahwa sebuah tuturan telah selesai bergantung pada mode VAD atau mode Manual yang dikonfigurasi melalui parameter turn_detection:

• Mode VAD (default): Klien terus-menerus mengirim event input_audio_buffer.append. Saat server mendeteksi awal/akhir ucapan, server mengembalikan event input_audio_buffer.speech_started dan input_audio_buffer.speech_stopped masing-masing, secara otomatis melakukan commit buffer audio, dan memicu terjemahan. Respons terjemahan dihasilkan secara sinkron dengan aliran audio dan biasanya dimulai selama input audio, tanpa menunggu ucapan berakhir.
• Mode Manual: Atur session.turn_detection ke null. Setelah klien selesai mengirim tuturan lengkap, klien mengirim event input_audio_buffer.commit untuk melakukan commit buffer audio. Setelah server mengembalikan event input_audio_buffer.committed sebagai konfirmasi, server secara otomatis mulai menghasilkan respons terjemahan; klien tidak perlu mengirim event lain untuk memicu respons. Untuk menghapus audio yang belum di-commit sebelum melakukan commit, kirim event input_audio_buffer.clear.

4. Terima respons model

Respons terjemahan dihasilkan secara sinkron dengan aliran audio dan biasanya tidak perlu menunggu ucapan berakhir (lihat deskripsi mode VAD/Manual pada bagian sebelumnya). Format respons bergantung pada modalitas output.

• Output hanya teks

  Server mengalirkan teks terjemahan inkremental (termasuk teks yang dikonfirmasi dan teks prediksi sementara) melalui event response.text.text; setelah selesai, teks terjemahan lengkap dikembalikan dalam event response.text.done.

• Output teks dan audio
  • Teks

    Server mengalirkan teks terjemahan inkremental melalui event response.audio_transcript.text; setelah selesai, teks terjemahan lengkap dikembalikan dalam event response.audio_transcript.done.

  • Audio

    Server mengembalikan data audio inkremental yang di-encode Base64 dalam event response.audio.delta.

5. Akhiri sesi

Setelah mengirim semua audio, kirim event Client events, lalu tunggu hingga server mengembalikan event session.finished sebelum menutup koneksi WebSocket.

Jika Anda menutup WebSocket tanpa mengirim session.finish, VAD server tidak dapat mendeteksi akhir segmen ucapan terakhir. Hal ini menyebabkan hasil terjemahan untuk segmen tersebut hilang sepenuhnya, dan koneksi mungkin hang tanpa batas waktu. Selalu kirim event ini sebelum memutus koneksi.

Model yang didukung

Model yang direkomendasikan

Model lawas

Model berikut masih tersedia tetapi tidak lagi menjadi pilihan yang direkomendasikan. Untuk kasus penggunaan baru, gunakan model yang lebih baru di atas untuk kualitas terjemahan dan efisiensi biaya yang lebih baik.

Mulai

1. Siapkan lingkungan

   Memerlukan Python 3.10 atau lebih baru.

   Pertama, instal pyaudio.

   macOS

   brew install portaudio && pip install pyaudio
   

   Debian/Ubuntu

   sudo apt-get install python3-pyaudio
   
   or
   
   pip install pyaudio
   

   CentOS

   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
   

   Windows

   pip install pyaudio
   

   Kemudian instal dependensi WebSocket:

pip install websocket-client==1.8.0 websockets

2. Buat klien

   Buat file bernama livetranslate_client.py dengan kode berikut:

   Kode klien - livetranslate_client.py

   import os
   import time
   import base64
   import asyncio
   import json
   import websockets
   import pyaudio
   import queue
   import threading
   import traceback
   
   class LiveTranslateClient:
       def __init__(self, api_key: str, target_language: str = "en", *, audio_enabled: bool = True):
           if not api_key:
               raise ValueError("API key cannot be empty.")
   
           self.api_key = api_key
           self.target_language = target_language
           self.audio_enabled = audio_enabled
           self.ws = None
           self.api_url = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"
   
           # Konfigurasi input audio (dari mikrofon)
           self.input_rate = 16000
           self.input_chunk = 1600
           self.input_format = pyaudio.paInt16
           self.input_channels = 1
   
           # Konfigurasi output audio (untuk pemutaran)
           self.output_rate = 24000
           self.output_chunk = 2400
           self.output_format = pyaudio.paInt16
           self.output_channels = 1
   
           # Manajemen status
           self.is_connected = False
           self.audio_player_thread = None
           self.audio_playback_queue = queue.Queue()
           self.pyaudio_instance = pyaudio.PyAudio()
           self.session_finished_event = asyncio.Event()
   
       async def connect(self):
           """Membuat koneksi WebSocket ke layanan terjemahan."""
           headers = {"Authorization": f"Bearer {self.api_key}"}
           try:
               self.ws = await websockets.connect(self.api_url, additional_headers=headers)
               self.is_connected = True
               print(f"Successfully connected to the server: {self.api_url}")
               await self.configure_session()
           except Exception as e:
               print(f"Connection failed: {e}")
               self.is_connected = False
               raise
   
       async def configure_session(self):
           """Mengonfigurasi sesi terjemahan, mengatur bahasa target, suara, dll."""
           config = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "session.update",
               "session": {
                   # 'modalities' mengontrol jenis output.
                   # ["text", "audio"]: Mengembalikan teks terjemahan dan audio sintesis (direkomendasikan).
                   # ["text"]: Hanya mengembalikan teks terjemahan.
                   "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                   "input_audio_format": "pcm",
                   "output_audio_format": "pcm",
                   # 'input_audio_transcription' mengonfigurasi pengenalan bahasa sumber.
                   # Atur 'model' ke 'qwen3-asr-flash-realtime' untuk juga mengeluarkan hasil pengenalan bahasa sumber.
                   # "input_audio_transcription": {
                   #     "model": "qwen3-asr-flash-realtime",
                   #     "language": "zh"  # bahasa sumber, default 'en'
                   # },
                   "translation": {
                       "language": self.target_language,
                       # 'corpus' mengonfigurasi hotword untuk meningkatkan akurasi terjemahan istilah tertentu.
                       # "corpus": {
                       #     "phrases": {
                       #         "Artificial Intelligence": "Artificial Intelligence",
                       #         "Machine Learning": "Machine Learning"
                       #     }
                       # }
                   }
               }
           }
           print(f"Sending session configuration: {json.dumps(config, indent=2, ensure_ascii=False)}")
           await self.ws.send(json.dumps(config))
   
       async def send_audio_chunk(self, audio_data: bytes):
           """Meng-encode dan mengirim potongan audio ke server."""
           if not self.is_connected:
               return
   
           event = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "input_audio_buffer.append",
               "audio": base64.b64encode(audio_data).decode()
           }
           await self.ws.send(json.dumps(event))
   
       async def send_image_frame(self, image_bytes: bytes, *, event_id: str | None = None):
           # Mengirim frame gambar ke server.
           if not self.is_connected:
               return
   
           if not image_bytes:
               raise ValueError("image_bytes cannot be empty.")
   
           # Encode ke Base64
           image_b64 = base64.b64encode(image_bytes).decode()
   
           event = {
               "event_id": event_id or f"event_{int(time.time() * 1000)}",
               "type": "input_image_buffer.append",
               "image": image_b64,
           }
   
           await self.ws.send(json.dumps(event))
   
       def _audio_player_task(self):
           stream = self.pyaudio_instance.open(
               format=self.output_format,
               channels=self.output_channels,
               rate=self.output_rate,
               output=True,
               frames_per_buffer=self.output_chunk,
           )
           try:
               while self.is_connected or not self.audio_playback_queue.empty():
                   try:
                       audio_chunk = self.audio_playback_queue.get(timeout=0.1)
                       if audio_chunk is None: # Sinyal terminasi
                           break
                       stream.write(audio_chunk)
                       self.audio_playback_queue.task_done()
                   except queue.Empty:
                       continue
           finally:
               stream.stop_stream()
               stream.close()
   
       def start_audio_player(self):
           """Memulai thread pemutar audio (hanya saat output audio diaktifkan)."""
           if not self.audio_enabled:
               return
           if self.audio_player_thread is None or not self.audio_player_thread.is_alive():
               self.audio_player_thread = threading.Thread(target=self._audio_player_task, daemon=True)
               self.audio_player_thread.start()
   
       async def handle_server_messages(self, on_text_received):
           """Menangani pesan masuk dari server dalam loop."""
           try:
               async for message in self.ws:
                   event = json.loads(message)
                   event_type = event.get("type")
                   if event_type == "response.audio.delta" and self.audio_enabled:
                       audio_b64 = event.get("delta", "")
                       if audio_b64:
                           audio_data = base64.b64decode(audio_b64)
                           self.audio_playback_queue.put(audio_data)
   
                   elif event_type == "response.done":
                       print("\n[INFO] Response round complete.")
                       usage = event.get("response", {}).get("usage", {})
                       if usage:
                           print(f"[INFO] token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                   elif event_type == "session.finished":
                       print("[INFO] Session finished.")
                       self.session_finished_event.set()
                   # Proses hasil pengenalan bahasa sumber (memerlukan pengaktifan input_audio_transcription.model)
                   # elif event_type == "conversation.item.input_audio_transcription.text":
                   #     stash = event.get("stash", "")  # Teks pengenalan yang tertunda
                   #     print(f"[Recognizing] {stash}")
                   # elif event_type == "conversation.item.input_audio_transcription.completed":
                   #     transcript = event.get("transcript", "")  # Hasil pengenalan lengkap
                   #     print(f"[Source language] {transcript}")
                   elif event_type == "response.text.text":
                       # Teks terjemahan streaming dalam modalitas hanya teks
                       text = event.get("text", "")
                       stash = event.get("stash", "")
                       print(f"\r[Translating] {text}{stash}", end="", flush=True)
                   elif event_type == "response.audio_transcript.done":
                       print("\n[INFO] Translation complete.")
                       text = event.get("transcript", "")
                       if text:
                           print(f"[INFO] Translated text: {text}")
                   elif event_type == "response.text.done":
                       print("\n[INFO] Translation complete.")
                       text = event.get("text", "")
                       if text:
                           print(f"[INFO] Translated text: {text}")
   
           except websockets.exceptions.ConnectionClosed as e:
               print(f"[WARNING] Connection closed: {e}")
               self.is_connected = False
           except Exception as e:
               print(f"[ERROR] An unexpected error occurred while processing messages: {e}")
               traceback.print_exc()
               self.is_connected = False
   
       async def start_microphone_streaming(self):
           """Mengambil audio dari mikrofon dan mengalirkannya ke server."""
           stream = self.pyaudio_instance.open(
               format=self.input_format,
               channels=self.input_channels,
               rate=self.input_rate,
               input=True,
               frames_per_buffer=self.input_chunk
           )
           print("Microphone is on. Start speaking...")
           try:
               while self.is_connected:
                   audio_chunk = await asyncio.get_event_loop().run_in_executor(
                       None, stream.read, self.input_chunk
                   )
                   await self.send_audio_chunk(audio_chunk)
           finally:
               stream.stop_stream()
               stream.close()
   
       async def close(self):
           """Menutup koneksi dan melepaskan sumber daya secara elegan."""
           # Kirim session.finish untuk memastikan server menyelesaikan terjemahan segmen ucapan terakhir
           if self.is_connected and self.ws:
               finish_event = {
                   "event_id": f"event_{int(time.time() * 1000)}",
                   "type": "session.finish",
               }
               await self.ws.send(json.dumps(finish_event))
               print("Sent session.finish, waiting for server to finish processing...")
               try:
                   await asyncio.wait_for(self.session_finished_event.wait(), timeout=15)
                   print("Server processing complete.")
               except asyncio.TimeoutError:
                   print("Timed out waiting for session.finished.")
           self.is_connected = False
           if self.ws:
               await self.ws.close()
               print("WebSocket connection closed.")
   
           if self.audio_player_thread:
               self.audio_playback_queue.put(None) # Kirim sinyal terminasi
               self.audio_player_thread.join(timeout=1)
               print("Audio player thread stopped.")
   
           self.pyaudio_instance.terminate()
           print("PyAudio instance released.")
   

3. Berinteraksi dengan model

   Di direktori yang sama, buat file bernama main.py dengan kode berikut:

   main.py

   import os
   import asyncio
   from livetranslate_client import LiveTranslateClient
   
   def print_banner():
       print("=" * 60)
       print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime")
       print("=" * 60 + "\n")
   
   def get_user_config():
       """Mendapatkan konfigurasi pengguna."""
       print("Select a mode:")
       print("1. Voice + Text [Default] | 2. Text Only")
       mode_choice = input("Enter your choice (press Enter for Voice + Text): ").strip()
       audio_enabled = (mode_choice != "2")
   
       if audio_enabled:
           lang_map = {
               "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt",
               "7": "es", "8": "it", "9": "ko", "10": "ja", "11": "yue"
           }
           print("Select the target language (Voice + Text mode):")
           print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Korean | 10. Japanese | 11. Cantonese")
       else:
           lang_map = {
               "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt", "7": "es", "8": "it",
               "9": "id", "10": "ko", "11": "ja", "12": "vi", "13": "th", "14": "ar",
               "15": "yue", "16": "hi", "17": "el", "18": "tr"
           }
           print("Select the target language (Text Only mode):")
           print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Indonesian | 10. Korean | 11. Japanese | 12. Vietnamese | 13. Thai | 14. Arabic | 15. Cantonese | 16. Hindi | 17. Greek | 18. Turkish")
   
       choice = input("Enter your choice (defaults to the first option): ").strip()
       target_language = lang_map.get(choice, next(iter(lang_map.values())))
   
       return target_language, audio_enabled
   
   async def main():
       """Titik masuk program utama."""
       print_banner()
   
       api_key = os.environ.get("DASHSCOPE_API_KEY")
       if not api_key:
           print("[ERROR] Please set the DASHSCOPE_API_KEY environment variable.")
           print("  For example: export DASHSCOPE_API_KEY='your_api_key_here'")
           return
   
       target_language, audio_enabled = get_user_config()
       print("\nConfiguration complete:")
       print(f"  - Target language: {target_language}")
       if not audio_enabled:
           print("  - Output mode: Text Only")
   
       client = LiveTranslateClient(api_key=api_key, target_language=target_language, audio_enabled=audio_enabled)
   
       # Definisikan fungsi callback.
       def on_translation_text(text):
           print(text, end="", flush=True)
   
       try:
           print("Connecting to the translation service...")
           await client.connect()
   
           # Mulai pemutaran audio berdasarkan mode.
           client.start_audio_player()
   
           print("\n" + "-" * 60)
           print("Connection successful! Speak into the microphone.")
           print("The program will translate your speech in real time and play the translated audio. Press Ctrl+C to exit.")
           print("-" * 60 + "\n")
   
           # Jalankan penanganan pesan dan perekaman mikrofon secara konkuren.
           message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
           tasks = [message_handler]
           # Ambil audio dari mikrofon untuk terjemahan, terlepas dari apakah output audio diaktifkan.
           microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
           tasks.append(microphone_streamer)
   
           await asyncio.gather(*tasks)
   
       except KeyboardInterrupt:
           print("\n\nUser interrupted. Exiting...")
       except Exception as e:
           print(f"\nA critical error occurred: {e}")
       finally:
           print("\nCleaning up resources...")
           await client.close()
           print("Program exited.")
   
   if __name__ == "__main__":
       asyncio.run(main())
   

   Jalankan main.py dan bicara ke mikrofon Anda. Model akan menerjemahkan ucapan Anda dan menghasilkan audio serta teks secara real-time.

Kloning suara

Model mengkloning suara pembicara dari audio input dan menggunakannya untuk output terjemahan. Gunakan profil suara yang telah dikloning sebelumnya atau biarkan server melakukan kloning secara real-time. Berguna untuk interpretasi konferensi, siaran langsung, dan dubbing video.

Atur parameter berikut dalam session.update untuk mengaktifkan kloning suara:

• session.enable_voice_clone: Atur ke true untuk mengaktifkan kloning suara.

• session.voice_clone_options.frequency: Mengontrol kapan kloning suara terjadi. Nilai yang diterima:

  • never: Tidak melakukan kloning di server. Menggunakan profil suara yang telah dikloning sebelumnya. Atur session.voice ke ID suara kloning kustom Anda.
  • once: Mengkloning suara dari audio input sekali saat sesi dimulai, lalu menggunakannya kembali untuk semua output berikutnya. Paling cocok untuk skenario pembicara tunggal. Atur session.voice ke default.
  • always: Mengkloning suara sebelum setiap respons, secara dinamis menyesuaikan dengan perubahan pembicara. Paling cocok untuk percakapan multi-pembicara. Atur session.voice ke default.

• session.voice: Menentukan suara output. Nilainya bergantung pada pengaturan frequency:

  • Atur ke default: Gunakan dengan frequency diatur ke once atau always. Server mengkloning suara pembicara dari audio input. Suara default digunakan hingga proses kloning selesai.
  • Atur ke ID suara kloning kustom (misalnya, qwen-translate-vc-xxx-yyy-zzz): Gunakan dengan frequency diatur ke never. Anda harus menyiapkan suara terlebih dahulu menggunakan Voice Cloning API dengan targetModel diatur ke qwen3.5-livetranslate-flash-realtime.

Saat frequency diatur ke once atau always, parameter voice harus diatur ke default. Nilai lain akan menyebabkan server mengembalikan error.

Contoh konfigurasi kloning suara

Profil suara yang telah dikloning sebelumnya (kualitas konsisten; direkomendasikan saat identitas suara yang stabil diperlukan):

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "qwen-translate-vc-xxx-yyy-zzz",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "never"
        }
    }
}

Kloning di sisi server, sekali per sesi (paling cocok untuk skenario pembicara tunggal):

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "once"
        }
    }
}

Kloning di sisi server, setiap respons (paling cocok untuk percakapan multi-pembicara):

{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "always"
        }
    }
}

Perbaiki terjemahan dengan gambar

Input gambar membantu menghilangkan ambiguitas homonim dan mengenali kata benda proper yang tidak umum selama terjemahan. Kirim tidak lebih dari 2 gambar per detik.

Unduh contoh gambar berikut: medical mask.png, masquerade mask.png

Unduh kode di bawah ini ke direktori yang sama dengan livetranslate_client.py dan jalankan. Ucapkan "What is mask?" ke mikrofon Anda. Model menggunakan gambar untuk menghilangkan ambiguitas: medical mask.png menghasilkan "What is a medical mask?" dan masquerade mask.png menghasilkan "What is a masquerade mask?".

import os
import time
import json
import asyncio
import contextlib
import functools

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "medical mask.png"
# IMAGE_PATH = "masquerade mask.png"

def print_banner():
    print("=" * 60)
    print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime — single-turn interaction example (mask)")
    print("=" * 60 + "\n")

async def stream_microphone_once(client: LiveTranslateClient, image_bytes: bytes):
    pa = client.pyaudio_instance
    stream = pa.open(
        format=client.input_format,
        channels=client.input_channels,
        rate=client.input_rate,
        input=True,
        frames_per_buffer=client.input_chunk,
    )
    print(f"[INFO] Recording started. Please speak...")
    loop = asyncio.get_event_loop()
    last_img_time = 0.0
    frame_interval = 0.5  # 2 fps
    try:
        while client.is_connected:
            data = await loop.run_in_executor(None, stream.read, client.input_chunk)
            await client.send_audio_chunk(data)

            # Tambahkan frame gambar setiap 0,5 detik
            now = time.time()
            if now - last_img_time >= frame_interval:
                await client.send_image_frame(image_bytes)
                last_img_time = now
    finally:
        stream.stop_stream()
        stream.close()

async def main():
    print_banner()
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("[ERROR] Please set the DASHSCOPE_API_KEY environment variable.")
        return

    client = LiveTranslateClient(api_key=api_key, target_language="zh", audio_enabled=True)

    def on_text(text: str):
        print(text, end="", flush=True)

    try:
        await client.connect()
        client.start_audio_player()
        message_task = asyncio.create_task(client.handle_server_messages(on_text))
        with open(IMAGE_PATH, "rb") as f:
            img_bytes = f.read()
        await stream_microphone_once(client, img_bytes)
        await asyncio.sleep(15)
    finally:
        await client.close()
        if not message_task.done():
            message_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await message_task

if __name__ == "__main__":
    asyncio.run(main())

Penerapan satu klik Function Compute

Untuk menerapkan aplikasi:

1. Buka Templat Function Compute, masukkan Kunci API Anda, lalu klik Create and Deploy Default Environment untuk menguji aplikasi.

2. Tunggu sekitar satu menit. Di Environment Details > Environment Context, ambil endpoint, ubah protokol dari http ke https (misalnya, https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/), lalu buka URL tersebut di browser untuk berinteraksi dengan model.

   PentingEndpoint ini menggunakan sertifikat tanda tangan sendiri dan hanya untuk pengujian sementara. Browser Anda akan menampilkan peringatan keamanan saat kunjungan pertama. Ini adalah perilaku yang diharapkan. Jangan gunakan endpoint ini di lingkungan produksi. Untuk melanjutkan, ikuti instruksi di layar (misalnya, klik Advanced → Proceed to (unsafe)).

Jika Anda diminta mengonfigurasi izin Resource Access Management, ikuti instruksi di layar.

Untuk melihat kode sumber proyek, buka Resource Information > Function Resources.

Baik Function Compute maupun Model Studio menyediakan kuota gratis untuk pengguna baru, cukup untuk debugging dasar. Setelah kuota gratis habis, penagihan bayar sesuai penggunaan berlaku.

Alur interaksi

Terjemahan menggunakan model WebSocket berbasis event. Cara menentukan batas ucapan bergantung pada mode VAD atau mode Manual (lihat 3. Input audio dan gambar). Tabel di bawah ini berdasarkan mode VAD (default) dan memberi anotasi event server yang berbeda dalam mode Manual.

Setelah mengirim semua audio, kirim event session.finish dan tunggu session.finished sebelum menutup WebSocket. Jika Anda menutup koneksi tanpa mengirim session.finish, server tidak dapat mengetahui bahwa input audio telah berakhir, dan hasil pengenalan serta terjemahan untuk segmen ucapan terakhir akan hilang.

API

• Qwen-Livetranslate-Realtime.
• AOQ client SDK
• Ikhtisar API Realtime (deskripsi protokol WebRTC)

Penagihan

• Audio: 7 token per detik audio input; 12,5 token per detik audio output.
• Gambar: Setiap 32×32 piksel mengonsumsi 0,5 token.
• Teks: Saat pengenalan ucapan bahasa sumber diaktifkan, layanan mengembalikan transkrip audio input selain terjemahan. Transkrip ini ditagih sebagai token teks output.

• Audio: Setiap detik audio input atau output mengonsumsi 12,5 token.
• Gambar: Setiap 28×28 piksel mengonsumsi 0,5 token.
• Teks: Saat pengenalan ucapan bahasa sumber diaktifkan, layanan mengembalikan transkrip audio input selain terjemahan. Transkrip ini ditagih sebagai token teks output.

Harga: Daftar model.

Batas laju

Untuk informasi tentang batas laju model, lihat Pembatasan laju.

Bahasa yang didukung

Gunakan kode bahasa berikut untuk menentukan bahasa sumber dan target.

Beberapa bahasa target hanya mendukung teks. Model lawas qwen3-livetranslate-flash-realtime hanya mendukung 18 bahasa berikut: en, zh, ru, fr, de, pt, es, it, id, ko, ja, vi, th, ar, yue, hi, el, tr.

Suara yang didukung

Untuk daftar suara yang didukung dan nilai parameter voice yang sesuai, lihat Daftar suara.