LangGraph 源码深潜：Checkpoint、Thread 与 Durable Execution 机制解剖

1
def compile(
2
    self,
3
    checkpointer: Checkpointer = None,
4
    *,
5
    store: BaseStore | None = None,
6
    interrupt_before: All | list[str] | None = None,
7
    interrupt_after: All | list[str] | None = None,
8
    debug: bool = False,
9
    name: str | None = None,
10
    cache: BaseCache | None = None,
11
) -> CompiledStateGraph:
12
    ...

1
用户代码
2
  ↓
3
builder.compile(checkpointer=checkpointer)
4
  ↓
5
StateGraph.validate()
6
  ↓
7
CompiledStateGraph(..., checkpointer=checkpointer, store=store, ...)
8
  ↓
9
compiled.attach_node / attach_edge / attach_branch
10
  ↓
11
compiled.validate()

1
def compile(self, checkpointer=None, store=None, **options):
2
    # 1. 校验图结构
3
    self.validate(
4
        interrupt_before=options.get("interrupt_before"),
5
        interrupt_after=options.get("interrupt_after"),
6
    )
7

8
    # 2. 标记 builder 已经编译
9
    self.compiled = True
10

11
    # 3. 构造可执行图
12
    compiled = CompiledStateGraph(
13
        builder=self,
14
        nodes={},
15
        channels={**self.channels, START: EphemeralValue(...)} ,
16
        input_channels=START,
17
        output_channels=self.output_channels,
18
        stream_channels=self.stream_channels,
19
        checkpointer=checkpointer,
20
        store=store,
21
        interrupt_before=...,
22
        interrupt_after=...,
23
        debug=...,
24
        name=...,
25
        cache=...,
26
    )
27

28
    # 4. 把 builder 中的 node/edge/branch 挂到 compiled graph
29
    compiled.attach_node(START, None)
30
    for name, node_spec in self.nodes.items():
31
        compiled.attach_node(name, node_spec)
32

33
    for start, end in self.edges:
34
        compiled.attach_edge(start, end)
35

36
    for start, branches in self.branches.items():
37
        for branch_name, branch_spec in branches.items():
38
            compiled.attach_branch(start, branch_name, branch_spec)
39

40
    # 5. 最终校验，并返回可执行图
41
    return compiled.validate()

1
1. checkpoint 需要覆盖所有节点和边，不应分散在业务节点。
2
2. checkpoint 必须与 Pregel super-step 边界对齐。
3
3. checkpoint 需要保存 versions_seen、channel_versions 等调度信息，业务节点无法可靠维护。
4
4. checkpoint 需要支持 get_state、history、time travel、interrupt resume 等统一能力。

1
class BaseCheckpointSaver(Generic[V]):
2
    def get(self, config: RunnableConfig) -> Checkpoint | None: ...
3
    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None: ...
4
    def list(
5
        self,
6
        config: RunnableConfig | None,
7
        *,
8
        filter: dict[str, Any] | None = None,
9
        before: RunnableConfig | None = None,
10
        limit: int | None = None,
11
    ) -> Iterator[CheckpointTuple]: ...
12
    def put(
13
        self,
14
        config: RunnableConfig,
15
        checkpoint: Checkpoint,
16
        metadata: CheckpointMetadata,
17
        new_versions: ChannelVersions,
18
    ) -> RunnableConfig: ...
19
    def put_writes(
20
        self,
21
        config: RunnableConfig,
22
        writes: Sequence[tuple[str, Any]],
23
        task_id: str,
24
        task_path: str = "",
25
    ) -> None: ...

1
Pregel runtime
2
  ↓
3
BaseCheckpointSaver.get_tuple / list / put / put_writes
4
  ↓
5
Concrete saver: InMemorySaver / PostgresSaver / SqliteSaver / custom saver

1
class BaseCheckpointSaver:
2
    def get(self, config):
3
        # 1. get 是 get_tuple 的简化包装
4
        checkpoint_tuple = self.get_tuple(config)
5
        if checkpoint_tuple is None:
6
            return None
7
        return checkpoint_tuple.checkpoint
8

9
    def get_tuple(self, config):
10
        # 2. 抽象方法：具体 saver 必须实现
11
        raise NotImplementedError
12

13
    def list(self, config, *, filter=None, before=None, limit=None):
