大模型WebSocket对接：从握手到流式输出的实操教程
大模型 WebSocket 对接，本质是让你的服务端与模型服务之间建立一条长连接，双方按约定格式发送 JSON 消息，模型把生成内容分片推回来。
适合需要打字机效果、实时语音转写后接大模型、或长回答边生成边展示的场景。
适用条件：你已经有模型服务的 WebSocket 接入地址和鉴权方式（不同厂商路径不同，以控制台文档为准），服务器能主动访问外网，并且你希望用流式方式降低首字等待时间。
如果只是普通问答且不要求实时，用 HTTP 流式接口通常更简单。
先确认连接地址和鉴权方式
对接前先拿到三样东西：WebSocket 地址、鉴权参数、请求消息格式。
多数服务采用 wss:// 开头，鉴权放在 URL 查询参数或首条消息里。
不要凭记忆拼地址，以控制台或接口文档为准。
一个可参考的连接形式如下（参数名以实际文档为准）：
wss://api.example.com/v1/realtime?model=your-model
Authorization: Bearer <你的API_KEY>如果文档要求把密钥放在首条 JSON 消息中，就不要同时写进请求头，避免重复鉴权。
用 Python 建立连接并发送首条消息
下面示例使用 websockets 库，安装命令：
pip install websockets最小连接与发送示例：
import asyncio, json, websockets
URL = "wss://api.example.com/v1/realtime?model=your-model"
HEADERS = {"Authorization": "Bearer <你的API_KEY>"}
async def main():
async with websockets.connect(URL, extra_headers=HEADERS) as ws:
await ws.send(json.dumps({
"type": "session.update",
"session": {"instructions": "你是一个简洁的助手"}
}))
await ws.send(json.dumps({
"type": "conversation.item.create",
"item": {"type": "message", "role": "user",
"content": [{"type": "input_text", "text": "用一句话解释WebSocket"}]}
}))
await ws.send(json.dumps({"type": "response.create"}))
async for msg in ws:
data = json.loads(msg)
print(data.get("type"), data)
asyncio.run(main())预期结果：连接成功后能收到服务端返回的事件消息，常见类型包括会话创建、内容增量、响应完成。
如果连接阶段就报错，先看下一节的排错顺序。
流式消息怎么读，才不会丢内容
模型返回通常是增量事件，例如 response.output_text.delta 携带 delta 文本片段，最后收到完成事件。
处理时按事件类型分支，不要把每条消息都当成完整回答。
if data.get("type") == "response.output_text.delta":
print(data["delta"], end="", flush=True)
elif data.get("type") == "response.completed":
break注意：增量片段需要自己拼接；
完成事件到达后再落库或返回前端，避免半截内容被当作最终结果。
心跳、超时和重连的配置要点
长连接会被中间代理或防火墙断开，需要保活。websockets 库可设置心跳间隔：
async with websockets.connect(URL, extra_headers=HEADERS,
ping_interval=20, ping_timeout=20) as ws:
...如果服务端要求应用层心跳，按文档发送 {"type":"ping"} 并处理 pong。
重连不要无限快速重试，建议指数退避，例如 1 秒、2 秒、4 秒，并设置上限。
常见报错怎么排查
- 401/403：密钥错误、过期或未放在文档要求的位置。检查请求头和 URL 参数是否重复或缺失。
- 连接超时：服务器出站被安全组或防火墙拦截，确认能访问
wss地址的 443 端口。 - 收到消息但无文本：事件类型判断写错，或请求消息缺少触发响应的字段，对照文档检查
type字段。 - 连接频繁断开：未配置心跳或代理空闲超时，先加
ping_interval，再检查反向代理的读超时设置。
上线前做一次完整验证
验证分三步：先用最小脚本确认能连上并收到完成事件；
再用长回答测试增量拼接是否完整；
最后模拟网络中断，确认重连后不会重复计费或重复写入。不要在生产环境直接用示例密钥，密钥放环境变量，日志里避免打印完整请求头。
如果对接的是语音实时场景，还需要确认音频分片格式和采样率与文档一致，这部分参数错误通常表现为连接正常但识别结果为空。
对接后回答不完整，先查哪里
优先检查是否在收到完成事件前就关闭了连接，其次检查增量拼接逻辑是否覆盖所有文本事件类型。
最后对比日志中的事件序列，确认没有漏掉结束标记。
能否用 HTTP 替代 WebSocket
可以。
如果不需要双向实时推送，HTTP 流式接口维护成本更低。
WebSocket 更适合需要持续双向交互的场景，例如语音对话或需要中途打断生成的业务。
多用户并发时要注意什么
每个用户会话建议独立连接，不要在多个请求间复用同一条连接发送不同会话的消息，否则容易串话。
连接数上来后关注服务端文件描述符上限和内存占用。
