07 状态、持久化与恢复

第一部分 速记页

第二部分 易混对照

第三部分 面试口述稿

3.1 “Agent 跑到一半崩了怎么办？”

要做到能恢复，需要三样东西。一是检查点：每完成一步就把状态写进持久存储，包括消息历史、当前步数、预算用量、做过的调用。二是恢复时核对外部状态：检查点里说装好了依赖，工作目录里是不是真的还在。三是处理有副作用的操作：如果崩溃正好发生在操作已经执行、但记录还没写下的时候，恢复后会再执行一次，所以写操作要么天然可以重复执行，要么用幂等键防重。

另外预算要延续，恢复后接着原来的步数算，不能每重启一次就重新给 40 步。

3.2 “你的项目怎么做断点续跑的？”

手写版不支持，崩了只能重跑。后来用 LangGraph 重写了同一个循环，拆成模型、工具、验收三个节点，用 SqliteSaver 做检查点，每次运行用 run-编号 作为 thread_id。命令行加 --resume run-编号 就从最后完成的步骤继续。

实测过一次：编 re2 跑到第 12 步时直接 kill -9，这时已经执行了 24 次工具调用、abseil 已经交叉编译装好了。续跑后从第 13 步开始，直接用装好的 abseil 配置 re2，第 18 步编译通过，运行记录里每个工具调用都只出现一次。

但我也要说清楚边界：LangGraph 检查点的粒度是节点，不是单次工具调用。那次杀的时候第 12 步的工具正好都执行完了。如果杀在工具节点执行到一半，恢复后这个节点里已经跑过的工具会再执行一次。编译命令重复执行问题不大，但如果是不能重复的操作就要另外做幂等。

3.3 “怎么保证工具不被重复执行？”

先按副作用分类。只读操作和天然幂等的操作不用管，重复执行没影响。有副作用的操作，常见做法是幂等键：每个逻辑操作生成一个唯一编号，比如运行编号加步数加调用编号，执行前查一下这个编号有没有做过。

但光靠自己记录还有一个时间窗口：操作已经发出去了、还没来得及记录”已完成”就崩了。真正可靠的做法是让下游服务本身支持幂等键，比如很多支付接口允许请求带一个幂等键，服务端看到重复的键直接返回上次结果。下游不支持时，恢复前先去查询状态，比如查一下邮件是否已经发出，再决定要不要重做。

3.4 “人工审批怎么实现？”

核心是暂停时不占用进程。遇到高风险操作，程序把当前状态和待确认的操作一起存下来，然后通知人，进程可以直接结束。人批准、拒绝或者修改参数之后，程序读回状态继续：批准就执行，拒绝就把原因作为工具结果回给模型，让它换别的办法。

用 LangGraph 的话，在节点里调 interrupt() 就会保存状态并暂停，恢复时用同一个 thread_id 传 Command(resume=决定)。有个坑要注意：恢复时整个节点会从头重新执行，interrupt() 之前的代码会再跑一遍，所以不能在它前面放有副作用的操作，应该放到它后面或者单独的节点里。

3.5 “为什么选 LangGraph 而不是自己写持久化？”

我两个都写过，结论是看需求。手写循环 65 行，一眼能看完，适合单 Agent、线性流程；LangGraph 版 126 行，控制流拆在几个函数里跳着看，但检查点、续跑、中断审批、查看历史状态这些都现成。评测上两者步数、token、通过率看不出差别。

如果任务短、不需要续跑，手写更直接；任务长、要人工审批、要能恢复，用框架省掉自己设计状态存储和恢复逻辑。但框架不会替你解决幂等，那个仍然要自己做。

第四部分 逐个详解

4.1 Agent 有哪些状态

官方定义：状态（state）是程序在某一时刻继续执行所需要的全部信息。持久化（persistence）是把状态写入进程退出后依然存在的存储（文件、数据库）。

打个比方：玩单机游戏时的”存档”。存档里不只有”当前在第几关”，还有血量、背包、任务进度。只记”第几关”，读档后就是满血空背包，和原来不是同一个局面。Agent 也是一样，只存聊天记录不够。

flowchart TB
    subgraph SG1["程序内部状态（可以存进检查点）"]
        M[消息历史]
        P[执行进度<br/>第几步 / 做过哪些调用 / 结果]
        B[预算用量<br/>步数 / token / 时间]
        Q[待确认的操作<br/>等人审批的调用]
        C[控制信息<br/>重复计数 / 当前计划]
    end
    subgraph SG2["外部环境状态（检查点存不下，只能核对）"]
        F[工作目录里的文件]
        D[数据库记录]
        X[已经发出的邮件 / 已调用的外部接口]
    end
    P -. "检查点说做过<br/>环境里是否真的还在？" .-> F
    Q -. "恢复前要确认<br/>是否已执行" .-> X

