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

Cara menggunakan

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

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-livetranslate-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"Terhubung ke server: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("Event diterima:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("Kesalahan:", error)

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

ws.run_forever()

Model yang didukung

qwen3-livetranslate-flash-realtime adalah model terjemahan audio dan video real-time multibahasa. Model ini dapat mengenali 18 bahasa dan menerjemahkannya menjadi audio dalam 10 bahasa secara real-time.

Fitur utama:

• Dukungan multibahasa: Mendukung 18 bahasa, seperti Tiongkok, Inggris, Prancis, Jerman, Rusia, Jepang, dan Korea, serta 6 dialek Tiongkok, termasuk Mandarin, Kanton, dan Sichuan.

• Peningkatan visi: Menggunakan konten visual untuk meningkatkan akurasi terjemahan. Model ini menganalisis gerakan bibir, tindakan, dan teks di layar guna meningkatkan terjemahan di lingkungan bising atau untuk kata-kata dengan banyak makna.

• Latensi 3 detik: Mencapai latensi interpretasi simultan serendah 3 detik.

• Interpretasi simultan tanpa kehilangan kualitas: Menggunakan teknologi prediksi unit semantik untuk menyelesaikan masalah urutan kata antarbahasa. Kualitas terjemahan real-time mendekati terjemahan offline.

• Suara alami: Menghasilkan ucapan alami seperti manusia. Model ini secara otomatis menyesuaikan nada dan emosi berdasarkan audio sumber.

Memulai

1. Persiapkan lingkungan

   Versi Python Anda harus 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

   Setelah instalasi selesai, gunakan pip untuk menginstal dependensi WebSocket yang diperlukan:

   pip install websocket-client==1.8.0 websockets

