Qwen-Audio 实时语音对话

# pip install websockets pyaudio
import json
import os
import base64
import threading
import time
import pyaudio
import websocket

API_KEY = os.getenv("DASHSCOPE_API_KEY")
# 以下为华北2（北京）地域的WebSocket URL，调用时请将{WorkspaceId}（含花括号）替换为真实的业务空间ID，各地域的URL不同。
API_URL = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-audio-3.0-realtime-plus"

pya = pyaudio.PyAudio()
out_stream = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)
ws_ref = [None]
resp_done = threading.Event()

def on_open(ws):
    ws_ref[0] = ws
    # 配置 push-to-talk 模式（turn_detection 设为 null）
    ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "modalities": ["audio", "text"],
            "voice": "longanqian",
            "turn_detection": None
        }
    }))

def on_message(ws, message):
    event = json.loads(message)
    event_type = event["type"]
    if event_type == "response.audio.delta":
        out_stream.write(base64.b64decode(event["delta"]))
    elif event_type == "conversation.item.input_audio_transcription.completed":
        print(f"[User] {event['transcript']}")
    elif event_type == "response.audio_transcript.done":
        print(f"[LLM] {event['transcript']}")
    elif event_type == "response.done":
        resp_done.set()
    elif event_type == "error":
        print(f"[Error] {event['error']['message']}")

def on_error(ws, error):
    print(f"Error: {error}")

def record_and_send(ws):
    mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    stop = threading.Event()

    def reader():
        while not stop.is_set():
            try:
                data = mic.read(3200, exception_on_overflow=False)
                ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(data).decode()
                }))
            except Exception:
                break

    t = threading.Thread(target=reader, daemon=True)
    t.start()
    input()
    stop.set()
    t.join(timeout=1.0)
    mic.close()

headers = ["Authorization: Bearer " + API_KEY]
ws = websocket.WebSocketApp(
    API_URL, header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)
threading.Thread(target=ws.run_forever, daemon=True).start()
time.sleep(2)

try:
    turn = 1
    while True:
        print(f"\n--- 第 {turn} 轮对话 ---")
        cmd = input("按 Enter 开始录音（输入 q 退出）...")
        if cmd.strip().lower() in ["q", "quit"]:
            break
        print("录音中... 再次按 Enter 停止。")
        record_and_send(ws_ref[0])
        resp_done.clear()
        # 提交音频并触发推理
        ws_ref[0].send(json.dumps({"type": "input_audio_buffer.commit"}))
        ws_ref[0].send(json.dumps({
            "type": "response.create",
            "response": {"modalities": ["audio", "text"]}
        }))
        print("等待模型回复...")
        resp_done.wait(timeout=30)
        turn += 1
except KeyboardInterrupt:
    pass
finally:
    ws.close()
    out_stream.close()
    pya.terminate()
    print("\n对话结束")

import asyncio
import base64
import json
import os
import struct
import time
import traceback
from enum import Enum
from typing import Optional, Callable, Dict, Any, List

import pyaudio
import websockets

from B64PCMPlayer import B64PCMPlayer


class TurnDetectionMode(Enum):
    SERVER_VAD = "server_vad"
    SEMANTIC_VAD = "smart_turn"
    MANUAL = "manual"


# ============ 工具函数定义 ============

def get_weather(city: str) -> str:
    """查询城市天气（生产环境替换为真实 API）。"""
    return json.dumps({"temperature": 18, "condition": "晴", "wind": "微风"})


def get_train_price(src: str, dst: str) -> str:
    """查询火车票价（生产环境替换为真实 API）。"""
    return json.dumps({"price": 350, "seat": "二等座", "note": "以 12306 为准"})


# ============ Tools Schema ============

tools: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气信息。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称，如北京、上海"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "查询两个城市之间的火车票价格。",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {"type": "string", "description": "出发城市"},
                    "dst": {"type": "string", "description": "到达城市"}
                },
                "required": ["src", "dst"]
            }
        }
    }
]

# 函数名 -> 可调用对象
functions: Dict[str, Callable] = {
    "get_weather": get_weather,
    "get_train_price": get_train_price,
}


class FunRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "longanqian",
            instructions: str = "",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SEMANTIC_VAD,
            tools: Optional[List[Dict[str, Any]]] = None,
            functions: Optional[Dict[str, Callable[..., Any]]] = None,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta_b64: Optional[Callable[[str], None]] = None,
            on_speech_started: Optional[Callable[[], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,
            on_output_transcript: Optional[Callable[[str], None]] = None,
            extra_event_handlers: Optional[Dict[str, Callable[[Dict[str, Any]], None]]] = None
    ):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.ws = None
        self.on_text_delta = on_text_delta
        # 回调参数为 base64 编码的 PCM 音频
        self.on_audio_delta_b64 = on_audio_delta_b64
        self.on_speech_started = on_speech_started
        self.on_input_transcript = on_input_transcript
        self.on_output_transcript = on_output_transcript
        self.turn_detection_mode = turn_detection_mode
        self.extra_event_handlers = extra_event_handlers or {}

        # Function Calling 配置
        self.tools = tools or []
        self.functions = functions or {}

        # 回复状态跟踪（用于打断处理和回声抑制）
        self._current_response_id = None
        self._current_item_id = None
        self._is_responding = False
        self._audio_suppressed = False
        self._print_input_transcript = True
        self._output_transcript_buffer = ""

    async def connect(self) -> None:
        """建立 WebSocket 连接并发送会话配置。"""
        url = f"{self.base_url}?model={self.model}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "x-dashscope-dataInspection": "disable",
        }
        self.ws = await websockets.connect(url, additional_headers=headers)

        session_config = {
            "modalities": ["text", "audio"],
            "voice": self.voice,
            "instructions": self.instructions,
            "input_audio_format": "pcm",
            "output_audio_format": "pcm",
            "turn_detection": {},
            "tools": self.tools,
            "max_history_turns": 50
        }

        if self.turn_detection_mode == TurnDetectionMode.MANUAL:
            session_config['turn_detection'] = None
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SERVER_VAD:
            session_config['turn_detection'] = {
                "type": "server_vad",
                "threshold": 0.1,
                "silence_duration_ms": 900
            }
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
            session_config['turn_detection'] = {
                "type": "smart_turn"
            }
            await self.update_session(session_config)
        else:
            raise ValueError(f"Invalid turn detection mode: {self.turn_detection_mode}")

    async def send_event(self, event) -> None:
        event['event_id'] = "event_" + str(int(time.time() * 1000))
        await self.ws.send(json.dumps(event))

    async def update_session(self, config: Dict[str, Any]) -> None:
        """更新会话配置。"""
        event = {
            "type": "session.update",
            "session": config
        }
        await self.send_event(event)

    async def stream_audio(self, audio_chunk: bytes) -> None:
        """向 API 流式发送原始音频数据。"""
        # 仅支持 16bit 16kHz 单声道 PCM
        audio_b64 = base64.b64encode(audio_chunk).decode()
        await self.send_event({
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        })

    async def commit_audio_buffer(self) -> None:
        """提交音频缓冲区以触发处理。"""
        await self.send_event({"type": "input_audio_buffer.commit"})

    async def create_response(self) -> None:
        """请求生成回复（手动模式或 Function Call 结果回传后调用）。"""
        await self.send_event({"type": "response.create"})

    async def cancel_response(self) -> None:
        """取消当前回复。"""
        await self.send_event({"type": "response.cancel"})

    async def handle_interruption(self):
        """处理用户对当前回复的打断。"""
        if not self._is_responding:
            return
        self._audio_suppressed = True
        if self._current_response_id:
            await self.cancel_response()
        self._is_responding = False
        self._current_response_id = None
        self._current_item_id = None

    @staticmethod
    def _format_event_for_log(event: Dict[str, Any]) -> str:
        """格式化事件用于日志输出，对音频数据脱敏。"""
        event_type = event.get("type")
        if event_type == "response.audio.delta":
            delta = event.get("delta", "")
            redacted = dict(event)
            redacted["delta"] = f"<audio b64 omitted, length={len(delta)}>"
            return json.dumps(redacted, ensure_ascii=False)
        return json.dumps(event, ensure_ascii=False)

    async def _handle_function_call(self, event: Dict[str, Any]) -> None:
        """处理 Function Call：解析参数、执行函数、回传结果、触发二轮推理。"""
        call_id = event.get("call_id")
        name = event.get("name")
        arguments_str = event.get("arguments", "{}")

        print(f"[FunctionCall] 调用: {name}, call_id: {call_id}, args: {arguments_str}")

        try:
            arguments = json.loads(arguments_str) if arguments_str else {}
        except json.JSONDecodeError:
            arguments = {}

        func = self.functions.get(name)
        if func is None:
            output = json.dumps({"error": f"未注册的函数: {name}"})
        else:
            try:
                if asyncio.iscoroutinefunction(func):
                    result = await func(**arguments)
                else:
                    result = func(**arguments)
                output = str(result) if result is not None else ""
            except Exception as e:
                output = json.dumps({"error": str(e)})
                traceback.print_exc()

        # 回传 function_call_output
        await self.send_event({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": output,
            }
        })

        # 触发二轮推理
        await self.create_response()

    async def handle_messages(self) -> None:
        try:
            async for message in self.ws:
                event = json.loads(message)
                event_type = event.get("type")

                print(self._format_event_for_log(event))

                if event_type == "error":
                    continue
                elif event_type == "response.created":
                    self._current_response_id = event.get("response", {}).get("id")
                    self._is_responding = True
                    self._audio_suppressed = False
                elif event_type == "response.output_item.added":
                    self._current_item_id = event.get("item", {}).get("id")
                elif event_type == "response.done":
                    self._is_responding = False
                    self._current_response_id = None
                    self._current_item_id = None
                elif event_type == "input_audio_buffer.speech_started":
                    print("----------------Speech Started----------------")
                    if self.on_speech_started:
                        self.on_speech_started()
                    if self._is_responding:
                        await self.handle_interruption()
                elif event_type == "response.audio.delta":
                    if self._audio_suppressed:
                        continue
                    if self.on_audio_delta_b64:
                        self.on_audio_delta_b64(event["delta"])
                elif event_type == "response.function_call_arguments.done":
                    await self._handle_function_call(event)
                elif event_type in self.extra_event_handlers:
                    self.extra_event_handlers[event_type](event)
                elif event_type == "input_audio_buffer.speech_stopped":
                    print("----------------Speech Stopped----------------")
        except websockets.exceptions.ConnectionClosed:
            print(" Connection closed")
        except Exception as e:
            print(" Error in message handling: ", str(e))
            traceback.print_exc()

    async def close(self) -> None:
        """关闭 WebSocket 连接。"""
        if self.ws:
            await self.ws.close()


