【AI Agent 编排全栈实战】第 06 篇LangGraph 核心架构 — StateGraph 与消息传递模型本系列定位面向 AI 应用开发者的 Agent 编排框架系统化教程从编排概念出发深入对比 LangGraph、Microsoft Agent Framework、CrewAI、Semantic Kernel、LlamaIndex 五大框架最终落地生产级多 Agent 系统。本篇你将学到理解 LangGraph 的 Pregel 超步super-step执行模型知道图到底是怎么一步步跑起来的掌握 StateGraph 的四要素State状态、Node节点、Edge边、Reducer合并器并能独立编译运行一张图学会用TypedDictAnnotated定义状态结构理解默认覆盖与自定义合并的区别能够读懂编译后的图结构并排查常见的KeyError、状态覆盖等新手坑学完本篇你将能够用 LangGraph 搭建出第一张有状态、有分支、能循环的 Agent 工作流图并为后续的条件路由、并行执行、持久化等高级特性打下坚实基础。一、为什么 Agent 编排需要一张图在第 15 篇里我们已经建立了一个共识一个实用的 Agent 系统几乎从来不是一次调用大模型那么简单而是由感知 → 规划 → 行动 → 反思等多个环节组成的编排流程。当环节变多、环节之间存在条件分支和循环时用裸if/else加函数调用的方式组织代码会迅速失控。LangGraph 给出的答案很直接把整个 Agent 工作流显式建模成一张状态图StateGraph。图中的每个节点是一个处理函数边是控制流而所有节点共享一份可演进的状态State。这套抽象有三个立竿见影的好处流程可视化图的拓扑本身就是文档谁先谁后、哪里循环一目了然。状态可控状态不再是散落在各函数里的局部变量而是一份被框架统一管理的结构可快照、可回滚、可持久化。天然支持复杂控制流条件分支、并行、循环、人工介入都是图的一等公民不需要自己造轮子。为了让你在进入 API 细节之前先有一个整体印象下面这张图展示了 LangGraph 的核心概念关系。图构建期compileinvoke/streamState SchemaTypedDictStateGraphNode A函数Node B函数Edge A→BConditional Edge条件边CompiledGraph可执行应用执行结果最终状态可以看到LangGraph 的使用节奏是先定义状态结构再把节点和边加进StateGraph编译后得到一个可执行应用最后调用它。下面我们逐层拆解。二、核心执行模型Pregel 超步要真正理解 LangGraph绕不开它底层的执行模型——Pregel 超步super-step。这个名字来自 Google 2010 年发表的一篇大规模图计算论文核心思想是批量同步并行BSPBulk Synchronous Parallel。2.1 超步是什么把图的执行想象成一轮一轮的循环每一轮就叫一个超步。在一个超步内部框架会找出当前所有可以被触发的节点它们的前驱都已完成并行执行这些节点。每个节点执行时只读取上一步的共享状态并把要修改的内容写入自己的本地输出不直接改全局状态。等这一超步里所有节点都跑完框架统一收集它们的输出按 Reducer 规则合并进全局状态进入下一个超步。节点B节点A共享状态Runtime调用方节点B节点A共享状态Runtime调用方 超步 1 开始 par[并行执行] 超步 1 结束 超步 2 开始 invoke(initial_state)读取当前状态派发 state返回局部更新 {x: 1}派发 state返回局部更新 {y: 2}按Reducer合并 {x:1, y:2}读取合并后状态派发新状态返回更新合并这种算的时候互不干扰写的时候统一合并的设计是 LangGraph 能够安全支持并行和循环的根本。2.2 为什么要用超步而不是直接顺序执行你可能会问既然 Python 本身是顺序执行的为什么不直接节点A(); 节点B()原因有三点并行安全同一个超步内的多个节点可以真正并行默认线程池它们的写操作不会互相踩踏因为合并是最后统一做的。循环可控循环本质上就是一个超步跑完根据状态决定下一个超步跑哪些节点天然契合 BSP 模型。可恢复每个超步结束时的状态都是一个天然的检查点配合 Checkpointer 可以实现时间旅行式调试第 11 篇详解。理解了超步你就理解了为什么 LangGraph 的状态需要合并器Reducer——因为多个并行节点可能同时往同一个字段写必须有一套规则来裁决最终值。这就是下一章的主角。三、四要素详解State、Node、Edge、Reducer3.1 State共享状态的结构定义State 是整张图的数据总线。在 LangGraph 里State 通常用一个TypedDict来定义它声明了所有节点可以读写的字段。fromtypingimportTypedDict,Annotatedfromoperatorimportadd# 最简单的状态用 TypedDict 声明字段classResearchState(TypedDict):topic:str# 研究主题summary:str# 研究摘要references:list[str]# 参考文献这里有一个关键约定默认情况下节点返回的字段会整体覆盖原状态中的同名字段。比如节点返回{summary: 新摘要}那么状态里的summary就被替换成新摘要。这种last write wins的语义对大多数字段是合理的。但有些字段我们希望累加而不是覆盖——比如对话历史、参考文献列表。这时就要用到Reducer。3.2 Reducer自定义合并规则Reducer 通过Annotated[类型, 合并函数]来声明。它告诉框架当多个节点或多个超步向这个字段写入时请用我指定的函数来合并而不是直接覆盖。fromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassResearchState(TypedDict):topic:strsummary:str# 用 add 作为 reducer新值会和旧值做列表拼接references:Annotated[list[str],add]operator.add作用于两个列表时就是拼接所以这行声明的效果是每个节点返回的references会被追加到已有列表后面而不是覆盖。LangGraph 还专门为消息历史提供了一个内置 reduceradd_messages第 7 篇会专门讲 Reducer 设计这里先有个印象即可。3.3 Node节点是普通函数节点就是普通的 Python 函数它接收当前状态返回一个部分状态字典只包含要更新的字段。这种只返回变化量的风格和 Pregel 超步模型完美契合。defresearch_node(state:ResearchState)-dict:负责检索的节点根据主题查找参考资料。topicstate[topic]# 这里用模拟数据实际可接入搜索 APIfound[f《{topic}综述》,f《{topic}前沿进展》]# 只返回要更新的字段注意 references 带 reducer会自动追加return{references:found}defsummarize_node(state:ResearchState)-dict:负责摘要的节点把参考资料压缩成一段摘要。refsstate[references]summaryf共找到{len(refs)}篇参考主题为{state[topic]}。return{summary:summary}两个要点节点不必返回完整状态只返回增量即可。未提到的字段保持不变。节点的入参类型注解state: ResearchState既是文档也方便 IDE 补全但运行时框架不会强制校验类型。3.4 Edge边定义控制流边决定了节点之间的执行顺序。LangGraph 有三种边边的类型API含义普通边add_edge(a, b)执行完 a 一定执行 b条件边add_conditional_edges(a, 路由函数, 映射)执行完 a 后由路由函数决定下一个节点入口/出口set_entry_point / set_finish_point或add_edge(START, ...)指定起点和终点第 8 篇会深入条件边和动态路由本篇我们先用最基础的普通边。四、动手实操搭建第一张图理论讲完现在把四要素串起来完整地搭一张研究 → 摘要的线性图。4.1 完整可运行代码# pip install langgraphfromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graphimportStateGraph,START,END# ① 定义状态结构classResearchState(TypedDict):topic:strsummary:strreferences:Annotated[list[str],add]# 列表累加# ② 定义节点函数defresearch_node(state:ResearchState)-dict:检索节点根据主题查找参考资料。topicstate[topic]found[f《{topic}综述》,f《{topic}前沿进展》]return{references:found}defsummarize_node(state:ResearchState)-dict:摘要节点把参考资料压缩成摘要。refsstate[references]summaryf围绕「{state[topic]}」共找到{len(refs)}篇参考。return{summary:summary}# ③ 构建图graph_builderStateGraph(ResearchState)graph_builder.add_node(research,research_node)graph_builder.add_node(summarize,summarize_node)# ④ 连接边START → research → summarize → ENDgraph_builder.add_edge(START,research)graph_builder.add_edge(research,summarize)graph_builder.add_edge(summarize,END)# ⑤ 编译成可执行应用appgraph_builder.compile()# ⑥ 运行resultapp.invoke({topic:大模型推理优化,references:[]})print(result)运行后你会得到类似这样的输出键的顺序可能不同{topic: 大模型推理优化, references: [《大模型推理优化综述》, 《大模型推理优化前沿进展》], summary: 围绕「大模型推理优化」共找到 2 篇参考。}注意一个细节invoke时我们传入了初始状态{topic: ..., references: []}其中的references是空列表。因为research_node返回了{references: [...]}而references字段带addreducer所以框架执行的是[] [...] [...]。如果初始状态不传references框架会把它当作[]处理Reducer 字段的默认值是累加器的单位元结果一样。4.2 图的可视化结构上面这张图的拓扑非常简单是一条直线STARTresearch检索节点summarize摘要节点END虽然简单但它已经体现了 LangGraph 的完整骨架。后续所有复杂特性——条件分支、循环、并行、人工审批——都是在这副骨架上加肉。4.3 用 stream 观察每一步的状态变化invoke是一次性返回最终结果而stream可以让你看到每个超步结束后的中间状态这在调试时非常有用。# 用 stream 逐步观察foreventinapp.stream({topic:大模型推理优化,references:[]}):# 每个 event 是 {节点名: 该节点返回的更新}fornode_name,updateinevent.items():print(f[{node_name}] 产出增量:{update})输出大致是[research] 产出增量: {references: [《大模型推理优化综述》, 《大模型推理优化前沿进展》]} [summarize] 产出增量: {summary: 围绕「大模型推理优化」共找到 2 篇参考。}可以看到stream默认模式stream_modevalues的近邻暴露的是每个节点的增量输出这正是 Pregel 超步模型的体现每个节点只产生自己的局部更新合并由框架在后台完成。如果你想在每一步看到完整的合并后状态可以加stream_modeupdates之外的参数这部分我们在第 11 篇调试环节会详细对比几种模式。五、新手常见坑与排查清单把第一张图跑通之后我们来盘点几个几乎所有新手都会踩的坑帮你少走弯路。僵尸 1字段被意外覆盖现象某个列表字段明明多个节点都往里写了最后却只剩最后一个节点的结果。原因忘记给该字段加 Reducer。默认语义是覆盖所以后写的会冲掉先写的。解决用Annotated[list[str], add]声明该字段。# ❌ 错误会被覆盖classBadState(TypedDict):logs:list[str]# ✅ 正确带 reducer自动累加classGoodState(TypedDict):logs:Annotated[list[str],add]坑 2节点返回了未声明的字段现象节点返回了一个 State 里没有的键调用时抛KeyError或被静默忽略取决于版本。原因State Schema 是状态的契约框架只认识 Schema 里声明过的字段。解决把字段加进 TypedDict。如果你确实想要一个灵活的附加字段容器可以声明一个extra: dict字段用自定义 reducer 合并。坑 3入口/出口设置缺失现象编译时报错ValueError: Node…is a dead-end或找不到起始节点。原因每个连通的图必须有明确的起点连到START和终点连到END否则框架不知道从哪开始、在哪结束。解决用add_edge(START, 第一个节点)和add_edge(最后一个节点, END)显式声明。坑 4把图当成普通函数理解循环现象想实现如果摘要质量不够就重做却不知道在哪里写while。原因在 LangGraph 里循环不是用 Python 的while/for实现的而是用边指回上游节点 条件边判断是否继续来实现的。框架会自动按超步一轮轮跑直到命中退出条件。解决这是第 9 篇的主题届时我们会手写一个 ReAct 循环。这里只需记住循环 条件边 回指边。本篇小结知识点核心内容Pregel 超步模型图的执行被切成一轮轮超步每轮内节点并行执行、输出在轮末按 Reducer 统一合并支撑并行与循环State状态用TypedDict定义所有共享字段是节点的数据总线Node节点普通函数接收当前状态返回增量更新字典不必返回完整状态Edge边普通边add_edge、条件边add_conditional_edges、入口出口START/ENDReducer合并器通过Annotated[类型, 函数]声明默认覆盖operator.add做列表累加add_messages专用于消息历史编译与运行StateGraph(...).compile()得到可执行应用invoke一次性运行stream逐步观察增量常见坑字段未加 reducer 被覆盖、返回未声明字段、入口出口缺失、误用 while 实现循环下篇预告第 07 篇状态管理与 Reducer 设计 — 合并策略与自定义通道上一篇里我们初步接触了Annotated[list, add]这样的 Reducer 声明但只是浅尝辄止。下一篇会把 Reducer 单独拎出来讲透内置 reducer 有哪些、如何自定义合并策略、消息历史的add_messages到底做了什么、以及多 Agent 共享状态时如何设计 Schema 才能避免冲突。如果本篇内容对你有帮助欢迎点赞收藏有任何疑问欢迎在评论区交流。