1. 为什么LangGraph的执行核心叫Pregel刚开始接触LangGraph源码的时候我最大的困惑在于为什么一个AI Agent编排框架执行引擎的名字会叫Pregel这么绕口后来翻了几遍源码才意识到这个命名不是随便起的它背后是整套设计思想的来源——Google在2010年发表的Pregel图计算模型一个大规模图计算框架核心思想叫BSPBulk Synchronous Parallel整体同步并行。简单说BSP模型把计算过程拆成一个个超步superstep每个超步里所有节点并行执行自己的逻辑然后通过消息传递把自己的结果同步给其他节点等所有节点都算完了再进入下一个超步。LangGraph把这一套几乎原封不动地搬到了LLM Agent的执行场景里。我一开始也怀疑这种“复古”图计算模型到底适不适合LLM场景。但读代码的过程中我逐渐理解了Agent的执行链条本质上就是一个有向图——节点是LLM调用、工具调用边是数据流转和条件分支。Pregel式的超步同步机制天然适合这种需要全局状态一致性的编排场景。这篇文章我会带你从源码层面把这个执行引擎完整拆一遍。适合的人群是已经对LangGraph有基本了解至少写过几个Graph、想深入理解“节点到底是怎么被调度执行的”“State是怎么在节点之间流转的”“Checkpointer到底做了什么”以及准备在LangGraph之上做二次开发或写自定义节点的开发者。这篇文章不会教你写Agent而是告诉你Agent在框架内部是怎么跑起来的。2. LangGraph整体架构与Pregel执行引擎的定位2.1 LangGraph代码仓库结构一览在深入Pregel之前我们先看一眼LangGraph整个项目是怎么组织的。源码主要在libs/langgraph/src/下面几个核心目录分别是graph/用户直接打交道的API层包括StateGraph、MessageGraph以及状态图的构建逻辑。这一层是“面向用户”的我们定义的节点、边、条件边都在这里被转换成统一的数据结构。pregel/执行引擎本体。我看源码时最大的认知刷新就在这里——LangGraph真正的核心不在graph/而在pregel/。StateGraph最终会编译成一个Pregel对象所有节点的调度、状态的读写、超步的推进全在这个包里面完成。核心文件包括runtime.py、loop.py、types.py、io.py。channels/状态通道层。LangGraph里的State本质上是多个Channel的集合每个Channel有自己独立的读写语义。比如LastValueChannel只保留最新值TopicChannel作为消息队列累积历史消息。这一层定义了“状态如何被写入”。与中文社区常见的理解不同LangGraph里的Channel并不是像NATS那样用于跨节点传递消息而是节点读取和写入状态的标准读写层节点之间不直接通信一律通过Channel读写共享状态。checkpoint/状态持久化与恢复。这里实现了快照语义也就是每个超步执行完整个状态如何被序列化保存下来以及出错之后如何从最近的快照恢复。types.py整个框架类型定义的地方包括StateSnapshot、Send等核心数据结构。我对这个架构的整体印象是“API层很薄执行层很厚”。graph/里的大多数代码都是做校验和数据结构转换真正的复杂度全部下沉到了pregel/和channels/里。所以你要真正理解LangGraph的执行模型花时间在pregel/上就是必须的。2.2 一次任务执行的完整流程总览我们用一个最典型的开始app.invoke(input)看看底层的调用链是怎么走的。源码核心是这样一条路径Pregel.invoke → Pregel._run_once → Pregel._run_graph → Pregel._run_loop我自己在实际读代码时习惯把整条链路画成一张图分5个阶段理解输入校验与输入写入阶段invoke收到的输入dict会被统一转换成InputChannelValues然后通过_write_input写入对应的输入Channel。图执行阶段核心入口是_run_graph它会调用_run_loop来进入执行主循环。超步推进阶段_run_loop会不停调用_run_step每调一次就是推进一个超步。每个超步里调度节点、收集写入。输出提取阶段当图中没有更多待执行节点执行循环结束_run_graph通过_write_output把输出Channel的值整理成最终结果返回给用户。任务提交与清理阶段_run_once负责管理task的submit和清理工作对应_run_step结束时调用的_write分支逻辑。个人体验是第2、3阶段就是全篇最核心的部分后面我们把这两块彻底拆开。3. 核心调度器Channel、Task与超步循环3.1 Channel机制状态的读写如何实现我们先看Channel。LangGraph里有一句话值得记住节点不直接读输入也不直接写输出它只是收到一个input参数返回一个output字典真正把output写进状态的是Pregel框架本身。这样做的好处在于如果节点直接操作状态那并行执行的多个节点之间就会出现读写竞争通过Channel统一管理写操作Pregel可以在一个超步结束时做冲突检测和合并。先看一个具体的Channel实现LastValueChannel。在channels/base.py里抽象基类BaseChannel定义了四个核心接口update写入新值、get读取当前值、checkpoint生成快照、from_checkpoint从快照恢复。接口理解清楚了我们看一下具体的LastValueChannel.update实现逻辑def update(self, values): if not values: return if len(values) ! 1: raise InvalidUpdateError(...) self.value values[-1]内部的严格校验是每个超步一个Channel只能有一个写入值。为什么因为LastValue语义就是“只保留最新值”如果多个节点在同一超步写同一个Channel会互相覆盖。LangGraph选择直接报错而不是“后者覆盖前者”这样能尽早暴露图配置里的读写冲突。再看另一个重要的TopicChannel在部分版本中也叫Topic它的update方法则是把新值追加到内部的deque里get返回所有累积的值。这里有一个关键点不同Channel语义决定了状态读写的冲突规则。读者在看源码时不需要背接口只要理解“Channel就是状态的最小读写单元”这一层后面看动态图重构时会轻松很多。3.2 Task与TaskProtocol单个节点的执行单元Channel解决的是“状态怎么读写”Task解决的是“节点怎么被调度执行”。在pregel/types.py里Task的定义大致如下不同版本字段有出入但核心字段稳定class Task: name: str input: Any None triggers: list[str] path: tuple[tuple[str, ...], ...] writes: list[tuple[str, Any]] node: str ...关键字段理解name任务节点名。input该节点本轮收到的输入数据由Pregel从Channel读取后实际注入。triggers触发条件记录了哪些Channel的更新触发了这个Task。writes节点返回的写入数据这里存储节点输出对应的Channel名和值。path动态图专用标记当前任务在动态图展开路径中的位置用于后续的路径去重和检查点恢复。在BSP模型下Pregel并不是“边到了就立刻触发下游”而是先为每个应该执行的节点构造一个Task收集节点输出然后在一个超步结束时统一apply所有的writes。这就保证了所有节点在同一个超步看到的状态是一致的——不会出现A节点看到的是更新前状态、B节点看到的是更新后状态的错乱。3.3_run_step的核心逻辑拆解_run_step是执行循环里最核心的函数。我摘出关键处理流程async def _run_step(self, input, config, *, stream, interrupt_events): ... # 1. 构造本轮task tasks self._prepare_next_tasks(...) ... # 2. 并发执行所有task tasks list(await asyncio.gather(*tasks)) ... # 3. 收集每个task的writes for t in tasks: ... self._write_all(t.writes, ...)第一步是_prepare_next_tasks这是整个执行引擎的决策大脑。它接收上一步执行产生的writes经过三个关键逻辑得到本轮要执行的Tasks根据边的配置确定哪些Channel更新后会触发哪些节点对条件边根据上一步的输出内容动态决策下一步走哪个分支根据writes的结果判断哪些节点被“唤起”了生成对应Task。第二步是并发执行。这一步用的是asyncio.gather所以LangGraph中节点函数可以是async也可以是普通函数框架做了兼容适配。第三步是_write_all按Channel的语义更新状态。重点是这个也是个异步过程因为要经历Channel的update规则校验。回到源码主线_run_loop长这样简化async def _run_loop(self, input, config, *, stream, interrupt_events): while True: ... if not tasks: break ... self._write(input, ...) ... self._apply_writes(...) # 保存checkpoint循环的退出条件是_prepare_next_tasks返回空列表也就是没有任何可执行的节点了整个图运行结束。这里我想强调一个初读源码容易踩坑的地方_run_loop里的while True不是“死循环”而是“BSP超步推进循环”。每一轮循环对应一个超步一个图通常跑几个超步就会结束。但如果不小心在图中设计了一个环A→B→A且无限触发这个循环就真的会一直跑下去。这也是为什么LangGraph官方强烈建议条件边写终止逻辑不然就是无限执行。4. 图构建状态图如何变成Pregel对象4.1 StateGraph构建双链表的过程现在我们看从StateGraph到Pregel的编译过程这一部分不复杂但对理解全貌很有帮助。StateGraph里最核心的数据结构是一张双链表。看过源码的应该有印象self.nodes: dict[str, StateNode] self.edges: set[tuple[str, str]] self.branches: dict[str, dict[str, Branch]]其中StateNode内部维护了incoming_edges和outgoing_edges两个集合。add_node只是往nodes里塞一个节点定义add_edge是把边的两个端点连接起来add_conditional_edges则是构建分支逻辑。源码里validate阶段干了这么几件事检查是否有悬空的边引用、检查是否存在无入口的孤立节点除了START、把完整的边映射补全。然后compile阶段的大流程是可以通过直接读源码看到的核心类就是StateGraph.compile()它把整个图构建成Pregel对象。我整理一下这个过程4.2 compile阶段做了什么compile做的主要事情总结下来是五大步第一步validate检查。校验图结构合法性。第二步构建节点配置列表。每个节点要包含节点名、触发该节点的Channel列表比如START触发的节点通常triggers为[__start__]、节点的调用函数等。第三步构建Channel映射。根据用户传入的State schema为每个字段创建对应的Channel对象。第四步构建边的映射。将我们代码里写的add_edge、add_conditional_edges转换成执行引擎能理解的Edge映射结构。第五步实例化Pregel对象。编译完成后我们拿到的app不再是StateGraph而是一个Pregel对象。源码里这一点也体现得很明确CompiledStateGraph内部持有一个Pregel实例并在invoke时直接转发给它。这就是为什么我们说LangGraph的真正执行引擎是Pregel。StateGraph只是用于构建静态图的API壳编译之后一切都由Pregel接管。4.3 Checkpointer如何与图编译交互Checkpointer在编译时也可以传入。它的核心接口是BaseCheckpointSaver定义了put保存检查点、put_writes保存节点中间写入、get_tuple获取检查点等方法。当传入checkpointer后Pregel会在每个超步结束时自动把状态写入到checkpointer里。注意这里有个细节检查点保存有两种粒度一种是put保存整个图的完整状态快照另一种是put_writes保存每个Task的写入记录。这一步是LangGraph能够支持interrupt和resume的关键——因为执行中途的状态都被持久化了恢复时只需要从最近的检查点重新应用未完成的写入即可。我自己调试多轮Agent时对这个体会很深如果某一轮工具调用出错了我能把checkpoint恢复到出错前一步修补输入后继续跑不用从头开始——这套机制底层全靠这里的检查点设计支撑。5. 图执行机制动态图、条件边与中断恢复5.1 动态图SendAPI与动态分支Send是LangGraph里一个非常有意思的API官方文档里叫“动态图”。先说结论Send本身不是一个Channel而是一个数据结构。它的作用是允许一个节点在执行过程中动态创建多个下游任务。源码定义大致如下class Send(NamedTuple): node: str arg: Any一个节点返回Send(next_node, {data: ...})Pregel在_prepare_next_tasks阶段看到这个返回值后会为每一项Send动态创建对应的Task并直接指定该Task的输入数据完全绕开静态路由。这个机制典型的使用场景是“Map-Reduce”。比如一个节点要把一篇长文拆分成10个chunk做并行摘要。在一个普通图里你得写10个显式的节点才能并行。用Send就优雅很多def splitter(state): chunks split(state[text]) return [Send(summarize, {chunk: c}) for c in chunks]这个Send列表会让Pregel在下一个超步创建10个summarize任务的Task每个Task输入各自chunk。它们会在一个超步里并发执行最后把10个摘要合并回状态。5.2 条件边的实现原理条件边是LangGraph里最常用的“图路由”机制。它的源码核心在graph/state.py的add_conditional_edges方法和pregel/runtime.py的_prepare_next_tasks实现中。理解条件边的关键在于它不是一个运行时“if/else”而是在超步开始时决定“下一个超步执行哪些节点”。源码里的核心逻辑是节点A执行完成后Pregel拿到A的输出把输出作为参数传给条件函数条件函数返回一个路由字符串比如tool或end然后Pregel根据这个路由字符串找到对应的下游节点集合生成Task。有一个实用细节值得注意条件函数的返回值还可以是一个Send列表。这就把条件路由和动态图组合起来了可以实现“先判断再分发”的复杂路由逻辑实际项目里这个组合我用得很频繁。5.3 interrupt与断点续跑机制interrupt是LangGraph的一个重要功能源码实现相当精彩它的核心工作在pregel/loop.py的_run_step中。当图中某个节点调用了interrupt函数Pregel检测到这个中断信号后会做三件事保存当前完整状态到checkpointer返回一个__interrupt__对象给调用方包含中断信息和当前状态结束当前invoke调用。之后开发者拿到这个中断对象处理完外部输入比如人工审批结果再调用invoke并传入新的输入Pregel会从最近的检查点恢复执行把新输入作为中断节点的返回值继续往下跑。这个机制让LangGraph能实现“等待人工审批后再继续”的Agent流程比如写邮件发送前的人工确认、高危操作前的二次确认。5.4 版本更新对执行模型的影响顺着源码往下看最新版本里执行模型还有一个重要的运行分支支持流式输出与流式更新。旧版本里的_run_step是“收集完所有writes统一写状态”新版本支持stream_modeupdates或stream_modemessages时writes会实时推送给调用方。模式之间的区别核心在于是否等着拿到全部结果才写Channel。这个区分实际使用中影响很大比如对话场景想要“边生成边打字”的效果就必须使用支持流式更新的模式。6. BSP模型在Agent执行中的优势与局限6.1 为什么Agent编排场景需要“全局状态一致”我们把话题拉回去为什么LLM Agent的执行引擎要用BSP这种偏“传统”的模型我自己的体会是因为LLM Agent场景对全局一致性的要求极高。一个Agent的执行链条里可能有多个并行节点比如多个工具的并发调用、有分支比如“判断是否需要调用工具”、有回环比如“工具结果不符合预期重新让LLM规划”。如果每个节点都直接修改共享状态一旦有多条执行路径状态就会乱。BSP模型下每个节点的读取和写入被严格地隔离开读是上一超步结束时的状态写是本超步结束才统一应用。这种“读写分离”给了图编排一个非常干净的执行语义。6.2 超步机制的代价与应对但BSP也有代价每个超步结束都有一个全局同步屏障barrier所有节点必须等最慢的那个执行完才能一起进入下一步。这意味着如果某个节点执行特别慢比如一个长超时的工具调用整个图都得等它浪费算力与时间。LangGraph方案是用“Task级并发”来对冲这个代价同一超步内的所有Task用asyncio.gather并发执行这样慢节点的等待时间至少和别的节点重叠。超步间的延迟无法消除但在大多数Agent场景里这种延迟是完全可以接受的——毕竟一个节点的执行本身就可能好几秒钟多等一个超步的调度开销毫秒级完全可以忽略。6.3 与传统异步事件循环的对比如果你熟悉其他Agent编排框架比如直接手写asyncio或用带事件循环的任务队列系统可能会觉得Pregel这套有点“重”。但手写异步事件循环的问题在于状态管理、失败恢复、断点续跑几乎都要自己重新造轮子。Pregel模型的“同步屏障”虽然看起来笨但它天然提供了清晰的“步骤边界”。更重要的是这个步骤边界恰好和Agent的可观察性、可恢复性需求完全对齐——每个超步就是一次天然的事件审计点每个检查点就是一次天然的状态快照。这种对齐是Pregel模型在Agent场景里最大的价值所在。7. Checkpointer机制与状态持久化解析7.1 检查点生命周期老规矩我们从源码看。在checkpoint/base.py里BaseCheckpointSaver有四个关键方法get_tuple(config)按配置获取一个检查点元组。put(config, checkpoint, metadata, new_versions)保存一个新的检查点。put_writes(config, writes, task_id, task_path)保存某个Task的中间写入。get_next_version为Channel版本号生成器默认是单调递增。在pregel/loop.py的_apply_writes里可以看到每个超步结束时会检查如果当前通道有更新就以checkpoint_id为标识推进一个新检查点。这个检查点里存的是每个Channel的当前值、每个Channel的版本号以及图的执行状态。这里有一个很关键的设计Channel版本号version。LangGraph依赖版本号来判断某个状态字段是否发生了变化从而决定哪些节点需要在下一个超步被唤醒。这个机制在_prepare_next_tasks里会频繁出现。7.2 从检查点恢复的执行细节当我们需要恢复一个中断的执行调用invoke时带上config{configurable: {thread_id: ..., checkpoint_id: ...}}Pregel会根据checkpoint_id或thread_id找到对应检查点用检查点里的状态恢复所有Channel从检查点记录的待执行Tasks继续执行。恢复过程的源码实现在pregel/runtime.py的_initialize_state里。其中最值得注意的部分是resuming的判定逻辑如果当前检查点还有未完成的writes就需要先应用这些writes再从后续tasks开始执行。个人经验排查恢复类问题核心就是盯着resuming这个标志位和checkpoint_id的传递思路会清晰很多。7.3 自定义Checkpointer的接口要求我自己实现过自定义的Checkpointer基于PostgreSQL做了一个生产级存储方案。接口要求其实很简单继承BaseCheckpointSaver实现上面说的四个方法就行。但坑在细节上默认的SqliteSaver里存储的checkpoint数据是序列化后的JSON字符串字段是channel_values和channel_versions。如果自定义存储你要保证序列化格式兼容否则恢复时解析就会失败。另外一个常见坑是metadata的健壮性问题比如并发读写、多个线程同时写同一个thread_id导致版本冲突。如果你只是在本地实验直接用MemorySaver就够了性能很好上生产再换持久化实现。千万别在生产环境里用MemorySaver进程一重启所有会话全部丢失。8. 调试Pregel执行引擎我踩过的坑8.1 断点调试时的关键观察点读这类执行引擎源码时直接一上来打断点会迷失。我建议按以下顺序观察关键节点第一优先_prepare_next_tasks的返回值。这是理解每一轮超步“为什么执行这几个节点”的关键。断点打在这个函数返回前观察tasks列表看它包含哪些节点、每个task的input是什么、triggers是什么。第二优先_write_all和_apply_writes。观察一次图执行中状态是怎么一步步发生变化的Channel的value在每个超步后变成什么。这一步对理解“状态更新”特别有帮助。第三优先_run_loop的循环退出条件。看最后一轮tasks为空时的状态理解“为什么图判定执行结束了”。按照这个顺序打断点比漫无目的地step into高效太多。我第一次调试时就是从入口一层层跟进去结果在asyncio的内部细节里迷路半天。8.2 异步执行下的日志追踪技巧用asyncio.gather并发执行节点时直接打印日志常常出现乱序。我自己的做法是给每个日志带上task.name和step编号。这就是调试LangGraph应用时的核心思路横向是超步编号纵向是节点名反复打印这两个信息执行的推进过程就会非常清楚。更实用的一个技巧是打印Channel版本变化print(snapshot.channel_versions)这样你能看到每个Channel是哪个超步发生变化的。这个信息对于定位“为什么这个节点没被触发”非常有帮助——大概率就是它的trigger channel版本没变。8.3 常见异常与解决方案速查下面这几种异常是我在调试LangGraph项目时遇到最多的InvalidUpdateError某个Channel在同一个超步内收到多个写值。这表明有多个并行节点往同一个LastValue类型的Channel写了数据。解决方法是要么把该Channel改成支持多个写值的类型如TopicChannel要么调整图结构避免同超步写冲突。GraphRecursionError超过了默认递归限制默认25步。通常是图中出现了环且缺少终止条件。解决方法是加条件边的退出分支或调大recursion_limit。我明确建议优先加退出分支调大一时的确是能“跑通”但内部隐藏的循环仍然在消耗预算。CheckpointError检查点恢复失败。最常见原因是自定义checkpointer的序列化格式不兼容或者thread_id/checkpoint_id传错。检查方向就两个存储的数据是否完整传入的配置是否与保存时一致。InvalidUpdateError还有个变种某个节点在超步1写了Channel A但超步2没有节点读取Channel A导致Channel A的值“悬空”。这种不算报错但往往意味着图配置有隐含问题——比如少加了一条边。遇到这种情况回到_prepare_next_tasks的断点处观察triggers会比较容易发现原因。8.4 调试实践中的一个完整案例说一个我之前做的LangGraph项目里的真实案例大致结构是入口节点判断用户问题是否涉及数据库查询是则调用工具节点查询工具结果送LLM生成最终答案最后校验答案是否满足要求不满足则重新规划。看起来很简单但实际运行中遇到一个问题第一次执行完成后第二次执行同样的输入结果状态里多出了上次执行的历史残留。排查下来发现状态Channel用的是LastValueChannel单看没问题但是图上有个回环让同一个Channel在同一个超步收到了多个写入值导致状态异常。最后解决方式拆出独立的临时状态Channel回环内写入用独立变量避免与总状态冲突。这类问题用上面说的“打印channel_versions”能快速定位你不试一次永远不知道这方法有多好用。9. 手写一个最小Pregel执行引擎读源码读到一定程度最有价值的验证方式是自己动手写一个简化版。我当初用不到200行Python实现了一个极简Pregel核心只保留最核心的超步调度、任务执行、状态读写逻辑。虽然没有LangGraph那么多功能但它验证了我对执行模型的理解是否真的通了。9.1 两百行代码实现核心调度核心数据结构就三个Channel存储池、Task列表、超步循环。核心代码如下关键部分import asyncio from collections import defaultdict class MiniPregel: def __init__(self, nodes, edges, channels): self.nodes nodes # {name: callable} self.edges edges # {src: [dst...]} self.channels channels # {name: value} self.pending_writes defaultdict(list) async def _execute_task(self, task): # 任务执行调用节点函数收集写入 result task[func](task[input]) if asyncio.iscoroutine(result): result await result return result def _prepare_next_tasks(self, writes): tasks [] for channel, value in writes: for src, dsts in self.edges.items(): if src in self._triggers_for(channel): for dst in dsts: tasks.append({ node: dst, func: self.nodes[dst], input: {channel: channel, value: value}, }) return tasks async def run(self, initial_input): # 初始输入写入 self.pending_writes[__input__].append(initial_input) step 0 while True: step 1 writes self.pending_writes if not writes: break self.pending_writes defaultdict(list) tasks self._prepare_next_tasks(writes) if not tasks: break results await asyncio.gather( *[self._execute_task(t) for t in tasks] ) for task, result in zip(tasks, results): self.pending_writes[task[node]].append(result) return self.channels这段代码把_run_loop/_run_step/_prepare_next_tasks的核心逻辑压缩到了极简。当然真实性上它跳过了很多细节——比如Channel版本管理、条件边、任务路径、检查点等。但超步的推进逻辑、Task的生成与执行、writes的收集与分发这三个核心循环全在。9.2 为极简实现补充条件边与状态版本如果要加条件边只需要在_prepare_next_tasks里加一个判断某个边的触发条件函数返回的字符串决定走哪个目标节点。如果要加版本管理给每个Channel挂一个version整数只在version变化时才生成任务。这些加完之后这个极简实现就有LangGraph核心功能的一个可运行影子了。9.3 从手写实现反观LangGraph的工程设计手写这个极简版本的过程中我最大的收获是LangGraph的名设计并非炫技而是每一个设计都在解决一个具体问题。_prepare_next_tasks与_execute_task分离让“执行”和“调度”解耦。准备任务时完全不执行任何节点函数执行时完全不修改状态。这种分离让超步语义非常纯粹。Channel读写分离节点返回的是“待写入数据”writes而不是“状态更新操作”。这使Pregel可以像数据库事务一样在一个超步结束时统一提交。asyncio.gather执行Task让并行成为默认行为而不需要用户显式写parallel的API。用户只需要用add_edge把多个节点挂在一个上游节点下它们就会自动并行。检查点与版本绑定每个Channel的版本号是执行空间里“敏感信息”的代理一切“是否需要触发”的判断都从这里出发。10. 阅读Pregel源码的实操路径建议如果你准备自己打开源码读一遍我提供一条基于我自身摸索过的路径按这条路线走效率会比对着官方文档硬啃好很多。第一步准备好工具和版本。建议直接clone源码仓库本地跑通一个官方的quickstart示例确保环境没问题。然后选择一个固定版本读避免版本漂移造成的理解冲突。第二步从StateGraph.compile()开始。先看API层怎么把用户代码转换成内部配置。看到Pregel的构造函数时先停下来把它的参数列表完整过一遍。第三步进入pregel/runtime.py找到_run_step就以它为圆心向外扩散阅读。先读_prepare_next_tasks然后读_execute_tasks最后读_apply_writes。第四步回到上层看pregel/loop.py关注_run_loop和_apply_writes中的checkpoint交互。第五步去channels/里挑两个典型ChannelLastValueChannel和TopicChannel精读实现。第六步配合做实验验证理解。每次改源码加打印信息后跑一个简单示例观察执行流程变化。我再补充三个对新手最实用的建议不要一开始读utils或constants里的代码意义不大。先忽略流式处理和中断处理相关的代码主线理解之后再回头补。源码不要只看一遍我读了三遍才把整体框架和细节串起来。11. 从源码分析到实际调优光读懂源码还不够我们要能把它转化成实际的调优能力。LangGraph的性能瓶颈通常在节点内部的LLM调用耗时而不是框架本身。但有几个框架层面的调优点值得注意第一减小Channel的读写量。每次状态写入都会触发序列化和检查点保存。如果State里放了很大的数据比如整个对话历史、全量文档每个超步的Checkpointer开销会很可观。实践经验上尽量让Channel存“摘要”或“引用”而不是“全量原文”能明显减少检查点写入时的瓶颈。第二适当调整recursion_limit。默认25步对于复杂Agent可能不够但也不要迷信调大这个参数——它本质上是执行安全阀。与其无脑调大不如检查图的逻辑是否可以减少超步数。第三利用asyncio.gather的天然并发。如果你有多个独立的工具调用尽量把它们设计成同一个父节点下的多个子节点让它们在同一个超步内并发执行而不是串行地在多个超步里逐个调用。第四Checkpointer的选择会影响整个执行链路的吞吐。MemorySaver适合开发调试生产场景建议用PostgreSQL或Redis做持久化存储并且定期清理老thread的检查点避免存储膨胀。第五合理使用流式模式。需要逐token输出时用stream_modemessages只需要获取最终更新时用stream_modeupdates。模式选错要么响应慢要么信息过载都很影响使用体验。12. 我读完整个执行引擎后的几个体会源码读到后期我反而觉得Pregel没那么神秘了。它核心就是两件事用超步把“计算”切成离散的节拍器用Channel把“状态”纳入统一的读写协议。LangGraph在这两者之上再选择性地引入检查点、中断、流式之后才变成一个适合Agent编排的执行引擎。我个人在实际使用中还有一个比较深的体会是Agent框架的演化一定会越来越重视底层执行引擎的表达能力。Pregel模型擅长静态图、并行、可恢复但如果Agent将自己的执行拆得足够细并且想在子任务之间引入细粒度的异步框架就需要在BSP范式里继续做扩展。最后分享一个小技巧如果你也是源码阅读型学习者建议给pregel/runtime.py里那句核心的while循环加一处临时日志打印每个超步后各Channel的版本号再跑一个简单的多节点示例。那个版本的输出信息比任何文档都更能让你直观地理解LangGraph在每个时间片里到底在做什么。我第一次看到超步版本推进的日志时之前所有关于这张执行引擎的模糊认知一下就全串起来了。 
长河君 · 2026/9/17 4:25:26 · 阅读 1.2k LangGraph核心执行引擎Pregel源码深度解析：BSP模型与超步调度机制
1. 为什么LangGraph的执行核心叫Pregel刚开始接触LangGraph源码的时候我最大的困惑在于为什么一个AI Agent编排框架执行引擎的名字会叫Pregel这么绕口后来翻了几遍源码才意识到这个命名不是随便起的它背后是整套设计思想的来源——Google在2010年发表的Pregel图计算模型一个大规模图计算框架核心思想叫BSPBulk Synchronous Parallel整体同步并行。简单说BSP模型把计算过程拆成一个个超步superstep每个超步里所有节点并行执行自己的逻辑然后通过消息传递把自己的结果同步给其他节点等所有节点都算完了再进入下一个超步。LangGraph把这一套几乎原封不动地搬到了LLM Agent的执行场景里。我一开始也怀疑这种“复古”图计算模型到底适不适合LLM场景。但读代码的过程中我逐渐理解了Agent的执行链条本质上就是一个有向图——节点是LLM调用、工具调用边是数据流转和条件分支。Pregel式的超步同步机制天然适合这种需要全局状态一致性的编排场景。这篇文章我会带你从源码层面把这个执行引擎完整拆一遍。适合的人群是已经对LangGraph有基本了解至少写过几个Graph、想深入理解“节点到底是怎么被调度执行的”“State是怎么在节点之间流转的”“Checkpointer到底做了什么”以及准备在LangGraph之上做二次开发或写自定义节点的开发者。这篇文章不会教你写Agent而是告诉你Agent在框架内部是怎么跑起来的。2. LangGraph整体架构与Pregel执行引擎的定位2.1 LangGraph代码仓库结构一览在深入Pregel之前我们先看一眼LangGraph整个项目是怎么组织的。源码主要在libs/langgraph/src/下面几个核心目录分别是graph/用户直接打交道的API层包括StateGraph、MessageGraph以及状态图的构建逻辑。这一层是“面向用户”的我们定义的节点、边、条件边都在这里被转换成统一的数据结构。pregel/执行引擎本体。我看源码时最大的认知刷新就在这里——LangGraph真正的核心不在graph/而在pregel/。StateGraph最终会编译成一个Pregel对象所有节点的调度、状态的读写、超步的推进全在这个包里面完成。核心文件包括runtime.py、loop.py、types.py、io.py。channels/状态通道层。LangGraph里的State本质上是多个Channel的集合每个Channel有自己独立的读写语义。比如LastValueChannel只保留最新值TopicChannel作为消息队列累积历史消息。这一层定义了“状态如何被写入”。与中文社区常见的理解不同LangGraph里的Channel并不是像NATS那样用于跨节点传递消息而是节点读取和写入状态的标准读写层节点之间不直接通信一律通过Channel读写共享状态。checkpoint/状态持久化与恢复。这里实现了快照语义也就是每个超步执行完整个状态如何被序列化保存下来以及出错之后如何从最近的快照恢复。types.py整个框架类型定义的地方包括StateSnapshot、Send等核心数据结构。我对这个架构的整体印象是“API层很薄执行层很厚”。graph/里的大多数代码都是做校验和数据结构转换真正的复杂度全部下沉到了pregel/和channels/里。所以你要真正理解LangGraph的执行模型花时间在pregel/上就是必须的。2.2 一次任务执行的完整流程总览我们用一个最典型的开始app.invoke(input)看看底层的调用链是怎么走的。源码核心是这样一条路径Pregel.invoke → Pregel._run_once → Pregel._run_graph → Pregel._run_loop我自己在实际读代码时习惯把整条链路画成一张图分5个阶段理解输入校验与输入写入阶段invoke收到的输入dict会被统一转换成InputChannelValues然后通过_write_input写入对应的输入Channel。图执行阶段核心入口是_run_graph它会调用_run_loop来进入执行主循环。超步推进阶段_run_loop会不停调用_run_step每调一次就是推进一个超步。每个超步里调度节点、收集写入。输出提取阶段当图中没有更多待执行节点执行循环结束_run_graph通过_write_output把输出Channel的值整理成最终结果返回给用户。任务提交与清理阶段_run_once负责管理task的submit和清理工作对应_run_step结束时调用的_write分支逻辑。个人体验是第2、3阶段就是全篇最核心的部分后面我们把这两块彻底拆开。3. 核心调度器Channel、Task与超步循环3.1 Channel机制状态的读写如何实现我们先看Channel。LangGraph里有一句话值得记住节点不直接读输入也不直接写输出它只是收到一个input参数返回一个output字典真正把output写进状态的是Pregel框架本身。这样做的好处在于如果节点直接操作状态那并行执行的多个节点之间就会出现读写竞争通过Channel统一管理写操作Pregel可以在一个超步结束时做冲突检测和合并。先看一个具体的Channel实现LastValueChannel。在channels/base.py里抽象基类BaseChannel定义了四个核心接口update写入新值、get读取当前值、checkpoint生成快照、from_checkpoint从快照恢复。接口理解清楚了我们看一下具体的LastValueChannel.update实现逻辑def update(self, values): if not values: return if len(values) ! 1: raise InvalidUpdateError(...) self.value values[-1]内部的严格校验是每个超步一个Channel只能有一个写入值。为什么因为LastValue语义就是“只保留最新值”如果多个节点在同一超步写同一个Channel会互相覆盖。LangGraph选择直接报错而不是“后者覆盖前者”这样能尽早暴露图配置里的读写冲突。再看另一个重要的TopicChannel在部分版本中也叫Topic它的update方法则是把新值追加到内部的deque里get返回所有累积的值。这里有一个关键点不同Channel语义决定了状态读写的冲突规则。读者在看源码时不需要背接口只要理解“Channel就是状态的最小读写单元”这一层后面看动态图重构时会轻松很多。3.2 Task与TaskProtocol单个节点的执行单元Channel解决的是“状态怎么读写”Task解决的是“节点怎么被调度执行”。在pregel/types.py里Task的定义大致如下不同版本字段有出入但核心字段稳定class Task: name: str input: Any None triggers: list[str] path: tuple[tuple[str, ...], ...] writes: list[tuple[str, Any]] node: str ...关键字段理解name任务节点名。input该节点本轮收到的输入数据由Pregel从Channel读取后实际注入。triggers触发条件记录了哪些Channel的更新触发了这个Task。writes节点返回的写入数据这里存储节点输出对应的Channel名和值。path动态图专用标记当前任务在动态图展开路径中的位置用于后续的路径去重和检查点恢复。在BSP模型下Pregel并不是“边到了就立刻触发下游”而是先为每个应该执行的节点构造一个Task收集节点输出然后在一个超步结束时统一apply所有的writes。这就保证了所有节点在同一个超步看到的状态是一致的——不会出现A节点看到的是更新前状态、B节点看到的是更新后状态的错乱。3.3_run_step的核心逻辑拆解_run_step是执行循环里最核心的函数。我摘出关键处理流程async def _run_step(self, input, config, *, stream, interrupt_events): ... # 1. 构造本轮task tasks self._prepare_next_tasks(...) ... # 2. 并发执行所有task tasks list(await asyncio.gather(*tasks)) ... # 3. 收集每个task的writes for t in tasks: ... self._write_all(t.writes, ...)第一步是_prepare_next_tasks这是整个执行引擎的决策大脑。它接收上一步执行产生的writes经过三个关键逻辑得到本轮要执行的Tasks根据边的配置确定哪些Channel更新后会触发哪些节点对条件边根据上一步的输出内容动态决策下一步走哪个分支根据writes的结果判断哪些节点被“唤起”了生成对应Task。第二步是并发执行。这一步用的是asyncio.gather所以LangGraph中节点函数可以是async也可以是普通函数框架做了兼容适配。第三步是_write_all按Channel的语义更新状态。重点是这个也是个异步过程因为要经历Channel的update规则校验。回到源码主线_run_loop长这样简化async def _run_loop(self, input, config, *, stream, interrupt_events): while True: ... if not tasks: break ... self._write(input, ...) ... self._apply_writes(...) # 保存checkpoint循环的退出条件是_prepare_next_tasks返回空列表也就是没有任何可执行的节点了整个图运行结束。这里我想强调一个初读源码容易踩坑的地方_run_loop里的while True不是“死循环”而是“BSP超步推进循环”。每一轮循环对应一个超步一个图通常跑几个超步就会结束。但如果不小心在图中设计了一个环A→B→A且无限触发这个循环就真的会一直跑下去。这也是为什么LangGraph官方强烈建议条件边写终止逻辑不然就是无限执行。4. 图构建状态图如何变成Pregel对象4.1 StateGraph构建双链表的过程现在我们看从StateGraph到Pregel的编译过程这一部分不复杂但对理解全貌很有帮助。StateGraph里最核心的数据结构是一张双链表。看过源码的应该有印象self.nodes: dict[str, StateNode] self.edges: set[tuple[str, str]] self.branches: dict[str, dict[str, Branch]]其中StateNode内部维护了incoming_edges和outgoing_edges两个集合。add_node只是往nodes里塞一个节点定义add_edge是把边的两个端点连接起来add_conditional_edges则是构建分支逻辑。源码里validate阶段干了这么几件事检查是否有悬空的边引用、检查是否存在无入口的孤立节点除了START、把完整的边映射补全。然后compile阶段的大流程是可以通过直接读源码看到的核心类就是StateGraph.compile()它把整个图构建成Pregel对象。我整理一下这个过程4.2 compile阶段做了什么compile做的主要事情总结下来是五大步第一步validate检查。校验图结构合法性。第二步构建节点配置列表。每个节点要包含节点名、触发该节点的Channel列表比如START触发的节点通常triggers为[__start__]、节点的调用函数等。第三步构建Channel映射。根据用户传入的State schema为每个字段创建对应的Channel对象。第四步构建边的映射。将我们代码里写的add_edge、add_conditional_edges转换成执行引擎能理解的Edge映射结构。第五步实例化Pregel对象。编译完成后我们拿到的app不再是StateGraph而是一个Pregel对象。源码里这一点也体现得很明确CompiledStateGraph内部持有一个Pregel实例并在invoke时直接转发给它。这就是为什么我们说LangGraph的真正执行引擎是Pregel。StateGraph只是用于构建静态图的API壳编译之后一切都由Pregel接管。4.3 Checkpointer如何与图编译交互Checkpointer在编译时也可以传入。它的核心接口是BaseCheckpointSaver定义了put保存检查点、put_writes保存节点中间写入、get_tuple获取检查点等方法。当传入checkpointer后Pregel会在每个超步结束时自动把状态写入到checkpointer里。注意这里有个细节检查点保存有两种粒度一种是put保存整个图的完整状态快照另一种是put_writes保存每个Task的写入记录。这一步是LangGraph能够支持interrupt和resume的关键——因为执行中途的状态都被持久化了恢复时只需要从最近的检查点重新应用未完成的写入即可。我自己调试多轮Agent时对这个体会很深如果某一轮工具调用出错了我能把checkpoint恢复到出错前一步修补输入后继续跑不用从头开始——这套机制底层全靠这里的检查点设计支撑。5. 图执行机制动态图、条件边与中断恢复5.1 动态图SendAPI与动态分支Send是LangGraph里一个非常有意思的API官方文档里叫“动态图”。先说结论Send本身不是一个Channel而是一个数据结构。它的作用是允许一个节点在执行过程中动态创建多个下游任务。源码定义大致如下class Send(NamedTuple): node: str arg: Any一个节点返回Send(next_node, {data: ...})Pregel在_prepare_next_tasks阶段看到这个返回值后会为每一项Send动态创建对应的Task并直接指定该Task的输入数据完全绕开静态路由。这个机制典型的使用场景是“Map-Reduce”。比如一个节点要把一篇长文拆分成10个chunk做并行摘要。在一个普通图里你得写10个显式的节点才能并行。用Send就优雅很多def splitter(state): chunks split(state[text]) return [Send(summarize, {chunk: c}) for c in chunks]这个Send列表会让Pregel在下一个超步创建10个summarize任务的Task每个Task输入各自chunk。它们会在一个超步里并发执行最后把10个摘要合并回状态。5.2 条件边的实现原理条件边是LangGraph里最常用的“图路由”机制。它的源码核心在graph/state.py的add_conditional_edges方法和pregel/runtime.py的_prepare_next_tasks实现中。理解条件边的关键在于它不是一个运行时“if/else”而是在超步开始时决定“下一个超步执行哪些节点”。源码里的核心逻辑是节点A执行完成后Pregel拿到A的输出把输出作为参数传给条件函数条件函数返回一个路由字符串比如tool或end然后Pregel根据这个路由字符串找到对应的下游节点集合生成Task。有一个实用细节值得注意条件函数的返回值还可以是一个Send列表。这就把条件路由和动态图组合起来了可以实现“先判断再分发”的复杂路由逻辑实际项目里这个组合我用得很频繁。5.3 interrupt与断点续跑机制interrupt是LangGraph的一个重要功能源码实现相当精彩它的核心工作在pregel/loop.py的_run_step中。当图中某个节点调用了interrupt函数Pregel检测到这个中断信号后会做三件事保存当前完整状态到checkpointer返回一个__interrupt__对象给调用方包含中断信息和当前状态结束当前invoke调用。之后开发者拿到这个中断对象处理完外部输入比如人工审批结果再调用invoke并传入新的输入Pregel会从最近的检查点恢复执行把新输入作为中断节点的返回值继续往下跑。这个机制让LangGraph能实现“等待人工审批后再继续”的Agent流程比如写邮件发送前的人工确认、高危操作前的二次确认。5.4 版本更新对执行模型的影响顺着源码往下看最新版本里执行模型还有一个重要的运行分支支持流式输出与流式更新。旧版本里的_run_step是“收集完所有writes统一写状态”新版本支持stream_modeupdates或stream_modemessages时writes会实时推送给调用方。模式之间的区别核心在于是否等着拿到全部结果才写Channel。这个区分实际使用中影响很大比如对话场景想要“边生成边打字”的效果就必须使用支持流式更新的模式。6. BSP模型在Agent执行中的优势与局限6.1 为什么Agent编排场景需要“全局状态一致”我们把话题拉回去为什么LLM Agent的执行引擎要用BSP这种偏“传统”的模型我自己的体会是因为LLM Agent场景对全局一致性的要求极高。一个Agent的执行链条里可能有多个并行节点比如多个工具的并发调用、有分支比如“判断是否需要调用工具”、有回环比如“工具结果不符合预期重新让LLM规划”。如果每个节点都直接修改共享状态一旦有多条执行路径状态就会乱。BSP模型下每个节点的读取和写入被严格地隔离开读是上一超步结束时的状态写是本超步结束才统一应用。这种“读写分离”给了图编排一个非常干净的执行语义。6.2 超步机制的代价与应对但BSP也有代价每个超步结束都有一个全局同步屏障barrier所有节点必须等最慢的那个执行完才能一起进入下一步。这意味着如果某个节点执行特别慢比如一个长超时的工具调用整个图都得等它浪费算力与时间。LangGraph方案是用“Task级并发”来对冲这个代价同一超步内的所有Task用asyncio.gather并发执行这样慢节点的等待时间至少和别的节点重叠。超步间的延迟无法消除但在大多数Agent场景里这种延迟是完全可以接受的——毕竟一个节点的执行本身就可能好几秒钟多等一个超步的调度开销毫秒级完全可以忽略。6.3 与传统异步事件循环的对比如果你熟悉其他Agent编排框架比如直接手写asyncio或用带事件循环的任务队列系统可能会觉得Pregel这套有点“重”。但手写异步事件循环的问题在于状态管理、失败恢复、断点续跑几乎都要自己重新造轮子。Pregel模型的“同步屏障”虽然看起来笨但它天然提供了清晰的“步骤边界”。更重要的是这个步骤边界恰好和Agent的可观察性、可恢复性需求完全对齐——每个超步就是一次天然的事件审计点每个检查点就是一次天然的状态快照。这种对齐是Pregel模型在Agent场景里最大的价值所在。7. Checkpointer机制与状态持久化解析7.1 检查点生命周期老规矩我们从源码看。在checkpoint/base.py里BaseCheckpointSaver有四个关键方法get_tuple(config)按配置获取一个检查点元组。put(config, checkpoint, metadata, new_versions)保存一个新的检查点。put_writes(config, writes, task_id, task_path)保存某个Task的中间写入。get_next_version为Channel版本号生成器默认是单调递增。在pregel/loop.py的_apply_writes里可以看到每个超步结束时会检查如果当前通道有更新就以checkpoint_id为标识推进一个新检查点。这个检查点里存的是每个Channel的当前值、每个Channel的版本号以及图的执行状态。这里有一个很关键的设计Channel版本号version。LangGraph依赖版本号来判断某个状态字段是否发生了变化从而决定哪些节点需要在下一个超步被唤醒。这个机制在_prepare_next_tasks里会频繁出现。7.2 从检查点恢复的执行细节当我们需要恢复一个中断的执行调用invoke时带上config{configurable: {thread_id: ..., checkpoint_id: ...}}Pregel会根据checkpoint_id或thread_id找到对应检查点用检查点里的状态恢复所有Channel从检查点记录的待执行Tasks继续执行。恢复过程的源码实现在pregel/runtime.py的_initialize_state里。其中最值得注意的部分是resuming的判定逻辑如果当前检查点还有未完成的writes就需要先应用这些writes再从后续tasks开始执行。个人经验排查恢复类问题核心就是盯着resuming这个标志位和checkpoint_id的传递思路会清晰很多。7.3 自定义Checkpointer的接口要求我自己实现过自定义的Checkpointer基于PostgreSQL做了一个生产级存储方案。接口要求其实很简单继承BaseCheckpointSaver实现上面说的四个方法就行。但坑在细节上默认的SqliteSaver里存储的checkpoint数据是序列化后的JSON字符串字段是channel_values和channel_versions。如果自定义存储你要保证序列化格式兼容否则恢复时解析就会失败。另外一个常见坑是metadata的健壮性问题比如并发读写、多个线程同时写同一个thread_id导致版本冲突。如果你只是在本地实验直接用MemorySaver就够了性能很好上生产再换持久化实现。千万别在生产环境里用MemorySaver进程一重启所有会话全部丢失。8. 调试Pregel执行引擎我踩过的坑8.1 断点调试时的关键观察点读这类执行引擎源码时直接一上来打断点会迷失。我建议按以下顺序观察关键节点第一优先_prepare_next_tasks的返回值。这是理解每一轮超步“为什么执行这几个节点”的关键。断点打在这个函数返回前观察tasks列表看它包含哪些节点、每个task的input是什么、triggers是什么。第二优先_write_all和_apply_writes。观察一次图执行中状态是怎么一步步发生变化的Channel的value在每个超步后变成什么。这一步对理解“状态更新”特别有帮助。第三优先_run_loop的循环退出条件。看最后一轮tasks为空时的状态理解“为什么图判定执行结束了”。按照这个顺序打断点比漫无目的地step into高效太多。我第一次调试时就是从入口一层层跟进去结果在asyncio的内部细节里迷路半天。8.2 异步执行下的日志追踪技巧用asyncio.gather并发执行节点时直接打印日志常常出现乱序。我自己的做法是给每个日志带上task.name和step编号。这就是调试LangGraph应用时的核心思路横向是超步编号纵向是节点名反复打印这两个信息执行的推进过程就会非常清楚。更实用的一个技巧是打印Channel版本变化print(snapshot.channel_versions)这样你能看到每个Channel是哪个超步发生变化的。这个信息对于定位“为什么这个节点没被触发”非常有帮助——大概率就是它的trigger channel版本没变。8.3 常见异常与解决方案速查下面这几种异常是我在调试LangGraph项目时遇到最多的InvalidUpdateError某个Channel在同一个超步内收到多个写值。这表明有多个并行节点往同一个LastValue类型的Channel写了数据。解决方法是要么把该Channel改成支持多个写值的类型如TopicChannel要么调整图结构避免同超步写冲突。GraphRecursionError超过了默认递归限制默认25步。通常是图中出现了环且缺少终止条件。解决方法是加条件边的退出分支或调大recursion_limit。我明确建议优先加退出分支调大一时的确是能“跑通”但内部隐藏的循环仍然在消耗预算。CheckpointError检查点恢复失败。最常见原因是自定义checkpointer的序列化格式不兼容或者thread_id/checkpoint_id传错。检查方向就两个存储的数据是否完整传入的配置是否与保存时一致。InvalidUpdateError还有个变种某个节点在超步1写了Channel A但超步2没有节点读取Channel A导致Channel A的值“悬空”。这种不算报错但往往意味着图配置有隐含问题——比如少加了一条边。遇到这种情况回到_prepare_next_tasks的断点处观察triggers会比较容易发现原因。8.4 调试实践中的一个完整案例说一个我之前做的LangGraph项目里的真实案例大致结构是入口节点判断用户问题是否涉及数据库查询是则调用工具节点查询工具结果送LLM生成最终答案最后校验答案是否满足要求不满足则重新规划。看起来很简单但实际运行中遇到一个问题第一次执行完成后第二次执行同样的输入结果状态里多出了上次执行的历史残留。排查下来发现状态Channel用的是LastValueChannel单看没问题但是图上有个回环让同一个Channel在同一个超步收到了多个写入值导致状态异常。最后解决方式拆出独立的临时状态Channel回环内写入用独立变量避免与总状态冲突。这类问题用上面说的“打印channel_versions”能快速定位你不试一次永远不知道这方法有多好用。9. 手写一个最小Pregel执行引擎读源码读到一定程度最有价值的验证方式是自己动手写一个简化版。我当初用不到200行Python实现了一个极简Pregel核心只保留最核心的超步调度、任务执行、状态读写逻辑。虽然没有LangGraph那么多功能但它验证了我对执行模型的理解是否真的通了。9.1 两百行代码实现核心调度核心数据结构就三个Channel存储池、Task列表、超步循环。核心代码如下关键部分import asyncio from collections import defaultdict class MiniPregel: def __init__(self, nodes, edges, channels): self.nodes nodes # {name: callable} self.edges edges # {src: [dst...]} self.channels channels # {name: value} self.pending_writes defaultdict(list) async def _execute_task(self, task): # 任务执行调用节点函数收集写入 result task[func](task[input]) if asyncio.iscoroutine(result): result await result return result def _prepare_next_tasks(self, writes): tasks [] for channel, value in writes: for src, dsts in self.edges.items(): if src in self._triggers_for(channel): for dst in dsts: tasks.append({ node: dst, func: self.nodes[dst], input: {channel: channel, value: value}, }) return tasks async def run(self, initial_input): # 初始输入写入 self.pending_writes[__input__].append(initial_input) step 0 while True: step 1 writes self.pending_writes if not writes: break self.pending_writes defaultdict(list) tasks self._prepare_next_tasks(writes) if not tasks: break results await asyncio.gather( *[self._execute_task(t) for t in tasks] ) for task, result in zip(tasks, results): self.pending_writes[task[node]].append(result) return self.channels这段代码把_run_loop/_run_step/_prepare_next_tasks的核心逻辑压缩到了极简。当然真实性上它跳过了很多细节——比如Channel版本管理、条件边、任务路径、检查点等。但超步的推进逻辑、Task的生成与执行、writes的收集与分发这三个核心循环全在。9.2 为极简实现补充条件边与状态版本如果要加条件边只需要在_prepare_next_tasks里加一个判断某个边的触发条件函数返回的字符串决定走哪个目标节点。如果要加版本管理给每个Channel挂一个version整数只在version变化时才生成任务。这些加完之后这个极简实现就有LangGraph核心功能的一个可运行影子了。9.3 从手写实现反观LangGraph的工程设计手写这个极简版本的过程中我最大的收获是LangGraph的名设计并非炫技而是每一个设计都在解决一个具体问题。_prepare_next_tasks与_execute_task分离让“执行”和“调度”解耦。准备任务时完全不执行任何节点函数执行时完全不修改状态。这种分离让超步语义非常纯粹。Channel读写分离节点返回的是“待写入数据”writes而不是“状态更新操作”。这使Pregel可以像数据库事务一样在一个超步结束时统一提交。asyncio.gather执行Task让并行成为默认行为而不需要用户显式写parallel的API。用户只需要用add_edge把多个节点挂在一个上游节点下它们就会自动并行。检查点与版本绑定每个Channel的版本号是执行空间里“敏感信息”的代理一切“是否需要触发”的判断都从这里出发。10. 阅读Pregel源码的实操路径建议如果你准备自己打开源码读一遍我提供一条基于我自身摸索过的路径按这条路线走效率会比对着官方文档硬啃好很多。第一步准备好工具和版本。建议直接clone源码仓库本地跑通一个官方的quickstart示例确保环境没问题。然后选择一个固定版本读避免版本漂移造成的理解冲突。第二步从StateGraph.compile()开始。先看API层怎么把用户代码转换成内部配置。看到Pregel的构造函数时先停下来把它的参数列表完整过一遍。第三步进入pregel/runtime.py找到_run_step就以它为圆心向外扩散阅读。先读_prepare_next_tasks然后读_execute_tasks最后读_apply_writes。第四步回到上层看pregel/loop.py关注_run_loop和_apply_writes中的checkpoint交互。第五步去channels/里挑两个典型ChannelLastValueChannel和TopicChannel精读实现。第六步配合做实验验证理解。每次改源码加打印信息后跑一个简单示例观察执行流程变化。我再补充三个对新手最实用的建议不要一开始读utils或constants里的代码意义不大。先忽略流式处理和中断处理相关的代码主线理解之后再回头补。源码不要只看一遍我读了三遍才把整体框架和细节串起来。11. 从源码分析到实际调优光读懂源码还不够我们要能把它转化成实际的调优能力。LangGraph的性能瓶颈通常在节点内部的LLM调用耗时而不是框架本身。但有几个框架层面的调优点值得注意第一减小Channel的读写量。每次状态写入都会触发序列化和检查点保存。如果State里放了很大的数据比如整个对话历史、全量文档每个超步的Checkpointer开销会很可观。实践经验上尽量让Channel存“摘要”或“引用”而不是“全量原文”能明显减少检查点写入时的瓶颈。第二适当调整recursion_limit。默认25步对于复杂Agent可能不够但也不要迷信调大这个参数——它本质上是执行安全阀。与其无脑调大不如检查图的逻辑是否可以减少超步数。第三利用asyncio.gather的天然并发。如果你有多个独立的工具调用尽量把它们设计成同一个父节点下的多个子节点让它们在同一个超步内并发执行而不是串行地在多个超步里逐个调用。第四Checkpointer的选择会影响整个执行链路的吞吐。MemorySaver适合开发调试生产场景建议用PostgreSQL或Redis做持久化存储并且定期清理老thread的检查点避免存储膨胀。第五合理使用流式模式。需要逐token输出时用stream_modemessages只需要获取最终更新时用stream_modeupdates。模式选错要么响应慢要么信息过载都很影响使用体验。12. 我读完整个执行引擎后的几个体会源码读到后期我反而觉得Pregel没那么神秘了。它核心就是两件事用超步把“计算”切成离散的节拍器用Channel把“状态”纳入统一的读写协议。LangGraph在这两者之上再选择性地引入检查点、中断、流式之后才变成一个适合Agent编排的执行引擎。我个人在实际使用中还有一个比较深的体会是Agent框架的演化一定会越来越重视底层执行引擎的表达能力。Pregel模型擅长静态图、并行、可恢复但如果Agent将自己的执行拆得足够细并且想在子任务之间引入细粒度的异步框架就需要在BSP范式里继续做扩展。最后分享一个小技巧如果你也是源码阅读型学习者建议给pregel/runtime.py里那句核心的while循环加一处临时日志打印每个超步后各Channel的版本号再跑一个简单的多节点示例。那个版本的输出信息比任何文档都更能让你直观地理解LangGraph在每个时间片里到底在做什么。我第一次看到超步版本推进的日志时之前所有关于这张执行引擎的模糊认知一下就全串起来了。