def _audio_energy(audio_data: bytes) -> float:
    count = len(audio_data) // 2
    if count == 0:
        return 0.0
    samples = struct.unpack(f'<{count}h', audio_data)
    return sum(abs(s) for s in samples) / count


async def record_and_send(client, player, echo_suppression=True):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    print("开始录音，请讲话...")
    if echo_suppression:
        print("提示：回声抑制已开启（AI 说话期间麦克风静音，不支持打断）。如使用耳机请设置 echo_suppression=False 以启用打断。")
    else:
        print("提示：耳机模式，支持语音打断。")
    playback_end_time = 0.0
    NOISE_GATE_THRESHOLD = 500
    try:
        while True:
            audio_data = await asyncio.to_thread(stream.read, 3200, False)
            if echo_suppression:
                is_active = client._is_responding or player.is_playing()
                if is_active:
                    playback_end_time = time.time()
                    await asyncio.sleep(0.02)
                    continue
                if time.time() - playback_end_time < 0.5:
                    await asyncio.sleep(0.02)
                    continue
            else:
                if client._is_responding or player.is_playing():
                    if _audio_energy(audio_data) < NOISE_GATE_THRESHOLD:
                        await asyncio.sleep(0.02)
                        continue
            await client.stream_audio(audio_data)
            await asyncio.sleep(0.02)
    finally:
        stream.stop_stream(); stream.close(); p.terminate()


async def main():
    pya = pyaudio.PyAudio()
    # 输出采样率 24kHz，匹配服务端音频格式
    player = B64PCMPlayer(pya, sample_rate=24000)

    client = FunRealtimeClient(
        # 以下为华北2（北京）地域的WebSocket URL，调用时请将{WorkspaceId}（含花括号）替换为真实的业务空间ID，各地域的URL不同。
        base_url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.0-realtime-plus",
        voice="longanqian",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        tools=tools,
        functions=functions,
        on_audio_delta_b64=player.add_data,
        # 语音打断时清空播放缓冲
        on_speech_started=player.cancel_playing,
    )

    await client.connect()
    print("连接成功，开始实时对话（已启用 Function Calling）...")

    try:
        await asyncio.gather(client.handle_messages(), record_and_send(client, player, echo_suppression=False))
    finally:
        await client.close()
        player.shutdown()
        pya.terminate()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程序已退出。")