外观
实时语音
WSS
WebSocket 接口,请在代码中调用wss://vip.zen-ai.top/v1/realtime?model=gpt-realtime通过 WebSocket 与 Realtime 模型建立长连接,实现低延迟的语音与文本实时交互。连接建立后,客户端与服务端通过 JSON 事件通信,事件协议与 OpenAI Realtime API 一致。
连接
text
wss://vip.zen-ai.top/v1/realtime?model=gpt-realtimemodelstring必填查询参数,Realtime 模型名称,如 gpt-realtime、gpt-realtime-mini。
Authorizationstring必填请求头 Bearer sk-xxxx(服务端连接时使用)。
Sec-WebSocket-Protocolstring浏览器无法设置自定义请求头时,改用子协议传递密钥:realtime, openai-insecure-api-key.sk-xxxx。注意这会在前端暴露密钥,仅用于测试。
常用事件
| 方向 | 事件 | 说明 |
|---|---|---|
| 发送 | session.update | 设置指令、音色、输入输出格式、语音活动检测等 |
| 发送 | input_audio_buffer.append | 追加 base64 编码的音频片段 |
| 发送 | conversation.item.create | 添加一条文本 / 音频消息 |
| 发送 | response.create | 请求模型生成回复 |
| 接收 | session.created | 连接建立,返回会话配置 |
| 接收 | response.*.delta | 文本、音频、转写文本的增量 |
| 接收 | response.done | 本轮回复结束,包含用量 |
| 接收 | error | 错误信息 |
提示
不同模型版本的事件名称略有差异(如 response.output_text.delta 与 response.text.delta),编写客户端时建议按事件类型后缀 .delta 统一处理增量。
Python 示例(文本对话)
python
# pip install websockets
import asyncio
import json
import os
import websockets
URL = "wss://vip.zen-ai.top/v1/realtime?model=gpt-realtime"
async def main():
headers = {"Authorization": f"Bearer {os.environ['VIP_API_KEY']}"}
async with websockets.connect(URL, additional_headers=headers) as ws:
await ws.send(json.dumps({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "你好,用一句话介绍一下你自己"}],
},
}))
await ws.send(json.dumps({"type": "response.create"}))
async for raw in ws:
event = json.loads(raw)
etype = event.get("type", "")
# 文本增量与语音转写增量(音频数据本身的 delta 为 base64,这里不打印)
if etype.endswith(("text.delta", "transcript.delta")):
print(event["delta"], end="", flush=True)
elif etype == "response.done":
print("\n用量:", event["response"].get("usage"))
break
elif etype == "error":
print("错误:", event["error"])
break
asyncio.run(main())OpenAI Python SDK
python
from openai import AsyncOpenAI
client = AsyncOpenAI(base_url="https://vip.zen-ai.top/v1", api_key="sk-xxxxxxxx")
# 较旧版本的 SDK 为 client.beta.realtime.connect
async with client.realtime.connect(model="gpt-realtime") as conn:
await conn.conversation.item.create(item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "你好"}],
})
await conn.response.create()
async for event in conn:
if event.type.endswith(("text.delta", "transcript.delta")):
print(event.delta, end="", flush=True)
elif event.type == "response.done":
break