★ FEATURED ARTICLE 点一下发送页面开始转圈。三秒五秒十秒……它到底是在读文档、检索知识库、调搜索工具还是已经死循环了这个问题我做 LangGraph Agent 项目时感触特别深用同步graph.invoke()跑一个多节点的 Agent你只能拿到最终答案中间那段“思考过程”对用户和前端来说完全是黑盒。LangGraph 的流式输出就是为了把这个黑盒开一个口子。核心是stream_mode、astream、token 级事件机制把节点的思考、工具调用、文本生成逐段吐给前端。这篇文章不打算复述官方文档的 Hello World而是把从后端到前端跑通“思考过程实时可见”的选型逻辑、完整实现和踩坑记录一次讲清楚。如果你正在搭 Agent 应用想让用户看到的不再是转圈菊花而是 Agent 真实的思考轨迹这篇应该能帮你少走弯路。1. 等结果的那个瞬间Agent 应用的体验缺口与流式的本质价值1.1 同步调用的黑盒感graph.invoke()会等整张图跑完才返回。假设你的链路是“意图识别 → 检索文档 → 工具调用 → 汇总答案”第二步检索如果花了 8 秒用户看到的就是 8 秒空白。问题在于这不是“慢”的问题而是“不知道在等什么”的问题。人可以对进度条有耐心但对一块没有任何信息的黑屏没有耐心。开发侧的麻烦更隐蔽。Agent 链路一旦变长哪里慢、哪里卡、工具有没有被反复调用用invoke()只能靠事后读日志猜测。日志虽然能保留最终状态但中间每一步的时序、每次 LLM 调用到底吐了什么内容很难还原出来。我之前做一个内部知识库问答 Agent 时用户反馈“有时候回答特别慢”用调用链日志查了半天没定位到瓶颈后来接上流式输出才发现是某个元数据节点反复触发了同一个文档检索工具单次调用本身不慢但被循环调了 6 次。所以“思考过程可见”不是锦上添花的功能而是 Agent 应用在体验和工程化上的一个基础需求。1.2 思考可见能解决的不只是“等待感”很多人以为流式输出的价值就是让用户早点看到字其实至少有三个层面的收益第一用户信任度。你看到的不是一句干巴巴的“正在生成”而是“正在理解问题”“正在搜索文档”“正在整理答案”。哪怕动作很快用户也能感知到系统在做事情。我做客服机器人时对比过接入思考过程展示后用户对回答耗时的容忍度肉眼可见地提高了因为等待被转化成了“信息接收”。第二开发调试。流式事件天然是一条时间线哪个节点先跑、哪个节点跑了多久、工具调用发生了几次、每次传了什么参数全部按时间顺序暴露。很多在invoke()模式下难以复现的问题在流式日志里一眼就能看出来。这个价值只要做一次 Agent 性能排查就会懂。第三产品差异化。如果你在做面向 C 端用户的 Agent 产品“思考过程”本身就是一种可感知的智能感。深度思考模式、工具调用过程、分步骤展示这些都是可以在产品层面做文章的点而它们全部依赖“服务端愿意把过程吐出来”。1.3 流式不是单一通道而是分层事件流LangGraph 的流式输出最需要先建立的一个认知是它不是“LLM 一边生成一边返回”那么简单而是一套分层的事件流机制。在执行层面LangGraph 把一次 Agent 运行拆成多个节点和边。流的来源至少有三种节点级事件节点开始、节点结束、状态如何变化对应stream_modeupdates能拿到每个节点完成后的增量状态。token 级事件LLM 逐 token 生成的内容对应stream_modemessages返回的是消息块chunk。自定义事件你在节点代码里主动写入的业务事件比如“正在调用搜索工具”“工具返回 3 条结果”对应stream_modecustom。把这个分层结构理解透后面才不会一头扎进messages模式里出不来的坑。我见过不少新手只盯着messages一个参数结果工具调用过程、节点跳转信息全都拿不到做出来的“思考过程”就只有一段干巴巴的文本在滚动。2. stream_mode 四种模式的选派逻辑values、updates、messages 怎么用不打架2.1 四种模式的对照LangGraph 的stream_mode可以传字符串也可以传一个组合列表。我先给个参数对照表后面再逐个解释stream_mode返回粒度典型用途是否需要前端逐字渲染values每次 State 更新后的全量状态拿完整状态快照做节点同步不需要updates每个节点完成后返回该节点的增量状态判断“哪一步跑完了”不需要messagesLLM 输出的 token 级消息块逐字渲染思考过程和最终答案需要custom节点内手动写入的自定义事件工具执行过程、业务日志上报按需默认情况下stream_modeupdates。如果你只是简单调用graph.stream(...)不传参你会拿到一个个节点更新对象而不是大模型输出的文字流。很多刚开始接触 LangGraph 的人会疑惑“为什么 stream 了半天看不到 token”多半就是没搞清这一层。values和updates的区别在于全量与增量。前者每次状态变化后都给你一个完整 State后者只给“这个节点改了什么”。对前端渲染来说一般用不到values因为全量状态太重了但如果你在做前端快照回放之类的功能values会很方便。2.2 messages 模式下 chunk 的特殊性为什么很多人第一次用stream_modemessages时会被绕晕因为它返回的东西不是纯文本而是AIMessageChunk。AIMessageChunk是一个消息块对象.content是当前累积的一段文本可以直接拼接。但它在流式过程中还有别的字段比如带工具调用信息时会有tool_call_chunks里面是一段段的工具调用参数增量。如果你把AIMessageChunk当成普通字符串去文本拼接没问题但工具调用信息就会拼出错。还有一个关键点每条消息块通常带有元数据里面包含langgraph_node也就是“这段 token 是从哪个节点流出来的”。这是做“思考过程”和“最终答案”分区展示的关键线索。不同版本里这个元数据可能是 chunk 的属性也可能作为流返回元组的第二项返回具体看你的依赖版本但思路一致拿到节点名再决定这段内容渲染到哪个区域。2.3 用 astream 同时拿多种模式的流实际项目里我几乎不会只用单一模式而是用组合模式。最常用的组合是[messages, custom]一个收 token 文本一个收工具和节点事件互不干扰。async for item in graph.astream( {messages: [(user, question)]}, stream_mode[messages, custom], config{configurable: {thread_id: demo-1}} ): if item[0] messages: message_chunk, metadata item[1] # 从 metadata 里拿节点名 node_id metadata.get(langgraph_node, unknown) text message_chunk.content # 这里把 text 和 node_id 一起推给前端 yield fdata: {json.dumps({type: token, node: node_id, content: text})}\n\n elif item[0] custom: event item[1] # 自定义事件比如工具 start/end yield fdata: {json.dumps({type: event, ...event})}\n\n这里有个使用习惯async for返回的item[0]是模式名item[1]是该模式对应的事件。用组合模式时你不需要关心内部到底是先吐 token 还是先吐事件只需要按模式分发处理即可。2.4 模式选择要跟着产品需求走模式选型的核心逻辑一句话前端需要逐字看到的东西用 messages需要感知流程变化的东西用 custom需要状态快照的东西用 values需要判断节点完成情况的东西用 updates。我见过一个产品需求只是“让用户看到搜索过程”结果后端把[messages]全量流开了。问题在于Agent 图里可能存在多个 LLM 节点messages模式会把所有节点的文本全吐出来包括内部的一些中间提示词前端一不小心就把半成品内容渲染出去了。我的习惯是外部用户界面默认只开custom事件 最终答案的完整文本把消息的 token 级流关掉Debug 后台再开[messages, custom]全量事件方便排查。这个习惯避免了很多次“把内部提示词暴露给用户”的尴尬。3. 最小可复现链路从 LangGraph 后端到浏览器逐字渲染3.1 一个带思考阶段的最简图先说清思路为了让“思考过程”和“最终答案”能在前端分开展示我把 Agent 拆成think节点和answer节点。think节点负责让模型先做一轮推理和规划answer节点基于思考结果做最终回答。实际业务里可能不需要这么刻意地拆但拆开有两个好处一是流式事件天然带着节点名前端可以按节点名分类渲染二是后续在思考阶段插入检索、工具调用等步骤时不用改前端协议。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] # 用你自己实际使用的模型服务这里只是占位 model ChatOpenAI(modelyour-model-id, temperature0.7) def think_node(state: AgentState): # 这里让模型先输出一段“思考”实际项目中会加入检索或工具调用 response model.invoke(state[messages]) return {messages: [response]} def answer_node(state: AgentState): # 基于思考结果做最终汇总 response model.invoke(state[messages] [{ role: system, content: 基于已有信息给出最终回答不需要再输出思考内容。 }]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(think, think_node) graph.add_node(answer, answer_node) graph.add_edge(START, think) graph.add_edge(think, answer) graph.add_edge(answer, END) compiled graph.compile()这个图很简单但已经具备了一个可流式的多节点骨架。如果你实际项目里只有一个 Agent 节点也没关系——把think的过程理解成“节点内部多次 LLM 调用”即可流的消费方式不变。3.2 用 SSE 把事件推给浏览器后端我用 FastAPI 的StreamingResponse出 SSE 格式。SSE 本质上就是text/event-stream的 HTTP 响应每次用data:开头推一行 JSON再跟一个空行分隔事件。选择 SSE 而不是 WebSocket是因为这个场景是“服务端单向推送”用户的输入在建立连接时已经通过 POST 传给服务端了后续基本不需要客户端再主动发消息。SSE 更简单也更容易被代理和日志中间件处理。import json from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() app.post(/api/chat) async def chat(request: Request): body await request.json() question body.get(input, ) async def event_generator(): async for item in compiled.astream( {messages: [HumanMessage(contentquestion)]}, stream_mode[messages, custom], ): if item[0] messages: message_chunk, metadata item[1] node_id metadata.get(langgraph_node, unknown) text message_chunk.content if text: event { type: token, node: node_id, content: text, seq: next_seq(), # 下面会解释序号 ts: current_time_ms(), } yield fdata: {json.dumps(event)}\n\n elif item[0] custom: event_data { type: event, **item[1], seq: next_seq(), ts: current_time_ms(), } yield fdata: {json.dumps(event_data)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这段代码里我加了seq和ts两个字段看起来是小事但后面避坑章节会专门讲没有序号的事件流在网络抖动时前端几乎无法恢复顺序也几乎没有断点续传能力。3.3 前端消费fetch 流式读取与逐字渲染前端不能直接用EventSource因为通常我们需要用 POST 发用户输入而浏览器原生EventSource只支持 GET。实际方案是fetchReadableStream手动读流按 SSE 的空行分隔符切事件。async function chatStream(input, handlers) { const res await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ input }), }); if (!res.ok || !res.body) { throw new Error(HTTP ${res.status}); } const reader res.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const events buffer.split(\n\n); // 最后一段可能是半截事件留到下一轮 buffer events.pop() || ; for (const raw of events) { const dataLine raw.split(\n).find(line line.startsWith(data: )); if (!dataLine) continue; const event JSON.parse(dataLine.slice(6)); handlers.onEvent?.(event); } } }拿到事件后渲染逻辑按node字段分类think节点的 token 追加到“思考过程区”answer节点的 token 追加到“最终答案区”。这样用户能看到两段内容同时推进而不是混在一起滚动。3.4 一份可用的事件协议长什么样事件协议这件事很多人是踩了坑才重视的。我现在的协议固定长这样{ type: token, node: think, content: 我先理解一下你的问题, seq: 12, ts: 1710000000123 }type区分事件类型node告诉前端这段内容属于哪个阶段content是实际内容seq让前端有能力排序和检测丢包ts用于调试和性能统计。有些团队会再加一个event_id用于幂等处理也不错。核心原则是协议里要带上“你是谁、你属于哪一步、你排在第几个”这三个信息。没有这三个字段的流式协议在并发节点加入后一定会乱。4. 工具调用链路的流式打通当 Agent 开始“动手”时4.1 工具调用的增量语义不能按字符串拼接文本流式是一行行拼工具调用流式是一段段“参数增量”完全不是一回事。模型决定要调用某个工具时流式 chunk 里会携带tool_call_chunks每一段只包含工具调用的一部分信息可能是id可能是函数名也可能是argsJSON 字符串的一个片段。如果你简单地做previous_args new_args那得到的结果在大多数情况下是能拼出一个完整 JSON 的因为模型通常从左到右生成参数。但问题在于你可能同时收到多个工具调用的片段而且这些片段是交错着来的——你不能假定chunk.tool_call_chunks[0]和chunk.tool_call_chunks[1]属于同一次调用。我的处理方式是按index分组累积def merge_tool_call_chunks(tool_call_chunks): collected {} for chunk in tool_call_chunks: index chunk.index # 这是第几个工具调用 entry collected.setdefault(index, {id: None, name: None, args: }) if chunk.id: entry[id] chunk.id if chunk.name: entry[name] chunk.name if chunk.args: entry[args] chunk.args return [collected[i] for i in sorted(collected.keys())]这段逻辑看起来简单但如果你不做index分组只按“每次收到的第一个 chunk 是新增工具调用”来处理很快会遇到两个工具同时调用时参数互相串的 bug。我之前写过一版上线后客户反馈“为什么工具参数里出现了搜索关键词和用户名贴在一起”查了半天就是这里拼错了。另一个细节是最终消息对象里会有一个聚合好的tool_calls字段你不需要自己拼。流式场景下你拼增量只是为了“实时展示给用户看”真正传给下一个节点执行工具时用 LangGraph 最终状态里的tool_calls或者聚合后的消息即可千万别拿自己前端拼的字符串去反序列化容易出事。4.2 用自定义事件上报工具执行过程工具执行过程中的状态变化用自定义事件上报最合适。文本流只负责“模型在想什么”工具流负责“系统在干什么”。两者通过事件类型区分视觉上也能做不同的 UI。在节点内部可以通过get_stream_writer()或adx.stream()写入自定义事件。核心思路是往stream_modecustom通道里塞数据from langgraph.config import get_stream_writer # 不同版本导入路径略有差异本质都是往 custom 通道写事件 def tool_orchestrator(state: AgentState): writer get_stream_writer() # 工具开始 writer({ event_type: tool_start, tool_name: document_search, query: state[question], }) result document_search.invoke(state[question]) # 工具结束附一个摘要不要把完整结果全推给前端 writer({ event_type: tool_end, tool_name: document_search, summary: f找到 {len(result)} 个候选片段, }) return {search_result: result}前端收到tool_start事件时可以在界面上渲染一个“正在搜索文档”的卡片收到tool_end后再把卡片状态更新成“找到 3 个候选片段”。整个过程不需要把检索结果全文发到前端既保护了内部数据也减少了渲染压力。有一点要特别注意自定义事件一定不要塞太多内容。它是给前端看的“状态信号”不是给用户读的“长文”。有些开发者图省事把整个中间结果 JSON 都塞进事件里推出去结果前端渲染表格卡顿网络流量也飙高。我的习惯是只推关键字段和摘要完整数据留在服务端状态里。4.3 别把完整思考链一股脑推给用户这个点很现实也容易被忽略模型产生的“思考过程”并不等于适合给用户看的“思考过程”。很早之前我把messages流的前端区域做成一个类似聊天窗口的滚动区把模型的所有中间输出都直接渲染。结果用户看到的是大段大段的推理草稿、自我纠正、甚至放弃上一个方案重来的过程体验很碎反而让用户觉得 Agent “不太聪明”。后来我改成了摘录式展示服务端对思考阶段的事件做归纳只推“正在理解问题”“正在检索相关信息”“正在组织答案”这样的轻量状态在 Debug 模式下才把完整 token 流推出来。介于两者之间也可以把思考阶段往前端推但只保留前 200 字作为“思考摘要”超过就折叠。这个取舍的度取决于你的产品定位如果卖点是“深度思考过程”那可以适当保留思考内容如果卖点是“回答质量”思考过程越精简越好让用户把注意力留给最终答案。5. 实测踩坑记录token 丢失、顺序错乱与性能损耗5.1 节点内部手动流式导致 token 丢失流式输出里最容易翻车的一个坑节点内部自己model.stream()了一遍结果外层graph.astream(..., stream_modemessages)反而拿不到完整 token。原因在于 LangGraph 的messages模式依赖回调机制的自动收集。当你在节点内部消费了一个流式事件流却没有让事件继续向外传播时外部捕获到的事件就不完整。更常见的情况是你在节点里手动for chunk in model.stream(...)拼接了一段文本用于判断又用model.invoke()重新调了一次模型导致一次用户提问实际产生了两次 LLM 调用流被稀释成本也翻倍。解决思路两个一是尽量不在节点内部手动流式。如果你只是需要在节点结束时输出一个结果直接model.invoke()外层stream_modemessages会帮你捕获 token 级事件。二是如果真的需要在节点内部做额外处理比如边生成边判断是否提前终止那么用astream_events手动收集并转发或者干脆把逻辑拆成独立的子节点在边上判断条件。我的经验是80% 的“流式丢 token”问题根源都在“内层流式和外层流式抢同一份事件”。让事件的产生和消费各归其位是最有效的规避手段。5.2 思考文本与最终答案乱序如果你在 LangGraph 里使用了并行节点或者你的图里有异步分支很容易出现think节点的 token 还没吐完answer节点的 token 已经先到了。前端如果按接收顺序渲染就会看到“最终答案先出来思考过程还在补”非常滑稽。我当时做的第一个并行实验就踩了这个坑。两个分支同时执行前端同时接收两个节点的 token界面上的“思考区”和“答案区”内容互相穿插。后来加了两层防护第一协议层给每个事件带上node和seq。前端维护一个按node分组的缓冲区同一节点的 token 按seq升序排列节点名称不同则等对应状态更新后再渲染。第二业务层如果能约束就尽量把关键节点设计成顺序执行。并不是所有环节都应该并行。思考阶段和最终回答阶段本身有前后依赖用顺序边把它们串起来既符合业务逻辑也省了前端排序的复杂度。5.3 全量 messages 流的性能开销messages模式很直观但很贵。每生成一个 token就是一个独立事件要经历后端编码、网络传输、前端解码、DOM 渲染。一个 10 秒的回复假设 300 个 token前端就要处理几百个高频事件。在弱网环境下用户会明显感到界面卡顿甚至事件积压。性能账要这样算token 流适合作为“主过程”展示不适合每个节点都开。我现在的生产配置是思考阶段只开custom事件展示工具调用状态、检索结果数量等轻量信息。最终生成阶段开messages流把模型生成最终答案的过程逐字展示给用户。Debug 环境开[messages, custom, updates]全量记录所有事件用于调试和分析。这样既保住了“实时可见”的核心体验又把事件量控制住了。如果你追求极致性能甚至可以只在服务端录制messages事件流生成回放日志前端只接收定时打包的状态快照但那属于另一个量级的需求了。5.4 代理缓冲导致的 SSE 假死最后提一个后端很少会主动踩到、但要上线一定会遇到的坑代理服务器的响应缓冲。SSE 本质是“长连接 不定时刷新”中间如果有一层 Nginx 或网关帮你做了响应缓冲它会等你的响应攒够一定大小再一次性发给前端。结果就是前端迟迟收不到流只有等你整段 LLM 生成完毕、缓冲区满了才一次性吐出一大坨。用户看到的体验跟没做流式几乎一样。解决办法在 SSE 响应上显式关闭缓冲比如在响应头里设置X-Accel-Buffering: no针对部分代理中间件或者在服务端配置里关闭该路由的响应缓冲。同时流式接口建议定时主动发一个空行或心跳事件避免连接空闲被闲时连接检测干掉保证连接稳定可靠。另外前端断线重连后的续传问题也需要考虑seq。连接断开后前端记录已经收到的最大seq重连时把last_seq传给服务端服务端能通过事件序号把缺失的补齐。这个方案不复杂但要求你从一开始就在协议里埋好seq字段临时加是加不出来的。最后分享一个我现在固定的工程习惯只要是用 LangGraph 搭 Agent不管产品最终要不要展示思考过程我都会至少保留一条messages流到调试后台。原因很简单——invoke模式看到的是结果流式模式看到的是过程很多玄学性能问题比如某次生成特别慢、某个工具被重复调用在流里一眼就暴露了。上面这些模式选择、协议设计和踩坑记录也都是从这条习惯里长出来的希望对正在做 Agent 界面或者调试链路的你有参考价值。
阅读完成 · 觉得有帮助？
