大模型服务平台百炼：Qwen-LiveTranslate-Realtime Python SDK

import os
import sys
import base64
import signal
import pyaudio
from concurrent.futures import ThreadPoolExecutor

import dashscope
from dashscope.v2 import DashScopeError

def main():
    # Check the API Key
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("请设置环境变量 DASHSCOPE_API_KEY")
        sys.exit(1)

    # The following is the configuration for the Singapore (ap-southeast-1) region. When calling, replace "{WorkspaceId}" with your actual workspace ID. Configurations differ by region.
    client = dashscope.Client(
        # If you have not configured the environment variable, replace the following line with your Alibaba Cloud Model Studio API Key: api_key="sk-xxx"
        api_key=os.environ["DASHSCOPE_API_KEY"],
        workspace="{WorkspaceId}",
        region="ap-southeast-1",
    )

    # Initialize PyAudio
    pya = pyaudio.PyAudio()

    # Initialize the speaker (for playing the translated speech)
    speaker = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True,
        frames_per_buffer=2400
    )

    # Initialize the microphone (for capturing the speech input)
    mic = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        frames_per_buffer=1600
    )

    # Create a real-time translation session: entering the with context automatically establishes the connection
    with client.audio.realtime.connect(
            model="qwen3.5-livetranslate-flash-realtime",
    ) as session:
        # Update the session configuration (recommended to call right after the connection is established)
        session.update_session(
            output_modalities=["text", "audio"],
            input_audio_transcription_model="qwen3-asr-flash-realtime",
            voice="Tina",
            translation_params={
                "language": "en",  # Translation target language: English
                "corpus": {
                    "phrases": {
                        "人工智能": "Artificial Intelligence",
                        "机器学习": "Machine Learning"
                    }
                }
            },
        )

        # Receiving thread: iterate over the server events
        def receive_events():
            try:
                for event in session.events():
                    event_type = event.get("type", "")
                    if event_type == "input_audio_buffer.speech_started":
                        print("====== 检测到语音输入 ======")
                    elif event_type == "input_audio_buffer.speech_stopped":
                        print("====== 语音输入结束 ======")
                    elif event_type == "conversation.item.input_audio_transcription.completed":
                        print(f"[原文] {event.get('transcript', '')}")
                    elif event_type == "response.audio_transcript.done":
                        print(f"[翻译结果] {event.get('transcript', '')}")
                    elif event_type == "response.audio.delta":
                        audio_b64 = event.get("delta", "")
                        if audio_b64:
                            speaker.write(base64.b64decode(audio_b64))
            except DashScopeError as e:
                print(f"[错误] {e.message}")

        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events)

            # Register the exit signal handler
            def on_exit(sig, frame):
                print("\n[正在退出...]")
                session.end_session()
                session.close()
                mic.stop_stream()
                mic.close()
                speaker.stop_stream()
                speaker.close()
                pya.terminate()
                sys.exit(0)

            signal.signal(signal.SIGINT, on_exit)

            print("[开始实时翻译] 请对着麦克风说话，按 Ctrl+C 退出")

            # Continuously capture microphone audio and send it (append_audio also accepts raw bytes, encoded automatically)
            try:
                while True:
                    audio_data = mic.read(1600, exception_on_overflow=False)
                    session.append_audio(audio_data)
            finally:
                session.end_session()
                receiver.result()

if __name__ == "__main__":
    main()