GPT-Live 的目标不是让传统聊天机器人多一个语音输入框，而是支持持续、自然的语音交互。它把“等用户说完再回答”的固定轮次，改造成能够持续接收声音、增量生成响应并处理打断的实时系统。要让这种体验成立，模型能力和传输架构必须同时满足低延迟要求。
从轮次对话转向连续事件流
传统语音助手通常按顺序执行一条完整流水线：
- 录制一段音频。
- 检测用户是否停止说话。
- 将整段音频交给语音识别。
- 把文本发送给语言模型。
- 等待完整回复，再执行语音合成。
这套架构容易实现，但每个阶段都会累积等待时间。更关键的是，它预设了清晰的“你说完，我再说”边界。真实对话并不总是如此：用户可能停顿、补充、改口，也可能在 AI 回答到一半时打断。
来源摘要提到 GPT-Live 使用 turnless speech model，也就是不依赖固定轮次边界的语音模型。对工程系统而言，这意味着输入和输出应被设计成持续事件流：音频片段不断进入，模型状态持续更新，响应也以小块形式尽快返回。
事件可以抽象为：
音频片段 -> 输入缓冲区 -> 实时模型 -> 文本或音频增量
\-> 打断事件 -------------> 取消当前生成
这里的重点不是删除所有边界，而是让边界从硬编码流程变成可更新的运行时判断。系统仍然需要判断停顿、打断和会话结束，只是不必为了等待一个绝对的“说完”信号而阻塞整条链路。
低延迟不是单个指标
语音 AI 的响应速度至少包含几种不同延迟：
- 上行延迟：麦克风采集到服务端收到音频片段所需的时间。
- 首响应延迟：服务端收到足够上下文后，到返回第一个可播放片段的时间。
- 持续生成延迟：后续音频块能否稳定到达，避免播放缓冲耗尽。
- 打断延迟：用户开始插话后，系统多久停止旧响应并转向新输入。
只优化总生成时间并不够。即使完整回答需要数秒，只要首个响应及时到达、后续输出稳定，交互仍然可能显得灵敏。相反，如果服务端一次性返回完整音频，模型再快也会被串行等待拖慢。
可以把端到端预算拆开记录，而不是只统计一次请求耗时：
capture_ms + upload_ms + queue_ms + model_first_chunk_ms
+ synth_first_chunk_ms + download_ms + playback_buffer_ms
生产系统还应按分位数观察这些指标。平均延迟可能很好看，但语音交互中的少量 P95 或 P99 卡顿会直接表现为沉默、抢话或断续播放。
可以这样实践：用 WebSocket 模拟可打断的流式会话
来源摘要没有披露 GPT-Live 的具体协议和内部实现。下面是一个可以直接运行的最小示例，用 WebSocket 演示三项关键机制：持续上传二进制音频块、增量返回响应，以及用 interrupt 事件取消正在生成的内容。示例中的“模型响应”是模拟数据，需要替换成实际的流式语音模型调用。
安装依赖：
python -m pip install fastapi 'uvicorn[standard]' websockets
创建 server.py：
import asyncio
import contextlib
import json
import time
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
async def stream_mock_response(ws: WebSocket, generation_id: int) -> None:
words = ["我", "已经", "收到", "你的", "实时", "语音", "输入"]
for index, word in enumerate(words):
await asyncio.sleep(0.08)
await ws.send_json({
"type": "response.delta",
"generation_id": generation_id,
"index": index,
"text": word,
"server_time_ms": int(time.time() * 1000),
})
await ws.send_json({
"type": "response.completed",
"generation_id": generation_id,
})
@app.websocket("/live")
async def live(ws: WebSocket) -> None:
await ws.accept()
generation_id = 0
response_task = None
received_bytes = 0
try:
while True:
message = await ws.receive()
if message.get("bytes") is not None:
received_bytes += len(message["bytes"])
await ws.send_json({
"type": "input.ack",
"received_bytes": received_bytes,
})
continue
if message.get("text") is None:
continue
event = json.loads(message["text"])
event_type = event.get("type")
if event_type == "respond":
if response_task and not response_task.done():
response_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await response_task
generation_id += 1
response_task = asyncio.create_task(
stream_mock_response(ws, generation_id)
)
elif event_type == "interrupt":
if response_task and not response_task.done():
response_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await response_task
await ws.send_json({
"type": "response.interrupted",
"generation_id": generation_id,
})
except WebSocketDisconnect:
if response_task and not response_task.done():
response_task.cancel()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
创建 client.py：
import asyncio
import json
import time
import websockets
async def receive_events(ws, started_at: float) -> None:
async for raw in ws:
event = json.loads(raw)
elapsed_ms = int((time.perf_counter() - started_at) * 1000)
print(f"{elapsed_ms:4d} ms {event}")
async def main() -> None:
uri = "ws://127.0.0.1:8000/live"
async with websockets.connect(uri, max_size=None) as ws:
started_at = time.perf_counter()
receiver = asyncio.create_task(receive_events(ws, started_at))
# 模拟连续上传 20 ms 音频帧；内容只是占位字节。
for _ in range(10):
await ws.send(bytes(640))
await asyncio.sleep(0.02)
await ws.send(json.dumps({"type": "respond"}))
await asyncio.sleep(0.25)
# 模拟用户在 AI 输出期间插话。
await ws.send(json.dumps({"type": "interrupt"}))
await asyncio.sleep(0.1)
receiver.cancel()
if __name__ == "__main__":
asyncio.run(main())
分别启动服务端和客户端：
python server.py
python client.py
接入真实模型时，需要替换 stream_mock_response()，并明确处理音频格式、采样率、鉴权、背压和断线重连。客户端发送的空字节不是有效音频，仅用于验证流式协议和取消语义。
真正困难的是会话控制
连续语音系统不能只靠一条长期 WebSocket 连接。工程复杂度集中在状态和并发控制上：
- 每次生成都应带有单调递增的
generation_id，客户端必须丢弃已取消生成的迟到数据。 - 打断要贯穿整个链路，包括模型推理、语音合成、网络发送和本地播放队列。
- 输入速度超过处理速度时，需要背压或有限缓冲，不能无限积压音频。
- 重连后要决定恢复会话、重放有限上下文，还是建立新会话。
- 服务端应设置会话时长、空闲时间、消息大小和并发生成上限。
还要区分“用户短暂停顿”和“用户已经结束表达”。turnless 不代表完全不做语音活动检测，而是不要让单一的静音阈值成为系统唯一的控制器。可以把声学信号、模型状态和客户端事件结合起来，允许判断结果在新音频到达后被修正。
上线前的取舍清单
六个月构建出响应迅速的实时语音 AI，反映的不是某一个组件足够快，而是模型、协议和播放控制形成了闭环。团队采用类似架构时，可以重点检查：
- 是否测量首音频包、持续输出和打断延迟，而不只是完整请求耗时。
- 是否支持取消，并能阻止旧响应在取消后继续播放。
- 是否用小块流式传输，同时避免过小数据包带来的调度和协议开销。
- 是否为网络抖动设置有限播放缓冲，而不是追求绝对零缓冲。
- 是否记录原始音频的隐私边界、保留期限和访问权限。
- 是否在噪声、重叠说话、长停顿和弱网条件下做端到端测试。
低延迟与稳定性之间存在直接取舍：缓冲越小，首响应越快，但越容易受到网络抖动影响；上下文越长，回答可能越连贯，但推理和传输成本也会增加。实际落地时，应先定义可感知的延迟预算和中断行为，再选择模型与基础设施，而不是让组件指标替用户体验做决定。