2. Buat klien

   Buat file Python baru secara lokal, beri nama livetranslate_client.py, lalu salin kode berikut ke dalam file tersebut:

   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", voice: str | None = "Cherry", *, audio_enabled: bool = True):
           if not api_key:
               raise ValueError("Kunci API tidak boleh kosong.")
               
           self.api_key = api_key
           self.target_language = target_language
           self.audio_enabled = audio_enabled
           self.voice = voice if audio_enabled else "Cherry"
           self.ws = None
           self.api_url = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-livetranslate-flash-realtime"
           
           # Konfigurasi masukan audio (dari mikrofon)
           self.input_rate = 16000
           self.input_chunk = 1600
           self.input_format = pyaudio.paInt16
           self.input_channels = 1
           
           # Konfigurasi keluaran 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()
   
       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"Berhasil terhubung ke server: {self.api_url}")
               await self.configure_session()
           except Exception as e:
               print(f"Koneksi gagal: {e}")
               self.is_connected = False
               raise
   
       async def configure_session(self):
           """Mengonfigurasi sesi terjemahan, mengatur bahasa target, suara, dan parameter lainnya."""
           config = {
               "event_id": f"event_{int(time.time() * 1000)}",
               "type": "session.update",
               "session": {
                   # 'modalities' mengontrol jenis keluaran.
                   # ["text", "audio"]: Mengembalikan teks terjemahan dan audio sintesis (disarankan).
                   # ["text"]: Hanya mengembalikan teks terjemahan.
                   "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                   **({"voice": self.voice} if self.audio_enabled and self.voice else {}),
                   "input_audio_format": "pcm16",
                   "output_audio_format": "pcm16",
                   "translation": {
                       "language": self.target_language
                   }
               }
           }
           print(f"Mengirim konfigurasi sesi: {json.dumps(config, indent=2, ensure_ascii=False)}")
           await self.ws.send(json.dumps(config))
   
       async def send_audio_chunk(self, audio_data: bytes):
           """Mengkodekan dan mengirim blok data 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):
           # Kirim data citra ke server
           if not self.is_connected:
               return
   
           if not image_bytes:
               raise ValueError("image_bytes tidak boleh kosong")
   
           # 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):
           """Mulai thread pemutar audio (hanya jika keluaran 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):
           """Loop untuk memproses pesan dari server."""
           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] Satu putaran respons telah selesai.")
                       usage = event.get("response", {}).get("usage", {})
                       if usage:
                           print(f"[INFO] Penggunaan token: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                   elif event_type == "response.audio_transcript.done":
                       print("\n[INFO] Terjemahan teks selesai.")
                       text = event.get("transcript", "")
                       if text:
                           print(f"[INFO] Teks terjemahan: {text}")
                   elif event_type == "response.text.done":
                       print("\n[INFO] Terjemahan teks selesai.")
                       text = event.get("text", "")
                       if text:
                           print(f"[INFO] Teks terjemahan: {text}")
   
           except websockets.exceptions.ConnectionClosed as e:
               print(f"[WARNING] Koneksi ditutup: {e}")
               self.is_connected = False
           except Exception as e:
               print(f"[ERROR] Terjadi kesalahan tak dikenal saat memproses pesan: {e}")
               traceback.print_exc()
               self.is_connected = False
   
       async def start_microphone_streaming(self):
           """Menangkap 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("Mikrofon aktif. Mulai berbicara...")
           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."""
           self.is_connected = False
           if self.ws:
               await self.ws.close()
               print("Koneksi WebSocket ditutup.")
           
           if self.audio_player_thread:
               self.audio_playback_queue.put(None) # Kirim sinyal terminasi
               self.audio_player_thread.join(timeout=1)
               print("Thread pemutar audio dihentikan.")
               
           self.pyaudio_instance.terminate()
           print("Instansi PyAudio dilepaskan.")

3. Berinteraksi dengan model

   Di folder yang sama dengan livetranslate_client.py, buat file Python lain bernama main.py, lalu salin kode berikut ke dalam file tersebut:

   main.py

   import os
   import asyncio
   from livetranslate_client import LiveTranslateClient
   
   def print_banner():
       print("=" * 60)
       print("  Didukung oleh Qwen qwen3-livetranslate-flash-realtime")
       print("=" * 60 + "\n")
   
   def get_user_config():
       """Dapatkan konfigurasi pengguna"""
       print("Pilih mode:")
       print("1. Suara + Teks [Default] | 2. Hanya Teks")
       mode_choice = input("Masukkan pilihan Anda (tekan Enter untuk Suara + Teks): ").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("Pilih bahasa terjemahan target (mode Suara + Teks):")
           print("1. Inggris | 2. Tiongkok | 3. Rusia | 4. Prancis | 5. Jerman | 6. Portugis | 7. Spanyol | 8. Italia | 9. Korea | 10. Jepang | 11. Kanton")
       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("Pilih bahasa terjemahan target (mode Hanya Teks):")
           print("1. Inggris | 2. Tiongkok | 3. Rusia | 4. Prancis | 5. Jerman | 6. Portugis | 7. Spanyol | 8. Italia | 9. Indonesia | 10. Korea | 11. Jepang | 12. Vietnam | 13. Thailand | 14. Arab | 15. Kanton | 16. Hindi | 17. Yunani | 18. Turki")
   
       choice = input("Masukkan pilihan Anda (default ke opsi pertama): ").strip()
       target_language = lang_map.get(choice, next(iter(lang_map.values())))
   
       voice = None
       if audio_enabled:
           print("\nPilih suara sintesis ucapan:")
           voice_map = {"1": "Cherry", "2": "Nofish", "3": "Sunny", "4": "Jada", "5": "Dylan", "6": "Peter", "7": "Eric", "8": "Kiki"}
           print("1. Cherry (Perempuan) [Default] | 2. Nofish (Laki-laki) | 3. Sunny (Perempuan Sichuan) | 4. Jada (Perempuan Shanghai) | 5. Dylan (Laki-laki Beijing) | 6. Peter (Laki-laki Tianjin) | 7. Eric (Laki-laki Sichuan) | 8. Kiki (Perempuan Kanton)")
           voice_choice = input("Masukkan pilihan Anda (tekan Enter untuk Cherry): ").strip()
           voice = voice_map.get(voice_choice, "Cherry")
       return target_language, voice, 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] Atur variabel lingkungan DASHSCOPE_API_KEY.")
           print("  Contoh: export DASHSCOPE_API_KEY='your_api_key_here'")
           return
           
       target_language, voice, audio_enabled = get_user_config()
       print("\nKonfigurasi selesai:")
       print(f"  - Bahasa target: {target_language}")
       if audio_enabled:
           print(f"  - Suara sintesis: {voice}")
       else:
           print("  - Mode keluaran: Hanya Teks")
       
       client = LiveTranslateClient(api_key=api_key, target_language=target_language, voice=voice, audio_enabled=audio_enabled)
       
       # Definisikan fungsi callback
       def on_translation_text(text):
           print(text, end="", flush=True)
   
       try:
           print("Menghubungkan ke layanan terjemahan...")
           await client.connect()
           
           # Mulai pemutaran audio berdasarkan mode
           client.start_audio_player()
           
           print("\n" + "-" * 60)
           print("Koneksi berhasil! Berbicaralah ke mikrofon.")
           print("Program akan menerjemahkan ucapan Anda secara real-time dan memutar hasilnya. Tekan Ctrl+C untuk keluar.")
           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]
           # Audio harus ditangkap dari mikrofon untuk terjemahan, terlepas dari apakah keluaran audio diaktifkan
           microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
           tasks.append(microphone_streamer)
   
           await asyncio.gather(*tasks)
   
       except KeyboardInterrupt:
           print("\n\nPengguna menghentikan. Keluar...")
       except Exception as e:
           print(f"\nTerjadi kesalahan kritis: {e}")
       finally:
           print("\nMembersihkan sumber daya...")
           await client.close()
           print("Program keluar.")
   
   if __name__ == "__main__":
       asyncio.run(main())

   Jalankan main.py dan ucapkan kalimat yang ingin Anda terjemahkan ke mikrofon. Model menyediakan audio dan teks terjemahan secara real-time. Sistem secara otomatis mendeteksi ucapan Anda dan mengirim audio ke server, sehingga tidak diperlukan tindakan manual.

