ARTICLE · 1209407

【LangGraph源码精读】第02期|并行执行不冲突?状态副本+确定序合并太巧妙了

发布时间：2026-10-19 10:41:06 最近访问：2026-10-19 10:41:09

【LangGraph源码精读】第02期｜并行执行不冲突？状态副本+确定序合并太巧妙了

30篇源码精读系列·第02篇（执行引擎层第二篇）。上一期讲了Pregel BSP模型的概念：Plan→Execute→Update三阶段。这一期深入源码，看这三个阶段在代码里到底是怎么实现的。主循环长什么样？prepare_next_tasks怎么从Channel更新反查出待执行节点？PregelRunner怎么保证并发安全？apply_writes怎么做到确定序合并？读完这篇，你能看懂PregelLoop的每一行核心代码。

一、开头：你的Agent并行执行时，内部到底在跑什么？

你写了一个LangGraph Agent，里面有"搜索"和"代码生成"两个独立节点，配置成并行执行。运行时你看到两个节点几乎同时完成，结果正确合并。

但你有没有想过：

● 两个节点同时读同一个状态字段，读到的是同一个值吗？
● 两个节点同时写同一个状态字段，谁覆盖谁？
● 如果一个节点失败了，另一个已经完成的结果会丢吗？
● 为什么同样的输入跑10次，结果永远一样？

这些问题的答案，都在PregelLoop的主循环里。

上一期讲了BSP三阶段的概念，这一期讲源码实现。 概念懂了不代表源码能看懂——很多人看Pregel的源码，看到loop.tick()、runner.tick()、after_tick()这三个方法就懵了，不知道谁先谁后、谁做什么。这篇把主循环彻底拆开。

二、主循环全景：一行代码看懂执行流程

Pregel的执行主循环，核心就是这一段（在pregel/main.py的stream方法里）：

步骤	方法	做什么	对应BSP阶段
1	loop.tick()	准备本轮任务：检查Channel更新，生成待执行任务列表	Plan
2	match_cached_writes()	缓存命中的任务直接复用结果，不重复执行	Plan（优化）
3	runner.tick(tasks)	并发执行所有待执行任务，结果写入缓冲区	Execute
4	loop.after_tick()	统一合并写入、更新Channel版本、保存Checkpoint	Update
5	循环回到1	检查是否有新任务，有则继续，无则终止	循环控制

关键理解：tick()是"准备"，runner.tick()是"执行"，after_tick()是"提交"。 三者严格按顺序执行，构成一个完整的Super-step。

这就是为什么LangGraph的执行是确定性的——不管节点并行执行的先后顺序如何，after_tick()总是在所有节点执行完之后才统一合并，合并顺序是确定的。

三、Plan阶段源码：prepare_next_tasks()怎么找出谁该干活？

prepare_next_tasks()在pregel/_algo.py里，是Plan阶段的核心。它的输入是当前Checkpoint和Channel状态，输出是待执行任务列表。

3.1 算法步骤

步骤	做什么	关键数据结构
1	检查所有Channel的版本号，找出上一个Super-step中被更新的Channel	channel_versions
2	通过trigger_to_nodes映射表，找到订阅这些Channel的节点	trigger_to_nodes（编译时构建的索引）
3	为每个节点构造一个PregelExecutableTask	PregelExecutableTask
4	将节点要读取的Channel快照、config、path、retry/cache信息装进task	task.input、task.config
5	合并恢复场景下已成功的writes，避免已完成节点重复执行	task.writes
6	返回task_id到PregelExecutableTask的映射	输出

3.2 PregelExecutableTask里装了什么？

PregelExecutableTask是执行的最小单位，它不是一个简单的函数调用，而是一个包含完整上下文的任务对象：

字段	说明	作用
name	节点名称	标识、日志、Checkpoint记录
input	节点要读取的Channel快照	Execute阶段节点只读这个快照，不读实时状态
proc	要执行的Runnable（用户节点函数包装后的对象）	真正被调用的执行体
writes	写入缓冲区（deque）	节点输出先存在这里，不直接写Channel
config	RunnableConfig（含thread_id、路径、retry/cache策略）	执行配置
triggers	触发该节点的Channel列表	记录"为什么这个节点本轮要执行"

最关键的设计：input是快照，writes是缓冲区。 节点执行期间，它看到的状态是冻结的（input快照），它产生的输出是隔离的（writes缓冲区）。这两个设计保证了并发安全。

3.3 trigger_to_nodes：编译时构建的触发索引