最容易被忽略的是外部环境状态。 检查点里记着”zlib 已经安装到 deps/“，但如果工作目录被清理了，恢复后按检查点继续一定出错。我项目真的遇到过：

评测时第 3 轮 re2 跑到第 28 步，DeepSeek 返回 402（余额不足）崩溃（run 208）。本想拿它演示续跑，但评测脚本跑完每个项目都会删掉工作目录，崩溃的也不例外。检查点里的状态还在，编了一半的源码没了，续不上。后来评测脚本改成崩溃时保留工作目录。

所以恢复有两个前提：程序状态存下来了，外部状态也还在（或者能重建）。

4.2 手写一个检查点：实验 1

先不用框架，自己实现一个最简单的”每步存档、崩溃后读档”。只用标准库，保存为 checkpoint_resume.py。

import json
import os
import tempfile
from pathlib import Path

PLAN = ["clone zlib", "build zlib", "install zlib", "configure libpng", "build libpng"]
MAX_STEPS = 5


def save(path, state):
    fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
    with os.fdopen(fd, "w") as f:
        json.dump(state, f, ensure_ascii=False)
    os.replace(tmp, path)


def load(path):
    if not path.exists():
        return {"done": [], "steps_used": 0}
    return json.loads(path.read_text())


def run(path, crash_at=None):
    state = load(path)
    print(f"  从检查点读到: 已完成 {state['done']}，已用 {state['steps_used']} 步")
    for action in PLAN:
        if action in state["done"]:
            continue
        if state["steps_used"] >= MAX_STEPS:
            return "budget_exceeded"
        if action == crash_at:
            raise RuntimeError(f"执行 {action} 时进程被杀")
        print(f"  执行 {action}")
        state["done"].append(action)
        state["steps_used"] += 1
        save(path, state)
    return "completed"


with tempfile.TemporaryDirectory() as d:
    ckpt = Path(d) / "run-217.json"
    print("第一次运行")
    try:
        run(ckpt, crash_at="configure libpng")
    except RuntimeError as e:
        print("  崩溃:", e)
    print("恢复运行")
    print("  结果:", run(ckpt))

实际输出：

第一次运行
  从检查点读到: 已完成 []，已用 0 步
  执行 clone zlib
  执行 build zlib
  执行 install zlib
  崩溃: 执行 configure libpng 时进程被杀
恢复运行
  从检查点读到: 已完成 ['clone zlib', 'build zlib', 'install zlib']，已用 3 步
  执行 configure libpng
  执行 build libpng
  结果: completed

逐段讲。

① with 语句。 with 表达式 as 变量: 是上下文管理器（context manager）语法，用来保证资源用完后一定被清理，即使中间出了异常。

• with tempfile.TemporaryDirectory() as d: 创建一个临时目录，路径存在 d 里；with 块结束时目录连同里面的文件自动删除
• with os.fdopen(fd, "w") as f: 打开一个文件用于写入，块结束时自动关闭

② 原子写入 save。 为什么不直接 open(path, "w") 写？因为如果写到一半进程被杀，文件只写了一半，JSON 就坏了，连原来的旧存档也没了。

正确做法是”先写临时文件，再改名替换”：

• tempfile.mkstemp(dir=path.parent, suffix=".tmp")：在同一个目录下创建一个临时文件，返回两个值：文件描述符 fd（操作系统层面代表这个打开文件的整数）和文件路径 tmp
• os.fdopen(fd, "w")：把文件描述符包装成 Python 文件对象
• json.dump(state, f)：把字典写成 JSON 到文件里（dumps 是返回字符串，dump 是直接写文件）
• os.replace(tmp, path)：把临时文件改名成正式文件名。在同一个文件系统上，这个操作是原子的（atomic），要么完全成功、要么完全没发生，不存在”改了一半”

所以任何时刻，path 要么是完整的旧存档，要么是完整的新存档。

sequenceDiagram
    participant P as 程序
    participant T as run-217.json.tmp
    participant F as run-217.json
    Note over F: 旧存档（完整）
    P->>T: 写入新状态
    Note over P,T: 此时崩溃：旧存档仍完整，临时文件作废
    P->>F: os.replace 原子替换
    Note over F: 新存档（完整）

③ 读档 load。 文件不存在就返回初始状态，否则读出来解析。path.read_text() 是 pathlib 提供的方法，一次读完整个文件。

④ 执行 run。

• 先读档，打印已经完成了什么
• 遍历计划，if action in state["done"]: continue 跳过已完成的步骤。这就是”恢复”的核心
• 预算检查用的是存档里的 steps_used，恢复后接着算
• raise RuntimeError(...) 模拟进程被杀
• 每做完一步，更新状态并立刻存档

