实时音视频翻译（Qwen-LiveTranslate）Python SDK-API参考

from dashscope.audio.qwen_omni import (
    OmniRealtimeConversation,
    OmniRealtimeCallback,
    MultiModality,
)
from dashscope.audio.qwen_omni.omni_realtime import TranslationParams

class MyCallback(OmniRealtimeCallback):
    """实时翻译回调处理"""
    def __init__(self, conversation=None):
        self.conversation = conversation
        self.handlers = {
            'session.created': self._handle_session_created,
            'response.audio_transcript.done': self._handle_translation_done,
            'response.audio.delta': self._handle_audio_delta,
            'response.done': lambda r: print('======Response Done======'),
            'input_audio_buffer.speech_started': lambda r: print('======Speech Start======'),
            'input_audio_buffer.speech_stopped': lambda r: print('======Speech Stop======'),
        }

    def on_open(self):
        print('Connection opened')

    def on_close(self, code, msg):
        print(f'Connection closed, code: {code}, msg: {msg}')

    def on_event(self, response):
        try:
            handler = self.handlers.get(response['type'])
            if handler:
                handler(response)
        except Exception as e:
            print(f'[Error] {e}')

    def _handle_session_created(self, response):
        print(f"Session created: {response['session']['id']}")

    def _handle_translation_done(self, response):
        print(f"Translation result: {response['transcript']}")

    def _handle_audio_delta(self, response):
        # 处理增量音频数据
        audio_b64 = response.get('delta', '')
        # 可将音频数据解码后播放或保存

conversation = OmniRealtimeConversation(
    model='qwen3.5-livetranslate-flash-realtime',
    url='wss://maas.qianwenaiapi.com/api-ws/v1/realtime',
    callback=MyCallback(conversation=None)  # 暂时传None，稍后注入
)
# 注入自身到回调
conversation.callback.conversation = conversation

import os
import sys
import base64
import signal
import pyaudio
from dashscope.audio.qwen_omni import (
    OmniRealtimeConversation,
    OmniRealtimeCallback,
    MultiModality,
)
from dashscope.audio.qwen_omni.omni_realtime import TranslationParams

class Callback(OmniRealtimeCallback):
    """实时翻译回调处理类"""

    def __init__(self, speaker):
        self.speaker = speaker

    def on_open(self):
        print("[连接已建立]")

    def on_close(self, code, msg):
        print(f"[连接已关闭] code: {code}, msg: {msg}")

    def on_event(self, response):
        event_type = response.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"[原文] {response.get('transcript', '')}")
        elif event_type == "response.audio_transcript.done":
            print(f"[翻译结果] {response.get('transcript', '')}")
        elif event_type == "response.audio.delta":
            audio_b64 = response.get("delta", "")
            if audio_b64:
                self.speaker.write(base64.b64decode(audio_b64))
        elif event_type == "error":
            print(f"[错误] {response.get('error', {}).get('message', '')}")

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

    # 初始化 PyAudio
    pya = pyaudio.PyAudio()

    # 初始化扬声器（用于播放翻译后的语音）
    speaker = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        output=True,
        frames_per_buffer=2400
    )

    # 初始化麦克风（用于采集语音输入）
    mic = pya.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        frames_per_buffer=1600
    )

    # 创建回调实例
    callback = Callback(speaker=speaker)

    # 创建实时会话
    conversation = OmniRealtimeConversation(
        model="qwen3.5-livetranslate-flash-realtime",
        url="wss://maas.qianwenaiapi.com/api-ws/v1/realtime",
        callback=callback
    )

    # 连接服务端
    conversation.connect()

    # 配置翻译参数
    translation_params = TranslationParams(
        language="en",  # 翻译目标语言：英语
        corpus=TranslationParams.Corpus(
            phrases={
                "人工智能": "Artificial Intelligence",
                "机器学习": "Machine Learning"
            }
        )
    )

    # 更新会话配置
    conversation.update_session(
        output_modalities=[MultiModality.TEXT, MultiModality.AUDIO],
        input_audio_transcription_model="qwen3-asr-flash-realtime",
        voice="Tina",
        translation_params=translation_params,
    )

    # 注册退出信号处理
    def on_exit(sig, frame):
        print("\n[正在退出...]")
        mic.stop_stream()
        mic.close()
        speaker.stop_stream()
        speaker.close()
        pya.terminate()
        conversation.end_session()
        conversation.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, on_exit)

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

    # 持续采集麦克风音频并发送
    while True:
        audio_data = mic.read(1600, exception_on_overflow=False)
        conversation.append_audio(base64.b64encode(audio_data).decode("ascii"))

if __name__ == "__main__":
    main()