14
        # 3. 抽象方法：按 thread / namespace / filter 列出 checkpoint
15
        raise NotImplementedError
16

17
    def put(self, config, checkpoint, metadata, new_versions):
18
        # 4. 抽象方法：保存完整 checkpoint
19
        raise NotImplementedError
20

21
    def put_writes(self, config, writes, task_id, task_path=""):
22
        # 5. 抽象方法：保存 task 级中间写入
23
        raise NotImplementedError

1
Pregel 执行 step
2
  ↓
3
节点产生 writes
4
  ↓
5
put_writes 保存 task writes
6
  ↓
7
writes 应用到 channels
8
  ↓
9
create checkpoint
10
  ↓
11
put 保存完整 checkpoint

1
开发调试：InMemorySaver
2
本地持久：SqliteSaver
3
生产服务：PostgresSaver / 托管 checkpointer / 自定义高可用 saver

1
class Checkpoint(TypedDict):
2
    v: int
3
    id: str
4
    ts: str
5
    channel_values: dict[str, Any]
6
    channel_versions: ChannelVersions
7
    versions_seen: dict[str, ChannelVersions]
8
    updated_channels: list[str] | None
9

10
class CheckpointTuple(NamedTuple):
11
    config: RunnableConfig
12
    checkpoint: Checkpoint
13
    metadata: CheckpointMetadata
14
    parent_config: RunnableConfig | None = None
15
    pending_writes: list[PendingWrite] | None = None

1
Concrete saver
2
  ↓
3
returns CheckpointTuple
4
  ↓
5
Pregel._prepare_state_snapshot
6
  ↓
7
StateSnapshot

1
def create_checkpoint(previous_checkpoint, channels, step):
2
    # 1. 生成 checkpoint id 与 timestamp
3
    checkpoint_id = generate_monotonic_id(step)
4
    timestamp = now_iso8601()
5

6
    # 2. 从 channel 中提取可 checkpoint 的值
7
    values = {}
8
    for channel_name, channel in channels.items():
9
        if channel_name not in previous_checkpoint["channel_versions"]:
10
            continue
11
        try:
12
            values[channel_name] = channel.checkpoint()
13
        except EmptyChannelError:
14
            pass
15

16
    # 3. 保存 channel version 与 versions_seen
17
    return {
18
        "v": LATEST_VERSION,
19
        "id": checkpoint_id,
20
        "ts": timestamp,
21
        "channel_values": values,
22
        "channel_versions": previous_checkpoint["channel_versions"],
23
        "versions_seen": previous_checkpoint["versions_seen"],
24
        "pending_sends": previous_checkpoint.get("pending_sends", []),
25
        "updated_channels": None,
26
    }

1
channels
2
  ↓
3
channel.checkpoint()
4
  ↓
5
Checkpoint.channel_values
6
  ↓
7
CheckpointTuple(checkpoint, metadata, parent_config, pending_writes)

1
{"plan": "生成旅行计划草稿"}

1
下一步该执行哪个节点？
2
哪些 channel 版本已经被哪些节点看过？
3
是否有 pending writes？
4
是否处于 interrupt？
5
是否可从某个 checkpoint fork？

1
class InMemorySaver(
2
    BaseCheckpointSaver[str],
3
    AbstractContextManager,
4
    AbstractAsyncContextManager,
5
):
6
    storage: defaultdict[str, dict[str, dict[str, tuple[...]]]]
7
    writes: defaultdict[tuple[str, str, str], dict[tuple[str, int], tuple[...]]]
8
    blobs: dict[tuple[str, str, str, str | int | float], tuple[str, bytes]]
9

10
    def __init__(
11
        self,
12
        *,
13
        serde: SerializerProtocol | None = None,
14
        factory: type[defaultdict] = defaultdict,
15
    ) -> None: ...

1
用户代码
2
  ↓
3
InMemorySaver()
4
  ↓
5
builder.compile(checkpointer=memory)
6
  ↓
7
Pregel runtime
8
  ↓
9
InMemorySaver.get_tuple / put / put_writes / list

1
class InMemorySaver(BaseCheckpointSaver):
2
    def __init__(self, serde=None, factory=defaultdict):
3
        # 1. 初始化序列化器
4
        super().__init__(serde=serde)
