门主踩坑最近在项目里做 AI 对话接口返回一次要等十几秒用户盯着空白屏幕发呆体验差到被产品经理追着问能不能像 ChatGPT 那样一个字一个字蹦出来。于是我开始研究 LangChain 的流式输出结果发现——光stream_mode就有 6 种值每种返回的数据结构还不一样。官方文档写得是挺全但对于刚上手的人来说哪一种该在什么场景下用真没说清楚。这篇文章就是我把这些坑都踩了一遍之后的总结。摘要LangChain/LangGraph 的流式输出系统提供了 6 种stream_mode对应不同的数据粒度和使用场景。本文结合真实代码和运行结果逐一拆解values、updates、messages、custom、checkpoints、debug的区别与适用场景同时给出 Agent 完整工作流程图、OpenAI SSE 格式输出方案以及企业级避坑指南。一、为什么会写这篇做 AI 应用开发流式输出不是锦上添花是刚需。大模型推理慢一个完整回复动辄 5~15 秒。如果用户每次都要等到全部生成完才能看到内容那这个产品基本没人愿意用。ChatGPT 之所以体验好核心之一就是逐 Token 流式输出——边生成边显示用户感知的等待时间从十几秒变成了不到一秒。LangChain/LangGraph 在底层帮我们实现了流式传输系统但暴露出来的 API 有多种模式每种模式返回的数据结构、粒度、适用场景完全不同。问题是官方文档把每种模式都讲了但没告诉你什么时候用哪种stream_modeupdates和stream_modevalues返回的东西看起来很像到底有啥区别stream_modemessages怎么实现类似 ChatGPT 的逐字输出stream_modecustom又是什么鬼什么时候需要自定义流式数据多种模式能不能同时用线上禁用流式输出是什么场景这些坑我一个一个踩了。下面把结果整理出来。二、流式输出的使用场景先回答最根本的问题为什么需要流式输出2.1 核心原因降低感知延迟大模型推理是逐 Token 生成的。如果用传统的invoke()同步调用必须等所有 Token 生成完才能拿到完整结果。流式输出利用这一点每生成一个 Token 就立即推送给客户端用户看到的是边想边说而不是想了半天突然冒出一大段。2.2 典型使用场景场景说明推荐模式聊天对话界面类似 ChatGPT 的逐字输出messagesAgent 执行监控实时展示 Agent 在做什么调了哪个工具、拿到什么结果updates调试与排错看 Agent 内部每一步的状态变化values/debug进度条/状态通知工具执行时推送进度如正在查询数据库…custom多 Agent 协作控制哪些 Agent 流式输出、哪些静默禁用流式OpenAI SSE 兼容前端用 OpenAI SDK 接收流式数据messages 手动转 SSE2.3 流式 vs 非流式对比门主提醒流式输出不改变总生成时间只改变用户的感知等待时间。总时间还是那么多但用户体验天差地别。三、Agent 完整工作流程在深入每种stream_mode之前必须先搞清楚 Agent 一次请求到底经历了什么。不然你看流式输出返回的数据会觉得云里雾里。3.1 Agent 执行流程图3.2 对应代码中的阶段以我们项目中的天气查询 Agent 为例from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain.tools import toolfrom langchain.messages import SystemMessage, HumanMessagechanAI ChatOpenAI(modelqwen3.7-plus, temperature0.7)tool(description查询城市天气的工具)def get_weather(city: str): return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])messages [ SystemMessage(content你是一个AI助手叫IT空门), HumanMessage(content今天长沙的天气怎么样)]当用户问今天长沙的天气怎么样Agent 会经历以下阶段阶段对应节点产出内容说明1. 规划modeltool_call: get_weather(city长沙)LLM 决定调用天气工具2. 工具执行tools今天长沙的天气晴,温度为30度执行get_weather工具3. 推理整合model今天长沙的天气是晴天温度30度…LLM 将工具结果整合为自然语言4. 返回-完整回答Agent 结束门主踩坑当时我一直在想为什么流式输出里model节点会出现两次后来才明白——第一次是 LLM 做规划决定调哪个工具第二次是 LLM 做整合把工具结果变成人话。中间夹着tools节点。整个流程是Model → Tool → Model的循环。四、stream_mode 参数详解这是本文的核心。6 种stream_mode一种一种拆开讲。4.1 总览对比stream_mode返回内容数据粒度典型用途values每步执行后的完整状态快照最粗理解 State 变化全貌updates每步执行后的状态增量中等监控 Agent 执行流程messagesLLM Token 元数据最细Token级逐字输出、SSE 推送custom自定义数据自定义工具执行进度通知checkpoints检查点事件中等持久化/回溯debug所有可用信息最全调试排错4.2 stream_mode“values” —— 完整状态快照是什么每执行完一个节点返回整个图的完整状态。返回结构每一步返回的chunk是一个 dictkey 是状态字段名如messagesvalue 是该字段的当前完整值。代码示例for chunk in agent.stream( {messages: messages}, stream_modevalues): for step, data in chunk.items(): print(fstep: {step}) print(fcontent: {data[messages][-1].content_blocks})运行结果step: messagescontent: [{type: text, text: 今天长沙的天气怎么样}] # ← 用户输入step: messagescontent: [{type: tool_call, name: get_weather, args: {city: 长沙}, id: call_50f37520...}] # ← LLM 决定调工具step: messagescontent: [{type: text, text: 今天长沙的天气晴,温度为30度}] # ← 工具返回结果step: messagescontent: [{type: text, text: 今天长沙的天气是晴天温度达到了30度。…}] # ← LLM 最终回答使用场景调试时看 State 的完整变化轨迹理解 LangGraph 状态机是怎么一步步演进的需要当前为止所有信息的场景注意事项每次返回的是完整状态不是增量。如果你只想知道这一步变了什么用updates更高效数据量较大时长对话历史每次返回完整messages列表会有性能开销门主建议如果你只是想在前端展示对话内容别用values用messages。values返回的是整个状态快照信息冗余前端处理起来也麻烦。4.3 stream_mode“updates” —— 状态增量是什么每执行完一个节点只返回该节点新产生的状态更新。与 values 的区别valuesupdates返回内容完整状态只有变化的部分信息量大包含所有历史小只有增量适合看全貌看流程代码示例for chunk in agent.stream( {messages: messages}, stream_modeupdates): print(chunk)运行结果# 第一步model 节点 —— LLM 决定调用工具{model: {messages: [AIMessage(tool_calls[{name: get_weather, args: {city: 长沙}}])]}}# 第二步tools 节点 —— 工具执行返回{tools: {messages: [ToolMessage(content今天长沙的天气晴,温度为30度)]}}# 第三步model 节点 —— LLM 整合最终回答{model: {messages: [AIMessage(content今天长沙的天气是晴天温度为30度。…)]}}使用场景Agent 执行监控面板实时显示正在调用 get_weather…“工具返回结果…”“正在生成回答…”日志记录只记录每一步的变化而不是全量快照理解 Agent 流程清楚看到 Model → Tool → Model 的执行链路门主踩坑当时我用updates做前端实时展示发现返回的 key 是节点名如model、tools不是固定的messages。一开始写成了chunk[messages]直接 KeyError。记住updates返回的 key 是节点名value 才是该节点的状态更新。4.4 stream_mode“messages” —— LLM Token 级流式是什么逐 Token词元流式输出 LLM 的生成内容。这是实现类似 ChatGPT 逐字输出效果的关键模式。返回结构每次迭代返回一个元组(message_chunk, metadata)message_chunk当前 Token 的内容块metadata元数据包含langgraph_node来源节点等信息代码示例for token, metadata in agent.stream( {messages: messages}, stream_modemessages): print(fnode: {metadata[langgraph_node]}) print(fcontent: {token.content_blocks})运行结果关键部分# 第一阶段Model 规划决定调工具node: modelcontent: [{type: tool_call_chunk, name: get_weather, args: , index: 0}]node: modelcontent: [{type: tool_call_chunk, name: None, args: {city: 长沙, index: 0}]node: modelcontent: [{type: tool_call_chunk, name: None, args: }, index: 0}]# 第二阶段Tool 执行 node: toolscontent: [{type: text, text: 今天长沙的天气晴,温度为30度}]# 第三阶段Model 整合最终回答逐Token输出node: modelcontent: [{type: text, text: 今天}]node: modelcontent: [{type: text, text: 长沙的天气是晴天}]node: modelcontent: [{type: text, text: 温度}]node: modelcontent: [{type: text, text: 30度。}]node: modelcontent: [{type: text, text: 天气比较热}]node: modelcontent: [{type: text, text: 出门}]node: modelcontent: [{type: text, text: 记得做好防晒哦}]node: modelcontent: [{type: text, text: }]使用场景前端逐字输出最核心的使用场景实现打字机效果OpenAI SSE 格式兼容转成data: {...}\n\n格式推给前端Token 级过滤只显示某个节点的输出或者排除工具调用的 Token关键细节Tool Call 也是流式的注意第一阶段tool_call_chunk是分片传输的name和args拆成多个 chunk 发送空 content 列表很多 chunk 的content_blocks是空列表[]这些通常是元信息 chunk如 role 声明可以在前端过滤掉metadata 中的langgraph_node这是区分 Token 来源的关键前端可以据此只显示model节点的最终回答前端逐字输出的精简写法for message, metadata in agent.stream( {messages: messages}, stream_modemessages): # 只保留 Model 节点的文本输出 if metadata.get(langgraph_node) model and message.content: print(message.content, end, flushTrue)门主提醒messages模式下即使你用的是model.invoke()非流式调用LangGraph 也会把结果拆成 Token 流式输出。这是框架层面的处理不需要你改调用方式。4.5 stream_mode“custom” —— 自定义更新是什么在工具执行过程中通过get_stream_writer()推送任意自定义数据。为什么需要默认的流式模式只能拿到 LLM 输出和节点状态变化。但实际项目中你经常需要在工具执行过程中推送进度信息比如“正在连接数据库…”“已查询 50/100 条记录”“数据下载中进度 30%…”这些信息不是 LLM 生成的也不是状态变化而是你自定义的业务数据。代码示例from langgraph.config import get_stream_writertool(description查询城市天气的工具)def get_weather(city: str): # 获取流式写入器 writer get_stream_writer() # 推送自定义进度信息 writer(fLooking up data for city: {city}) writer(fAcquired data for city: {city}) return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])for chunk in agent.stream( {messages: messages}, stream_modecustom): print(chunk)运行结果Looking up data for city: 长沙Acquired data for city: 长沙使用场景场景writer 推送内容前端展示数据库查询“正在查询第 N 页…”进度文字文件处理“已处理 30/100 行”进度条API 调用“正在调用第三方 API…”状态提示多步骤任务“步骤 2/5数据清洗”步骤指示器重要限制⚠️ 在工具内部添加get_stream_writer后该工具将无法在 LangGraph 执行上下文之外调用。也就是说你不能单独测试这个工具函数必须通过 Agent 的stream()来触发。踩坑记录# ❌ 错误写法chunk 是字符串不是 dictfor chunk in agent.stream(..., stream_modecustom): for step, data in chunk.items(): # AttributeError: str object has no attribute items ...# ✅ 正确写法custom 模式直接返回你 writer() 传入的内容for chunk in agent.stream(..., stream_modecustom): print(chunk) # chunk 就是字符串本身门主踩坑我一开始照着values模式的写法对chunk调.items()结果直接 AttributeError。custom模式返回的就是你writer()传进去的原始数据不是 dict。这个坑浪费了我半小时。4.6 流式传输多种模式 —— stream_mode 传列表是什么同时使用多种流式模式把stream_mode参数从字符串改成列表。为什么需要实际项目中你往往既想知道 Agent 的执行进度updates又想推送自定义进度custom。只用一种模式满足不了。代码示例from langgraph.config import get_stream_writertool(description查询城市天气的工具)def get_weather(city: str): writer get_stream_writer() writer(fLooking up data for city: {city}) writer(fAcquired data for city: {city}) return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])for chunk, stream_mode in agent.stream( {messages: messages}, stream_mode[updates, custom] # ← 传列表): print(fstream_mode: {stream_mode}) print(fcontent: {chunk})运行结果# updates 模式的数据stream_mode: updatescontent: {model: {messages: [AIMessage(tool_calls[...])]}}# custom 模式的数据stream_mode: customcontent: Looking up data for city: 长沙stream_mode: customcontent: Acquired data for city: 长沙# updates 模式的数据stream_mode: updatescontent: {tools: {messages: [ToolMessage(content今天长沙的天气晴,温度为30度)]}}# updates 模式的数据stream_mode: updatescontent: {model: {messages: [AIMessage(content今天长沙的天气是晴天...)]}}返回结构变化单一模式迭代变量是(data)多模式迭代变量变成(data, stream_mode)第二个值标识当前 chunk 来自哪种模式使用场景前端同时展示 Agent 执行步骤updates和工具执行进度custom监控面板需要多维度信息4.7 stream_mode“checkpoints” 与 stream_mode“debug”这两种模式在日常开发中用得相对少但值得了解checkpoints需要配置 checkpointer持久化存储每次状态持久化时返回检查点事件格式与get_state()一致场景长时间运行的任务、故障恢复、时间旅行调试debug返回所有可用信息是checkpointstasks 额外元数据的组合场景开发调试、排查复杂问题门主建议生产环境一般用不到debug模式但开发阶段遇到 Agent 行为诡异时开debug看一遍完整执行轨迹往往能快速定位问题。五、禁用流式传输Disable Streaming5.1 是什么在某些场景下你需要主动关闭某个 LLM 的流式输出能力让它走传统的同步调用路径。5.2 为什么需要禁用你可能会问流式输出不是更好吗为什么要禁用因为不是所有场景都适合流式场景原因多 Agent 系统只想让某个 Agent 流式输出其他 Agent 静默执行内部 LLM 调用Agent 内部用 LLM 做结构化输出/分类结果不需要推给前端批量处理非交互式场景不需要逐 Token 推送兼容性某些下游系统不支持流式接收5.3 禁用方式方式一nostream 标签给不需要流式输出的 LLM 打上nostream标签stream_model ChatOpenAI(modelgpt-4o-mini)internal_model ChatOpenAI(modelgpt-4o-mini).with_config( {tags: [nostream]})# stream_model 的输出会出现在 messages 流中# internal_model 的输出不会出现在 messages 流中但仍然正常执行方式二模型配置在模型初始化时直接禁用 streaming# ChatOpenAI 默认支持 streaming可以在调用时关闭model ChatOpenAI(modelgpt-4o-mini, streamingFalse)5.4 使用场景门主提醒nostream不是不执行只是不在 messages 流中输出。LLM 该调还是调结果该拿还是拿只是对前端不可见。别搞混了。六、OpenAI SSE 格式输出6.1 为什么需要如果你的前端用的是 OpenAI 官方 SDK或者兼容 OpenAI 协议的组件它期望的流式格式是SSEServer-Sent Events也就是data: {id:chatcmpl-xxx,choices:[{delta:{content:今天,role:assistant},...}]}\n\ndata: {id:chatcmpl-yyy,choices:[{delta:{content:长沙天气是晴天},...}]}\n\ndata: {id:chatcmpl-zzz,choices:[{delta:{},finish_reason:stop},...]}\n\ndata: [DONE]\n\nLangChain/LangGraph 的流式输出格式跟这个不一样所以需要手动转换。6.2 完整转换代码import timeimport uuidfrom openai.types.chat.chat_completion_chunk import ( Choice, ChoiceDelta, ChatCompletionChunk,)def build_chunk(*, text, roleNone, finish_reasonNone): 构造 OpenAI 官方 ChatCompletionChunk return ChatCompletionChunk( idfchatcmpl-{uuid.uuid4().hex}, objectchat.completion.chunk, createdint(time.time()), modelqwen3.7-plus, choices[ Choice( index0, deltaChoiceDelta(rolerole, contenttext), finish_reasonfinish_reason, ) ], )def stream_openai(messages): LangChain Agent → OpenAI Chunk → SSE first_chunk True for message, metadata in agent.stream( {messages: messages}, stream_modemessages, ): # 只保留 Model 节点的输出 if metadata.get(langgraph_node) ! model: continue if not message.content: continue # 构造 OpenAI 格式的 chunk chunk build_chunk( textmessage.content, roleassistant if first_chunk else None, ) first_chunk False # SSE 格式输出 yield fdata: {chunk.model_dump_json()}\n\n # 结束标记 finish_chunk build_chunk(finish_reasonstop) yield fdata: {finish_chunk.model_dump_json()}\n\n yield data: [DONE]\n\n6.3 输出效果data: {id:chatcmpl-6195c858...,choices:[{delta:{content:今天,role:assistant,...},...}],...}data: {id:chatcmpl-c62f3903...,choices:[{delta:{content:长沙的天气是晴天,...},...}],...}data: {id:chatcmpl-0cd9ea1c...,choices:[{delta:{content:温度为30,...},...}],...}...逐Token输出data: {id:chatcmpl-041be5b3...,choices:[{delta:{content:,finish_reason:stop,...},...}],...}data: [DONE]6.4 转换流程图6.5 OpenAI 流式格式的优缺点优点优点说明行业事实标准几乎所有前端 AI 组件都兼容 OpenAI 格式生态丰富Vercel AI SDK、OpenAI 官方 SDK 都能直接消费结构清晰每个 chunk 都有id、model、choices元信息完整增量传输delta设计天然适合流式拼接缺点缺点说明格式冗余每个 Token 都要包一层完整的 JSON 结构带宽开销大转换成本LangChain 的流式输出不能直接用需要手动转格式无进度信息标准格式里没有正在调用工具这类进度字段需要自己扩展工具调用复杂tool_calls在流式模式下是分片传输的拼接逻辑比较繁琐七、stream_mode 完整决策指南当你不确定该用哪种stream_mode时按这个决策树走快速参考表# 1. 前端逐字输出最常用for msg, meta in agent.stream(inputs, stream_modemessages): if meta[langgraph_node] model and msg.content: print(msg.content, end, flushTrue)# 2. 监控 Agent 执行流程for chunk in agent.stream(inputs, stream_modeupdates): for node_name, state in chunk.items(): print(f节点 {node_name} 执行完毕: {state})# 3. 理解状态全貌for chunk in agent.stream(inputs, stream_modevalues): print(chunk) # 完整状态快照# 4. 工具执行进度for chunk in agent.stream(inputs, stream_modecustom): print(chunk) # 自定义进度信息# 5. 多模式组合for chunk, mode in agent.stream(inputs, stream_mode[updates, custom]): if mode updates: print(f执行更新: {chunk}) elif mode custom: print(f进度通知: {chunk})# 6. 调试模式for chunk in agent.stream(inputs, stream_modedebug): print(chunk) # 最全信息八、v2 流式输出格式LangGraph 1.1 引入了versionv2的统一输出格式。强烈建议新项目使用 v2。8.1 v1 vs v2 对比v1 的问题单一模式返回原始数据多模式返回(mode, data)元组有子图时返回(namespace, data)元组格式随参数变化类型不固定v2 的改进所有 chunk 统一为StreamPart字典{ type: values | updates | messages | custom | checkpoints | tasks | debug, ns: (), # 命名空间元组子图事件时填充 data: ..., # 实际数据类型根据 type 不同而不同}8.2 v2 使用示例for part in agent.stream( {messages: messages}, stream_mode[values, updates, messages, custom], versionv2,): if part[type] values: # ValuesStreamPart — 每步后的完整状态快照 print(fState: {part[data]}) elif part[type] updates: # UpdatesStreamPart — 每步的状态增量 for node_name, state in part[data].items(): print(f节点 {node_name} 更新: {state}) elif part[type] messages: # MessagesStreamPart — LLM Token 元数据 msg, metadata part[data] print(msg.content, end, flushTrue) elif part[type] custom: # CustomStreamPart — 自定义数据 print(f进度: {part[data]})门主建议新项目直接用versionv2代码更清晰类型更安全。老项目迁移也不复杂就是返回值的解包方式变了。九、总结LangChain/LangGraph 的流式输出系统核心就记住一句话stream_mode决定了你能看到什么粒度的数据。想要什么用什么逐字输出messagesAgent 执行流程updates状态全貌values工具执行进度custom全都要传列表[updates, custom]调试debug学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
LangChain 流式输出终于讲透了：6 种 stream_mode 一篇全搞懂
门主踩坑最近在项目里做 AI 对话接口返回一次要等十几秒用户盯着空白屏幕发呆体验差到被产品经理追着问能不能像 ChatGPT 那样一个字一个字蹦出来。于是我开始研究 LangChain 的流式输出结果发现——光stream_mode就有 6 种值每种返回的数据结构还不一样。官方文档写得是挺全但对于刚上手的人来说哪一种该在什么场景下用真没说清楚。这篇文章就是我把这些坑都踩了一遍之后的总结。摘要LangChain/LangGraph 的流式输出系统提供了 6 种stream_mode对应不同的数据粒度和使用场景。本文结合真实代码和运行结果逐一拆解values、updates、messages、custom、checkpoints、debug的区别与适用场景同时给出 Agent 完整工作流程图、OpenAI SSE 格式输出方案以及企业级避坑指南。一、为什么会写这篇做 AI 应用开发流式输出不是锦上添花是刚需。大模型推理慢一个完整回复动辄 5~15 秒。如果用户每次都要等到全部生成完才能看到内容那这个产品基本没人愿意用。ChatGPT 之所以体验好核心之一就是逐 Token 流式输出——边生成边显示用户感知的等待时间从十几秒变成了不到一秒。LangChain/LangGraph 在底层帮我们实现了流式传输系统但暴露出来的 API 有多种模式每种模式返回的数据结构、粒度、适用场景完全不同。问题是官方文档把每种模式都讲了但没告诉你什么时候用哪种stream_modeupdates和stream_modevalues返回的东西看起来很像到底有啥区别stream_modemessages怎么实现类似 ChatGPT 的逐字输出stream_modecustom又是什么鬼什么时候需要自定义流式数据多种模式能不能同时用线上禁用流式输出是什么场景这些坑我一个一个踩了。下面把结果整理出来。二、流式输出的使用场景先回答最根本的问题为什么需要流式输出2.1 核心原因降低感知延迟大模型推理是逐 Token 生成的。如果用传统的invoke()同步调用必须等所有 Token 生成完才能拿到完整结果。流式输出利用这一点每生成一个 Token 就立即推送给客户端用户看到的是边想边说而不是想了半天突然冒出一大段。2.2 典型使用场景场景说明推荐模式聊天对话界面类似 ChatGPT 的逐字输出messagesAgent 执行监控实时展示 Agent 在做什么调了哪个工具、拿到什么结果updates调试与排错看 Agent 内部每一步的状态变化values/debug进度条/状态通知工具执行时推送进度如正在查询数据库…custom多 Agent 协作控制哪些 Agent 流式输出、哪些静默禁用流式OpenAI SSE 兼容前端用 OpenAI SDK 接收流式数据messages 手动转 SSE2.3 流式 vs 非流式对比门主提醒流式输出不改变总生成时间只改变用户的感知等待时间。总时间还是那么多但用户体验天差地别。三、Agent 完整工作流程在深入每种stream_mode之前必须先搞清楚 Agent 一次请求到底经历了什么。不然你看流式输出返回的数据会觉得云里雾里。3.1 Agent 执行流程图3.2 对应代码中的阶段以我们项目中的天气查询 Agent 为例from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain.tools import toolfrom langchain.messages import SystemMessage, HumanMessagechanAI ChatOpenAI(modelqwen3.7-plus, temperature0.7)tool(description查询城市天气的工具)def get_weather(city: str): return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])messages [ SystemMessage(content你是一个AI助手叫IT空门), HumanMessage(content今天长沙的天气怎么样)]当用户问今天长沙的天气怎么样Agent 会经历以下阶段阶段对应节点产出内容说明1. 规划modeltool_call: get_weather(city长沙)LLM 决定调用天气工具2. 工具执行tools今天长沙的天气晴,温度为30度执行get_weather工具3. 推理整合model今天长沙的天气是晴天温度30度…LLM 将工具结果整合为自然语言4. 返回-完整回答Agent 结束门主踩坑当时我一直在想为什么流式输出里model节点会出现两次后来才明白——第一次是 LLM 做规划决定调哪个工具第二次是 LLM 做整合把工具结果变成人话。中间夹着tools节点。整个流程是Model → Tool → Model的循环。四、stream_mode 参数详解这是本文的核心。6 种stream_mode一种一种拆开讲。4.1 总览对比stream_mode返回内容数据粒度典型用途values每步执行后的完整状态快照最粗理解 State 变化全貌updates每步执行后的状态增量中等监控 Agent 执行流程messagesLLM Token 元数据最细Token级逐字输出、SSE 推送custom自定义数据自定义工具执行进度通知checkpoints检查点事件中等持久化/回溯debug所有可用信息最全调试排错4.2 stream_mode“values” —— 完整状态快照是什么每执行完一个节点返回整个图的完整状态。返回结构每一步返回的chunk是一个 dictkey 是状态字段名如messagesvalue 是该字段的当前完整值。代码示例for chunk in agent.stream( {messages: messages}, stream_modevalues): for step, data in chunk.items(): print(fstep: {step}) print(fcontent: {data[messages][-1].content_blocks})运行结果step: messagescontent: [{type: text, text: 今天长沙的天气怎么样}] # ← 用户输入step: messagescontent: [{type: tool_call, name: get_weather, args: {city: 长沙}, id: call_50f37520...}] # ← LLM 决定调工具step: messagescontent: [{type: text, text: 今天长沙的天气晴,温度为30度}] # ← 工具返回结果step: messagescontent: [{type: text, text: 今天长沙的天气是晴天温度达到了30度。…}] # ← LLM 最终回答使用场景调试时看 State 的完整变化轨迹理解 LangGraph 状态机是怎么一步步演进的需要当前为止所有信息的场景注意事项每次返回的是完整状态不是增量。如果你只想知道这一步变了什么用updates更高效数据量较大时长对话历史每次返回完整messages列表会有性能开销门主建议如果你只是想在前端展示对话内容别用values用messages。values返回的是整个状态快照信息冗余前端处理起来也麻烦。4.3 stream_mode“updates” —— 状态增量是什么每执行完一个节点只返回该节点新产生的状态更新。与 values 的区别valuesupdates返回内容完整状态只有变化的部分信息量大包含所有历史小只有增量适合看全貌看流程代码示例for chunk in agent.stream( {messages: messages}, stream_modeupdates): print(chunk)运行结果# 第一步model 节点 —— LLM 决定调用工具{model: {messages: [AIMessage(tool_calls[{name: get_weather, args: {city: 长沙}}])]}}# 第二步tools 节点 —— 工具执行返回{tools: {messages: [ToolMessage(content今天长沙的天气晴,温度为30度)]}}# 第三步model 节点 —— LLM 整合最终回答{model: {messages: [AIMessage(content今天长沙的天气是晴天温度为30度。…)]}}使用场景Agent 执行监控面板实时显示正在调用 get_weather…“工具返回结果…”“正在生成回答…”日志记录只记录每一步的变化而不是全量快照理解 Agent 流程清楚看到 Model → Tool → Model 的执行链路门主踩坑当时我用updates做前端实时展示发现返回的 key 是节点名如model、tools不是固定的messages。一开始写成了chunk[messages]直接 KeyError。记住updates返回的 key 是节点名value 才是该节点的状态更新。4.4 stream_mode“messages” —— LLM Token 级流式是什么逐 Token词元流式输出 LLM 的生成内容。这是实现类似 ChatGPT 逐字输出效果的关键模式。返回结构每次迭代返回一个元组(message_chunk, metadata)message_chunk当前 Token 的内容块metadata元数据包含langgraph_node来源节点等信息代码示例for token, metadata in agent.stream( {messages: messages}, stream_modemessages): print(fnode: {metadata[langgraph_node]}) print(fcontent: {token.content_blocks})运行结果关键部分# 第一阶段Model 规划决定调工具node: modelcontent: [{type: tool_call_chunk, name: get_weather, args: , index: 0}]node: modelcontent: [{type: tool_call_chunk, name: None, args: {city: 长沙, index: 0}]node: modelcontent: [{type: tool_call_chunk, name: None, args: }, index: 0}]# 第二阶段Tool 执行 node: toolscontent: [{type: text, text: 今天长沙的天气晴,温度为30度}]# 第三阶段Model 整合最终回答逐Token输出node: modelcontent: [{type: text, text: 今天}]node: modelcontent: [{type: text, text: 长沙的天气是晴天}]node: modelcontent: [{type: text, text: 温度}]node: modelcontent: [{type: text, text: 30度。}]node: modelcontent: [{type: text, text: 天气比较热}]node: modelcontent: [{type: text, text: 出门}]node: modelcontent: [{type: text, text: 记得做好防晒哦}]node: modelcontent: [{type: text, text: }]使用场景前端逐字输出最核心的使用场景实现打字机效果OpenAI SSE 格式兼容转成data: {...}\n\n格式推给前端Token 级过滤只显示某个节点的输出或者排除工具调用的 Token关键细节Tool Call 也是流式的注意第一阶段tool_call_chunk是分片传输的name和args拆成多个 chunk 发送空 content 列表很多 chunk 的content_blocks是空列表[]这些通常是元信息 chunk如 role 声明可以在前端过滤掉metadata 中的langgraph_node这是区分 Token 来源的关键前端可以据此只显示model节点的最终回答前端逐字输出的精简写法for message, metadata in agent.stream( {messages: messages}, stream_modemessages): # 只保留 Model 节点的文本输出 if metadata.get(langgraph_node) model and message.content: print(message.content, end, flushTrue)门主提醒messages模式下即使你用的是model.invoke()非流式调用LangGraph 也会把结果拆成 Token 流式输出。这是框架层面的处理不需要你改调用方式。4.5 stream_mode“custom” —— 自定义更新是什么在工具执行过程中通过get_stream_writer()推送任意自定义数据。为什么需要默认的流式模式只能拿到 LLM 输出和节点状态变化。但实际项目中你经常需要在工具执行过程中推送进度信息比如“正在连接数据库…”“已查询 50/100 条记录”“数据下载中进度 30%…”这些信息不是 LLM 生成的也不是状态变化而是你自定义的业务数据。代码示例from langgraph.config import get_stream_writertool(description查询城市天气的工具)def get_weather(city: str): # 获取流式写入器 writer get_stream_writer() # 推送自定义进度信息 writer(fLooking up data for city: {city}) writer(fAcquired data for city: {city}) return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])for chunk in agent.stream( {messages: messages}, stream_modecustom): print(chunk)运行结果Looking up data for city: 长沙Acquired data for city: 长沙使用场景场景writer 推送内容前端展示数据库查询“正在查询第 N 页…”进度文字文件处理“已处理 30/100 行”进度条API 调用“正在调用第三方 API…”状态提示多步骤任务“步骤 2/5数据清洗”步骤指示器重要限制⚠️ 在工具内部添加get_stream_writer后该工具将无法在 LangGraph 执行上下文之外调用。也就是说你不能单独测试这个工具函数必须通过 Agent 的stream()来触发。踩坑记录# ❌ 错误写法chunk 是字符串不是 dictfor chunk in agent.stream(..., stream_modecustom): for step, data in chunk.items(): # AttributeError: str object has no attribute items ...# ✅ 正确写法custom 模式直接返回你 writer() 传入的内容for chunk in agent.stream(..., stream_modecustom): print(chunk) # chunk 就是字符串本身门主踩坑我一开始照着values模式的写法对chunk调.items()结果直接 AttributeError。custom模式返回的就是你writer()传进去的原始数据不是 dict。这个坑浪费了我半小时。4.6 流式传输多种模式 —— stream_mode 传列表是什么同时使用多种流式模式把stream_mode参数从字符串改成列表。为什么需要实际项目中你往往既想知道 Agent 的执行进度updates又想推送自定义进度custom。只用一种模式满足不了。代码示例from langgraph.config import get_stream_writertool(description查询城市天气的工具)def get_weather(city: str): writer get_stream_writer() writer(fLooking up data for city: {city}) writer(fAcquired data for city: {city}) return f今天{city}的天气晴,温度为30度agent create_agent(modelchanAI, tools[get_weather])for chunk, stream_mode in agent.stream( {messages: messages}, stream_mode[updates, custom] # ← 传列表): print(fstream_mode: {stream_mode}) print(fcontent: {chunk})运行结果# updates 模式的数据stream_mode: updatescontent: {model: {messages: [AIMessage(tool_calls[...])]}}# custom 模式的数据stream_mode: customcontent: Looking up data for city: 长沙stream_mode: customcontent: Acquired data for city: 长沙# updates 模式的数据stream_mode: updatescontent: {tools: {messages: [ToolMessage(content今天长沙的天气晴,温度为30度)]}}# updates 模式的数据stream_mode: updatescontent: {model: {messages: [AIMessage(content今天长沙的天气是晴天...)]}}返回结构变化单一模式迭代变量是(data)多模式迭代变量变成(data, stream_mode)第二个值标识当前 chunk 来自哪种模式使用场景前端同时展示 Agent 执行步骤updates和工具执行进度custom监控面板需要多维度信息4.7 stream_mode“checkpoints” 与 stream_mode“debug”这两种模式在日常开发中用得相对少但值得了解checkpoints需要配置 checkpointer持久化存储每次状态持久化时返回检查点事件格式与get_state()一致场景长时间运行的任务、故障恢复、时间旅行调试debug返回所有可用信息是checkpointstasks 额外元数据的组合场景开发调试、排查复杂问题门主建议生产环境一般用不到debug模式但开发阶段遇到 Agent 行为诡异时开debug看一遍完整执行轨迹往往能快速定位问题。五、禁用流式传输Disable Streaming5.1 是什么在某些场景下你需要主动关闭某个 LLM 的流式输出能力让它走传统的同步调用路径。5.2 为什么需要禁用你可能会问流式输出不是更好吗为什么要禁用因为不是所有场景都适合流式场景原因多 Agent 系统只想让某个 Agent 流式输出其他 Agent 静默执行内部 LLM 调用Agent 内部用 LLM 做结构化输出/分类结果不需要推给前端批量处理非交互式场景不需要逐 Token 推送兼容性某些下游系统不支持流式接收5.3 禁用方式方式一nostream 标签给不需要流式输出的 LLM 打上nostream标签stream_model ChatOpenAI(modelgpt-4o-mini)internal_model ChatOpenAI(modelgpt-4o-mini).with_config( {tags: [nostream]})# stream_model 的输出会出现在 messages 流中# internal_model 的输出不会出现在 messages 流中但仍然正常执行方式二模型配置在模型初始化时直接禁用 streaming# ChatOpenAI 默认支持 streaming可以在调用时关闭model ChatOpenAI(modelgpt-4o-mini, streamingFalse)5.4 使用场景门主提醒nostream不是不执行只是不在 messages 流中输出。LLM 该调还是调结果该拿还是拿只是对前端不可见。别搞混了。六、OpenAI SSE 格式输出6.1 为什么需要如果你的前端用的是 OpenAI 官方 SDK或者兼容 OpenAI 协议的组件它期望的流式格式是SSEServer-Sent Events也就是data: {id:chatcmpl-xxx,choices:[{delta:{content:今天,role:assistant},...}]}\n\ndata: {id:chatcmpl-yyy,choices:[{delta:{content:长沙天气是晴天},...}]}\n\ndata: {id:chatcmpl-zzz,choices:[{delta:{},finish_reason:stop},...]}\n\ndata: [DONE]\n\nLangChain/LangGraph 的流式输出格式跟这个不一样所以需要手动转换。6.2 完整转换代码import timeimport uuidfrom openai.types.chat.chat_completion_chunk import ( Choice, ChoiceDelta, ChatCompletionChunk,)def build_chunk(*, text, roleNone, finish_reasonNone): 构造 OpenAI 官方 ChatCompletionChunk return ChatCompletionChunk( idfchatcmpl-{uuid.uuid4().hex}, objectchat.completion.chunk, createdint(time.time()), modelqwen3.7-plus, choices[ Choice( index0, deltaChoiceDelta(rolerole, contenttext), finish_reasonfinish_reason, ) ], )def stream_openai(messages): LangChain Agent → OpenAI Chunk → SSE first_chunk True for message, metadata in agent.stream( {messages: messages}, stream_modemessages, ): # 只保留 Model 节点的输出 if metadata.get(langgraph_node) ! model: continue if not message.content: continue # 构造 OpenAI 格式的 chunk chunk build_chunk( textmessage.content, roleassistant if first_chunk else None, ) first_chunk False # SSE 格式输出 yield fdata: {chunk.model_dump_json()}\n\n # 结束标记 finish_chunk build_chunk(finish_reasonstop) yield fdata: {finish_chunk.model_dump_json()}\n\n yield data: [DONE]\n\n6.3 输出效果data: {id:chatcmpl-6195c858...,choices:[{delta:{content:今天,role:assistant,...},...}],...}data: {id:chatcmpl-c62f3903...,choices:[{delta:{content:长沙的天气是晴天,...},...}],...}data: {id:chatcmpl-0cd9ea1c...,choices:[{delta:{content:温度为30,...},...}],...}...逐Token输出data: {id:chatcmpl-041be5b3...,choices:[{delta:{content:,finish_reason:stop,...},...}],...}data: [DONE]6.4 转换流程图6.5 OpenAI 流式格式的优缺点优点优点说明行业事实标准几乎所有前端 AI 组件都兼容 OpenAI 格式生态丰富Vercel AI SDK、OpenAI 官方 SDK 都能直接消费结构清晰每个 chunk 都有id、model、choices元信息完整增量传输delta设计天然适合流式拼接缺点缺点说明格式冗余每个 Token 都要包一层完整的 JSON 结构带宽开销大转换成本LangChain 的流式输出不能直接用需要手动转格式无进度信息标准格式里没有正在调用工具这类进度字段需要自己扩展工具调用复杂tool_calls在流式模式下是分片传输的拼接逻辑比较繁琐七、stream_mode 完整决策指南当你不确定该用哪种stream_mode时按这个决策树走快速参考表# 1. 前端逐字输出最常用for msg, meta in agent.stream(inputs, stream_modemessages): if meta[langgraph_node] model and msg.content: print(msg.content, end, flushTrue)# 2. 监控 Agent 执行流程for chunk in agent.stream(inputs, stream_modeupdates): for node_name, state in chunk.items(): print(f节点 {node_name} 执行完毕: {state})# 3. 理解状态全貌for chunk in agent.stream(inputs, stream_modevalues): print(chunk) # 完整状态快照# 4. 工具执行进度for chunk in agent.stream(inputs, stream_modecustom): print(chunk) # 自定义进度信息# 5. 多模式组合for chunk, mode in agent.stream(inputs, stream_mode[updates, custom]): if mode updates: print(f执行更新: {chunk}) elif mode custom: print(f进度通知: {chunk})# 6. 调试模式for chunk in agent.stream(inputs, stream_modedebug): print(chunk) # 最全信息八、v2 流式输出格式LangGraph 1.1 引入了versionv2的统一输出格式。强烈建议新项目使用 v2。8.1 v1 vs v2 对比v1 的问题单一模式返回原始数据多模式返回(mode, data)元组有子图时返回(namespace, data)元组格式随参数变化类型不固定v2 的改进所有 chunk 统一为StreamPart字典{ type: values | updates | messages | custom | checkpoints | tasks | debug, ns: (), # 命名空间元组子图事件时填充 data: ..., # 实际数据类型根据 type 不同而不同}8.2 v2 使用示例for part in agent.stream( {messages: messages}, stream_mode[values, updates, messages, custom], versionv2,): if part[type] values: # ValuesStreamPart — 每步后的完整状态快照 print(fState: {part[data]}) elif part[type] updates: # UpdatesStreamPart — 每步的状态增量 for node_name, state in part[data].items(): print(f节点 {node_name} 更新: {state}) elif part[type] messages: # MessagesStreamPart — LLM Token 元数据 msg, metadata part[data] print(msg.content, end, flushTrue) elif part[type] custom: # CustomStreamPart — 自定义数据 print(f进度: {part[data]})门主建议新项目直接用versionv2代码更清晰类型更安全。老项目迁移也不复杂就是返回值的解包方式变了。九、总结LangChain/LangGraph 的流式输出系统核心就记住一句话stream_mode决定了你能看到什么粒度的数据。想要什么用什么逐字输出messagesAgent 执行流程updates状态全貌values工具执行进度custom全都要传列表[updates, custom]调试debug学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