⑤ 主流程。 第一次运行在第 4 步崩溃，前 3 步已存档；恢复运行读到 3 步完成、用了 3 步，从第 4 步继续，最终完成。

这个实验隐藏了一个问题：崩溃发生在”执行之前”。如果崩溃发生在”执行完成、存档之前”呢？

        print(f"  执行 {action}")
        state["done"].append(action)
        state["steps_used"] += 1
        save(path, state)

假设 执行 configure libpng 真的跑完了（外部世界已经改变），但在 save 之前进程被杀，存档里没有这一步。恢复后会再执行一次。这就是下一节要讲的问题。

自己改一改：

• 把 MAX_STEPS 改成 4，恢复运行会怎样？预算延续起作用了吗
• 把 raise 那两行移到 print(f" 执行 {action}") 之后，看看恢复后 configure libpng 被执行了几次

4.3 崩溃时间窗口和幂等

一个有副作用的步骤，从开始到记录完成，可以拆成四个时刻。在不同时刻崩溃，恢复后的结果不同：

flowchart LR
    A[① 记录：准备执行] --> B[② 执行副作用<br/>文件写入 / 发邮件 / 扣款]
    B --> C[③ 收到结果]
    C --> D[④ 记录：已完成]
    A -. "在 ①② 之间崩溃" .-> R1[没执行过<br/>恢复后执行：正确]
    B -. "在 ②③ 之间崩溃" .-> R2[执行了但不知道结果<br/>状态不明]
    C -. "在 ③④ 之间崩溃" .-> R3[执行成功但没记下<br/>恢复后会重复执行]

②到④之间崩溃都会导致重复执行。这个窗口没法消除，只能让”重复执行”变得无害。

先说为什么会有”幂等键”这种做法。 “幂等”本来是数学词，HTTP 规范很早就规定 GET、PUT、DELETE 这类方法应当是幂等的，客户端没收到响应时可以放心重发；但 POST（比如”创建一笔扣款”）不是，重发一次就可能多扣一次钱。支付公司最怕的就是这个场景：请求发出去了，网络断了，客户端不知道扣没扣成功。不重试，钱可能没扣；重试，钱可能扣两遍。Stripe 的做法是让客户端给每个逻辑操作生成一个唯一编号，放在请求头 Idempotency-Key 里，服务端记下这个编号和结果，同一个编号再来就直接返回上次的结果，不再执行。Stripe 工程师 Brandur Leach 在 2017 年 2 月的文章 Designing robust and predictable APIs with idempotency 里系统讲了这套设计，之后不少 API 也用了同名的请求头，IETF 的 HTTP API 工作组也在把 Idempotency-Key 请求头写成标准草案。下面第 2 种办法就是这个思路。

幂等（idempotent）：一个操作执行一次和执行多次，最终效果相同。

让非幂等操作变安全的三种办法：

1. 改成幂等的写法：追加改成写入完整内容；INSERT 改成 upsert
2. 幂等键（idempotency key）：每个逻辑操作一个唯一编号，执行前查是否做过
3. 执行前查询：恢复时先去外部系统确认是否已经执行（邮件是否已发出、订单是否已创建）

动手实验 2：模拟”发出去了但回执没收到”，对比有无幂等键。保存为 idempotency.py。

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE emails (idempotency_key TEXT PRIMARY KEY, to_addr TEXT, status TEXT)")
outbox = []


def send_email_unsafe(to_addr, crash_after_send=False):
    outbox.append(to_addr)
    if crash_after_send:
        raise ConnectionError("发出去了，但回执没收到")
    return "sent"


def send_email_safe(key, to_addr, crash_after_send=False):
    row = db.execute("SELECT status FROM emails WHERE idempotency_key = ?", (key,)).fetchone()
    if row and row[0] == "sent":
        return "already_sent"
    if row is None:
        db.execute("INSERT INTO emails VALUES (?, ?, 'pending')", (key, to_addr))
        db.commit()
    outbox.append(to_addr)
    db.execute("UPDATE emails SET status = 'sent' WHERE idempotency_key = ?", (key,))
    db.commit()
    if crash_after_send:
        raise ConnectionError("发出去了，但回执没收到")
    return "sent"


def retry(func, *args):
    for attempt in range(1, 4):
        try:
            return func(*args, crash_after_send=(attempt == 1))
        except ConnectionError as e:
            print(f"    第 {attempt} 次失败: {e}，重试")


print("不做幂等:")
outbox.clear()
print("    结果:", retry(send_email_unsafe, "boss@example.com"), "| 实际发出", len(outbox), "封")