你在StateGraph里写的add_edge("A", "B")，编译后变成了什么？

你写的代码	编译后的结果
add_edge("A", "B")	B订阅A写入的Channel；trigger_to_nodes中记录"channel_of_A → [B]"
add_edge(START, "A")	A订阅输入Channel；trigger_to_nodes中记录"input_channel → [A]"
add_conditional_edges("A", router)	router函数执行后决定写入哪个Channel，不同Channel触发不同节点

节点不是被"调用"的，是被"触发"的。 A执行完写入Channel，Channel版本号变化，下一个Super-step的prepare_next_tasks()发现这个Channel被更新了，通过trigger_to_nodes找到B，把B加入待执行列表。

这就是为什么LangGraph支持循环——A→B→A的边，编译后就是"A的Channel触发B，B的Channel触发A"，每一轮Super-step自然形成循环。

四、Execute阶段源码：PregelRunner怎么保证并发安全？

PregelRunner在pregel/_runner.py里，负责Execute阶段的并发执行。它不决定图结构，只负责把已经选好的任务执行起来。

4.1 执行流程

步骤	做什么	关键机制
1	把tasks转成tuple，建立FuturesDict	每个task对应一个future
2	单任务、无timeout时走fast path，直接run_with_retry()	避免并发开销
3	多任务时通过submit()并发提交	线程池（Sync）或asyncio（Async）
4	使用FIRST_COMPLETED等待，一个任务完成就yield一次	支持流式输出边执行边输出
5	每个任务完成时回调commit()	将结果转为writes，不直接改Channel
6	所有future完成后，_panic_or_proceed()决定是否抛异常	错误处理

4.2 commit()：并发安全的核心

commit()是PregelRunner最关键的方法。它的语义是：节点执行结果不直接修改共享Channel，而是转为task writes，调用loop.put_writes(task_id, writes)。

传统做法（有问题）	LangGraph做法（安全）
节点执行完直接state["field"] = result	节点执行完把result放入task.writes缓冲区
并发写同一个dict → 数据竞争	每个task有独立的writes缓冲区 → 无竞争
需要加锁保护共享状态	不需要锁，缓冲区是线程私有的
执行顺序影响最终结果	执行顺序不影响，after_tick统一合并

这就是"状态副本+输出缓冲"的具体实现。 不是什么高深的技术，就是"每个线程写自己的缓冲区，最后统一合并"这个朴素的思想。但它解决了并发安全这个大问题。

4.3 错误处理

错误类型	处理方式
节点抛出可重试异常	run_with_retry()按RetryPolicy重试，不影响其他节点
节点抛出不可重试异常	commit()记录ERROR写入，after_tick时检查
配置了error_handler	错误被标记为handled，调度handler task执行
未处理的异常	_panic_or_proceed()抛出，整个Super-step失败

重试的粒度是task，不是整个graph。 一个节点失败重试，不会导致同Super-step中其他已成功的节点重新执行。这是BSP模型的另一个优势——失败隔离。

五、Update阶段源码：after_tick()怎么做到确定序合并？

after_tick()在pregel/_loop.py里，是Update阶段的核心。它的执行顺序非常关键，每一步都有严格的先后。

5.1 after_tick()的8步执行顺序

步骤	做什么	为什么这个顺序
1	收集所有task的writes	先收集全部，再统一处理
2	调用apply_writes()，将writes应用到Channel	核心合并逻辑
3	如果输出Channel被更新，发出values事件	合并后才能输出最终状态
4	保存exit durability需要的delta writes	延迟持久化模式
5	清空pending writes，关闭replay状态	准备下一轮
6	_put_checkpoint()保存新Checkpoint	合并完成后才保存，保证一致性
7	检查interrupt_after	Checkpoint保存后才能安全中断
8	清掉resuming标记	恢复流程结束

顺序不能乱。 比如Checkpoint必须在apply_writes之后保存——如果先保存再合并，Checkpoint里存的就是旧状态。interrupt_after必须在Checkpoint之后检查——如果先中断再保存，恢复时就找不到中断点了。

5.2 apply_writes()：确定序合并的实现

apply_writes()在pregel/_algo.py里，负责把所有task的writes合并到Channel。

步骤	做什么	关键机制
1	遍历所有task的writes	按task_id排序，保证确定顺序
2	对每个写入，找到对应的Channel	Channel名→Channel对象映射
3	调用Channel.update()方法，传入新值	Channel根据自己的类型和reducer合并
4	Channel版本号递增	每个被更新的Channel版本+1
5	检查冲突	两个节点写同一个LastValue Channel且无reducer → InvalidUpdateError
6	返回被更新的Channel列表	供下一轮Plan阶段使用

