Qwen3.8-LiveTranslate: Names the speaker. Carries the meaning.

API Demo Simultaneous interpretation is not only about translating fast — it must also hear clearly and translate accurately. Qwen3.8-LiveTranslate rebuilds real-time simultaneous interpretation with an Interleave architecture, improving faithfulness, fluency, and conciseness across the board, while average lagging (LAAL) drops from 2.8 seconds to 2.3 seconds. We want simultaneous interpretation to convey not only language but also the people and context behind a conversation. Building on support for 60 languages, Qwen3.

Simultaneous interpretation is not only about translating fast — it must also hear clearly and translate accurately. Qwen3.8-LiveTranslate rebuilds real-time simultaneous interpretation with an Interleave architecture, improving faithfulness, fluency, and conciseness across the board, while average lagging (LAAL) drops from 2.8 seconds to 2.3 seconds. We want simultaneous interpretation to convey not only language but also the people and context behind a conversation. Building on support for 60 languages, Qwen3.8-LiveTranslate adds three capabilities that bring simultaneous interpretation to a wider range of real-world scenarios: real-time speaker separation, with clear attribution for every sentence and more stable voice cloning; synchronized source-and-translation output on one bilingual screen; and long-context disambiguation that reads the present in light of what came before, making names and terminology more precise.

Qwen3.8-LiveTranslate adopts a Hybrid-MoE-based Thinker–Talker two-module design, connecting streaming understanding, text output, and speech generation through interleaving. Here, the Thinker arranges video, audio, source text, and translation into a single causal sequence — interleaved in temporal order and produced end to end — so that understanding and translation happen within one sequence; the Talker then combines the translation and the source audio to synthesize the translation into speech that preserves the original speaker’s timbre.

On Omnilingua-MSpeaker, a multi-speaker long-audio evaluation set covering 14 language directions, Qwen3.8-LiveTranslate outperforms today’s mainstream real-time interpretation systems across four dimensions: translation faithfulness, fluency, and conciseness, together with the Diarization Error Rate (DER).

We evaluate real-time interpretation performance across 70 language directions on the public FLEURS audio test set. Qwen3.8-LiveTranslate leads both the previous generation and current mainstream real-time interpretation systems across four dimensions: translation quality, average lagging, speech recognition accuracy, and speech synthesis quality.

Using two dialogues from Journey to the West and a set of homophone examples, the demos below showcase speaker attribution and voice cloning, synchronized bilingual output, and how context and visual information help with translation disambiguation.

Below is a complete real-time interpretation client example: it captures microphone audio, streams it to the server, and receives and plays back the translation. This generation’s new speaker separation and synchronized source-and-translation output capabilities are both enabled through session configuration; the server returns the speaker identifier and the source-language text alongside the translation, with no need to call a separate ASR interface.

