Qwen3.8-LiveTranslate：知其人，传其义

import os
import time
import base64
import asyncio
import json
import websockets
import pyaudio
import queue
import threading
import traceback


class LiveTranslateClient:
    """与 DashScope 实时翻译服务交互的客户端：采集麦克风音频、发送至服务端、接收并播放译文。"""

    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
        # 华北 2（北京）地域接入地址；{workspace_id} 为百炼业务空间 ID，其它地域 URL 不同
        self.api_url = (
            f"wss://{workspace_id}.cn-beijing.maas.aliyuncs.com"
            "/api-ws/v1/realtime?model=qwen3.8-livetranslate-flash-realtime"
        )

        # 音频输入参数（麦克风采集）
        self.input_rate = 16000
        self.input_chunk = 1600
        self.input_format = pyaudio.paInt16
        self.input_channels = 1

        # 音频输出参数（本地播放）
        self.output_rate = 24000
        self.output_chunk = 2400
        self.output_format = pyaudio.paInt16
        self.output_channels = 1

        # 运行状态与播放资源
        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):
        """与翻译服务建立 WebSocket 连接。"""
        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"成功连接到服务端: {self.api_url}")
            await self.configure_session()
        except Exception as e:
            print(f"连接失败: {e}")
            self.is_connected = False
            raise

    async def configure_session(self):
        """配置翻译会话：目标语言、音频格式与可选能力。"""
        config = {
            "event_id": f"event_{int(time.time() * 1000)}",
            "type": "session.update",
            "session": {
                # output_modalities 决定服务端返回内容的形式：
                #   ["text", "audio"] —— 同时返回译文文本和合成语音（推荐）
                #   ["text"]          —— 仅返回译文文本
                "output_modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                "input_audio_format": "pcm",
                "output_audio_format": "pcm",
                # input_audio_transcription：开启源语言识别（ASR）。
                # 将 model 设为 'qwen3-asr-flash-realtime'，可在翻译的同时拿到原文识别结果。
                # "input_audio_transcription": {
                #     "model": "qwen3-asr-flash-realtime",
                #     "language": "zh"  # 指定源语言；缺省为 'en'
                # },
                "translation": {
                    "language": self.target_language,
                    # corpus：注入热词，对专有名词、行业术语等可显著提升识别与翻译准确率。
                    # "corpus": {
                    #     "phrases": {
                    #         "人工智能": "Artificial Intelligence",
                    #         "机器学习": "Machine Learning"
                    #     }
                    # }
                }
            }
        }
        print(f"发送会话配置: {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 编码后发送至服务端。"""
        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):
        """将一帧图像发送至服务端，作为视觉上下文辅助翻译。"""
        if not self.is_connected:
            return
        if not image_bytes:
            raise ValueError("image_bytes 不能为空")

        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):
        """后台线程任务：从播放队列读取 PCM 数据，写入扬声器输出流。"""
        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:  # 收到结束信号，退出播放循环
                        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):
        """启动后台音频播放线程（仅在启用音频输出时生效）。"""
        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):
        """持续接收并分发服务端推送的事件消息。"""
        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)

                # 语音+文本模式下，译文文本随合成语音流式返回，通过 response.audio_transcript.delta 累加
                elif event_type == "response.audio_transcript.delta":
                    on_text_received(event.get("delta", ""))
                # 仅文本模式下，译文文本通过 response.text.delta 流式返回
                elif event_type == "response.text.delta":
                    on_text_received(event.get("delta", ""))

                elif event_type == "response.done":
                    print("\n[INFO] 一轮响应完成。")
                    usage = event.get("response", {}).get("usage", {})
                    if usage:
                        print(f"[INFO] Token 使用情况: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                # 收到 session.finished 表示服务端已处理完 session.finish 请求
                elif event_type == "session.finished":
                    print("\n[INFO] 会话已结束。")
                    self.session_finished_event.set()
                # 接收源语言识别结果（需先启用 input_audio_transcription.model）
                # elif event_type == "conversation.item.input_audio_transcription.delta":
                #     print(event.get("delta", ""), end="", flush=True)  # 流式识别增量
                # elif event_type == "conversation.item.input_audio_transcription.completed":
                #     transcript = event.get("transcript", "")  # 一段语音的最终识别文本
                #     print(f"[源语言] {transcript}")

        except websockets.exceptions.ConnectionClosed as e:
            print(f"[WARNING] 连接已关闭: {e}")
            self.is_connected = False
        except Exception as e:
            print(f"[ERROR] 消息处理时发生未知错误: {e}")
            traceback.print_exc()
            self.is_connected = False

    async def start_microphone_streaming(self):
        """持续从麦克风采集音频，并实时流式发送至服务端。"""
        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("麦克风已启动，请开始说话...")
        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):
        """优雅关闭 WebSocket 连接，并释放音频相关资源。"""
        # 发送 session.finish 并等待 session.finished，确保最后一段语音的翻译不丢失
        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("已发送 session.finish，等待服务端完成处理...")
            try:
                await asyncio.wait_for(self.session_finished_event.wait(), timeout=15)
                print("服务端已完成处理。")
            except asyncio.TimeoutError:
                print("等待 session.finished 超时。")

        self.is_connected = False
        if self.ws:
            await self.ws.close()
            print("WebSocket 连接已关闭。")

        if self.audio_player_thread:
            self.audio_playback_queue.put(None)  # 通知播放线程退出
            self.audio_player_thread.join(timeout=1)
            print("音频播放线程已停止。")

        self.pyaudio_instance.terminate()
        print("PyAudio 实例已释放。")