确定序合并的关键：按task_id排序遍历。 不管节点执行的先后顺序如何（因为并行），apply_writes总是按task_id的字典序来合并writes。所以同样的输入，永远产生同样的输出。

冲突检测是安全网。 如果两个节点在同一个Super-step里都写了同一个LastValue类型的Channel（覆盖语义），而且没有reducer，apply_writes会直接抛InvalidUpdateError。这不是bug，是设计——它逼着你在设计状态时就考虑好并行写入的合并策略。

5.3 Channel.update()：不同Channel类型的合并行为

Channel类型	update()行为	并行写入结果
LastValue	新值覆盖旧值	两个节点都写 → 冲突报错（无reducer时）
BinaryOperatorAggregate	用reducer函数合并（old, new）→ new	两个节点都写 → 按确定序依次reducer合并
Topic	追加到列表	两个节点都写 → 都追加，按确定序排列
EphemeralValue	只存一个Super-step	不持久化，下一轮自动清空
DeltaChannel	只存增量delta	减少序列化开销

最常用的是BinaryOperatorAggregate。 比如messages: Annotated[list, add_messages]，底层就是BinaryOperatorAggregate，reducer是add_messages。两个节点同时写messages，apply_writes会按确定序依次调用add_messages合并，结果是确定的。

六、恢复和幂等性：put_writes和pending writes

这是很多人忽略但非常重要的机制。LangGraph的持久化不是"每个Super-step保存一次Checkpoint"这么简单，它还有一层更细粒度的pending writes机制。

6.1 两层持久化

层级	时机	存什么	作用
第一层：put_writes	每个节点执行完立即保存	task级别的writes（节点名、Channel名、值）	节点执行到一半崩溃，已完成节点的结果不丢
第二层：put（Checkpoint）	每个Super-step结束（after_tick）	所有Channel的完整快照	Super-step边界的持久化，支持时间旅行

为什么需要两层？ 因为Super-step是持久化的最小单位——如果一个Super-step里有3个并行节点，其中2个成功了1个失败了，整个Super-step没有Checkpoint。如果只有Checkpoint这一层，恢复时3个节点都要重跑。有了put_writes，恢复时已经成功的2个节点的writes可以从pending writes恢复，只重跑失败的1个。

6.2 恢复时发生了什么？

步骤	做什么
1	PregelLoop启动时，从Checkpoint加载Channel状态
2	tick()调用prepare_next_tasks时，把pending writes中已成功的writes挂回对应task
3	有writes的task不需要执行（writes已经有结果了）
4	没有writes的task（失败的那个）正常执行
5	after_tick时，所有writes（恢复的+新执行的）统一合并

控制类writes不重放。 ERROR、INTERRUPT、RESUME这些控制信号不能按普通状态写入重放——它们代表的是异常状态，需要重新触发而不是直接恢复。

6.3 节点必须幂等

因为节点可能被重新执行（重试、恢复、时间旅行），所以节点函数必须是幂等的：

非幂等操作（有问题）	幂等做法
直接发邮件	先查是否已发，再发；或用唯一键去重
直接写数据库	用UPSERT或唯一键约束
调用外部API产生副作用	加幂等键，或把副作用移到interrupt之后
生成随机数作为ID	用输入派生的确定性ID

这是BSP模型的固有约束。 Super-step是原子单位，节点是可重放的。LangGraph不保证节点只执行一次，它保证的是"Super-step恰好提交一次"。节点层面是"至少一次"，所以幂等性是你的责任。

七、常见bug排查指南：6个问题定位到源码

读源码最大的价值是排查问题。以下是6个常见问题，以及对应的源码定位：

问题现象	第一跳	第二跳	重点看什么
节点没有执行	StateGraph.compile	prepare_next_tasks	Channel trigger是否正确、START边是否配置
节点执行顺序不对	Pregel.stream	tick/after_tick	BSP边界、哪些Channel被更新了
Agent无限循环	call_model	_are_more_steps_needed	模型是否持续返回tool_calls、路由是否到END
中断无法恢复	SyncPregelLoop	put_writes、checkpointer	thread_id是否正确、pending writes是否保存
stream没有token输出	Pregel.stream	StreamMessagesHandler	stream_mode是否正确、callback是否传递
并行结果不稳定	PregelRunner.tick	apply_writes	reducer是否可交换可结合、Super-step边界