Gunakan citra untuk meningkatkan akurasi terjemahan

Model qwen3-livetranslate-flash-realtime dapat menerima masukan citra untuk membantu terjemahan audio. Ini berguna untuk skenario yang melibatkan homonim atau pengenalan kata benda proper yang tidak umum. Anda dapat mengirim maksimal dua citra per detik.

Unduh contoh citra berikut secara lokal: face_mask.png dan mask.png.

Unduh kode berikut ke folder yang sama dengan livetranslate_client.py, lalu jalankan. Ucapkan "What is mask?" ke mikrofon. Saat Anda memasukkan citra masker medis, model menerjemahkan frasa tersebut menjadi “What is medical mask?”. Saat Anda memasukkan citra masker pesta, model menerjemahkan frasa tersebut menjadi “What is masquerade mask?”.

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

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "mask_medical.png"
# IMAGE_PATH = "mask_masquerade.png"

def print_banner():
    print("=" * 60)
    print("  Didukung oleh Qwen qwen3-livetranslate-flash-realtime — Contoh interaksi satu putaran (masker)")
    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] Perekaman dimulai. Silakan berbicara...")
    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 citra 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] Pertama, konfigurasikan KUNCI API di variabel lingkungan DASHSCOPE_API_KEY.")
        return

    client = LiveTranslateClient(api_key=api_key, target_language="zh", voice="Cherry", 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 menggunakan Function Compute

Konsol saat ini tidak mendukung demo ini. Anda dapat menerapkannya dengan satu klik sebagai berikut:

1. Buka Templat Function Compute, masukkan Kunci API Anda, lalu klik Create and Deploy Default Environment untuk mencobanya secara online.

2. Tunggu sekitar satu menit. Di Environment Details > Environment Context, ambil titik akhirnya. Ubah http menjadi https pada titik akhir (misalnya, https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/) dan gunakan tautan tersebut untuk berinteraksi dengan model.

   Penting

   Tautan ini menggunakan sertifikat tanda tangan sendiri dan hanya untuk pengujian sementara. Saat pertama kali mengaksesnya, browser Anda akan menampilkan peringatan keamanan. Ini adalah perilaku yang diharapkan. Jangan gunakan ini di lingkungan produksi. Untuk melanjutkan, ikuti instruksi browser Anda, seperti mengklik "Advanced" → "Proceed to (unsafe)".

Untuk mengaktifkan izin Resource Access Management (RAM), ikuti petunjuk di layar.

Anda dapat melihat kode sumber proyek di bawah Resource Information – Function Resources.

Function Compute dan Alibaba Cloud Model Studio keduanya menawarkan kuota gratis untuk pengguna baru. Kuota ini dapat menutupi biaya pengujian sederhana. Setelah kuota gratis habis, Anda dikenai biaya berdasarkan bayar sesuai penggunaan. Biaya hanya dikenakan saat layanan diakses.

Alur interaksi

Alur interaksi untuk terjemahan ucapan real-time mengikuti model berbasis event WebSocket standar, di mana server secara otomatis mendeteksi awal dan akhir ucapan serta merespons.

Referensi API

Untuk informasi selengkapnya, lihat Terjemahan audio dan video real-time (Qwen-Livetranslate).

Penagihan

• Audio: Setiap detik audio masukan atau keluaran menghabiskan 12,5 token.

• Citra: Setiap masukan ukuran 28 × 28 piksel menghabiskan 0,5 token.

Untuk harga berdasarkan token, lihat Daftar Model.

Bahasa yang didukung

Kode bahasa dalam tabel berikut dapat digunakan untuk menentukan bahasa sumber dan target.

Beberapa bahasa target hanya mendukung keluaran teks, bukan keluaran audio.

Suara yang didukung