5

6
        # 2. thread_id -> checkpoint_ns -> checkpoint_id -> checkpoint entry
7
        self.storage = factory(lambda: defaultdict(dict))
8

9
        # 3. (thread_id, checkpoint_ns, checkpoint_id) -> task writes
10
        self.writes = factory(dict)
11

12
        # 4. (thread_id, checkpoint_ns, channel, version) -> serialized blob
13
        self.blobs = factory()
14

15
        # 5. 支持 context manager 管理
16
        self.stack = ExitStack()

1
thread_id = "travel-thread-001"
2
  ↓
3
checkpoint_ns = ""
4
  ↓
5
checkpoint_id = "..."
6
  ↓
7
storage[thread_id][checkpoint_ns][checkpoint_id]
8
  ↓
9
writes[(thread_id, checkpoint_ns, checkpoint_id)]
10
  ↓
11
blobs[(thread_id, checkpoint_ns, channel, version)]

1
单元测试
2
本地 debug
3
理解 checkpoint 数据结构
4
演示 HITL / time travel

1
生产服务
2
多进程部署
3
重启恢复
4
审计留存
5
高可用任务恢复

1
def get_state(self, config: RunnableConfig, *, subgraphs: bool = False) -> StateSnapshot: ...

1
{"configurable": {"thread_id": "travel-thread-001"}}

1
用户代码
2
  ↓
3
graph.invoke / graph.get_state
4
  ↓
5
ensure_config / merge_configs
6
  ↓
7
config["configurable"]["thread_id"]
8
  ↓
9
checkpointer.get_tuple / put / list

1
def normalize_checkpoint_config(graph, config):
2
    # 1. 确保 config 结构存在
3
    config = ensure_config(config)
4

5
    # 2. 合并 graph 自身默认 config
6
    if graph.config:
7
        config = merge_configs(graph.config, config)
8

9
    # 3. 从 configurable 中读取 thread_id
10
    thread_id = config["configurable"]["thread_id"]
11

12
    # 4. thread_id 作为存储 key，统一转成 str
13
    if not isinstance(thread_id, str):
14
        config["configurable"]["thread_id"] = str(thread_id)
15

16
    # 5. 保留可选 checkpoint_id / checkpoint_ns
17
    checkpoint_id = config["configurable"].get("checkpoint_id")
18
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
19

20
    return config

1
config={"configurable": {"thread_id": "travel-thread-001"}}
2
  ↓
3
ensure_config
4
  ↓
5
thread_id="travel-thread-001"
6
  ↓
7
checkpointer 使用 thread_id 存取 checkpoint

1
user_id + conversation_id
2
workflow_run_id
3
order_id + task_id
4
UUID

1
def stream(self, input: InputT | Command | None, config: RunnableConfig | None = None, **kwargs) -> Iterator[Any]: ...

1
graph.invoke
2
  ↓
3
graph.stream
4
  ↓
5
Pregel loop
6
  ↓
7
prepare_next_tasks
8
  ↓
9
run tasks
10
  ↓
11
apply_writes
12
  ↓
13
create_checkpoint
14
  ↓
15
checkpointer.put / put_writes

1
def pregel_run(input_value, config):
2
    # 1. 准备 checkpoint 定位信息
3
    config = normalize_checkpoint_config(config)
4
    checkpointer = resolve_checkpointer(config)
5

6
    # 2. 读取已有 checkpoint；没有则创建空 checkpoint
7
    saved = checkpointer.get_tuple(config) if checkpointer else None
8
    checkpoint = saved.checkpoint if saved else empty_checkpoint()
9

10
    # 3. 如果传入 input，则作为 INPUT 写入
11
    if input_value is not None:
12
        input_writes = map_input_to_channel_writes(input_value)
13
        apply_writes(checkpoint, channels, input_writes)
14
        checkpoint = create_checkpoint(checkpoint, channels, step=-1)
15
        if checkpointer:
16
            config = checkpointer.put(config, checkpoint, metadata={"source": "input"}, new_versions=...)
17

18
    # 4. 进入 Pregel super-step 循环
19
    while True:
20
        tasks = prepare_next_tasks(checkpoint, channels, nodes, config)
21
        if not tasks:
22
            break
23

24
        # 5. 执行当前 super-step 中所有 task
