第 18 讲 · 断点续跑 / 时间旅行 / Human-in-the-Loop

学习目标

理解 interrupt() / Command(resume=...) 的中断-恢复全链路。
理解断点续跑、时间旅行（fork）、人审三者如何共用 checkpoint 机制。
看懂 checkpoint_ns / checkpoint_map 在子图中的作用。

这一讲是前三讲（执行内核 + 检查点）的高潮：它们组合出 LangGraph 最有特色的三大能力。

一、Human-in-the-Loop：interrupt 与 resume

interrupt()：在节点中暂停

types.py 的 interrupt()（811–934 行）。在节点里调用它，可以暂停整个图，把控制权交还给调用方，等人审批后再继续。

from langgraph.types import interrupt, Command

def review(state):
    decision = interrupt({"draft": state["draft"]})  # 暂停, 把 draft 抛给人
    return {"approved": decision}

机制（901–934 行）：

读 scratchpad（第 10 讲注入的 CONFIG_KEY_SCRATCHPAD），递增 interrupt 计数。
若已有 resume 值（说明是恢复执行）→ 把值写 RESUME channel 并返回它。
否则 → 抛 GraphInterrupt(Interrupt.from_ns(...))，这是一种 GraphBubbleUp，一路冒泡到 loop，loop 把它"吞掉"并把图状态停在此处（第 12 讲 commit 写 INTERRUPT channel）。

resume：带着答案回来

config = {"configurable": {"thread_id": "t1"}}
app.invoke(inputs, config)                    # 跑到 interrupt 停下
# ... 人看到 draft, 做决定 ...
app.invoke(Command(resume="approve"), config) # 带答案恢复

恢复流程（_loop.py _first，836–1062 行）：

__enter__ 用 get_tuple 加载最新 checkpoint，恢复 checkpoint_pending_writes。
Command(resume=...) → put_writes 写 RESUME channel（负索引 -4，第 15 讲）。
tick() 里 _reapply_writes_to_succeeded_nodes 跳过 ERROR/INTERRUPT/RESUME， 让被中断的任务重跑——这次 interrupt() 看到 resume 值，直接返回，不再抛异常。

sequenceDiagram
    participant U as 调用方/人
    participant G as 图
    participant C as Checkpointer
    U->>G: invoke(inputs)
    G->>G: 节点调 interrupt()
    G->>C: 写 INTERRUPT, 存 checkpoint
    G-->>U: 返回(暂停), __interrupt__ 信息
    Note over U: 人审批
    U->>G: invoke(Command(resume="approve"))
    G->>C: 写 RESUME, 加载 checkpoint
    G->>G: 中断任务重跑, interrupt() 返回 resume 值
    G-->>U: 继续到结束

关键认知：interrupt 不是把函数挂起在内存里等待，而是把状态存档、整个 run 结束返回； resume 是用新输入重新进入、靠 scratchpad 计数让 interrupt() 直接返回。 所以服务重启、跨进程都能恢复——因为状态在 checkpoint 里。

也可以在 compile(interrupt_before=[...], interrupt_after=[...]) 做静态中断点 （第 10/11 讲 should_interrupt），用于调试/审批某节点前后。

二、断点续跑（Resume after failure）

如果 run 因异常/崩溃中断，重新用同一 thread_id invoke 即可续跑：

__enter__ 加载最新 checkpoint，pending writes 里已成功任务的写被恢复（不重复执行）。
只有失败/未完成的任务重跑。
这依赖第 14 讲"重试前 clear writes"+ 第 15 讲 pending writes 的配合。

触发"续跑/恢复"语义的条件（_first，839–860 行）：存在历史 checkpoint，且输入是"继续"信号（input is None / Command(...) / 同 run_id 重入 / 显式 RESUMING）。

三、时间旅行（Time Travel）

时间旅行 = 回到某个历史 checkpoint，从那里重新分叉执行。

# 列出历史
history = list(app.get_state_history(config))
# 选一个过去的 checkpoint
past = history[3]
# 从它继续（可先改状态）
app.invoke(None, past.config)

机制（_first，866–959 行）：

config 带 checkpoint_id（指向历史点）→ is_replaying=True。
get_tuple 精确加载该 checkpoint。
Fork：写一个 source="fork" 的新 checkpoint，从历史点分叉出新分支，而不是覆盖历史——这样原时间线和新时间线都保留。
update_state（main.py）也能在某历史点改状态后 fork（source="update"/"fork"）。

用途：调试（回到出错前一步改输入重跑）、A/B（从同一点尝试不同分支）、纠错（人工修正某步状态后继续）。

注意：ShallowPostgresSaver（第 17 讲）只存最新 checkpoint，不支持时间旅行。

四、子图与 checkpoint_ns

当图嵌套子图时，每个作用域有自己的 checkpoint namespace（checkpoint_ns）。格式形如 {parent_ns}|{subgraph_name}:{task_id}|...（main.py 1203–1205 行）。

checkpoint_ns 区分"同一 thread 下不同子图/任务"的 checkpoint。
checkpoint_map（ns → checkpoint_id，_config.py 62–79 行）记录 fork 后各层的父链，让时间旅行能精确定位到"某个子图的某个历史点"。
recast_checkpoint_ns 去掉 task_id 后缀用于定位子图 API。

这套机制让"在嵌套 Agent 里也能中断/续跑/时间旅行"成为可能。

使用场景

审批流：生成内容 → interrupt 给人审 → Command(resume=决定) 继续/打回。
工具调用确认：危险操作（转账、删库）前 interrupt，人确认后再执行（第 20 讲 ReAct v2 可按 tool 粒度中断）。
崩溃恢复：长任务服务重启后用同 thread_id 续跑，不从头来。
调试历史：线上出问题，用 get_state_history 回到出错前，改输入复现/验证修复。

动手实验

写一个含 interrupt() 的节点，invoke 跑到暂停，打印返回里的 __interrupt__，再用 Command(resume=...) 恢复。
跑完一个多步图后 get_state_history，挑一个历史点 invoke(None, past.config)，再次 get_state_history 观察 fork 出的新分支。
在 _first 打断点，对比"普通续跑"与"时间旅行"两条分支的判定（839–959 行）。

阅读作业

精读 types.py 的 interrupt()（811–934 行）与 Command（758–808 行）。
精读 _loop.py 的 _first（836–1062 行）与 _suppress_interrupt（1294–1352 行）。
浏览 _config.py 的 patch_checkpoint_map（62–79 行）。

小结

interrupt = 存档并返回；resume = 重新进入 + scratchpad 计数让 interrupt() 直接返回（跨进程可恢复）。
续跑靠 pending writes 跳过已成功任务；时间旅行靠加载历史 checkpoint 并 fork 新分支。
子图用 checkpoint_ns / checkpoint_map 支撑嵌套场景的中断与时间旅行。

模块五完结。最后一模块：上层 API（Send/函数式）与流式输出、ReAct Agent 拆解。