Agent 打字机是怎么来的：SSE 与 WebSocket 打通实时响应与中间状态

Agent 一跑就是几十秒，前端却只有一个转圈的 loading，用户不知道模型是在思考、在调用工具，还是已经卡死了——黑盒等待几乎是所有 Agent 产品的第一道体验硬伤。

常规做法是轮询接口或者等最终结果一次性返回，前者要自己写重试与状态合并，还多浪费请求；后者把思考过程、工具调用这些真正体现价值的中间环节全部丢掉，出了问题也分不清是模型慢还是链路断。

本文分享一套可以直接落地的方案：事件协议 + SSE 流式推送 + WebSocket 双向控制 + 前端状态机，从后端如何吐事件讲到前端如何展示中间状态，可直接落地。

一、痛点：黑盒等待让用户以为产品卡死

先把 Agent 一次长任务的时间线摊开，看看用户眼里它到底是什么样子：

• 整体返回的假象：一次性接口要等整段生成才响应，用户盯着转圈十几秒，最常见的动作就是直接关掉页面。
• 中间过程全部丢失：思考、检索、工具调用这些体现产品价值的环节，最终聚合结果里一句都看不到。
• 排障只能靠猜：前端只有成功或失败两个信号，出问题时分不清是模型慢、网关断还是渲染卡住。

**核心结论：**把「等待」变成「过程」，是 Agent 产品体验的第一道分水岭。

二、选型：SSE 与 WebSocket 的适用边界

两条通道都能把服务端数据推给浏览器，但脾气完全不同，先按场景对号入座：

• 单向吐字优先 SSE：模型输出本质是一条单向流，SSE 复用现有 HTTP 基础设施，还自带重连。
• 需要上行才用 WebSocket：只有当客户端要中途发指令时，引入 WS 的复杂度才划算。
• 混合方案最省心：出流走 SSE，控制命令另开一个轻量接口，比全量改造 WS 便宜得多。

**核心结论：**默认用 SSE，只有真正需要客户端上行时才升级 WebSocket。

三、契约：先钉死事件类型再动手写代码

流式接口最容易翻车的不是传输，而是前后端对事件的约定，先把协议写在纸上：

• 一条流多种事件：token 只管文字，status 只管阶段，tool_result 只管产物，职责互不串台。
• 每个事件带 id：断线重连时用 Last-Event-ID 续传，前端按 id 去重避免重复渲染。

**核心结论：**协议先于代码，事件类型定不清楚，后面全是返工。

四、后端：FastAPI 逐条吐出 token 与阶段事件

协议定完就写服务端，下面这段代码可直接跑在 FastAPI 环境里，模拟 Agent 的逐字输出与工具调用：

import asyncio, json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def sse(event: str, data: dict) –> str:
# 按 SSE 规范拼一条事件，结尾必须是两个换行
return f"event: {event}\\ndata: {json.dumps(data, ensure_ascii=False)}\\n\\n"

async def agent_stream(prompt: str):
yield sse("status", {"stage": "thinking"})
for ch in "我先查资料，再分点回答":
await asyncio.sleep(0.05)
yield sse("token", {"text": ch})
yield sse("status", {"stage": "tool_call", "name": "search"})
yield sse("tool_result", {"name": "search", "rows": 5})
yield sse("done", {"tokens": 312})

@app.get("/chat")
async def chat(prompt: str):
return StreamingResponse(
agent_stream(prompt),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)

• 事件分行发送：每条事件以空行结尾，浏览器才能识别为一条完整消息立即触发回调。
• 顺手关掉网关缓冲：X-Accel-Buffering: no 让 Nginx 直接放行，省一次踩坑。

异步生成器逐条 yield → SSE 规范格式 → 浏览器按事件名分发

五、前端：EventSource 驱动渲染状态机

服务端会吐了，前端要把它还原成打字机效果和阶段徽标：

const state = { answer: "", stage: "thinking", tools: [] };
const es = new EventSource("/chat?prompt=" + encodeURIComponent(q));

["token", "status", "tool_result", "done"].forEach((name) => {
es.addEventListener(name, (e) => {
const d = JSON.parse(e.data);
if (name === "token") state.answer += d.text;
if (name === "status") state.stage = d.stage;
if (name === "tool_result") state.tools.push(d);
if (name === "done") es.close();
render(state); // 只做最小更新，避免整段重排
});
});
es.onerror = () => (state.stage = "reconnecting");