25
        all_task_writes = []
26
        for task in run_ready_tasks(tasks):
27
            try:
28
                writes = task.node.invoke(task.input, task.config)
29
                all_task_writes.append((task.id, writes))
30
                if checkpointer:
31
                    checkpointer.put_writes(config, writes, task_id=task.id, task_path=task.path)
32
            except Exception as exc:
33
                error_write = make_error_write(exc)
34
                all_task_writes.append((task.id, error_write))
35
                if checkpointer:
36
                    checkpointer.put_writes(config, [error_write], task_id=task.id, task_path=task.path)
37
                raise
38

39
        # 6. super-step 边界：应用 writes，更新 channel versions
40
        apply_writes(checkpoint, channels, all_task_writes)
41

42
        # 7. 创建并保存新的 checkpoint
43
        checkpoint = create_checkpoint(checkpoint, channels, step=current_step)
44
        if checkpointer:
45
            config = checkpointer.put(config, checkpoint, metadata={"source": "loop", "step": current_step}, new_versions=...)
46

47
        # 8. 继续下一 super-step
48
        current_step += 1
49

50
    # 9. 从 channels 读取输出
51
    return read_output_channels(channels)

1
read checkpoint
2
  ↓
3
prepare tasks
4
  ↓
5
run tasks
6
  ↓
7
put_writes
8
  ↓
9
apply_writes
10
  ↓
11
put checkpoint
12
  ↓
13
next super-step

1
checkpoint 不是“每个节点调用前后随便存一下”。
2
checkpoint 是 Pregel 调度语义下的状态快照。

1
def get_state(
2
    self,
3
    config: RunnableConfig,
4
    *,
5
    subgraphs: bool = False,
6
) -> StateSnapshot:
7
    ...

1
用户代码 graph.get_state(config)
2
  ↓
3
Pregel.get_state
4
  ↓
5
ensure_config / merge_configs
6
  ↓
7
checkpointer.get_tuple(config)
8
  ↓
9
Pregel._prepare_state_snapshot
10
  ↓
11
StateSnapshot

1
def get_state(self, config, *, subgraphs=False):
2
    # 1. 从 config 或 graph 自身拿 checkpointer
3
    config = ensure_config(config)
4
    checkpointer = config["configurable"].get("checkpointer", self.checkpointer)
5

6
    # 2. 如果没有 checkpointer，不能读取 state
7
    if not checkpointer:
8
        raise ValueError("No checkpointer set")
9

10
    # 3. 处理 subgraph checkpoint namespace
11
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
12
    if checkpoint_ns and "checkpointer" not in config["configurable"]:
13
        subgraph = find_subgraph_by_checkpoint_ns(checkpoint_ns)
14
        return subgraph.get_state(
15
            patch_configurable(config, {"checkpointer": checkpointer}),
16
            subgraphs=subgraphs,
17
        )
18

19
    # 4. 合并默认 config，并规范 thread_id 类型
20
    config = merge_configs(self.config, config) if self.config else config
21
    thread_id = config["configurable"]["thread_id"]
22
    if not isinstance(thread_id, str):
23
        config["configurable"]["thread_id"] = str(thread_id)
24

25
    # 5. 从 checkpointer 读取 checkpoint tuple
26
    saved = checkpointer.get_tuple(config)
27

28
    # 6. 转换为 StateSnapshot
29
    return self._prepare_state_snapshot(
30
        config=config,
31
        saved=saved,
32
        recurse=checkpointer if subgraphs else None,
33
        apply_pending_writes="checkpoint_id" not in config["configurable"],
34
    )

1
graph.get_state(config)
2
  ↓
3
checkpointer.get_tuple(config)
4
  ↓
5
_prepare_state_snapshot
6
  ↓
7
StateSnapshot(values, next, config, metadata, tasks, interrupts)

1
values：当前 state 值
2
next：下一步将执行的节点名
3
tasks：下一步 task 详情
4
interrupts：当前中断信息
5
metadata：checkpoint 元数据
6
parent_config：父 checkpoint

1
HITL 审批页面
2
debug 面板
3
失败恢复检查
4
运行进度展示
5
time travel 入口