import osimport timeimport base64import asyncioimport jsonimport websocketsimport pyaudioimport queueimport threadingimport tracebackclass LiveTranslateClient:    """Client for the DashScope live-translation service: captures mic audio, sends it to the server, and plays back the translated speech."""
    def __init__(self, api_key: str, workspace_id: str, target_language: str = "en", *, audio_enabled: bool = True):        if not api_key:            raise ValueError("API key cannot be empty.")        if not workspace_id:            raise ValueError("Workspace ID cannot be empty.")
        self.api_key = api_key        self.workspace_id = workspace_id        self.target_language = target_language        self.audio_enabled = audio_enabled        self.ws = None        # URL for the China (Beijing) region; replace {workspace_id} with your Model Studio workspace ID. URLs differ by region.        self.api_url = (            f"wss://{workspace_id}.cn-beijing.maas.aliyuncs.com"            "/api-ws/v1/realtime?model=qwen3.8-livetranslate-flash-realtime"        )
        # Audio input parameters (microphone capture)        self.input_rate = 16000        self.input_chunk = 1600        self.input_format = pyaudio.paInt16        self.input_channels = 1
        # Audio output parameters (local playback)        self.output_rate = 24000        self.output_chunk = 2400        self.output_format = pyaudio.paInt16        self.output_channels = 1
        # Runtime state and playback resources        self.is_connected = False        self.audio_player_thread = None        self.audio_playback_queue = queue.Queue()        self.pyaudio_instance = pyaudio.PyAudio()        # Set once the server returns session.finished, so close() can wait for a clean shutdown.        self.session_finished_event = asyncio.Event()
    async def connect(self):        """Open a WebSocket connection to the translation service."""        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 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):        """Configure the translation session: target language, audio formats, and optional features."""        config = {            "event_id": f"event_{int(time.time() * 1000)}",            "type": "session.update",            "session": {                # `output_modalities` decides what the server returns:                #   ["text", "audio"] — both translated text and synthesized speech (recommended)                #   ["text"]          — translated text only                "output_modalities": ["text", "audio"] if self.audio_enabled else ["text"],                "input_audio_format": "pcm",                "output_audio_format": "pcm",                # `input_audio_transcription`: enable source-language ASR.                # Setting `model` to 'qwen3-asr-flash-realtime' also streams back the source transcript.                # "input_audio_transcription": {                #     "model": "qwen3-asr-flash-realtime",                #     "language": "zh"  # source language; defaults to 'en'                # },                "translation": {                    "language": self.target_language,                    # `corpus`: register hotwords to boost accuracy on proper nouns and domain-specific terms.                    # "corpus": {                    #     "phrases": {                    #         "人工智能": "Artificial Intelligence",                    #         "机器学习": "Machine Learning"                    #     }                    # }                }            }        }        print(f"Sending session config: {json.dumps(config, indent=2, ensure_ascii=False)}")        await self.ws.send(json.dumps(config))
    async def send_audio_chunk(self, audio_data: bytes):        """Base64-encode an audio chunk and send it to the 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):        """Send an image frame to the server as visual context for translation."""        if not self.is_connected:            return        if not image_bytes:            raise ValueError("image_bytes cannot be empty")
        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):        """Background thread task: drain PCM chunks from the playback queue and write them to the speaker output stream."""        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:  # sentinel: stop the playback loop                        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):        """Spin up the background audio playback thread (no-op when audio output is disabled)."""        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):        """Continuously receive and dispatch event messages pushed by the 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] Response complete.")                    usage = event.get("response", {}).get("usage", {})                    if usage:                        print(f"[INFO] Token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}")                # Receive source-language ASR results (requires input_audio_transcription.model to be enabled)                # elif event_type == "conversation.item.input_audio_transcription.delta":                #     delta = event.get("delta", "")  # incremental transcript                #     print(f"[Recognizing] {delta}", end="", flush=True)                # elif event_type == "conversation.item.input_audio_transcription.completed":                #     transcript = event.get("transcript", "")  # final transcript for an utterance                #     print(f"[Source] {transcript}")                # In voice + text mode, the incremental translation text arrives alongside synthesized audio                elif event_type == "response.audio_transcript.delta":                    on_text_received(event.get("delta", ""))                # In text-only mode, the incremental translation arrives via response.text.delta                elif event_type == "response.text.delta":                    on_text_received(event.get("delta", ""))                # The server acknowledges session.finish and completes the shutdown handshake                elif event_type == "session.finished":                    print("\n[INFO] Session finished.")                    self.session_finished_event.set()
        except websockets.exceptions.ConnectionClosed as e:            print(f"[WARNING] Connection closed: {e}")            self.is_connected = False        except Exception as e:            print(f"[ERROR] Unknown error during message handling: {e}")            traceback.print_exc()            self.is_connected = False
    async def start_microphone_streaming(self):        """Continuously capture microphone audio and stream it to the server in real time."""        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 started, please begin 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):        """Gracefully close the WebSocket connection and release audio resources."""        # Ask the server to finish the session, then wait for its session.finished acknowledgement.        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 the server to finish processing...")            try:                await asyncio.wait_for(self.session_finished_event.wait(), timeout=15)                print("Server finished processing.")            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)  # signal the playback thread to exit            self.audio_player_thread.join(timeout=1)            print("Audio playback thread stopped.")
        self.pyaudio_instance.terminate()        print("PyAudio instance released.")def print_banner():    print("=" * 60)    print("  Powered by Qwen qwen3.8-livetranslate-flash-realtime")    print("=" * 60 + "\n")def get_user_config():    """Collect runtime parameters from the user via CLI: output mode and target language."""    print("Select mode:")    print("1. Voice + Text [default] | 2. Text only")    mode_choice = input("Enter option (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 target translation 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 target translation 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 option (default is the first one): ").strip()    target_language = lang_map.get(choice, next(iter(lang_map.values())))
    return target_language, audio_enabledasync def main():    """Program entry point: connect, configure the session, and drive the live-translation loop."""    print_banner()
    api_key = os.environ.get("DASHSCOPE_API_KEY")    if not api_key:        print("[ERROR] Please set the environment variable DASHSCOPE_API_KEY")        print("  Example: export DASHSCOPE_API_KEY='your_api_key_here'")        return
    workspace_id = os.environ.get("DASHSCOPE_WORKSPACE_ID")    if not workspace_id:        print("[ERROR] Please set the environment variable DASHSCOPE_WORKSPACE_ID (your Model Studio workspace ID)")        print("  Example: export DASHSCOPE_WORKSPACE_ID='your_workspace_id_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, workspace_id=workspace_id, target_language=target_language, audio_enabled=audio_enabled)
    # Callback fired as translated text arrives — stream it to stdout, character by character    def on_translation_text(text):        print(text, end="", flush=True)
    try:        print("Connecting to the translation service...")        await client.connect()
        # Launch the audio playback thread (only does real work when audio output is enabled)        client.start_audio_player()
        print("\n" + "-" * 60)        print("Connected! Please speak into the microphone.")        print("The program will translate your speech in real time and play the results. Press Ctrl+C to exit.")        print("-" * 60 + "\n")
        # Run two coroutines concurrently: server-message handling + microphone audio upload        message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))        tasks = [message_handler]        # Microphone capture is the translation input source — required regardless of output mode        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"\nFatal error occurred: {e}")    finally:        print("\nCleaning up resources...")        await client.close()        print("Program exited.")if __name__ == "__main__":    asyncio.run(main())

Simultaneous interpretation is far from its final destination. Guided by “hear it more fully, convey it more faithfully, use it more widely,” we are looking toward the following directions:

Feel free to cite the following article if you find Qwen3.8-LiveTranslate helpful:

@misc{qwen38livetranslateblog,    title = {Qwen3.8-LiveTranslate: Names the speaker. Carries the meaning.},    url = {https://qwen.ai/blog?id=qwen3.8-livetranslate},    author = {Qwen Team},    month = {September},    year = {2026}}