LangGraph — 架构与原理

30 秒导读: LangGraph 是一个底层 agent 编排框架。你把工作流画成一张图(节点=干活的函数,边=控制流),它负责把这张图跑起来——而且天生有状态、能暂停、能从崩溃点精确恢复。它的内核不是调用链,而是一套借鉴自 Google Pregel 的超步(superstep)批量执行模型。

1. 这是什么(零基础也能懂)

一句话定义: LangGraph 让你用图来组织一个 agent 或工作流——节点是函数,它们通过读写一块共享状态来协作,框架负责调度谁先跑、谁后跑,并在每一步把状态存盘。

解决什么问题 / 给谁用? 假设你在写一个 LLM agent:它要先想、再调工具、看工具结果、再想、可能要停下来问人、人答了再继续——而且这中间任何一步都可能崩(网络断、进程挂)。

用普通的 while 循环写,你会发现三件事很难:

• 暂停再恢复:agent 跑到一半要等人类输入,怎么把现场存下来、过几小时再原样接上?
• 崩溃恢复:第 7 步挂了,怎么不从头重跑,而是从第 7 步继续?
• 并行分支:同时对 10 个子任务跑同一个节点,再把结果汇总(map-reduce),怎么写不乱?

LangGraph 把这三件事变成框架内建能力。它主要给构建长时间运行、有状态 agent 的工程师用(README 自述:low-level orchestration framework for building stateful agents)。

它能做什么(功能):

• 用 StateGraph 声明节点与边,编译成可执行图。
• 持久化(checkpoint):每一步自动存盘,支持暂停/恢复/时间旅行(回到任意历史步)。
• 人在环路(human-in-the-loop):节点里调 interrupt() 即可暂停等人输入。
• 动态控制流:节点返回 Send 做并行 map-reduce,返回 Command 做既改状态又跳转。
• 可观测的流式输出:逐步、逐 token 地 stream 出中间过程。

用起来什么样: 一个最小可运行例子(基于 graph/state.py:158-198 的官方 docstring 示例简化——删去了原示例的 runtime/context_schema 参数,并把输入从 {'x': 0.5} 改成 {'x': [0.5]},以聚焦 reducer 累加这一点):

from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph

# 状态是一个 TypedDict;x 这个键用 reducer 累加
def reducer(a: list, b: int | None) -> list:
    return a + [b] if b is not None else a

class State(TypedDict):
    x: Annotated[list, reducer]   # Annotated[类型, reducer] = 这个键怎么合并多次写入

def node(state: State) -> dict:
    last = state["x"][-1]
    return {"x": last * 3 * (1 - last)}   # 节点返回"对状态的增量更新"

graph = (
    StateGraph(State)
    .add_node("A", node)
    .set_entry_point("A")
    .set_finish_point("A")
    .compile()                    # 必须 compile 才能执行
)
graph.invoke({"x": [0.5]})        # -> {'x': [0.5, 0.75]}

注意两个反直觉点,它们是理解 LangGraph 的钥匙:

• 节点不返回新状态,只返回增量更新({"x": ...}),框架用 reducer 把它合并进共享状态。
• 必须 .compile()——StateGraph 只是图纸(builder),编译后才是能跑的 CompiledStateGraph(graph/state.py:139-144 的 warning 明说)。

一句话直觉/类比: 把 LangGraph 想成一个回合制的协作白板。每一回合(超步),框架先看白板(共享状态)决定这一回合谁该上场,让这些节点同时读白板、各自算出要写的内容;回合结束时,框架把所有人的笔记一次性合并写回白板,并给白板拍张照(checkpoint)。下一回合再看新白板决定谁上场。崩了就从最近一张照片恢复。

本节不碰底层代码。记住:节点 + 共享状态(channel)+ 回合制超步 + 每步存盘,就抓住了全部。

2. 顶层全景(它大概怎么转)

LangGraph 有两层:你写的高层 API(StateGraph),和它底下真正的执行引擎(Pregel)。compile() 是这两层的桥——它把你的图翻译成引擎能跑的 channel 拓扑。

2.1 两层结构图