• 按事件名分发：一个监听器管一类事件，状态机只维护 answer / stage / tools 三个字段。
• 注意方法限制：EventSource 只支持 GET，需要 POST 时改用 fetch 配合 ReadableStream 手动解析。

事件驱动渲染 → 单一状态源 → DOM 最小更新

六、升级：WebSocket 承载可中途干预的会话

当用户会随时喊停、追问或注入新指令时，SSE 的单向车道就不够用了：

@app.websocket("/ws")
async def ws_channel(ws: WebSocket):
await ws.accept()
running = None
while True:
cmd = await ws.receive_json()
if cmd["type"] == "ask":
# run_agent 内部按 token 调用 ws.send_json 推送事件
running = asyncio.create_task(run_agent(cmd["prompt"], ws))
elif cmd["type"] == "cancel" and running:
running.cancel()
await ws.send_json({"event": "status", "stage": "cancelled"})
elif cmd["type"] == "ping":
await ws.send_json({"event": "pong"})

• 下行复用同一套事件：WS 帧里仍然发 token / status / done，前端消费逻辑一行都不用改。
• 上行只承载控制指令：ask / cancel / ping 三类消息就覆盖了绝大多数交互需求。

下行流式 + 上行控制，才是真正闭环的 Agent 会话

七、中间状态：把思考与工具调用做成可视进度

拉开产品差距的不是文字本身，而是让用户看见 Agent 此刻在干什么：

• 阶段徽标：thinking / searching / writing 三态用不同色点提示，用户一眼知道卡在哪一步。
• 工具卡片：tool_result 折叠展示入参与返回行数，比一句「已为你查询」更有说服力。
• 思考可展开：推理片段默认收起，展开后能看到完整链路，也方便事后复盘。

中间状态不是装饰，它是用户建立信任的唯一证据。

八、可靠性：断线重连与事件续传排错清单

流式链路一拉长，断连、重复、丢事件都会找上门，按这张表逐一排查：

• 心跳保活：每 15 秒发一条 : ping 注释帧，既不触发渲染也不算业务事件。
• 断点续传：重连请求带上 Last-Event-ID，服务端从该 id 之后补发，前端按 id 去重。

**核心结论：**排流式问题先确认数据有没有出服务器，再看前端有没有正确解析。

九、性能：缓冲、背压与反代的三处坑

流量一上来最先暴露的就是缓冲和背压，三处配置决定它是不是真的实时：

• 网关缓冲：默认 proxy_buffering on 会把 token 攒成整块再下发，用户看到的就是「卡一下然后刷屏」。
• 背压控制：生成快于消费时用有界队列，超限就合并旧 token，而不是无限堆积吃内存。
• 压缩干扰：对 /chat 关闭 gzip，避免压缩把逐条事件粘在一起，破坏逐字节奏。

下面这段 Nginx 配置适配最常见的反向代理部署环境：

location /chat {
proxy_pass http://agent_backend;
proxy_http_version 1.1;
proxy_buffering off;
proxy_read_timeout 300s;
gzip off;
}

关缓冲、限背压、关压缩，三件事做完流式才真的「流」起来

十、可观测：给流式链路装上监控探针

上线后总要回答用户为什么中途离开，先把这几类指标埋进链路里：

• 首 token 延迟：从请求发出到第一个字出现的时间，是留存的生死线，务必按路由分位数统计。
• 流中断率：done 之前连接被关闭的比例，异常升高通常指向网关超时或缓冲配置回退。
• 阶段耗时分布：thinking 与 tool_call 各占多久，用来定位到底是模型慢还是工具链慢。

**核心结论：**没有指标的流式链路，出问题时只能靠用户反馈告诉你。

结语

整套方案的价值在于把 Agent 的「过程」变成了产品的一部分：后端用事件协议把 token、阶段、工具产物拆开推送，前端用状态机统一消费，中间状态从黑盒变成了可信的进度展示；再叠加断线续传、网关关缓冲和三项核心指标，链路从能跑到跑得稳。

落地时建议按 SSE 优先的顺序推进，先跑通单向流拿到体验收益，再按需引入 WebSocket 的上行控制，改动面最小、验证成本也最低。

先让数据流起来，再让过程被看见，最后让链路可度量。