想象你是一位美食直播导演,后厨里经验丰富的厨师正照着一张菜谱做菜。过去你只能等全部菜肴上齐,再向观众展示最终成品;观众等得焦虑,你也无法回答“现在做到哪一步了”。于是你架起了多台摄像机:一台对准切菜动作给特写,一台拍下整个灶台的全景,还有一台实时生成字幕提醒“出锅啦”。LangGraph 的 Stream 模式,就是把这样一套直播系统装进 AI Agent 的执行引擎里。
Agent 在处理复杂任务时会依次经过多个节点,每个节点都会读取状态、执行逻辑、更新状态。过去调用完整个图才拿到结果,过程是黑箱;流式输出让你像看直播一样,按节点顺序接收更新。stream_mode 参数则像镜头切换旋钮:"updates" 输出当前节点带来的增量变化,"values" 输出此刻完整的全局状态。两种镜头各有用途,配合事件订阅机制,你还能只关注特定的环节。
掌握 LangGraph 流式输出,等于给 Agent 装上了可观测仪表盘,让你快速定位卡点、改善用户体验,是构建生产级多智能体应用的必备技能。
可视化演示
展示KMP算法利用失败函数避免回溯的匹配过程,文本指针不回退,模式串根据部分匹配表智能跳转。
本节配有交互动效(进入视口后自动加载,约 3KB)。
KMP匹配过程 文本指针从左到右扫描,模式串根据失败函数跳转避免回溯 时间复杂度 O(n+m) 预处理失败函数后匹配过程线性完成
流式输出概览
想象你在美食直播间担任导演,后厨里厨师正按菜谱做菜。过去你只能等全部菜肴上齐后,才向观众展示最终成品;观众等得焦虑,你也无法回答“现在做到哪一步了”。于是你架起摄像机,把切菜、下锅、翻炒的过程实时传到屏幕上,后厨从此不再神秘。LangGraph 的 Stream 模式就扮演了这组摄像机的角色,让 Agent 每一步的执行变化都能被实时推送到调用方眼前。
LangGraph 用状态机来编排 Agent 的决策流程。开发者先定义状态的数据结构,再写若干个节点函数,最后用边把这些节点连接起来。StateGraph.compile() 会把这张由节点和边组成的状态图变成一个可调用的 App。调用 app.stream() 时,执行过程会以迭代器的形式逐段返回,而不是等全部结束后一次性返回。其中 stream_mode 参数决定返回的数据是增量还是全量快照。
一次完整的图执行,本质上是状态不断转移的过程。节点读取上一步的状态,结合当前输入事件,计算出新的状态。例如一个回答用户问题的 Agent,会经过“理解问题”“检索资料”“生成回答”三个节点,状态从空字典逐步填充成完整答案。状态转移关系可以写成:
到了这里你已经明白,流式输出就是把原来藏在函数调用内部的“状态转移过程”暴露出来。就像直播镜头让后厨的每个动作都清晰可见,LangGraph 的 app.stream() 让状态每次跳变的瞬间都成为可以被外部接收的事件。
- 上一步状态 $state_{t-1}$节点开始执行前累积下来的全部信息,比如已经获得的历史记录、中间变量等。
- 当前事件 $event_t$本轮传入的新输入,比如用户消息、工具调用返回值,它驱动节点产生变化。
- 状态转移函数 $f$节点函数本身,负责读取旧状态和事件,经过业务逻辑后输出新状态。
updates增量
导演注意到,如果全景镜头一直拍摄整个厨房,观众很难看清厨师手上的精细动作。于是你切换成特写镜头:厨师切完一根胡萝卜,镜头马上推近案板,让观众看到刚刚切好的那几片。LangGraph 的 stream_mode="updates" 就是这枚特写镜头,它只返回当前执行节点带来的状态变化量,而不是完整的全局状态。
当图执行到某个节点时,这个节点会返回一个字典,描述它想要更新哪些字段。updates 模式会把这次更新的内容包装成 {节点名: {字段: 值}} 的形式推送给调用方。比如检索节点执行完,你收到的 chunk 可能是:
这样做的好处是信息密度高,调用方不需要自己计算哪部分状态变了,拿来即用。尤其在节点多、状态大的图中,增量输出能明显减少每次传输的数据量。这种“只给变化部分”的思路,可以用下面的公式概括:
回到直播间的比喻:特写镜头虽然画面小,却最擅长呈现细节。updates 模式让开发者像盯着案板一样,实时感知每一个节点的最新动作,尤其适合需要逐步核对与逐节点监控的场景。
- 节点输出 $node\_output$节点函数返回的字典,只包含这次执行希望写入状态的键值对。
- 提取更新 $\text{update}(\cdot)$把节点输出转换成标准增量格式,保留键名和值,不引入历史状态。
- 增量状态 $\Delta_s$只描述“从上一个状态到当前状态发生了什么变化”的最小信息单元。
{"retrieve_node": {"documents": ["文档1", "文档2"]}}values快照
特写镜头看得清细节,却容易让人忘记整桌菜摆成了什么样。所以导演在厨房上方又装了一台全景摄像机,每隔几十秒把整个灶台、所有餐盘拍一遍,让观众知道当前全貌。stream_mode="values" 就是这台全景摄像机:每次状态更新后,它都会把当前完整的 state 对象作为一帧快照推送出去。
和 updates 的“只报增量”不同,values 模式每次推送的是全部状态字段。比如图刚执行完第一个节点,你收到的是包含所有初始字段加第一个节点修改后的完整状态;执行完第二个节点,你又收到一份包含前两个节点修改的新快照。虽然每次数据量更大,但调用方不需要维护任何中间拼接逻辑,拿到的永远是一份可直接展示的完整状态。
这种周期性的整体输出很适合用来渲染前端界面或记录执行日志:前端拿到一次快照就能刷新整个页面,不需要理解内部增量关系。快照的本质是对当前时刻状态的一次完整投影,可以表示为:
全景镜头让观众既有全局感又不遗漏信息。values 模式的价值正在于它把复杂的执行过程压缩成一个个清晰的“整桌菜照片”,适合汇报进度、渲染界面等需要整体视角的场景。
values 模式输出的是状态在 t 时刻的一张“照片”。照片里包含全部信息,但它是静态的,后续状态变化不会改变已经生成的照片。- 当前状态 $state_t$t 时刻图中所有状态字段的集合,包括用户输入、节点结果、临时变量等。
- 快照函数 $\text{snapshot}(\cdot)$把当前状态对象转成不可变的输出快照,不修改原状态。
- 快照结果 $V_t$交付给调用方的完整状态视图,可能比 updates 增量大,但胜在直观。
事件订阅
观众看直播时,可能并不想盯着每个镜头,只想在“菜出锅”那一刻收到提醒。于是平台推出了订阅功能:你选定一个环节,当它触及时,系统立刻推送一条通知。LangGraph 的事件订阅机制差不多是这个意思——它允许你在流式中只关注特定节点或特定类型的事件,而不是被动接收所有输出。
LangGraph 提供了更细粒度的 astream_events API,它会把图执行过程中的各种事件(如节点开始、节点结束、LLM 生成 token)都暴露出来。事件对象中包含 event 类型、node 名称、data 数据等字段。你可以像订阅频道一样,只处理关心的节点输出,其他事件则忽略,从而减少无效计算和噪声。
事件驱动本质上是一种回调映射:当某个事件发生时,调用对应的处理函数。订阅关系可以抽象为:
订阅功能让观众告别无效等待,只看自己想看的关键节点。事件订阅机制也让开发者从海量流式输出中解放出来,做到“按需关注、精准响应”,这是构建复杂 Agent 时非常重要的一环。
- 事件对象 $e$流中每个事件都携带类型字段、节点字段和数据字段,描述“什么时间、在哪个节点、发生了什么”。
- 数据提取 $e.\text{data}$从事件对象中取出真正需要的内容,如节点返回值或生成的文本片段。
- 回调函数 $\text{cb}(\cdot)$开发者为关心的事件注册的处理逻辑,可以写入日志、更新前端或触发下一个动作。
对比选型
作为直播导演,你手里有特写镜头和全景镜头,什么时候用哪一个,取决于直播的目的。如果观众想学做菜,你应该多用特写展示手法;如果观众只是想知道整桌菜齐不齐,全景镜头更有效率。updates 和 values 的选型逻辑完全一样。
从数据粒度看,updates 输出小、频率高,每个节点执行后都可能推送一次;values 输出大、频率可能相对更低,因为每次都要序列化整个状态。性能上,状态越庞大,values 的传输成本越高;节点越密集,updates 的事件流越琐碎。实际工程中,两者没有绝对好坏,只有匹配度。对延迟敏感、需要逐节点调试的场景用 updates;对展示总进度、记录完整轨迹的场景用 values。
如果用一个式子来对比总传输成本,可以这样看:
镜头选型没有标准答案,关键看你想要细节还是要点。updates 和 values 也常被组合使用:前端用 values 刷新整体界面,后端用 updates 做细粒度日志,两者并行并不冲突。
- 增量传输耗时 $t_i$第 i 个节点增量数据从图内部到达调用方所需的时间,通常与增量大小正相关。
- 快照次数 $N$values 模式下发送全量快照的次数,约等于状态更新的次数。
- 快照间隔 $\Delta T$相邻两次快照之间的平均时间间隔,间隔越小,实时性越高但开销越大。
多智能体协作
大型宴会的后厨不会只有一个厨师,而是多个厨位同时开工:有人切配、有人掌勺、有人摆盘。导演要给每个厨位都安上镜头,才能实时掌握每个岗位的进度。多智能体应用也是这样:多个 Agent 各自处理子任务,主控节点需要同时观察它们的运行状态,这时流式输出就成了协调多路的“多镜头系统”。
LangGraph 支持在一个状态图里编排多个智能体节点,每个节点可能就是一个完成独立任务的 Agent。通过流式输出,主控代码可以一边执行一边接收每个智能体发来的中间结果,提前发现某个子任务卡住或出错,而不必等所有 Agent 都跑完后才知道结果。这种并行协作让整个系统的响应时间和容错能力都得到提升。
把多个智能体的状态流汇聚起来,最终能得出一个综合结果。这个汇聚过程可以抽象为:
每个厨位的镜头合成一台完整的宴席直播,让总导演对所有岗位尽在掌握。多智能体配合流式输出,正是现代复杂 Agent 应用从单点执行走向并行协同的关键一步。
- 智能体状态流 $S_i$第 i 个智能体在流式输出中产生的状态序列,可能包含多轮更新。
- 汇聚函数 $\text{aggregate}(\cdot)$把各智能体的状态按业务规则合并,处理冲突、去重或加权。
- 最终结果 $R$聚合后的整体输出,可能是一个答案、一份报告或一个决策结果。
工程实践
直播不能只在实验室里搭好设备就结束,还要把它接入监控中心、存储录像、处理各种突发状况。LangGraph 流式输出落到工程上,同样涉及如何配置、如何消费、如何监控三个环节。
在配置环节,stream_mode 参数是入口。你可以使用 "updates" 或 "values",也可以传入列表同时启用两者。在消费环节,推荐用 astream_events 订阅事件,按节点名或事件类型过滤,再把结果写入日志或推送前端。在监控环节,可以把每次流式 chunk 的耗时、大小记录成指标,作为 Agent 健康度的依据。三者配合,才能让流式输出从“能跑”变成“好用”。
为了在代码里清晰声明流式策略,可以把相关参数集中成配置对象:
设备归位、镜头校准、信号连通,直播才算真正上线。工程实践中的 LangGraph 同样如此——把流式配置、事件处理和监控日志串联成体系,你的 Agent 才能真正做到“越跑越清晰”。
- 输出模式 $mode$字符串
"updates"或"values",也可以传列表同时启用。 - 节点过滤 $nodes$可选参数,只订阅列表中指定节点的输出,减少无关数据。
- 配置对象 $\text{stream\_cfg}$将模式与过滤条件包装成标准参数,方便复用和测试。
AI 应用场景
智能客服实时可见
阶段:多智能体协作与事件订阅
解决的问题:传统客服 Agent 接到用户提问后要经过意图识别、知识检索、答案生成等多个步骤,用户只能等待最终回复,不知道系统是否已理解自己;调试人员也难以定位响应慢的节点。
具体案例:某电商平台的客服系统使用 LangGraph 编排三个 Agent:意图识别 Agent、商品检索 Agent、话术生成 Agent。前端通过 stream_mode="updates" 实时展示“正在识别你的需求…”“正在查找商品…”,并用 astream_events 订阅“话术生成完成”事件,让用户等待时始终有反馈,投诉率下降 35%。
研究报告生成总监控
阶段:values 快照与对比选型
解决的问题:多智能体并行撰写研究报告时,主控需要掌握每个写作模块的产出情况;若等全部完成再合并,一旦某个子任务失败,整份报告都要返工。
具体案例:金融科技公司用 LangGraph 让资料收集 Agent、数据分析 Agent、风险提示 Agent 并行运行。主控用 stream_mode="values" 每隔几秒获取一次完整工作区快照,将进度渲染到内部看板;同时用 updates 模式接收每个 Agent 的增量,若发现某 Agent 卡在“数据接口超时”,立即重启该子任务,最终报告生成成功率提升 50%。
长文档审阅事件订阅
阶段:事件订阅机制
解决的问题:审阅长文档时,用户只关心某些关键节点(如查重、合规检查),并不需要看到所有中间步骤,全量输出会产生大量无用信息。
具体案例:法律科技公司的合同审阅 Agent 包含格式校验、条款抽取、风险标注、合规判断四个节点。团队通过 astream_events 只订阅“合规判断”节点的事件,其余节点用后台日志记录。这样业务人员收到的是精简的风险报告,工程师又能从日志还原完整执行轨迹。
Python 演示
from langgraph.graph import StateGraph, START, ENDfrom typing import TypedDict class OrderState(TypedDict): tasks: list log: list def add_tags(state: OrderState) -> dict: return { "tasks": state["tasks"] + ["打标签"], "log": state["log"] + ["打标签完成"], } def check_stock(state: OrderState) -> dict: return { "tasks": state["tasks"] + ["查库存"], "log": state["log"] + ["查库存完成"], } def generate_reply(state: OrderState) -> dict: return { "tasks": state["tasks"] + ["生成回复"], "log": state["log"] + ["生成回复完成"], } graph = StateGraph(OrderState)graph.add_node("add_tags", add_tags)graph.add_node("check_stock", check_stock)graph.add_node("generate_reply", generate_reply)graph.add_edge(START, "add_tags")graph.add_edge("add_tags", "check_stock")graph.add_edge("check_stock", "generate_reply")graph.add_edge("generate_reply", END) app = graph.compile() print("===== updates 模式:每个节点增量 =====")for chunk in app.stream({"tasks": [], "log": []}, stream_mode="updates"): print(chunk) print("===== values 模式:每次更新全量快照 =====")for chunk in app.stream({"tasks": [], "log": []}, stream_mode="values"): print(chunk)
stream_mode="updates" 与 "values" 两种模式输出形态的差异。class OrderState(TypedDict):定义状态结构,tasks和log两个字段会被节点逐步填充。graph.add_node("add_tags", add_tags)注册一个节点,节点名"add_tags"会出现在 updates 输出中。graph.add_edge(START, "add_tags")声明执行顺序,让图从入口边启动后先进入打标签节点。app = graph.compile()把状态图编译成可调用对象,之后所有流式输出都通过app发起。app.stream({...}, stream_mode="updates")以增量模式启动流式执行,每次迭代返回{节点名: 增量数据}。for chunk in app.stream(...):迭代 chunk 即可实时处理每一步状态变化,不必等待全图执行完。stream_mode="values"切换成快照模式,每次迭代返回整个state字典,方便前端直接渲染。
常见误区
误区我以为 `updates` 和 `values` 只是把同样的结果换个包装,一个叫增量一个叫全量,实际返回的数据信息量差别不大,所以选哪个都行。
正解两者的信息粒度完全不同。updates 只返回当前节点写入状态的那几个字段,像做菜特写镜头只拍案板;values 返回当前整个 state 对象,像全景镜头拍下所有灶台。数据量、推送频率、处理逻辑都不一样,不能混用。
为何常见:初学者常被“流式输出”这个大概念吸引,忽略了 stream_mode 参数的语义差异,加上示例代码经常只展示其中一种,就容易想当然。
误区我担心开启 Stream 模式后,边执行边输出会产生额外开销,导致整个 Agent 的响应时间变长,所以宁可等全部结果拿到再处理。
正解流式输出本质是执行引擎在状态转移后多做一些序列化与推送工作,并没有改变图内部的节点计算顺序。就像直播多了一个导播切换画面,但厨师切菜翻炒的速度并不变。真正的影响主要在传输层,而不是计算层。
为何常见:人们容易把“看到过程的成本”错当成“过程的成本”,尤其当网络传输、序列化耗时明显时,会误以为 Agent 变慢了。
误区我觉得 `values` 一次输出完整状态,`updates` 要输出很多次增量,所以后者更耗资源,总想优先用 `values`。
正解资源开销要看总数据量而非输出次数。values 每次全量推送,如果 state 很大,传输和序列化成本会很高;updates 虽然推送频繁,但每次只有很小的增量,总体可能更省。如果还要算上前端拼接逻辑,updates 反而更能减少带宽压力。
为何常见:直觉上“次数多”等于“开销大”,忽略了单次数据规模这个更重要的变量,导致选型时做出相反判断。
误区我以为 `updates` 和 `values` 是一套 API,事件订阅又是另一套 API,两者互不相通,用了事件订阅就不能再用流式模式。
正解astream_events 本身就是流式事件 API,它的事件对象里包含节点名称和事件类型,开发者可以只订阅关心的节点,等价于在更大的事件流上做过滤。在 LangGraph 中你甚至可以在一次执行中同时拿到 updates 和 values 两种输出,它们不是互斥关系。
为何常见:文档往往分章节介绍两种能力,加上示例代码通常单独演示,缺少“组合使用”的引导,让读者误以为它们是二选一。
误区我担心客户端看到中间状态后,会因为某些外部副作用(比如手动触发重试)影响后续执行,导致最终结果和关闭流式时不一样。
正解流式输出是只读观测通道,不会把客户端的数据注入回执行流程。LangGraph 的状态转移只由节点函数和传入的输入决定,与是否有人观察无关。直播镜头不会改变菜的味道,观测不会改变 Agent 的行为。
为何常见:观察者效应在量子物理里很出名,人们容易映射到软件系统上,实际上绝大多数软件观测都是无副作用的,除非你显式写了注入逻辑。
面试题
Q1. LangGraph 中 stream_mode="updates" 返回的每次 chunk 内容是?
- A.当前节点的增量状态← 正确答案
- B.全部节点的历史状态
- C.模型生成的原始 token
- D.状态图的配置信息
答案:A。updates 模式只返回当前节点执行后写入状态字段的增量,如 {"node1": {"msg": "..."}}。values 返回全量状态,token 属于 LLM 流式输出,配置信息不属于执行期的 chunk。
Q2. 下面哪种模式适合只想看每步细节、不想传全量状态的场景?
- A.updates 模式← 正确答案
- B.values 模式
- C.debug 模式
- D.sync 模式
答案:A。updates 以节点为单位推送增量,数据量小、细节清晰,适合逐节点监控;values 是整体快照,数据量更大;debug 和 sync 都不是 LangGraph 的流式模式名。
Q3. values 模式相比 updates 模式的主要特点是?
- A.只返回最新节点增量
- B.返回整个状态的快照← 正确答案
- C.只在出错时返回状态
- D.返回的是历史运行日志
答案:B。values 模式每次状态更新后都会推送当前完整状态对象,是一次全量快照;增量是 updates 的职责,出错触发和日志输出都不是流式模式的默认行为。
Q4. 使用 astream_events 订阅事件时,可以按哪些维度过滤?
- A.节点名称← 正确答案
- B.事件类型← 正确答案
- C.用户浏览器版本
- D.图运行版本
答案:AB。事件对象包含 node 字段和 event 字段,分别对应节点名称和事件类型,因此可以按这两者过滤。浏览器版本与后端执行无关,图运行版本也不是事件过滤的维度。
Q5. 假设一个无分支、无循环的 LangGraph 图有 5 个节点,每个节点只更新一次状态,values 模式会推送几次快照?
- A.1 次,仅最终状态
- B.2 次,开始和结束
- C.约 5 次,每次更新后← 正确答案
- D.10 次,翻倍推送
答案:C。values 模式在每次状态变化后都会推送一次全量快照。无分支且每个节点只更新一次时,快照次数约等于节点数,因此 5 个节点约推送 5 次,选 C。
Q6. 在多智能体协作中,使用流式输出能带来哪些收益?
- A.实时观察各智能体进度← 正确答案
- B.自动修复智能体代码缺陷
- C.更早发现执行卡顿点← 正确答案
- D.降低模型训练数据成本
答案:AC。流式输出让主控程序实时看到每个智能体的中间状态,能快速定位卡点、提前处理异常,但不能自动改代码,也与训练数据成本无关,因此选 A 和 C。
Q7. updates 模式推送的状态更新属于下面哪一类?
- A.整体状态快照
- B.节点产生的增量← 正确答案
- C.最终汇总答案
- D.配置文件差异
答案:B。updates 模式把每个节点返回的字典包装后推送,强调“变化的部分”,比如 {"retrieve_node": {"docs": [...]}},因此属于节点增量,不是全量快照或最终答案。
Q8. 关于 LangGraph 流式输出,以下说法正确的有?
- A.updates 比 values 推送频率更高← 正确答案
- B.values 每次包含状态全量← 正确答案
- C.两种模式只能同时启用一种
- D.两种模式都不影响最终结果← 正确答案
答案:ABD。updates 每个节点都可能推送增量,values 每次更新推送全量;LangGraph 支持通过列表参数同时启用多种模式,所以 C 错误。流式输出只是观测通道,不改变最终结果,因此 A、B、D 正确。
Q9. 使用 astream_events 时,要仅获取某个特定节点的输出,最适合的过滤字段是?
- A.event_type
- B.node_name← 正确答案
- C.graph_id
- D.state_schema
答案:B。事件对象中的 node 字段(即 node_name)用于标识节点,按它过滤可精确获取指定节点输出;event_type 过滤的是事件种类,graph_id 与 state_schema 都不是节点过滤字段。
Q10. 在 LangGraph 中启用流式输出,正确的位置是?
- A.编译后调用 stream 方法时指定参数← 正确答案
- B.创建 StateGraph 时修改初始化参数
- C.在节点函数内部设置全局变量
- D.修改 Python 解释器的环境变量
答案:A。流式输出通过 app.stream(..., stream_mode=...) 或 astream_events 在运行时启用,属于编译后的调用阶段。图定义和节点函数不负责流式策略,环境变量也无法直接开启流式模式。