print("用幂等键:")
outbox.clear()
print("    结果:", retry(send_email_safe, "run-217:step-9:call-2", "boss@example.com"), "| 实际发出", len(outbox), "封")

实际输出：

不做幂等:
    第 1 次失败: 发出去了，但回执没收到，重试
    结果: sent | 实际发出 2 封
用幂等键:
    第 1 次失败: 发出去了，但回执没收到，重试
    结果: already_sent | 实际发出 1 封

逐段讲。

① SQLite 数据库。 sqlite3 是 Python 自带的模块，SQLite 是一个存在单个文件里的轻量数据库，我项目的 runs.db 就是它。

• sqlite3.connect(":memory:")：:memory: 是特殊名字，表示数据库只在内存里，程序结束就没了，适合做实验
• db.execute("CREATE TABLE ...")：执行 SQL 语句建表。idempotency_key TEXT PRIMARY KEY 表示这一列是主键，同一个值只能出现一次
• outbox = [] 是一个列表，模拟”真正发出去的邮件”

② 不安全的发送。 先把邮件放进 outbox（副作用已经发生），然后模拟回执丢失抛异常。调用方以为失败了，重试，于是又发了一封。

③ 带幂等键的发送 send_email_safe。

• db.execute("SELECT ... WHERE idempotency_key = ?", (key,))：查询这个键的记录。? 是占位符，实际值通过后面的元组传入。(key,) 是只有一个元素的元组，逗号不能省，否则括号只是普通括号。用占位符而不是把值拼进 SQL 字符串，是为了防止 SQL 注入
• .fetchone()：取一行结果，没有就返回 None
• if row and row[0] == "sent"：已经发过就直接返回 already_sent
• 没有记录就先插入一条 pending（准备发送），db.commit() 提交，让写入真正生效
• 发送，然后把状态更新为 sent 并提交
• 模拟回执丢失

④ 重试 retry(func, *args)。 最多试 3 次。crash_after_send=(attempt == 1) 让第一次模拟回执丢失，后面正常。except ConnectionError as e 只接住网络类错误。

⑤ 结果。 不做幂等发了 2 封；有幂等键时第二次查到已经是 sent，直接返回，只发了 1 封。

这个实现仍然有漏洞，面试时要能说出来：如果崩溃发生在 outbox.append（真正发出）之后、UPDATE ... 'sent' 之前，记录停在 pending，重试时会再发一次。自己的数据库和外部邮件服务之间，没法用一次操作同时完成，这个窗口永远存在。可靠的做法：

• 下游支持幂等键：把键一起发给邮件或支付服务，服务端看到重复的键直接返回上次结果。很多支付接口都提供这种请求参数
• 遇到 pending 先查询：恢复时发现状态是 pending，先去邮件服务查”这个键对应的邮件发了没有”，再决定

幂等键怎么生成？要保证同一个逻辑操作每次重试都得到同一个键，不同操作得到不同的键。Agent 里常用”运行编号 + 步数 + 调用编号”，比如 run-217:step-9:call-2。不能用随机数或当前时间，那样每次重试键都不一样，等于没有。

4.4 恢复流程：不只是读档

把前面的内容合起来，一次完整的恢复应该这样：

flowchart TD
    S([收到恢复请求：run-217]) --> L[读取最新检查点]
    L --> E{检查点存在吗}
    E -- 不存在 --> X([报错：无法恢复])
    E -- 存在 --> V{外部环境还在吗<br/>工作目录 / 依赖 / 分支}
    V -- 不在 --> RB[重建环境或报告无法恢复]
    V -- 在 --> P{有 pending 状态的<br/>副作用操作吗}
    P -- 有 --> Q[去外部系统查询是否已执行<br/>更新记录]
    P -- 没有 --> BG
    Q --> BG{预算还有吗<br/>用检查点里的用量}
    BG -- 没有 --> BX([停止：budget_exceeded])
    BG -- 有 --> ENV[恢复运行环境<br/>环境变量 / 连接 / 工具配置]
    ENV --> C([从下一步继续执行])

我项目的 graph_agent.py 在恢复时做了其中几步：

• graph.get_state(config).values 读检查点，读不到就 raise ValueError(f"找不到检查点 {resume}")
• 从检查点里取出 project、target、arch、run_id，重新设置 ZIG_TARGET、ZIG_ARCH、CMAKE_TOOLCHAIN_FILE 环境变量。环境变量存在进程里，新进程不会自动有，漏了这一步恢复后工具链就找不到
• 步数 step 存在状态里，恢复后从原来的步数接着算
• 运行记录复用原来的 run_id，不新开一条

没做的：没有检查工作目录是否还在（依赖评测脚本崩溃时保留目录）；工具都是文件和构建命令，没有需要查询外部系统的操作。

