《AI Agent 高级编程》6.3 与自研检查点对照：轮子与现成品边界

6.3 与自研检查点对照：轮子与现成品边界

到这里我们已经手写过两版「持久化」：卷一 7.3 检查点与快照 把 Session 序列化成本地 JSON，只够单进程、一次 run_turn 内的续跑；本卷 6.1 用 append-only 日志把粒度推到步骤级，6.2 又用 tape 解决了确定性重放。这三块拼起来已经是一个能用的 Durable Execution 雏形。那么问题来了：现成框架（LangGraph 检查点、Temporal 类模型）到底比自研多做了什么？多出来的部分值不值得你让渡控制权？本节用真跑的 LangGraph 行为来回答。

本节把「自研 vs 现成」从感觉之争变成对照表：先真跑 LangGraph 的 InMemorySaver 看崩溃续跑与人在环 interrupt 的实际行为，再把自研日志引擎的存储从文件换成 SQLite 证明「换介质只是换后端」，最后给出四套方案逐项对照与一张选型边界表。

6.3.1 先把四套方案摆上桌

在对照之前，先把要比较的对象定清楚。它们不是「一个东西的不同版本」，而是四个抽象层级不同的方案：

它们的核心差别在**「谁负责决定状态边界」：自研方案里，边界由你的代码**（step 的名字、save_checkpoint 的时机）定义；LangGraph 里，边界由图的节点定义；Temporal 里，边界由工作流引擎在编译期切分。边界定义得越自动，你写的样板代码越少，但对执行模型的锁定越深。

6.3.2 真跑 LangGraph 检查点：崩溃后续跑

空谈不如实测。下面这个 LangGraph 图有 plan → search → report 三个节点，search 在第一次执行时抛异常模拟崩溃。用 InMemorySaver 做检查点，看第二次 invoke 会发生什么：

