import asyncio
import base64
import json
import os
import sys
import wave
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlencode
from websockets.asyncio.client import connect
async def send(ws, event):
await ws.send(json.dumps(event, ensure_ascii=False))
async def receive(ws):
event = json.loads(await ws.recv())
if event["type"] == "error":
raise RuntimeError(json.dumps(event.get("error"), ensure_ascii=False))
return event
async def wait_for(ws, event_type):
while True:
event = await receive(ws)
if event["type"] == event_type:
return event
async def read_response(ws):
pcm = bytearray()
while True:
event = await receive(ws)
if event["type"] in (
"response.output_text.delta",
"response.output_audio_transcript.delta",
):
print(event["delta"], end="", flush=True)
elif event["type"] == "response.output_audio.delta":
pcm.extend(base64.b64decode(event["delta"]))
elif event["type"] == "response.done":
response = event["response"]
print()
if response["status"] != "completed":
raise RuntimeError(json.dumps(
response.get("status_details") or response["status"],
ensure_ascii=False,
))
print("本轮用量：", json.dumps(response.get("usage"), ensure_ascii=False))
return response, pcm
async def run(mode, input_path):
key = os.environ["MODELHUB_API_KEY"]
model = os.getenv("MODELHUB_REALTIME_MODEL", "gpt-realtime-2.1")
base = os.getenv(
"MODELHUB_REALTIME_URL", "wss://model-api.skyengine.com.cn/v1/realtime"
)
url = base + "?" + urlencode({"model": model})
session = {
"type": "realtime",
"instructions": "请用简体中文简洁回答。",
"output_modalities": ["audio" if mode == "audio" else "text"],
"reasoning": {"effort": "low"},
"max_output_tokens": 2048,
"audio": {"input": {"turn_detection": None}},
}
if mode == "audio":
session["audio"] = {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"turn_detection": None,
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "marin",
},
}
if mode == "tools":
session["tools"] = [{
"type": "function",
"name": "get_beijing_time",
"description": "查询当前北京时间。",
"parameters": {
"type": "object", "properties": {},
"required": [], "additionalProperties": False,
},
}]
# 总超时覆盖建连、上传及生成；按业务需要调整。
async with asyncio.timeout(120):
async with connect(
url, additional_headers={"api-key": key},
open_timeout=15, close_timeout=5, max_size=10 * 1024 * 1024,
) as ws:
created = await wait_for(ws, "session.created")
print("会话模型：", created["session"].get("model"))
await send(ws, {"type": "session.update", "session": session})
await wait_for(ws, "session.updated")
if mode == "audio":
data = Path(input_path).read_bytes()
if len(data) < 4800 or len(data) % 2:
raise ValueError("请提供至少 100 毫秒、完整 16-bit 采样的裸 PCM。")
# 100 毫秒一片；这里只演示文件上传，不模拟麦克风采集。
for offset in range(0, len(data), 4800):
await send(ws, {
"type": "input_audio_buffer.append",
"audio": base64.b64encode(data[offset:offset + 4800]).decode(),
})
await send(ws, {"type": "input_audio_buffer.commit"})
await wait_for(ws, "input_audio_buffer.committed")
else:
prompt = "请用一句话介绍你能如何帮助我。"
if mode == "tools":
prompt = "请调用工具查询现在的北京时间，然后告诉我结果。"
content = [{"type": "input_text", "text": prompt}]
if mode == "image":
content[0]["text"] = "请描述这张图片。"
content.append({
"type": "input_image",
"image_url": "data:image/png;base64," +
base64.b64encode(Path(input_path).read_bytes()).decode(),
})
await send(ws, {
"type": "conversation.item.create",
"item": {"type": "message", "role": "user", "content": content},
})
config = {}
if mode == "tools":
config["tool_choice"] = {"type": "function", "name": "get_beijing_time"}
await send(ws, {"type": "response.create", "response": config})
response, pcm = await read_response(ws)
if mode == "tools":
calls = [item for item in response["output"]
if item["type"] == "function_call"]
if not calls:
raise RuntimeError("没有收到预期的函数调用。")
for call in calls:
if call["name"] != "get_beijing_time" or json.loads(call["arguments"]) != {}:
raise ValueError("未知工具或不符合约定的工具参数。")
now = datetime.now(timezone(timedelta(hours=8))).isoformat()
await send(ws, {
"type": "conversation.item.create",
"item": {
"type": "function_call_output", "call_id": call["call_id"],
"output": json.dumps({"beijing_time": now}),
},
})
await send(ws, {
"type": "response.create", "response": {"tool_choice": "none"},
})
await read_response(ws)
if mode == "audio":
if not pcm:
raise RuntimeError("没有收到音频数据。")
with wave.open("reply.wav", "wb") as output:
output.setnchannels(1)
output.setsampwidth(2)
output.setframerate(24000)
output.writeframes(pcm)
print("语音回答已保存为 reply.wav")
if __name__ == "__main__":
mode = sys.argv[1] if len(sys.argv) > 1 else "text"
if mode not in {"text", "audio", "image", "tools"}:
raise SystemExit("模式必须是 text、audio、image 或 tools。")
if mode in {"audio", "image"} and len(sys.argv) < 3:
raise SystemExit("audio / image 模式需要指定输入文件。")
asyncio.run(run(mode, sys.argv[2] if len(sys.argv) > 2 else None))