你写的代码(声明层)                编译产物 / 运行时(执行层)
┌─────────────────────────┐        ┌───────────────────────────────────┐
│ StateGraph (图纸/builder) │        │ CompiledStateGraph  --is-a-->Pregel │
│  - add_node / add_edge   │        │  - nodes: {name -> PregelNode}      │
│  - add_conditional_edges │        │  - channels: {name -> BaseChannel}  │
│  - state_schema(状态形状) │        │  - checkpointer / store / cache     │
└────────────┬─────────────┘        └──────────────────┬────────────────┘
             │  .compile()                              │  .invoke / .stream
             │  节点译成 PregelNode,                     │  驱动 PregelLoop
             │  边译成 channel + 触发器                   ▼
             └────────────────────────────▶  ┌─────────────────────────┐
                                              │ PregelLoop(超步循环)     │
                                              │  while tick():           │
                                              │   1 选出本步要跑的 task   │
                                              │   2 并行执行 task         │
                                              │   3 apply_writes 合并     │
                                              │   4 存 checkpoint         │
                                              └─────────────────────────┘

怎么读这张图: 左边是你声明的图纸,.compile() 把它编译成右边的执行层。执行层的核心不是图,而是 channel(状态片段)+ PregelNode(订阅了某些 channel 的节点);PregelLoop 一圈圈地转,每圈就是一个超步。

2.2 部件一句话职责

部件	干什么	在哪个文件
StateGraph	声明式 builder:加节点、加边、定状态 schema	graph/state.py:130
CompiledStateGraph	编译产物,继承 Pregel,实现 invoke/stream	graph/state.py:1391
Pregel	执行引擎本体,持有 nodes/channels/checkpointer	pregel/main.py:449
PregelLoop	超步循环:tick() 选任务、after_tick() 提交	pregel/_loop.py:157
BaseChannel	共享状态的最小单元(一个键=一个 channel)	channels/base.py:19
apply_writes	把一步所有节点写入批量合并进 channel 并升版本	pregel/_algo.py:232
prepare_next_tasks	算出下一步该跑哪些节点	pregel/_algo.py:392
checkpointer	把每步 checkpoint 落盘(内存/SQLite/Postgres)	libs/checkpoint*

2.3 主线走一遍(高层,不进代码)

以 graph.invoke({"x": [0.5]}) 为例,引擎做的事:

1. 接收输入 → 写进特殊的 START channel。
2. tick(选任务):看哪些 channel 在上一步被更新了,推出被触发的节点(第一步是订阅了 START 的节点)。
3. 执行任务:并行跑这些节点函数,它们各自读状态、返回增量更新(写入暂存,不立即生效)。
4. after_tick(提交):apply_writes 把这一步所有写入,按 channel 分组、用 reducer 合并、给每个被改的 channel 递增版本号;然后存一个 checkpoint。
5. 回到第 2 步:新版本号让下一批节点被触发……如此循环,直到某一步没有任何节点被触发(tick 返回 tasks 为空 → status="done",见 _loop.py:646-649)。
6. 输出:从输出 channel 读出最终状态返回。

这就是 BSP(Bulk Synchronous Parallel,批量同步并行):一步之内大家并行干、互不可见彼此的写入;步与步之间有一道同步栅栏,所有写入在栅栏处一次性生效。下一章把这个循环拆开看。

3. 阅读地图(按这个顺序读)

这套文档由浅入深拆成 5 章。建议顺序:

1. 01-pregel-bsp.md — 先读这章。讲清楚那个回合制超步循环到底怎么转:tick 怎么选任务、apply_writes 怎么用版本号判断谁该被唤醒。这是 LangGraph 的心脏。
2. 02-channels-reducers.md — 状态不是一个大 dict,而是一堆 channel。讲 5 种核心 channel 各自的合并语义,以及 StateGraph 怎么把边编译成 channel + 触发器(branch:to:X、join:... 屏障)。
3. 03-persistence-hil.md — checkpoint 的数据结构(channel_values / channel_versions / versions_seen),以及 interrupt() 为什么靠整个节点重放来实现暂停-恢复。
4. 04-control-flow.md — 两个动态控制流原语:Send(并行 map-reduce)和 Command(节点同时改状态加跳转)。
5. 05-clever-boundaries-map.md — 精华提炼、刻意不做什么、与同 shelf 兄弟框架的对比、完整代码地图。

如果你只想抓核心:读完 §1、§2 和第 01 章,你就能给别人讲清楚 LangGraph 是什么、内核怎么转。