4.5 LangGraph 的持久化机制

先说为什么会有 LangGraph。 Harrison Chase 在 2022 年 10 月发布了 LangChain，最初是一个把”提示词模板 → 模型 → 解析输出”这类步骤串起来的 Python 库，串起来的东西叫链（chain），是一条从头走到尾的直线。Agent 需要的却是循环：调模型、执行工具、再调模型，直到停下，中间还要记住走到哪了。LangChain 早期用一个现成的 AgentExecutor 把循环包起来，想改循环里某一步的逻辑、加一个审批节点、中途存档，都很别扭。2024 年 1 月，LangChain 团队发布 LangGraph，把 Agent 写成一张允许有环的图：节点是步骤，边决定下一步去哪，整张图共享一份状态。状态既然集中在一处，每一步结束把它存下来就是检查点，这也是本节要讲的持久化机制的来源。

官方说明（LangGraph Persistence 文档）：

• 检查点保存器（checkpointer）：负责把图的状态快照持久化。编译图时传入：graph = builder.compile(checkpointer=...)
• 线程（thread）：一条独立的执行和状态历史，用 thread_id 标识，调用时通过 config = {"configurable": {"thread_id": "..."}} 传入。同一个 thread_id 的多次调用共享状态
• 超级步（super-step）：图执行的一个”回合”，这一回合里可以运行的节点都执行完算一个超级步。每个超级步结束后自动保存一个检查点
• 常用保存器：InMemorySaver（内存，重启丢失，测试用）、SqliteSaver（文件，开发和单机用）、PostgresSaver / AsyncPostgresSaver（生产用）
• 节点在一个超级步中途失败时，这个超级步里未完成的写入不会进入检查点，保证状态一致
• 可以用 get_state 看当前状态、get_state_history 看历史快照，从旧检查点重放或修改状态后继续（时间回溯）

我项目的 LangGraph 版每一步的检查点在哪里：

sequenceDiagram
    participant G as 图执行
    participant A as agent 节点
    participant T as tools 节点
    participant DB as checkpoints.db
    G->>A: 超级步 k：调模型
    A-->>G: 新消息 + step+1
    G->>DB: 保存检查点
    G->>T: 超级步 k+1：执行这一步所有工具调用
    Note over T: 调用 1 执行完<br/>调用 2 执行完<br/>调用 3 执行中...
    T-->>G: 全部工具结果
    G->>DB: 保存检查点
    Note over T,DB: 如果在"调用 3 执行中"被杀：<br/>检查点停在 agent 节点之后<br/>恢复会重新执行整个 tools 节点<br/>调用 1、2 会再执行一次

这就是”检查点粒度是节点”的含义。框架的检查点只能保证图的状态一致，保证不了节点内部已经发生的外部副作用不重复。

真实的续跑实验（eval/REPORT.md 记录）：

flowchart LR
    subgraph SG3["第一次运行 run-217"]
        S1[第 1-11 步] --> S12[第 12 步<br/>工具都执行完<br/>检查点已写入]
        S12 --> K[kill -9]
    end
    subgraph SG4["续跑 --resume run-217"]
        R13[第 13 步<br/>读到 abseil 已装好] --> R14[...] --> R18[第 18 步<br/>re2 编译通过]
    end
    K -. "读检查点<br/>24 次工具调用都已记录" .-> R13

• 杀进程时已执行 24 次工具调用，abseil 已经交叉编译并装进项目目录
• 续跑从第 13 步开始，直接用装好的 abseil 配置 re2，第 18 步编译通过
• 运行记录里每一步的工具调用都只出现一次

要诚实说明的是：kill 的时机正好在第 12 步的工具节点执行完、检查点也写进去之后。这次实验证明了”检查点之间的状态能正确恢复”，没有证明”节点执行中途崩溃也不重复”，后者按机制分析是会重复的。

实际踩到的两个坑（REPORT 记录）：

• 线程：同步 stream 会在线程池里跑节点，原来 Trace 的 SQLite 连接不允许跨线程使用，要设 check_same_thread=False。sqlite3.connect 默认只允许创建连接的那个线程使用它
• 递归上限：很多教程说 LangGraph 默认只允许 25 个超级步，那是旧版本；实测 1.2 版默认值是 10007。项目里仍按 MAX_STEPS × 2 + 10 显式设了 recursion_limit，防止路由写错时死循环

4.6 人工审批：暂停不占进程

为什么不能用 input() 阻塞等待？ 审批可能要几小时甚至几天。阻塞等待意味着进程一直挂着，服务器重启就全丢了，而且同时有 100 个任务在等审批就要挂 100 个进程。正确做法是存档后退出，有决定时再读档继续。

