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

Key Highlights#

• Interleave architecture: audio and text woven together, higher quality and lower latency. This generation recasts simultaneous interpretation as a single audio-text interleaved stream; both the audio already heard and the translation already produced can be cached and reused, so translation quality improves markedly over the previous generation while average lagging (LAAL) drops from 2.8 seconds in the previous generation to 2.3 seconds.

• Real-time speaker separation: clearer content attribution, more stable voice cloning. When several people speak in turn, the model distinguishes different speakers and what each of them says, and helps the translated speech preserve each speaker’s timbre more accurately and stably.

• Synchronized source and translation. Bilingual alignment during interpretation supports both instant comprehension and source checking, laying a foundation for follow-on features such as subtitle display, content organization, and retrieval — with broader room for application.

• Long-context disambiguation: linking history across turns for more precise translation. By drawing on prior text and historical context, the model eases the ambiguity of proper nouns, references, and the like in complex scenarios, keeping expression consistent.

Model Architecture#

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.

Performance#

Multi-Speaker Long-Audio#

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

Multilingual Performance#

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.

🎬 See It in Action#

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.

Supported Languages#

Using Qwen3.8-LiveTranslate via DashScope API#

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 os
import time
import base64
import asyncio
import json
import websockets
import pyaudio
import queue
import threading
import traceback


class 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_enabled


async 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())

Future Directions#

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:

• Approaching the latency limit of simultaneous interpretation: keep compressing end-to-end latency, pushing the gap between “hearing” and “translating” toward its limit.
• Long-term memory across sessions: carry memory across sessions into the next conversation for the same project and the same group of people, getting better the more it is used.
• Covering more languages: keep extending coverage to more long-tail languages and regional dialects, so real-time interpretation can serve everyone in the world.

Citation#

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