《AI Agent 开发入门》7.3 检查点与快照

7.3 检查点与快照

7.2 让 Session 能 snapshot 成纯 JSON，但那只在内存里绕了一圈。真实的长任务——一次跑几十步、调十几次工具、耗时几分钟——最怕的是跑到一半进程没了：内存里的 messages、steps、工具中间结果全部蒸发，重启后只能从第一个问题重新问起，之前花掉的 token 与时间全部作废。要解决它，得在运行途中把状态写到磁盘，并在重启后从断点接上。

本节把 minagent 推进到「崩溃可续跑」：把 Session（消息 + 步骤 + 工具中间结果）连同模型游标序列化成 JSON 存进临时目录，用一个可暂停的单步执行器把循环切成片段；崩溃后 restore 读盘、从断点续跑，并证明已完成的工具调用不会被重复执行。

7.3.1 崩溃时我们丢了什么

先算清损失。一次 run_turn 中途崩溃，内存里丢掉三类东西：

第三类最危险。「重跑」听起来只是浪费，但如果工具带副作用——发了一封邮件、扣了一笔款、写了一条数据库记录——重跑就等于重复执行。所以续跑的目标不只是「省时间」，更是「不重做已经做完的事」。这就要求检查点必须精确记录「哪些工具调用已经完成」，恢复后从下一个未完成的动作接着走，而不是从头发车。

7.3.2 检查点里该装什么

一个检查点 = 一份状态 + 一个进度指针：

def save_checkpoint(path, session, model, steps):
    data = {
        "session": session.snapshot(),          # 消息、id、metadata、created_at
        "model_cursor": model.cursor,           # 模型消费到脚本第几项
        "steps": [
            {"index": s.index, "kind": s.kind, "detail": s.detail}
            for s in steps
        ],
    }
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

三块内容各有来历：

• session.snapshot()：复用 7.2 的成果，把消息历史整体存下来。工具中间结果就在 messages 的 tool 消息里，天然被覆盖。
• model_cursor：模型消费到脚本第几项。这是续跑不重放的关键——它告诉恢复后的模型「下一个该吐哪条脚本」。
• steps：步骤轨迹，用于续跑后接着编号，也让日志连续。

model_cursor 是本节唯一为假模型量身定制的字段，值得说清楚。真实模型没有「脚本游标」——它是无状态的函数，每次调用都要你把完整历史喂进去，它自己决定下一步。假模型为了确定性，把「下一步说什么」写死成一份线性脚本，所以必须额外记一个指针，才知道「演到第几幕了」。假模型省掉的是推理，代价就是得自己管这个游标。

还有一条工程纪律：写盘要原子。直接 open(path, "w") 再 json.dump，若写到一半崩了，磁盘上留下的是个残缺文件——下次恢复读到半个 JSON，比没有检查点更糟。稳妥做法是先写临时文件、再 os.replace 原子改名：改名在同一文件系统内是原子操作，结果要么是完整的旧检查点，要么是完整的新检查点，绝不会是半个。本节为聚焦主线没有加这层，但真实系统务必补上。

7.3.3 单步执行：把循环切成可暂停的片段

要能「跑到一半存盘」，循环就不能是一口气跑完的 for。把它切成「一次模型调用 = 一步」：

def step_once(agent, session, model, steps) -> str:
    """执行一次模型调用；返回 'paused'（还要继续）或 'done'（收敛）。"""
    messages = list(session.messages)
    resp = model.complete(messages, agent.registry.schemas())
    msg = resp.message
    messages.append(msg)
    if not msg.tool_calls:
        steps.append(Step(index=len(steps), kind="final", detail=msg.content))
        session.messages = messages
        return "done"
    for tc in msg.tool_calls:
        result = agent.registry.call(tc)
        messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
        steps.append(Step(index=len(steps), kind="tool_call", detail=f"{tc.name} -> {result}"))
    session.messages = messages
    return "paused"

和 run_turn 的区别：每次只调用一次模型。带工具的那一步执行完工具、把结果灌回 messages、返回 paused；只有模型给出纯文本时才返回 done。这样「跑一半」就有了明确边界——paused 之后、下一次 step_once 之前，就是可以安全存盘的时刻。

注意 steps 的 index 用 len(steps) 自增，而不是 for i in range(...) 的循环变量。因为循环被切碎后，i 每次从 0 重来；用 len(steps) 才能让步骤编号跨多次 step_once 连续。

7.3.4 load：从磁盘恢复

恢复是存盘的逆运算：

def load_checkpoint(path, script):
    with open(path, encoding="utf-8") as f:
        data = json.load(f)
    session = Session.restore(data["session"])
    model = FakeModel(script)
    model.cursor = data["model_cursor"]          # 把游标拨回断点位置
    steps = [Step(**s) for s in data["steps"]]
    return session, model, steps

