1. 从 AGUI 协议说起为什么流式渲染是 AI Agent 交互的命门第一次接触 AGUI 协议这个概念是在给一个内部知识库做 AI 问答助手的时候。当时前端同事问我一句话“大模型的回答能不能像打字机一样一个字一个字蹦出来而不是等十秒钟突然刷出一整段”这个问题看似简单背后却牵扯到一整套从模型输出到前端渲染的链路设计。AGUI全称可以理解为 Agent Graphical User Interface 协议它要解决的核心问题就是AI Agent 在运行过程中产生的中间状态、工具调用结果、文本增量如何以标准化、可流式消费的方式传递给前端并实时渲染。如果你用过 Vercel AI SDK 的useChat或者自己用 SSE 手写过流式接口你会发现一个尴尬的现实文本流式输出早就不是难题真正难的是结构化内容的流式渲染。比如 Agent 调用了一个天气查询工具返回的是 JSON前端想把它渲染成一张卡片Agent 又调用了搜索返回的是一组链接前端想渲染成列表。这些内容如果等整个响应结束再解析用户体验就退化成“转圈十秒然后一次性出现”流式的意义就丢了一半。AGUI 协议的价值就在这里。它定义了一套消息分片规范让文本增量、工具调用参数增量、工具执行结果、甚至 UI 组件描述都能以 chunk 的形式推送到前端前端根据 chunk 的类型决定是追加文本、更新卡片还是插入新组件。这套思路和 Vercel AI SDK 的 Generative UI、LangChain 的 streaming events 是一脉相承的只是 AGUI 更强调“协议”层面的标准化试图让不同 Agent 框架、不同前端框架之间有一个共同的对话语言。这篇文章适合谁看如果你正在做 AI Agent 应用开发前端需要实时展示 Agent 的思考过程、工具调用和最终回答如果你在用 LangChain 或 LangGraph 搭建 Agent但苦于流式输出只能拿到纯文本如果你听说过 Vercel AI SDK 的 Generative UI 但不知道底层怎么实现——那这篇内容应该能帮你把整条链路串起来。我会从协议设计、技术选型、实操实现到踩坑排查完整讲一遍我自己的落地经验。2. AGUI 协议的核心设计与技术选型考量2.1 为什么不能直接用纯文本 SSE很多人第一反应是SSE 推文本不就行了吗我一开始也这么想直到遇到三个绕不过去的问题。第一个问题是工具调用的中间态无法表达。Agent 在决定调用工具之前会先输出一段“思考”文本然后输出工具名和参数。如果只推文本前端无法区分“这是最终回答”还是“这是即将调用工具的铺垫”。用户看到一段文字以为结束了结果两秒后又冒出一张卡片体验很割裂。第二个问题是结构化数据的渲染时机。工具返回的 JSON 需要被解析成 UI 组件但 JSON 是逐步生成的你不能等完整 JSON 再解析否则又变成阻塞式。AGUI 的做法是把工具结果也拆成 chunk前端维护一个缓冲区边收边尝试解析解析成功就更新组件。第三个问题是多 Agent 协作时的消息归属。LangGraph 支持多节点、多 Agent 协作每个节点都可能产生输出。如果所有输出混在一个文本流里前端根本分不清哪段是哪个 Agent 说的。AGUI 协议里每条消息都带agentId和messageId前端可以按 Agent 分组渲染这对调试和用户体验都很关键。2.2 协议消息类型的设计我在实际项目中把 AGUI 的消息类型收敛成五类这个分类参考了 Vercel AI SDK 的 data stream protocol 和 LangChain 的 event stream但做了简化更适合中小团队快速落地。消息类型用途关键字段text-delta文本增量delta、messageIdtool-call-start工具调用开始toolName、toolCallIdtool-call-delta工具参数增量argsDelta、toolCallIdtool-result工具执行结果result、toolCallIdui-componentUI 组件描述componentType、props这个设计的好处是前端可以用一个switch就处理完所有情况不需要为每种框架写不同的解析逻辑。tool-call-delta这个类型是我踩坑之后加的——最初只有tool-call-start和tool-result结果参数生成过程完全黑盒用户看不到 Agent 正在准备什么参数体验很差。加上增量之后前端可以实时显示“正在查询北京天气”用户就知道 Agent 在干活。2.3 技术栈选型的取舍后端 Agent 框架我选的是 LangGraph而不是裸 LangChain。原因很直接LangGraph 的astream_events能拿到细粒度的事件流包括on_chat_model_stream、on_tool_start、on_tool_end这些事件天然对应 AGUI 的消息类型。裸 LangChain 的astream只能拿到文本增量工具调用信息要自己从 callback 里捞麻烦很多。前端我选的是 Vercel AI SDK 的useChat做基础但重写了消息解析层。useChat默认只处理文本流我通过它的onResponse钩子拦截原始流按 AGUI 协议解析后再喂给自定义的 state。这样既复用了useChat的连接管理和 abort 能力又能支持结构化渲染。传输层用 SSE 而不是 WebSocket理由是 Agent 场景下主要是服务端单向推送SSE 足够且更简单浏览器原生支持EventSource配合AbortController做中断也很方便。WebSocket 的双向能力在这个场景里用不上反而增加连接管理复杂度。提示如果你的 Agent 需要支持用户中途插话、多轮打断WebSocket 会更合适。但大多数问答式 AgentSSE 加 abort 已经够用。3. 流式渲染组件的实操实现细节3.1 后端从 LangGraph 事件到 AGUI 消息后端核心工作是把 LangGraph 的事件流转成 AGUI 的 SSE 消息。我用 FastAPI 写了一个流式接口关键代码如下from fastapi import FastAPI from fastapi.responses import StreamingResponse from langgraph.graph import StateGraph import json app FastAPI() async def agui_stream(agent, input_data): async for event in agent.astream_events(input_data, versionv2): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield sse_pack({ type: text-delta, messageId: event[run_id], delta: chunk.content }) elif kind on_tool_start: yield sse_pack({ type: tool-call-start, toolCallId: event[run_id], toolName: event[name] }) elif kind on_tool_end: yield sse_pack({ type: tool-result, toolCallId: event[run_id], result: event[data][output] }) def sse_pack(data): return fdata: {json.dumps(data, ensure_asciiFalse)}\n\n app.post(/agent/stream) async def stream(input_data: dict): return StreamingResponse( agui_stream(agent, input_data), media_typetext/event-stream )这里有个细节要注意astream_events的versionv2必须显式指定否则事件结构不一样run_id的层级也会变。我一开始没加结果toolCallId对不上前端渲染工具结果时找不到对应的调用卡片排查了半天。另一个坑是on_chat_model_stream里chunk.content可能是空字符串。某些模型在工具调用前会输出一个空的 content chunk如果不判断直接推前端会收到一堆空 delta虽然不影响渲染但浪费带宽。加个if chunk.content就解决了。3.2 前端消息缓冲与增量解析前端最复杂的部分不是渲染而是增量 JSON 的解析。工具参数是逐步生成的比如{city: 北到{city: 北京}你不能每来一个 delta 就JSON.parse会抛异常。我的做法是维护一个缓冲区每次追加后尝试解析失败就等下一个 delta。class ToolCallBuffer { constructor(toolCallId, toolName) { this.toolCallId toolCallId; this.toolName toolName; this.argsBuffer ; this.parsedArgs null; } append(delta) { this.argsBuffer delta; try { this.parsedArgs JSON.parse(this.argsBuffer); return true; } catch { return false; } } }这个缓冲区配合 React 的 state 更新就能实现“参数边生成边显示”的效果。比如搜索工具的参数里有query字段用户能看到查询词一个字一个字出现心理上感觉 Agent 在实时工作。文本增量的渲染相对简单但要注意批量更新。如果每个 delta 都触发一次setState高频输出时 React 会卡。我的做法是用requestAnimationFrame做节流把 100ms 内的多个 delta 合并成一次更新。实测下来输出速度从每秒 50 个 delta 降到每秒 10 次渲染流畅度明显提升。3.3 中断控制AbortController 的正确用法流式输出必须支持中断否则用户等不及想重新提问时只能刷新页面。Vercel AI SDK 的useChat自带stop()方法但如果你自己解析流就要手动管理AbortController。const controller new AbortController(); fetch(/agent/stream, { method: POST, body: JSON.stringify(input), signal: controller.signal }).then(response { const reader response.body.getReader(); // 读取流... }); // 用户点击停止 function handleStop() { controller.abort(); // 清理缓冲区保留已渲染内容 }这里有个容易忽略的点abort之后后端也要能感知到。FastAPI 的StreamingResponse在客户端断开时会抛asyncio.CancelledError你需要在生成器里捕获它把 LangGraph 的 Agent 执行也停掉否则后端还在跑浪费资源。async def agui_stream(agent, input_data): try: async for event in agent.astream_events(...): yield ... except asyncio.CancelledError: # 客户端断开停止 Agent raise注意LangGraph 的astream_events在取消时不一定能立即停止底层模型调用如果用的是按 token 计费的 API建议在 Agent 层面加一个max_tokens限制避免中断后还在烧钱。4. 常见问题与排查技巧实录4.1 工具结果渲染错位现象前端收到了tool-result但渲染到了错误的卡片上或者干脆找不到对应卡片。排查思路先检查toolCallId是否一致。LangGraph 的on_tool_start和on_tool_end的run_id是同一个但如果你在中间做了消息转发或格式转换很容易把 id 弄丢。我的做法是在后端加日志把每个事件的run_id和name打出来和前端收到的 id 对比。另一个可能是前端缓冲区没清理。如果同一个toolCallId被复用了某些框架会复用旧缓冲区还在新 delta 追加进去就乱了。解决方法是tool-call-start时先检查是否已有同 id 的缓冲区有就先清空。4.2 流式输出卡顿或断流现象文本输出到一半突然不动了或者每隔几秒卡一下。排查思路先看是不是代理层缓冲。Nginx 默认会缓冲 SSE 响应需要加proxy_buffering off;和X-Accel-Buffering: no响应头。我一开始在本地开发没问题部署到服务器就卡查了半天是 Nginx 的锅。如果代理层没问题检查后端是否有同步阻塞操作。比如在生成器里调了一个同步的数据库查询整个事件循环被卡住SSE 就断流了。所有 IO 操作都要用async版本。还有一种情况是模型本身输出慢。某些模型在长文本生成时会有明显的停顿这不是流式的问题是模型推理的特性。可以在前端加一个“正在思考”的动画让用户知道没断。4.3 增量 JSON 解析失败现象工具参数缓冲区一直解析失败卡片显示不出来。排查思路先打印原始argsBuffer看是不是 JSON 格式本身有问题。有些模型输出的 JSON 带 markdown 代码块标记比如json {...}直接JSON.parse会失败。需要在追加前做清洗去掉代码块标记。另一个常见问题是转义字符。模型输出的字符串里可能有\n或\在增量拼接时如果处理不当会破坏 JSON 结构。我的做法是在后端就把参数序列化成字符串再推前端只做拼接和解析不做任何转义处理。问题现象可能原因解决方法工具卡片错位toolCallId 不一致后端日志核对 run_id流式卡顿Nginx 缓冲关闭 proxy_buffering断流同步阻塞操作改用 async IOJSON 解析失败代码块标记或转义后端序列化前端只拼接中断后仍计费Agent 未停止捕获 CancelledError 并终止4.4 多 Agent 场景下的消息归属混乱现象LangGraph 多节点协作时前端分不清哪段输出是哪个 Agent 的。排查思路astream_events的事件里有tags和metadata可以在定义节点时给每个节点打标签比如tags[researcher]然后在事件里读取event[tags]映射成agentId推给前端。前端按agentId分组渲染每个 Agent 一个气泡或一个面板。这个方案我在一个多 Agent 调研项目里用过效果不错。用户能看到“研究员”在搜索、“分析师”在总结整个过程透明信任感也更强。5. 一些实操心得与扩展思路流式渲染这件事做完之后回头看最大的体会是协议设计比技术实现更重要。技术实现无非是 SSE、缓冲区、状态更新但协议设计决定了你的系统能不能扩展。我最初图省事消息类型只有text和tool结果后来想加“引用来源”“思考步骤”“错误重试”这些状态时发现根本塞不进去只能重构。另一个心得是前端要保留完整的消息历史。流式渲染时很多人只维护当前正在输出的消息输出完就丢弃中间状态。但用户可能想回看 Agent 的思考过程或者复制工具调用的参数。我的做法是每条消息都存完整的事件序列渲染时按需回放这样即使刷新页面也能恢复现场。扩展方向上AGUI 协议还可以和 Generative UI 结合得更深。现在我的ui-component消息类型只支持预定义的几种组件未来可以让 Agent 直接输出组件描述前端动态加载。Vercel AI SDK 的streamUI已经在做这件事但依赖 React Server Components对非 Next.js 项目不太友好。如果你用的是 Vue 或 Svelte可能需要自己实现一套类似的机制。最后分享一个小技巧在开发阶段我会在 SSE 消息里加一个debug字段包含事件的时间戳和原始类型。前端加一个开关打开后显示每个 chunk 的到达时间和类型。排查流式问题时这个时间线比任何日志都直观能一眼看出是后端推得慢还是前端渲染卡。这个内容后续还可以这样扩展把 AGUI 协议和 MCP 结合让 Agent 的工具调用不仅流式渲染还能流式发现和注册。MCP 解决的是工具接入标准化AGUI 解决的是交互标准化两者结合就是完整的 Agent 应用协议栈。我目前还在实验阶段等跑通了再单独写一篇。
