AI Agent 开发:SSE 流式协议——让用户＂看见＂ AI 思考的核心技术

第 13 个文章了，拖更了很久，圆上。上一篇讲了让 AI 自己画图。图能画出来了，下一个问题来了：Agent 思考 30 秒，这 30 秒你打算给用户看什么？

最简单的方案：转一个 loading 圈，等结果出来一次性显示。Web 1.0 的做法。在 ChatGPT 把"打字机效果"做成全民标配的今天，让用户对着 loading 圈发呆 30 秒，等于告诉他你的产品还停留在上个时代。

很多公司谈 Agent，PPT 上 demo 跑得飞起，真上线就被吐槽"卡顿"、"假死"。说实话，问题不在模型，问题是你没有呈现层。

这篇就聊一下"模型能跑"和"产品能用"中间到底隔着什么。重点放在那个真正决定 Agent 体验的协议——SSE（Server-Sent Events）。

先看一张图把全局讲清楚——左边是前端的视觉变化，右边是后端 SDK 在每个时刻发什么 SSE 事件：

后面章节就是把每一步拆开讲——后端怎么产出这些事件、前端怎么解析、踩了哪些坑。

为什么 Agent 不能用传统 HTTP 请求-响应

Agent 一次完整查询要走好几步：解析意图、查数据、选图表、生成洞察、合成回复。每一步都是 LLM 跟 Tool 来回拉锯，整轮 10 到 30 秒是常态。走传统 HTTP，三个问题躲不掉。

问题一：30 秒静默等于产品死亡。用户对 AI 等待时间的容忍度，跟"看不看得到进展"成正比。看得到进展，等 60 秒不烦；看不到进展，5 秒就开始焦虑。

问题二：失败成本极高。请求-响应模式下，30 秒后报错——前面所有 Tool 调用、所有 LLM token 全部白烧。生产环境一天跑几千次，错误率 5% 就是几百美元蒸发。更难受的是用户根本不知道哪一步错了，只能给他一句"出错了，请重试"，他重试还可能再错。

问题三：没办法中断。用户改主意了？切对话了？关页面了？传统模式没办法告诉服务端"别跑了"。Agent 在后端默默烧完最后 10 秒的 Tool 调用，把结果写进一个永远没人看的 buffer——你的钱在烧，价值是零。

OpenAI 在 2024 年开发者大会上说过一句话：流式输出不是锦上添花，是 Agent 产品的基础设施。问题不在 SSE 复杂，而在很多团队把它当成"前端展示问题"，没意识到它是贯穿 Agent 全栈的协议。

SSE 协议：5 种事件覆盖整个生命周期

SSE（Server-Sent Events）是 HTML5 标准的服务端推送协议。先看选型：

WebSocket 是全双工的，听上去更"高级"。但 Agent 的通信模式天然是单向的——用户发一句话，Agent 持续往外吐结果。用户不需要在 Agent 思考过程中"插嘴"。要中断的话，一个 AbortController 够了。

SSE 的实际好处：走标准 HTTP 协议（不需要握手升级，CDN/Nginx/负载均衡都直接处理）；浏览器原生 EventSource 内置断线重连；原生支持 event 字段，一个连接里传不同类型的事件。

OpenAI、Anthropic、DeepSeek 的流式 API 全部用 SSE。这不是巧合——LLM 的输出模式就是"服务端持续推送 token"，SSE 是最自然的匹配。主流 Agent SDK 的底层都是 SSE，区别只在你在哪一层包装它。

我们设计了 5 种事件覆盖 Agent 整个生命周期：

event: systemdata: {"session_id":"31e70ba6-ad75-4e1d-8cbb-d1fdce323e17","model":"deepseek-v4-pro"}event: thinkingdata: {"node":"cube_query","label":"正在查询数据...","status":"running"}event: thinkingdata: {"node":"cube_query","label":"正在查询数据...","status":"done"}event: text_deltadata: {"content":"根据数据分析，华东区"}event: text_deltadata: {"content":"的RTT在过去三个月呈下降趋势"}event: completedata: {"success":true,"chartConfig":{...},"queryData":[...]}