关键在 model.cursor = data["model_cursor"] 这一行：把游标直接拨到断点，而不是用 script[cursor:] 去切片。两者结果一样，但切片会让游标变成「相对剩余脚本」的坐标，下次存盘就存了个错位的值，第二次崩溃再恢复就会跳步。把游标做成相对完整脚本的绝对坐标，才能反复存、反复恢复而不漂移。本节实测的收尾检查点里 model_cursor 正好等于 3（脚本总长），就是绝对坐标的证明。

7.3.5 跑一遍：跑到一半 → 崩溃 → 续跑

脚本设成「add → multiply → 文本收敛」，故意在第一步之后存盘、然后模拟崩溃：

CKPT = "/tmp/agentbook_ckpt/session.json"

# --- 跑到一半 ---
session = init_session(agent, "先算 2+3，再把结果乘 4")
steps: list[Step] = []
status = step_once(agent, session, model, steps)      # 只走一步
save_checkpoint(CKPT, session, model, steps)
print("半程状态:", status, "| 消息条数:", len(session.messages))

# --- 模拟进程崩溃：丢掉内存中的对象 ---
del session, model, steps

# --- 重启：从检查点恢复 ---
print("检查点文件大小:", os.path.getsize(CKPT), "字节")
session, model, steps = load_checkpoint(CKPT, SCRIPT)
print("恢复后消息条数:", len(session.messages), "| 恢复的步骤:", [s.kind for s in steps])

# --- 续跑至收敛 ---
while True:
    status = step_once(agent, session, model, steps)
    save_checkpoint(CKPT, session, model, steps)
    if status == "done":
        break

tool_ids = [m.tool_call_id for m in session.messages if m.role == "tool"]
print("续跑后消息条数:", len(session.messages))
print("工具结果条数:", len(tool_ids), "| 工具调用 id:", tool_ids)
print("无重复工具调用:", len(tool_ids) == len(set(tool_ids)))

本机实测（Python 3.14.6）：

半程状态: paused | 消息条数: 4 | 步骤: ['tool_call']
检查点文件大小: 985 字节
恢复后消息条数: 4 | 恢复的步骤: ['tool_call']
续跑后消息条数: 7
工具结果条数: 2 | 工具调用 id: ['c1', 'c2']
无重复工具调用: True
最终回答: 先算 2+3 得 5，再乘 4 得 20。
步骤序列: ['tool_call', 'tool_call', 'final']

逐行读这份输出，续跑的正确性一目了然：

• 半程 4 条消息（system + user + assistant 的 add 调用 + add 的结果），说明存盘时确实「跑到一半」。
• 恢复后仍是 4 条、步骤仍是 ['tool_call']：load_checkpoint 无损还原了断点。
• 续跑后 7 条：补齐了 multiply 的一对往返和最终回答，与从头跑的结果完全一致。
• 工具调用 id 是 ['c1', 'c2']，且 无重复 为 True：add（c1）只执行了一次，没有被续跑重放。这是「不重做已完成动作」的直接证据。

model_cursor 的绝对值设计也经受住了检验：整个过程中存了三次盘（半程一次、续跑两次），游标从 1 走到 2 再走到 3，全程没有错位。哪怕在续跑途中再崩一次、再恢复，接着走的结果也不变。

7.3.6 为什么不会重复执行工具

有人会问：恢复后从 messages 接着跑，凭什么保证 add 不再被调一次？答案是循环的推进方向。step_once 只从模型的下一次响应里读工具调用，而断点处的 messages 已经包含了 add 的调用与结果；模型被拨到游标 1 后，下一个吐出的脚本项是 multiply，根本不提 add。工具调用由模型发起，模型不重提，工具就不重跑。

这个结论能推广到真实模型：真实模型看的是完整 messages，它看到 add 的结果已经躺在历史里，自然不会再调一次。所以「恢复不重放」的本质不是游标，而是历史完整性——只要 messages 无损耗地存下来，模型的行为就是确定的。游标只是假模型为了对齐脚本而多出来的补丁。真正要防的，是「检查点写了一半就崩」导致 messages 残缺，那才会让模型看到矛盾的历史、做出错误决策。

7.3.7 恢复后的完整性校验

存盘到读盘之间，状态可能被写坏——写盘写到一半断电、文件被外部截断、序列化遇到不可编码的值。恢复时若不校验，坏掉的状态会静默带进续跑，最后以一个莫名其妙的结果收场。所以恢复后应当立刻校验一条结构不变量：每个 tool 结果都对应一个已发出的工具调用。

def validate(session: Session) -> bool:
    """检查消息序列的不变量：每个 tool 结果都对应一个已发出的工具调用。"""
    pending: set[str] = set()
    for m in session.messages:
        for tc in m.tool_calls:
            pending.add(tc.id)
        if m.role == "tool":
            assert m.tool_call_id in pending, f"孤儿工具结果: {m.tool_call_id}"
            pending.discard(m.tool_call_id)
    return not pending