1
def _prepare_state_snapshot(
2
    self,
3
    config: RunnableConfig,
4
    saved: CheckpointTuple | None,
5
    recurse: BaseCheckpointSaver | None = None,
6
    apply_pending_writes: bool = False,
7
) -> StateSnapshot:
8
    ...

1
Pregel.get_state / get_state_history
2
  ↓
3
Pregel._prepare_state_snapshot
4
  ↓
5
channels_from_checkpoint
6
  ↓
7
prepare_next_tasks
8
  ↓
9
apply_writes(optional pending writes)
10
  ↓
11
read_channels
12
  ↓
13
StateSnapshot

1
def _prepare_state_snapshot(config, saved, recurse=None, apply_pending_writes=False):
2
    # 1. 没有 checkpoint 时，返回空 snapshot
3
    if saved is None:
4
        return StateSnapshot(
5
            values={},
6
            next=(),
7
            config=config,
8
            metadata=None,
9
            created_at=None,
10
            parent_config=None,
11
            tasks=(),
12
            interrupts=(),
13
        )
14

15
    # 2. 迁移旧格式 checkpoint
16
    self._migrate_checkpoint(saved.checkpoint)
17

18
    # 3. 根据 checkpoint 恢复 channels / managed values
19
    step = saved.metadata.get("step", -1) + 1
20
    stop = step + 2
21
    channels, managed = channels_from_checkpoint(
22
        self.channels,
23
        saved.checkpoint,
24
        saver=self.checkpointer,
25
        config=saved.config,
26
    )
27

28
    # 4. 计算该 checkpoint 之后的 next tasks
29
    next_tasks = prepare_next_tasks(
30
        checkpoint=saved.checkpoint,
31
        pending_writes=saved.pending_writes or [],
32
        nodes=self.nodes,
33
        channels=channels,
34
        managed=managed,
35
        config=saved.config,
36
        step=step,
37
        stop=stop,
38
        for_execution=True,
39
        store=self.store,
40
        checkpointer=self.checkpointer,
41
    )
42

43
    # 5. 处理子图状态
44
    task_states = {}
45
    if recurse:
46
        for task in next_tasks.values():
47
            if task.name in self.get_subgraphs():
48
                task_states[task.id] = get_subgraph_state(task, recurse)
49

50
    # 6. 可选应用 pending writes，得到更贴近当前状态的视图
51
    if apply_pending_writes and saved.pending_writes:
52
        for task_id, channel, value in saved.pending_writes:
53
            if channel in ("ERROR", "INTERRUPT"):
54
                continue
55
            if task_id in next_tasks:
56
                next_tasks[task_id].writes.append((channel, value))
57

58
        tasks_with_writes = [task for task in next_tasks.values() if task.writes]
59
        if tasks_with_writes:
60
            apply_writes(saved.checkpoint, channels, tasks_with_writes, ...)
61

62
    # 7. 组装 task 可读信息和 interrupts
63
    tasks = tasks_w_writes(
64
        next_tasks.values(),
65
        saved.pending_writes,
66
        task_states,
67
        self.stream_channels_asis,
68
    )
69

70
    # 8. 从 channels 读取用户可见 values
71
    values = read_channels(channels, self.stream_channels_asis)
72

73
    # 9. 返回 StateSnapshot
74
    return StateSnapshot(
75
        values=values,
76
        next=tuple(t.name for t in next_tasks.values() if not t.writes),
77
        config=patch_checkpoint_map(saved.config, saved.metadata),
78
        metadata=saved.metadata,
79
        created_at=saved.checkpoint["ts"],
80
        parent_config=patch_checkpoint_map(saved.parent_config, saved.metadata),
81
        tasks=tasks,
82
        interrupts=tuple(i for task in tasks for i in task.interrupts),
83
    )

1
CheckpointTuple
2
  ↓
3
channels_from_checkpoint
4
  ↓
5
prepare_next_tasks
6
  ↓
7
apply_pending_writes(optional)
8
  ↓
9
read_channels
10
  ↓
11
StateSnapshot

1
graph.get_state(config)
2
graph.get_state_history(config)

1
def get_state_history(
2
    self,
3
    config: RunnableConfig,
4
    *,
5
    filter: dict[str, Any] | None = None,
6
    before: RunnableConfig | None = None,
7
    limit: int | None = None,
8
) -> Iterator[StateSnapshot]:
9
    ...