这个排查表的价值： 遇到问题不用瞎猜，直接按表定位到具体源码文件和函数，读对应代码就能找到根因。

八、设计权衡：为什么是"准备-执行-提交"三段式？

PregelLoop的三段式设计（tick→runner.tick→after_tick）不是随意的，它是BSP模型的直接实现，但也有明确的权衡。

设计选择	好处	代价
Plan和Execute分离	Plan是纯计算（不涉及IO），可以快速判断是否终止	多了一次遍历Channel的开销
Execute和Update分离	Execute可以并行，Update是单线程确定序	快节点必须等慢节点（同步屏障）
输出缓冲不直接写Channel	并发安全、无锁、可复现	内存占用增加（每个task有独立缓冲区）
两层持久化（put_writes+Checkpoint）	细粒度恢复，已完成节点不重跑	实现复杂度增加、存储开销增加
确定序合并（按task_id排序）	结果100%可复现	不能利用执行顺序做优化

核心权衡：用内存和等待换确定性和可恢复性。 对于Agent编排来说，确定性比极致性能重要得多——Agent的瓶颈在大模型调用（秒级），不在执行引擎的同步屏障（毫秒级）。

九、面试关联：节点调度的5道高频题

面试题	答题要点	考察深度
LangGraph的主循环是什么结构？	while loop.tick() → runner.tick() → after_tick()，对应Plan/Execute/Update	整体流程
节点是怎么被触发执行的？	编译时构建trigger_to_nodes索引；Plan阶段检查Channel版本变化，通过索引找到待执行节点	触发机制
并发执行为什么不需要锁？	每个task有独立的input快照和writes缓冲区；commit()不直接改Channel，只写缓冲区；after_tick统一合并	并发安全
确定序合并是怎么实现的？	apply_writes按task_id排序遍历writes；Channel.update()根据reducer合并；冲突检测抛InvalidUpdateError	合并机制
节点失败恢复时，已成功节点会重跑吗？	不会。put_writes保存task级pending writes；恢复时已成功的writes挂回task，只重跑失败节点	恢复机制

十、本期小结

节点调度源码，核心就6句话：

1. 主循环是三段式：tick()准备→runner.tick()执行→after_tick()提交。 对应BSP的Plan/Execute/Update三阶段，严格按顺序执行，构成一个Super-step。
2. Plan阶段的核心是prepare_next_tasks()。 它检查Channel版本变化，通过编译时构建的trigger_to_nodes索引反查出待执行节点，为每个节点构造PregelExecutableTask（含input快照、writes缓冲区、config）。节点不是被调用的，是被Channel更新触发的。
3. Execute阶段的核心是PregelRunner和commit()。 多任务并发提交，FIRST_COMPLETED等待支持流式输出。commit()不直接修改共享Channel，而是把结果写入task.writes缓冲区——这是并发安全的关键，不需要锁。
4. Update阶段的核心是after_tick()和apply_writes()。 after_tick有严格的8步顺序（收集→合并→输出→清空→保存Checkpoint→检查中断）。apply_writes按task_id排序遍历，保证确定序合并；Channel.update()根据reducer合并；冲突检测抛InvalidUpdateError。
5. 两层持久化保证细粒度恢复。 put_writes在节点执行完立即保存task级writes，Checkpoint在Super-step结束保存完整快照。恢复时已成功节点的writes从pending writes恢复，只重跑失败节点。但节点必须幂等——LangGraph保证Super-step恰好提交一次，节点是至少一次执行。
6. 三段式设计的核心权衡是"用内存和等待换确定性和可恢复性"。 同步屏障让快节点等慢节点，输出缓冲增加内存占用，但换来了100%可复现的执行结果和完善的中断恢复能力。对于Agent编排，确定性比极致性能重要。

十一、下一期预告

下一期（源码03），执行引擎层第三篇，进入Channel系统：

【LangGraph源码精读】第03期｜你写的每个State字段，底层都是一个Channel

从LastValue到BinaryOperatorAggregate，从Topic到DeltaChannel，5种Channel类型逐个拆解，搞懂reducer和版本号是怎么驱动整个执行引擎的。

你在使用LangGraph时遇到过并行写入冲突或状态合并的问题吗？评论区聊聊，我优先写你最关心的场景。

本文为《LangGraph源码精读30讲》系列第02篇。内容基于LangGraph v1.2.11源码结构梳理，核心文件位于libs/langgraph/langgraph/pregel/目录下的_loop.py、_algo.py、_runner.py。版本更新后部分细节可能变化。如有错漏欢迎指正。