这条不变量能抓住两类损坏：孤儿工具结果（tool_call_id 找不到对应的 assistant.tool_calls，说明工具调用那条消息丢了）与悬空工具调用（模型发了调用却没有结果，说明检查点停在工具执行中途）。前者触发 assert 失败，后者让函数返回 False。实测中 validate(session) 返回 True，证明恢复后的消息序列是自洽的。

真实系统里，这种校验还会升级成校验和：存盘时对序列化后的字节算一个哈希一起存，恢复时重算比对。哈希对不上就拒绝恢复、退回上一个检查点——宁可丢一点进度，也不让损坏的状态污染续跑。

7.3.8 检查点频率与体积的取舍

检查点写多勤，是一场权衡：

频率越高，崩溃时丢的越少，但每步都要序列化 + 写盘——大会话的 snapshot 可能几百 KB，每步全量写会拖慢循环。一个务实的折中是每步都更新内存快照、每 N 步才落盘一次，并且在「即将执行有副作用的工具」之前强制落盘——因为那一刻之后的重放代价最高。

把「每 N 步落盘」写成策略，就是给循环套一层薄壳：

def run_with_checkpoints(agent, session, model, steps, path, every: int = 2) -> None:
    n = 0
    while True:
        status = step_once(agent, session, model, steps)
        n += 1
        if status == "done" or n % every == 0:
            save_checkpoint(path, session, model, steps)
        if status == "done":
            return

every=2 表示每两步落一次盘。收敛那一步无条件落盘——因为那是终点，必须存。这个壳与 step_once 正交：换频率只改一个数字，换执行器、换存储介质都不用动它。实测用它跑完同一段脚本，validate(session) 仍返回 True，消息条数仍是 7——降低落盘频率不改变最终结果，只改变崩溃时最多丢几步。

体积也有讲究。全量写最简单也最稳，但会话一大，每次写都是几百 KB。要省，可以只存增量消息（新增的那几条）而非整份 messages，恢复时「上次全量 + 若干增量」合并还原。代价是恢复逻辑变复杂，还得定期做一次全量压缩。工程上通常先上全量写把正确性立住，等体积真的成了瓶颈再优化——别为还没出现的性能问题提前复杂化。

7.3.9 真实模型下会怎样

真实场景的检查点，比假模型多两处复杂度：

• 存盘的对象更大、更杂。真实 messages 里有长系统提示、图片引用、工具返回的大段 JSON，序列化时要处理编码与体积。本节用 ensure_ascii=False 保证中文可读，真实系统还常要压缩（gzip）后再落盘。
• 状态可能落在外部系统。生产环境很少把检查点写在本地临时文件——进程可能被调度到另一台机器，本地盘根本读不到。真实做法是存进 Redis、对象存储或数据库，session_id 就是那把钥匙。本节的 /tmp/agentbook_ckpt/session.json 只是把「序列化 + 恢复」的机制讲清楚，换存储介质不改变这套逻辑。

还有一个假模型演示不出的风险：并发写同一个 session_id。两个请求同时恢复同一个会话、各自续跑、再各自存盘，后写的会覆盖先写的，丢掉一方的进展。真实系统要靠乐观锁（版本号）或悲观锁（处理中标记）来防。这又回到 7.2 结尾埋下的那个并发伏笔，第 13、14 章会正面处理。

第三个差异是真实任务的步骤数不可预知。假模型脚本长度固定，检查点总数可以提前算清；真实 Agent 可能三步收敛，也可能跑三十步。所以检查点策略必须自适应——按「距离上次落盘的时间」或「累积的 token 数」触发，而不是写死步数。时间驱动的策略尤其适合「单步很慢」的场景（每次模型调用要好几秒），能在崩溃损失与写盘开销之间取得更好的平衡。

把本节落到一个真实系统，检查点这一层至少要满足五条：

• 可恢复：任何一次 load_checkpoint 都能还原出与存盘时逐字段相同的状态。
• 幂等：同一份检查点恢复多次，续跑结果完全一致。
• 原子：写盘要么成功、要么不写，绝不留半个文件。
• 可校验：恢复后能判断状态是否自洽（如 validate 的那条不变量）。
• 有版本：检查点格式带 version 字段，将来改结构时能识别旧数据、做迁移。

这五条不是锦上添花，而是「崩溃续跑」从演示走向生产的分水岭。

一句话：假模型证明了「断点续跑」的逻辑闭环——状态可序列化、恢复无损、已完成的工具不重放；真实系统要做的是把这份状态搬到可靠的存储上，并加一层并发保护。

至此，第 7 章三节连成一条线：7.1 让运行可订阅，7.2 让状态可携带，7.3 让状态可落盘。minagent 现在能流式输出、能多会话隔离、能崩溃续跑。但 Session 里的 messages 依然只增不减——对话一长，历史就撑爆上下文。下一节 8.1 对话缓冲与摘要记忆 会引入记忆机制，给这份只涨不落的 messages 装上闸门。想回看状态对象是怎么定义的，翻 7.2 会话状态对象 。

阅读导航：上一节：7.2 会话状态对象 · 下一节：8.1 对话缓冲与摘要记忆 。