Terjemahan audio dan video real-time - Qwen

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

# 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

• 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

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

• 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

• 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

​

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

   cURL

   Copy

   brew install portaudio && 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

   Copy

   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

   Copy

   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

• 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

{
    "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"
        }
    }
}

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

{
    "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

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

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.
   Endpoint 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

​

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.

​

Batas laju

​

Bahasa yang didukung

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