sequenceDiagram
    participant U as 用户 / 审批人
    participant API as 后端服务
    participant G as Agent（LangGraph）
    participant DB as 检查点存储
    U->>API: 发起任务
    API->>G: invoke(输入, thread_id=run-1)
    G->>G: propose 节点生成补丁
    G->>G: approve 节点调用 interrupt()
    G->>DB: 保存状态
    G-->>API: 返回 __interrupt__（待审批内容）
    API-->>U: 通知：请审批这个补丁
    Note over G: 进程可以退出<br/>几小时后
    U->>API: 批准
    API->>G: invoke(Command(resume="yes"), thread_id=run-1)
    G->>DB: 读取状态
    G->>G: approve 节点从头重新执行<br/>interrupt() 返回 "yes"
    G->>G: apply 节点写入补丁
    G-->>API: 最终结果

动手实验 3：用 LangGraph 实现”生成补丁 → 人工审批 → 应用补丁”。不需要 API key，节点都是普通函数。需要 pip install langgraph langgraph-checkpoint-sqlite（我用的是 langgraph 1.2.11）。保存为 lg_interrupt.py。

import sqlite3
from typing import TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    patch: str
    approved: bool
    applied: bool


def propose(state: State):
    print("  [propose] 生成补丁")
    return {"patch": "CMakeLists.txt: option(PNG_TESTS OFF)"}


def approve(state: State):
    print("  [approve] 节点开始执行")
    decision = interrupt({"question": "应用这个补丁吗？", "patch": state["patch"]})
    print(f"  [approve] 收到人工决定: {decision}")
    return {"approved": decision == "yes"}


def apply_patch(state: State):
    print("  [apply] 写入补丁")
    return {"applied": True}


def route(state: State):
    return "apply" if state["approved"] else END


g = StateGraph(State)
g.add_node("propose", propose)
g.add_node("approve", approve)
g.add_node("apply", apply_patch)
g.add_edge(START, "propose")
g.add_edge("propose", "approve")
g.add_conditional_edges("approve", route, ["apply", END])
g.add_edge("apply", END)

conn = sqlite3.connect(":memory:", check_same_thread=False)
graph = g.compile(checkpointer=SqliteSaver(conn))
config = {"configurable": {"thread_id": "run-1"}}

print("第一次调用")
out = graph.invoke({"patch": "", "approved": False, "applied": False}, config)
print("  暂停，等待:", out["__interrupt__"][0].value)
print("  下一个要执行的节点:", graph.get_state(config).next)

print("人工批准后恢复")
out = graph.invoke(Command(resume="yes"), config)
print("  最终状态:", out)
print("  检查点数量:", len(list(graph.get_state_history(config))))

实际输出：

第一次调用
  [propose] 生成补丁
  [approve] 节点开始执行
  暂停，等待: {'question': '应用这个补丁吗？', 'patch': 'CMakeLists.txt: option(PNG_TESTS OFF)'}
  下一个要执行的节点: ('approve',)
人工批准后恢复
  [approve] 节点开始执行
  [approve] 收到人工决定: yes
  [apply] 写入补丁
  最终状态: {'patch': 'CMakeLists.txt: option(PNG_TESTS OFF)', 'approved': True, 'applied': True}
  检查点数量: 5

逐段讲。

① TypedDict。 class State(TypedDict): 定义一个”带类型说明的字典”。它在运行时就是普通字典，类型注解只是说明每个键应该存什么类型。LangGraph 用它知道状态里有哪些字段。

② 节点函数。 每个节点是一个普通函数，参数是当前状态，返回一个字典，只写要更新的字段。LangGraph 把返回的字典合并进状态。比如 propose 只返回 patch，其他字段不变。

③ interrupt(...)。 在节点内部调用时：

• 第一次执行到这里：保存状态，暂停整个图，invoke 返回的结果里多一个 __interrupt__ 键，里面是传给 interrupt 的内容（要给审批人看的信息）
• 恢复时：interrupt(...) 这个函数调用返回 Command(resume=...) 里传的值

④ 建图。

• StateGraph(State)：创建一个以 State 为状态结构的图
• add_node("名字", 函数)：添加节点
• add_edge(A, B)：固定边，A 执行完就执行 B。START 和 END 是特殊的起点和终点
• add_conditional_edges("approve", route, ["apply", END])：条件边。approve 执行完调用 route(state)，根据返回值决定去哪；第三个参数列出所有可能的去向
• g.compile(checkpointer=SqliteSaver(conn))：编译成可执行的图，并指定用 SQLite 存检查点

⑤ 两次 invoke。