def print_banner():
    print("=" * 60)
    print("  基于千问 qwen3.8-livetranslate-flash-realtime")
    print("=" * 60 + "\n")


def get_user_config():
    """通过命令行交互收集运行参数：输出模式与目标语言。"""
    print("请选择模式:")
    print("1. 语音+文本 [默认] | 2. 仅文本")
    mode_choice = input("请输入选项 (直接回车选择语音+文本): ").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("请选择翻译目标语言 (音频+文本 模式):")
        print("1. 英语 | 2. 中文 | 3. 俄语 | 4. 法语 | 5. 德语 | 6. 葡萄牙语 | 7. 西班牙语 | 8. 意大利语 | 9. 韩语 | 10. 日语 | 11. 粤语")
    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("请选择翻译目标语言 (仅文本 模式):")
        print("1. 英语 | 2. 中文 | 3. 俄语 | 4. 法语 | 5. 德语 | 6. 葡萄牙语 | 7. 西班牙语 | 8. 意大利语 | 9. 印尼语 | 10. 韩语 | 11. 日语 | 12. 越南语 | 13. 泰语 | 14. 阿拉伯语 | 15. 粤语 | 16. 印地语 | 17. 希腊语 | 18. 土耳其语")

    choice = input("请输入选项 (默认取第一个): ").strip()
    target_language = lang_map.get(choice, next(iter(lang_map.values())))

    return target_language, audio_enabled


async def main():
    """程序主入口：完成连接与会话配置，并启动实时翻译循环。"""
    print_banner()

    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("[ERROR] 请设置环境变量 DASHSCOPE_API_KEY")
        print("  例如: export DASHSCOPE_API_KEY='your_api_key_here'")
        return

    workspace_id = os.environ.get("DASHSCOPE_WORKSPACE_ID")
    if not workspace_id:
        print("[ERROR] 请设置环境变量 DASHSCOPE_WORKSPACE_ID（百炼业务空间 ID）")
        print("  例如: export DASHSCOPE_WORKSPACE_ID='your_workspace_id_here'")
        return

    target_language, audio_enabled = get_user_config()
    print("\n配置完成:")
    print(f"  - 目标语言: {target_language}")
    if not audio_enabled:
        print("  - 输出模式: 仅文本")

    client = LiveTranslateClient(api_key=api_key, workspace_id=workspace_id, target_language=target_language, audio_enabled=audio_enabled)

    # 翻译文本到达时的回调：按字流式打印到终端
    def on_translation_text(text):
        print(text, end="", flush=True)

    try:
        print("正在连接到翻译服务...")
        await client.connect()

        # 启动音频播放线程（仅当启用音频输出时实际工作）
        client.start_audio_player()

        print("\n" + "-" * 60)
        print("连接成功！请对着麦克风说话。")
        print("程序将实时翻译您的语音并播放结果。按 Ctrl+C 退出。")
        print("-" * 60 + "\n")

        # 并发执行：处理服务端消息 + 上传麦克风音频
        message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
        tasks = [message_handler]
        # 麦克风采集是翻译的输入源，无论是否输出语音都必须开启
        microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
        tasks.append(microphone_streamer)

        await asyncio.gather(*tasks)

    except KeyboardInterrupt:
        print("\n\n用户中断，正在退出...")
    except Exception as e:
        print(f"\n发生严重错误: {e}")
    finally:
        print("\n正在清理资源...")
        await client.close()
        print("程序已退出。")


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

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