设计的关键是「过程事件」和「结果事件」的分离：过程事件丢了不要紧，最多少显示一个步骤；结果事件不能丢，最终图表必须出来。过程事件用来缓解焦虑，结果事件用来交付价值。

为什么是 5 种不是 10 种？事件越少，前端状态机越简单，bug 越少。我们一开始设计了 11 种事件，跑两周砍到 5 种，bug 数掉了 70%。事件设计的原则就一句：宁可在 data 里塞字段，也不要新增事件类型。一个 complete 携带 chartConfig + queryData + textReply，比拆成 3 个事件简单得多。

服务端：AsyncGenerator + 原生 HTTP 响应

后端我喜欢用 TS 比较原生，任务语言是一样，我这例子用 Koa.js。这里有个反直觉的地方：Koa 默认会缓冲响应体，等所有中间件执行完才发送。SSE 要的是"写一条发一条"，得绕过去。办法是直接操作 Node.js 原生的 res 对象：

router.post('/api/agent/stream', async (ctx) => {  const { query, sessionId } = ctx.request.body;  ctx.respond = false;  // 关键: 告诉 Koa "我自己管响应,你别插手"  const res = ctx.res;  res.writeHead(200, {    'Content-Type': 'text/event-stream',    'Cache-Control': 'no-cache',    'Connection': 'keep-alive',    'X-Accel-Buffering': 'no',  // Nginx 反代必须加这个  });  const abortController = new AbortController();  ctx.req.on('close', () => abortController.abort());  ctx.req.on('error', () => abortController.abort());  // 处理 ECONNRESET  try {    forawait (constevent of runSdkAgent({ query, sessionId, abortSignal: abortController.signal })) {      if (abortController.signal.aborted || res.writableEnded) break;      res.write(`event: ${event.event}\ndata: ${JSON.stringify(event.data)}\n\n`);    }  } finally {    if (!res.writableEnded) res.end();  }});

runSdkAgent 内部是 SDK 的 query()，关键参数：

for await (const raw of query({  prompt: userQuery,  options: {    mcpServers: { 'ops-tools': opsServer },    allowedTools: ['mcp__ops-tools__cube_query', 'mcp__ops-tools__chart_generate'],    includePartialMessages: true,  // 必须开,否则没有 token 级流式    maxTurns: 6,  },})) {  const adapted = adaptSdkMessage(raw, toolIdMap);  if (adapted) yield* transformToSSE(adapted, cache);}

includePartialMessages: true 这一行是流式体验的开关。SDK 默认只发整块 assistant 消息，开了之后才会发 stream_event 类型的细粒度事件，里面 event.type === 'content_block_delta' 携带每个 token。默认关闭其实是合理的——批处理任务、自动化流水线确实不需要 token 流。但只要你做的是有界面的 Agent 产品，这个开关就得打开。

消息适配：从 SDK 格式到 SSE 事件

SDK 的消息流和我们这 5 种 SSE 事件不是一一对应的。先说 SDK 实际产出的几种 message：

适配代码：

function adaptSdkMessage(raw: any, toolIdMap: Map<string, string>): SDKMessage | null {  // 流式 token (需要开 includePartialMessages)  if (raw.type === 'stream_event' && raw.event?.type === 'content_block_delta') {    const delta = raw.event.delta;    if (delta?.type === 'text_delta') {      return { type: 'partial_assistant', content: delta.text };    }    return null;  }  // assistant 消息: 遍历 content blocks 提 tool_use  if (raw.type === 'assistant') {    for (const block of raw.message?.content ?? []) {      if (block.type === 'tool_use') {        // mcp__ops-tools__cube_query → cube_query        const skillName = block.name.split('__').pop();        toolIdMap.set(block.id, skillName);        return { type: 'tool_use', toolName: skillName, toolInput: block.input };      }    }    return null;  }  // user 消息里包 tool_result  if (raw.type === 'user') {    for (const block of raw.message?.content ?? []) {      if (block.type === 'tool_result') {        // 通过 toolIdMap 反查真实名称        const skillName = toolIdMap.get(block.tool_use_id);        return { type: 'tool_result', toolName: skillName };      }    }    return null;  }  if (raw.type === 'result') {    return { type: 'result', content: raw.result };  }  return null;}

注意 toolIdMap 这个 Map。Anthropic Messages API 用 tool_use_id 把 tool_use 和 tool_result 关联起来，tool 名称只在 tool_use 阶段出现。所以服务端必须在 tool_use 时记下 id → name，tool_result 时再查回来——不然前端不知道哪个 thinking 步骤可以打勾。

适配到 SSE 事件时还有个值得加的小设计：把 toolInput 和 toolOutput 也透到前端去。LLM 决策的入参、Tool 真实的返回值，让用户点开 thinking 步骤就能看到。这一步的价值不只是"好看"，是直接回答了用户那个潜在的怀疑——这玩意儿是不是写死的脚本？

Session 降级

Agent 多轮对话靠 Session 串起来。第一轮建 Session，后续在同一个里继续，Agent 才能记住之前聊过什么。SDK 的 Session 有过期时间，用户隔了几小时再发消息，resume 可能直接报 SESSION_NOT_FOUND。处理方式是降级——自动开新 Session：

catch (err) {  if (err.message?.includes('session') || err.message?.includes('resume')) {    console.warn('resume 失败,降级为新 Session');    yield* runSdkAgent({ ...options, sessionId: `new-${Date.now()}` });    return;  }  yield { event: 'error', data: { message: err.message } };}

原则就一条：别让用户看到技术错误。Session 过期是技术问题，不是用户的问题。

前端：fetch + ReadableStream 手动解析

前端不能用浏览器原生的 EventSource——它只支持 GET。Agent 查询要带 query、sessionId、filters，必须 POST，所以用 fetch + ReadableStream 手动解析 SSE。OpenAI 官方 SDK 也是这么干的：

async function queryAgentWithStream(request, callbacks) {  const response = await fetch('/api/agent/stream', {    method: 'POST',    headers: { 'Content-Type': 'application/json' },    body: JSON.stringify({ query: request.query, sessionId }),  });  const reader = response.body.getReader();  const decoder = new TextDecoder();  let buffer = '';  let accumulatedText = '';  while (true) {    const { done, value } = await reader.read();    if (done) break;    buffer += decoder.decode(value, { stream: true });    // SSE 格式: 每个事件以双换行分隔    const lines = buffer.split('\n');    buffer = lines.pop() || '';  // 最后一行可能不完整,留到下次    let currentEvent = '';    for (const line of lines) {      if (line.startsWith('event: ')) {        currentEvent = line.slice(7);      } else if (line.startsWith('data: ') && currentEvent) {        const data = JSON.parse(line.slice(6));        if (currentEvent === 'thinking') {          callbacks.onThinkingStep(data);        } else if (currentEvent === 'text_delta') {          accumulatedText += data.content;        } else if (currentEvent === 'complete') {          await new Promise(r => setTimeout(r, 50));  // 让 thinking 先渲染          callbacks.onComplete({ ...data, textReply: accumulatedText });        }        currentEvent = '';      }    }  }}

这段代码最值得说的是 buffer 累积。reader.read() 返回的 chunk 不保证是完整的 SSE 消息——一个携带几百条数据的 complete 事件可能被 TCP 拆成多个 chunk，直接 JSON.parse 必崩。所以 buffer 累积、按换行符切分、最后一行留到下次。

complete 回调里那个 setTimeout(r, 50) 是为了避开和最后一个 thinking: done 的渲染竞态——下面会展开讲。

前端拿到事件流之后，状态管理上有一个不算技术但很重要的小设计：占位消息。发查询的瞬间先在消息列表插一条 status: 'loading' 的 assistant 消息，显示"正在分析您的问题..."，让 thinking 步骤有地方挂着展示。Agent 完成后把这条消息替换成真实回复。这样用户从发出查询的第一秒到最后图表弹出，视觉上是连续的，不会出现"列表突然多一条消息"的跳变。

三个真正消耗时间的坑

前面在代码注释里提过的（ECONNRESET 监听、chunk 累积），读 MDN 和 Node 文档能避免，这里不重复。下面三个是"读文档读不到、要把整条管道串起来才会撞的"。

坑一：Koa 缓冲吞事件

最早我用 ctx.body 写 SSE 事件，结果 Koa 把所有输出缓冲住，等中间件链跑完才发。用户看到的是：等了 20 秒，所有 thinking 步骤和最终结果同时蹦出来。

修复：ctx.respond = false + res.writeHead + res.write，绕开 Koa 直接操作原生响应。

ctx.respond = false 在 Koa 文档里只有一句话："To bypass Koa's built-in response handling"。一句话讲完一个 API，但你不知道它存在，就永远绕不过去。

坑二：Nginx 二次缓冲

本地开发好好的，部署到生产 SSE 又不实时了——一摸一样的症状，原因换了：本地是 Koa，生产是 Nginx。

Nginx 反代默认开 proxy_buffering，等响应"差不多了"才往下游推。SSE 这种永远不会"差不多"的长连接，被它一直憋到连接关闭。

修复：响应头加 X-Accel-Buffering: no。Nginx 的私有响应头，专门用来让后端"通知"它别缓冲。代码里加一行就完事，前提是你得知道有这么个头。

坑三：complete 和最后一个 thinking: done 渲染打架

服务端可能在同一个 TCP chunk 里先发 thinking: done 再发 complete。两个事件几乎同时到前端，立即处理 complete 的话——thinking 步骤的 DOM 更新还没被浏览器画到屏幕上。

效果就是：图表"咔"地弹出来的瞬间，最后一个 thinking 步骤还在转圈，下一帧才变成对勾。视觉上像闪了一下。

修复：complete 回调延迟 50ms 执行，让浏览器先把 thinking 那一帧画完。这种坑没法靠读文档避免——是事件序列化、浏览器渲染调度、DOM diff 三者凑一块的微妙时序，得真有用户用了之后才能注意到。

写在最后

很多人谈 Agent，PPT 上 demo 跑得飞起，真上线就死掉了。我做了一年下来，越来越确信：很多 Agent 项目死，不是因为模型不行，是因为他们没把呈现层当核心技术看。

回头看一个能上线的 Agent，技术栈大概长这样：

Agent 产品 = LLM 推理引擎            + Tool 执行层            + 流式呈现协议  ← 这一层最容易被忽视            + 前端状态机

前两层大家都在卷，第三、第四层很多团队是用周末时间随便写的。结果就是模型能力越强，呈现层越成为短板——LLM 越聪明，思考时间越长，30 秒静默的折磨就越严重。

SSE 这套协议放在 Agent 上特别贴合，是因为 Agent 的通信模式天然就是单向流。从这个协议派生出去的所有设计——5 种事件、过程/结果分离、占位消息、Session 降级、toolInput/toolOutput 透传——都是同一个目标的不同侧面：让用户看到 AI 在工作，并且相信它真的在工作。

最后一个不算技术的观察：用户对 AI 的信任不来自"模型有多大"，来自"看得见在做什么"。把 LLM 真实的入参、Tool 真实的出参透传到前端，让用户点开就能看到——这是从"AI 噱头"到"AI 产品"的分水岭。

另外，如果大家要 demo 可以联系我。

下一篇是这个系列的下一篇，讲怎么把这套 Agent 装进 Electron 桌面应用——让它从"浏览器里的网页"，变成"用户系统里常驻的同事"。