import asyncio
import base64
import json
import os
import sys
import uuid
import wave
from urllib.parse import urlencode
from websockets.asyncio.client import connect
async def send(ws, kind, **fields):
await ws.send(json.dumps({
"type": kind, "event_id": uuid.uuid4().hex, **fields
}, ensure_ascii=False))
async def receive(ws):
event = json.loads(await ws.recv())
if event["type"] == "error":
raise RuntimeError(json.dumps(event["error"], ensure_ascii=False))
return event
async def wait_for(ws, kind):
while True:
event = await receive(ws)
if event["type"] == kind:
return event
async def user_text(ws, text):
await send(ws, "conversation.item.create", item={
"type": "message", "role": "user",
"content": [{"type": "input_text", "text": text}],
})
await wait_for(ws, "conversation.item.added")
async def read_response(ws):
pcm = bytearray()
while True:
event = await receive(ws)
kind = event["type"]
if kind in ("response.output_text.delta",
"response.output_audio_transcript.delta"):
print(event["delta"], end="", flush=True)
elif kind == "response.output_audio.delta":
pcm.extend(base64.b64decode(event["delta"]))
elif kind.startswith("modelhub."):
print("扩展事件：", kind)
elif kind == "response.done":
response = event["response"]
if response["status"] != "completed":
raise RuntimeError(json.dumps(response, ensure_ascii=False))
print("\n本轮用量：", json.dumps(response.get("usage"), ensure_ascii=False))
if response.get("search_info"):
print("搜索来源：", json.dumps(response["search_info"], ensure_ascii=False))
return response, pcm
async def feed_audio(ws, pcm, rate, auto):
# 每帧20ms；自动模式按实时速度发送，末尾追加静音。
payload = pcm + (bytes(rate * 2 * 3) if auto else b"")
chunk = rate * 2 // 50
for offset in range(0, len(payload), chunk):
await send(ws, "input_audio_buffer.append",
audio=base64.b64encode(payload[offset:offset + chunk]).decode())
if auto:
await asyncio.sleep(0.02)
async def run(mode, path):
audio_mode = mode in ("audio", "vad")
pcm, rate = b"", 24000
if audio_mode:
with wave.open(path, "rb") as source:
rate = source.getframerate()
if (source.getnchannels(), source.getsampwidth()) != (1, 2) or rate not in (16000, 24000):
raise ValueError("需要16kHz或24kHz、16位、单声道WAV")
pcm = source.readframes(source.getnframes())
session = {
"type": "realtime",
"instructions": "请用中文简洁回答，工具结果返回后再回答用户。",
"output_modalities": ["audio" if audio_mode else "text"],
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": rate},
"turn_detection": {"type": "semantic_vad"} if mode == "vad" else None,
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "longanqian",
},
},
"enable_speech_emotion": True,
"max_history_turns": 20,
}
if mode == "tools":
session["tools"] = [{
"type": "function", "name": "get_weather",
"description": "查询城市天气；天气问题必须先查询该工具。",
"parameters": {"type": "object", "properties": {
"city": {"type": "string"}}, "required": ["city"]},
}]
if mode == "search":
session.update(enable_search=True, search_options={"enable_source": True})
base = os.getenv("MODELHUB_REALTIME_URL", "wss://model-api.skyengine.com.cn/v1/realtime")
model = os.getenv("MODELHUB_REALTIME_MODEL", "qwen-audio-3.0-realtime-plus")
async with connect(base + "?" + urlencode({"model": model}),
additional_headers={"api-key": os.environ["MODELHUB_API_KEY"]},
open_timeout=20, close_timeout=3) as ws:
await wait_for(ws, "session.created")
await send(ws, "session.update", session=session)
await wait_for(ws, "session.updated")
if audio_mode:
if mode == "vad":
task = asyncio.create_task(feed_audio(ws, pcm, rate, True))
try:
response, output = await read_response(ws)
finally:
task.cancel()
await asyncio.gather(task, return_exceptions=True)
else:
await feed_audio(ws, pcm, rate, False)
await send(ws, "input_audio_buffer.commit")
await send(ws, "response.create")
response, output = await read_response(ws)
else:
prompt = {"text": "用一句话介绍你自己。", "tools": "北京天气怎么样？",
"search": "请联网查询阿里云百炼最近的产品更新，并提供来源。"}[mode]
await user_text(ws, prompt)
await send(ws, "response.create")
response, output = await read_response(ws)
calls = [item for item in response.get("output", []) if item["type"] == "function_call"]
if mode == "tools":
if not calls:
raise RuntimeError("本轮没有工具调用；模型只支持自动工具选择")
for call in calls:
if call["name"] != "get_weather":
raise ValueError("未知工具")
arguments = json.loads(call["arguments"])
# 演示固定结果；生产应用应查询自己的天气服务。
await send(ws, "conversation.item.create", item={
"type": "function_call_output", "call_id": call["call_id"],
"output": json.dumps({"city": arguments["city"], "weather": "晴",
"temperature": 26}, ensure_ascii=False),
})
await wait_for(ws, "conversation.item.added")
await send(ws, "response.create")
response, output = await read_response(ws)
if output:
with wave.open("qwen-output.wav", "wb") as target:
target.setnchannels(1)
target.setsampwidth(2)
target.setframerate(24000)
target.writeframes(output)
if __name__ == "__main__":
mode = sys.argv[1] if len(sys.argv) > 1 else "text"
if mode not in ("text", "audio", "vad", "tools", "search"):
raise SystemExit("模式：text / audio / vad / tools / search")
path = sys.argv[2] if len(sys.argv) > 2 else "input.wav"
asyncio.run(asyncio.wait_for(run(mode, path), timeout=120))