• 第一次传入初始状态和 config，执行 propose，进入 approve，遇到 interrupt 暂停
• graph.get_state(config).next 显示下一个要执行的是 ('approve',)，说明 approve 节点还没算完成
• 第二次传入 Command(resume="yes") 和同一个 config（同一个 thread_id）

⑥ 最重要的观察：[approve] 节点开始执行 打印了两次。 恢复时，整个 approve 节点从头重新执行，interrupt() 之前的代码又跑了一遍，这次 interrupt() 直接返回 "yes"，继续往下。

官方文档（Interrupts）明确列出了由此带来的规则：

⑦ 检查点数量 5。 我额外把 get_state_history 的每条记录打印出来看过：分别是收到输入时、进入 propose 前、propose 完成后（下一个是 approve）、approve 完成后（下一个是 apply）、apply 完成后。暂停时停在”propose 完成后”那个检查点上，approve 在它完成之前不会产生新检查点，这和”恢复时 approve 整个重跑”是一致的。每个超级步一份快照，也是”时间回溯”的基础。

审批还可以更细：批准、拒绝、修改参数后批准、要求模型补充说明。拒绝时要把理由作为工具结果回给模型，而不是直接结束任务，模型可能有别的办法。

4.7 什么操作需要审批

不是所有工具调用都要人确认，那样 Agent 就没意义了。按风险分级（和 04 篇的副作用分级对应）：

审批界面要给人看足够做决定的信息：要执行什么操作、参数是什么、为什么要做（模型给出的理由）、影响范围。只显示”Agent 请求执行 send_email，是否批准？“，人只能盲目点”是”。

4.8 什么时候需要持久化，什么时候不需要

我项目手写版不支持续跑，因为简单和中等项目几十秒就跑完；困难项目一次两三分钟、几十万 token，续跑才开始有实际价值。这是为什么后来要做 LangGraph 版的原因之一。

跑几小时、跨好几个上下文窗口的任务怎么接力。 上面讲的检查点解决”进程崩了接着跑”。任务长到一个上下文窗口装不下时，还有另一个问题：新开的会话不记得前面做了什么。Anthropic 的 Effective harnesses for long-running agents（2025-11）总结了两种典型失败：一是贪多，一次想做完所有功能，做到一半窗口满了，留下半成品也没留说明；二是后面的会话看到已经有不少代码，就以为做完了，提前宣布完成。他们的做法都是把状态写成文件：

这和本篇的思路一致：状态放在上下文之外、能被下一个执行者读到的地方，只是这里的”执行者”是新的模型会话，而不是重启的进程。

把会话记录和执行进程拆开。 Anthropic 2026 年 4 月的 Scaling Managed Agents 讲了他们托管 Agent 服务的架构，文中叫”把大脑和手分开”：

凭据不放进沙箱，由沙箱外的代理去取，模型写的代码拿不到。拆开之后，外层循环不用等容器启动就能开始调模型，首 token 延迟的中位数降了约 60%。这就是 12-Factor Agents 第 12 条”把 Agent 做成无状态的归约函数”在生产里的样子，和 pi 的只追加会话文件也是同一个方向。

第五部分 对照项目

建议阅读：agent/graph_agent.py 的 build_events 函数（续跑分支 if resume:），对照 4.4 节的恢复流程图。tests/test_graph_agent.py 里有续跑不重复执行工具的测试。

对照开源实现：pi

pi（05 篇第五部分介绍过）没有用 LangGraph 这类框架，会话持久化是自己写的，思路和本篇实验 1 的”快照”完全不同：只追加的日志。源码以 commit 7b4cfd6 为准，格式说明见 coding-agent/docs/session-format.md。

存什么、怎么存。 每个会话一个 JSONL 文件，放在 ~/.pi/agent/sessions/--<工作目录>--/<时间>_<会话编号>.jsonl。一行一条记录，只往文件末尾追加，从不改写已有的行：

{"type":"session","version":3,"id":"...","cwd":"/path/to/project"}
{"type":"message","id":"a1b2c3d4","parentId":null,"message":{"role":"user",...}}
{"type":"message","id":"b2c3d4e5","parentId":"a1b2c3d4","message":{"role":"assistant","content":[{"type":"toolCall",...}],...}}
{"type":"message","id":"c3d4e5f6","parentId":"b2c3d4e5","message":{"role":"toolResult",...}}
{"type":"model_change","id":"d4e5f6g7","parentId":"c3d4e5f6","provider":"openai","modelId":"..."}
{"type":"compaction","id":"e5f6g7h8","parentId":"d4e5f6g7","summary":"...","firstKeptEntryId":"c3d4e5f6"}

（字段有省略。）几个设计点：