"""LangGraph 检查点最小实测：InMemorySaver + thread_id + 崩溃后从检查点续跑。"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver


class S(TypedDict):
    n: int
    log: list[str]


CALLS = {"plan": 0, "search": 0}


def plan(state: S):
    CALLS["plan"] += 1
    return {"n": state["n"] + 1, "log": state["log"] + ["plan"]}


def search(state: S):
    CALLS["search"] += 1
    if CALLS["search"] == 1:                       # 第一次执行时"崩溃"
        raise RuntimeError("模拟第 2 步崩溃")
    return {"n": state["n"] + 10, "log": state["log"] + ["search"]}


def report(state: S):
    return {"log": state["log"] + ["report"]}


g = StateGraph(S)
g.add_node("plan", plan); g.add_node("search", search); g.add_node("report", report)
g.add_edge(START, "plan"); g.add_edge("plan", "search")
g.add_edge("search", "report"); g.add_edge("report", END)

saver = InMemorySaver()
app = g.compile(checkpointer=saver)
cfg = {"configurable": {"thread_id": "run-1"}}

try:
    app.invoke({"n": 0, "log": []}, cfg)
except RuntimeError as e:
    print("异常:", e)
print("检查点里的状态:", app.get_state(cfg).values)
print("下一次待执行节点:", app.get_state(cfg).next)
print("调用计数:", CALLS)

out = app.invoke(None, cfg)                        # 输入为 None => 从断点继续
print("最终状态:", out)
print("调用计数:", CALLS)

本机实测（langgraph 1.2.14）：

异常: 模拟第 2 步崩溃
检查点里的状态: {'n': 1, 'log': ['plan']}
下一次待执行节点: ('search',)
调用计数: {'plan': 1, 'search': 1}
最终状态: {'n': 11, 'log': ['plan', 'search', 'report']}
调用计数: {'plan': 1, 'search': 2}

这份输出把 LangGraph 的行为讲透了：

• 崩溃后检查点里 log 是 ['plan']：plan 节点的结果已经落盘，get_state().next 告诉引擎「下一个该跑 search」。
• 第二次 invoke(None, cfg) 从断点续跑：plan 的调用计数保持 1，没有被重跑——这就是「跳过已完成步骤」，与 6.1 的自研引擎效果一致。
• search 的调用计数变成 2：崩溃发生在 search 节点内部，LangGraph 的检查点粒度是节点级，节点没跑完就不算完成，所以整个 search 节点被从头重跑一次。

最后一条是最关键的边界：LangGraph 保证的是「节点级至少一次执行」，不保证「节点内部不重放」。如果 search 节点里已经发了邮件、写了库，重跑会重复执行——这正是 6.2 讲的副作用隔离问题，LangGraph 把它留给了你。所以「用框架 = 万事大吉」是误解，节点内副作用仍要自己幂等。

再补充一个实测事实：langgraph.checkpoint.memory.InMemorySaver 是内置的，而 PostgresSaver / SqliteSaver 属于独立包 langgraph-checkpoint-postgres / langgraph-checkpoint-sqlite，本机 venv 里只有基座包 langgraph-checkpoint 4.2.0，直接 import 会 ModuleNotFoundError。也就是说，框架替你做好了「接口」，但生产级的持久化后端要你自己装、自己运维。

检查点是怎么分布的？把 saver 里存的所有检查点列出来就清楚了：

for t in saver.list(cfg):
    cid = t.config["configurable"]["checkpoint_id"]
    print("checkpoint", cid, "->", t.checkpoint["channel_values"].get("log"))

实测输出（省略 id 前缀）：

checkpoint 1f1c4e50-4796 -> ['plan', 'search', 'report']
checkpoint 1f1c4e50-4795 -> ['plan', 'search']
checkpoint 1f1c4e50-4791 -> ['plan']
checkpoint 1f1c4e50-4790 -> []
checkpoint 1f1c4e50-478a -> None

每个 superstep 落一条检查点——从初始空状态、到 plan 之后、到 search 之后、到 report 之后，一路可追溯。这份历史还能支撑「时间旅行」：拿到任意一个 checkpoint_id，就能把图恢复到那一刻。这是自研方案里要额外写不少代码才能有的能力。每个 Checkpoint 对象带 v 字段（本机实测 v=4）和 channel_versions，正是用来做状态版本迁移的——格式升级时靠它识别旧数据。

6.3.3 真跑 LangGraph interrupt：人在环的持久化暂停

LangGraph 比自研多出来的一个实打实的能力，是 interrupt() + Command(resume=...)：让图在一个节点里暂停，把控制权交出去，等外部（人）给答复后再恢复。这是「跨天审批」这类场景的直接支持：

from langgraph.types import interrupt, Command

def approval_node(state: S):
    ans = interrupt({"ask": "是否批准草稿?", "draft": state["draft"]})
    return {"approved": bool(ans)}

实测流程（完整图与 6.3.2 同构，中间节点换成 approval_node）：

--- 第一次 invoke：跑到 approval 暂停 ---
interrupt 负载: [Interrupt(value={'ask': '是否批准草稿?', 'draft': '研究草稿 v1'}, id='651cf8b54424eb91574e9d2825a43776', response_schema=None)]
暂停后状态: {'draft': '研究草稿 v1', 'approved': False} | next: ('approval',)
--- 恢复：Command(resume=True) ---
最终状态: {'draft': '研究草稿 v1 [已发布]', 'approved': True}

interrupt() 让图停在 approval 节点，状态被完整落进检查点；恢复时用 Command(resume=True) 把「人的答复」喂回去，图从断点继续跑到 publish。整个过程进程可以完全退出，暂停状态靠检查点存活——这正是自研方案要自己搭一大套才能做到的事，LangGraph 用两个原语就给了。

不过要看清它给的到底是什么：interrupt 给你的是**「暂停/恢复」的机制**，但「谁在什么时候恢复」「审批超时怎么办」「多个待审批任务怎么排队」这些工作流治理问题，框架不负责，仍要你自己在业务层解决。

6.3.4 换存储介质：自研的边界在哪

回到自研方案，一个常见的质疑是：「自研的日志写在本地文件里，进程换台机器就读不到了，所以自研不可用。」这个质疑把机制和存储介质混为一谈了。下面证明：把 6.1 引擎的 journal 从文件换成 SQLite，只改存储后端，引擎逻辑一行不动：

"""同一个 Durable 引擎，把 journal 从文件换成 SQLite —— 只改存储后端。"""
import sqlite3, json


class SqliteJournal:
    def __init__(self, path):
        self.db = sqlite3.connect(path)
        self.db.execute("CREATE TABLE IF NOT EXISTS steps("
                        "run TEXT, step TEXT, output TEXT, PRIMARY KEY(run, step))")
        self.db.commit()

    def get(self, run, step):
        row = self.db.execute("SELECT output FROM steps WHERE run=? AND step=?",
                              (run, step)).fetchone()
        return json.loads(row[0]) if row else None

    def put(self, run, step, output):
        self.db.execute("INSERT OR REPLACE INTO steps VALUES(?,?,?)",
                        (run, step, json.dumps(output, ensure_ascii=False)))
        self.db.commit()

实测两次「进程」（第二次模拟重启）：

第一次: 真执行 plan 并落库
第二次: 命中 SQLite, 跳过 plan 真执行 -> 拆成两个子问题

把 sqlite3 换成 psycopg（PostgreSQL）或 redis（Redis），把 run 列换成 thread_id，就是一套「状态存在外部数据库、可跨机器读取」的方案——本机的 ab-pg（PostgreSQL 17.11）与 ab-redis（Redis 7.4.11）容器正是干这个用的。所以自研方案的真正边界不在「能不能换存储」，而在「愿不愿意自己写并发控制、日志压缩、状态迁移、可观测」这些工程配套。

6.3.5 逐项对照表

把四套方案放在同一张表里逐项比：

表格里的每一格都能在本章前两节找到出处，不是空泛断言：粒度、恢复、重放、幂等来自 6.1/6.2 的实现；interrupt、thread_id、channel_versions 来自 6.3.2/6.3.3 的实测；「需用户自己保证副作用幂等」来自 6.3.2 里 search 被重跑两次的观察。

6.3.6 选型边界：什么时候自研够用

对照表给出了「有哪些差异」，但选型要看「你的场景落在哪一档」。下面这张表直接给结论：

一条务实的经验法则：先自研到「痛点出现」再上框架。如果你的任务能在单机、几分钟内跑完，自研一个 append-only 日志（6.1 那 20 行）就够了；只有当「人在环暂停」「多会话并发隔离」「跨服务编排」这些具体痛点真的出现时，LangGraph 或 Temporal 的复杂度才划算。反过来，如果你已经确定要做一个「跑几天、跨多个审批环节、多个用户共享」的平台，那从一开始就上框架，避免把自研的并发与迁移问题留给未来的自己。

还要警惕一个反向陷阱：上了框架不等于拿到了「恰好一次」语义。6.3.2 的实测已经证明，LangGraph 的节点在崩溃后会被重跑，节点内的副作用照样会重复。所以无论自研还是上框架，幂等这道功课都逃不掉——框架能替你管状态、管调度，但管不了「你那个工具到底能不能安全地执行两次」。

6.3.7 这套框架替你做了什么、你又让渡了什么

现成框架（以 LangGraph 为例）替你做的：把「状态在每个节点后自动落盘」「按 thread_id 隔离会话」「interrupt 暂停 / Command 恢复」「检查点版本号与序列化」这些机制都封装好了，你不用再手写 journal、tape、幂等账本这一套。

你让渡的：任务必须写成图的形状（节点 + 边），而不是任意函数；要接受节点级的检查点粒度——节点内的副作用仍可能被重放，幂等还得自己保证；生产级持久化后端（PostgreSQL/Redis saver）要自己装、自己运维；整个执行模型被锁定在框架的抽象上，换框架等于重写。

自研方案的对称面是：你保留了全部控制权与零锁定，但并发控制、日志压缩、状态迁移、可观测这些「框架白送」的东西，都要自己造、自己修。边界不在「自研能不能用」，而在「你的团队愿意为这些工程配套投入多少」。本章三节给出的自研引擎，正是让你在「上框架」之前，先亲手把 Durable Execution 的每条不变量都走一遍——走过之后，无论选哪条路，你都清楚自己换来的和让渡的是什么。

下一章 7.1 Computer Use 基础 会换一个战场：让 Agent 直接操作屏幕与浏览器，并处理随之而来的安全与成本问题。想先回看框架源码里检查点是怎么实现的，见 5.1 读 LangGraph 。

阅读导航：上一节：6.2 确定性重放与副作用：模型调用变可重放步骤 · 下一节：7.1 Computer Use 基础：截图/坐标动作/循环最小实现 。