快照 vs 追加日志。 本篇实验 1 每步把整个状态覆盖写一次，所以要”先写临时文件再原子改名”防止写一半坏掉。pi 只追加不覆盖，一次崩溃最多坏最后一行，读的时候跳过就行，不需要原子改名。代价是文件只增不减，恢复时要从头读一遍再重建。

崩溃在工具执行中间会怎样？ 这正是本篇 4.3 节说的危险窗口。模型发起工具调用的那条回复已经写进文件了，工具结果还没写。恢复后再发请求时，pi 在转换消息的环节（ai/src/api/transform-messages.ts L158-L222）做两件事：

• 找出”有调用、没结果”的工具调用，每个补一条假的结果，内容是 No result provided，标记为错误。不补的话 API 会因为调用和结果不配对直接拒绝请求
• 出错或被中止的模型回复整条跳过，不再发给模型

也就是说，pi 不判断这个工具到底执行了没有，也没有幂等键，直接告诉模型”没拿到结果”，让模型自己决定要不要重做。 这在它的场景下能接受：人在终端前看着；主要工具是读文件、改文件、跑命令，改文件用的是”精确查找替换”，已经改过的再改一次通常会报”找不到原文”，模型能发现。但如果工具是发邮件、扣款，这种做法就会重复执行，必须用 4.3 节的办法。

pi 正在补这一块。 agent 包里有一套还在开发的新运行时（代码在 agent/src/harness/，命令行产品目前只在实验性功能里用到它，默认流程还是前面讲的那套），设计文档 agent/docs/tool-durability.md 几乎就是本篇 4.3 节的工程版：

文档里还讲了一个本篇没提到的并行场景：一批三个调用 A、B、C，B 和 C 先跑完，A 还在跑，这时崩溃。工具结果必须按 A、B、C 的顺序放进对话，所以 B、C 的结果还没来得及写，只在内存里，恢复后会被当成”没完成”而重跑。新设计的做法是：谁先跑完就先把完整结果单独存下来（状态 outcome_ready），等排在前面的都好了再按顺序放进对话。目标写得很直接：结果已经存下来的工具，永远不再重跑。

和我的项目对比：

第六部分 追问清单

第七部分 闭卷自测

1. 只把消息历史存下来，崩溃后能完整恢复吗？还缺什么？

2. 实验 1 的 save 为什么先写临时文件再 os.replace？

3. 一个步骤在”执行完成、存档之前”崩溃，恢复后会发生什么？怎么应对？

4. 判断是否幂等：（a）写入完整文件内容（b）往日志末尾追加一行（c）数据库 upsert（d）发短信（e）cmake —build

5. 幂等键能不能用 uuid4() 随机生成？为什么？

6. 实验 2 的 send_email_safe 还有什么情况会发两封？怎么彻底解决？

7. LangGraph 需要哪两样东西才能支持检查点和恢复？

8. 实验 3 里 [approve] 节点开始执行 为什么打印了两次？这对写代码有什么要求？

9. 为什么人工审批不用 input() 阻塞等待？

10. 你项目的 kill -9 续跑实验证明了什么？没有证明什么？

11. run 208 为什么续不上？给了什么教训？

12. 恢复后预算应该怎么算？为什么？

13. pi 的会话文件只追加、不覆盖。和实验 1 的”每步覆盖写快照”比，各有什么好处？为什么 pi 不需要”先写临时文件再原子改名”？

14. 工具调用已经写进会话、工具结果还没写就崩溃了。pi 恢复后怎么处理？这种做法适合什么场景，不适合什么场景？

延伸阅读

1. LangGraph：Persistence — 检查点、线程、超级步、时间回溯、各种保存器
2. LangGraph：Interrupts — interrupt() 和 Command(resume=...)，节点重新执行的规则和常见错误
3. 12-Factor Agents — 第 5 条”统一执行状态和业务状态”、第 6 条”用简单的接口启动 / 暂停 / 恢复”、第 12 条”把 Agent 做成无状态的归约函数”
4. Anthropic：Building effective agents — Agent 在检查点或遇到阻碍时暂停等待人工反馈的建议
5. 项目 eval/REPORT.md 的”LangGraph 重写对比”一节 — 续跑实测和踩过的坑的原始记录
6. pi：Session File Format（commit 7b4cfd6）— 只追加的 JSONL 会话格式、树形分支、压缩记录和上下文重建规则
7. Anthropic：Effective harnesses for long-running agents（2025-11）— 跨多个上下文窗口的任务怎么用清单、进度文件和 git 接力
8. Anthropic：Scaling Managed Agents（2026-04）— 会话日志、无状态外层循环、沙箱三者分开后怎么恢复

下一篇：08 RAG 全链路——给 Agent 接上知识库，从切片、向量到检索评测。