Checkpoint 源码:Agent 执行到一半怎么保存(第80篇-E66)

系列「企业级 AI Agent 实现拆解」E66 篇，Part 14 记忆篇第四章。上一篇讲记忆多了怎么衰减淘汰——那是长期知识。这篇讨论另一种持久化：Agent 正在执行时被中断了（用户关页面、服务器重启、手动暂停），怎么保存状态、下次从哪接着跑。
这篇回归 Eino 主线代码。compose.CheckPointStore 是 Eino 框架里正式定义的接口（不像前几篇记忆那样是项目自己设计的）。它的核心只有两个方法——Get 和 Set 两块 []byte，但中间发生的事值得拆开看。

读完这篇你会知道

CheckPointStore
只有两个方法：Get 和 Set，参数都是 string + []byte——框架不关心你存哪、用什么格式
checkpoint 里实际存的是什么：通道状态、各节点输入、图级 State、嵌套子图中断信息，共七种类型
Go 里 interface{} 的 typed nil 问题：一个看起来是 nil 的值，序列化后回去不是 nil——eino 在底层做了转换但你可以绕过
流式管道怎么被序列化：存 checkpoint 前把 streamReader 展开成普通值，恢复时再包回去
WriteToCheckPointID != CheckPointID
：从一个旧 checkpoint 恢复，把新的进度存到另一个地方
中断怎么跟 checkpoint 协同：InterruptID2Addr 记录谁在哪个节点暂停了，恢复时框架知道该从哪继续
StateModifier
是个钩子：每次 checkpoint 读写时你都有一次机会修改图 State

一、接口就是两块 `[]byte`

CheckPointStore 定义在 eino/internal/core/interrupt.go：

type CheckPointStore interface {    Get(ctx context.Context, checkPointID string) ([]byte, bool, error)    Set(ctx context.Context, checkPointID string, checkPoint []byte) error}

两个方法，参数全是字符串和字节数组。**没有 schema、没有版本号、没有 lastModified、没有任何索引能力。**四个字：字节存在哪由你决定。

这很 Go——框架只定协议，不帮你挑存储。map[string][]byte 能当 CheckPointStore，Redis 能，S3 能，PostgreSQL 也能。

框架另外定义了一个可选接口：

type CheckPointDeleter interface {    Delete(ctx context.Context, checkPointID string) error}

store 如果没实现 Delete，过期的 checkpoint 就永远不会被框架自动清理。注释说得很直白——「自己管生命周期，比如 TTL、外部清理、或者实现这个接口」。

你可以用一个 map 当 store，用另一个 Redis 当 store，但生产上务必加过期清理。

二、checkpoint 里到底存了什么

框架往里塞的不是随机的 []byte，而是一个 checkpoint 结构体的序列化：

type checkpoint struct {    Channels       map[string]channel               // ① 通道状态    Inputs         map[string]any                   // ② 各节点本轮输入    State          any                               // ③ 图级 State    SkipPreHandler map[string]bool                   // ④ 跳过哪些节点的前处理    RerunNodes     []string                          // ⑤ 需重跑的节点    SubGraphs map[string]*checkpoint                 // ⑥ 嵌套子图的 checkpoint    InterruptID2Addr  map[string]Address            // ⑦ 中断 ID → 地址    InterruptID2State map[string]core.InterruptState //   同一中断的内部状态}

七种信息，各自在恢复流程里扮演不同角色：

① Channels：DAG 和 Pregel 的通道——节点之间传数据的东西。保存下来意味着恢复时不需要重新计算已完成的节点，直接从通道里取值继续。这也决定了 checkpoint 的写入时机：一个超级步完成后，不是每执行一个节点就写一次。

② Inputs：每个节点本轮收到了什么输入。类型是 map[string]any——key 是节点名，值可以是任意类型。这里面就有 Go typed nil 的坑，第四节会展开。

③ State：你在 NewGraph 时用 WithGenLocalState 创建的那个状态对象。Eino 框架提供的唯一一个「把业务数据带进 checkpoint」的钩子。如果你的 Agent 需要在中断恢复后知道「上次做到了第几步」，数据放这里是最合适的。

④ SkipPreHandler / ⑤ RerunNodes：哪几个节点重跑时不走预处理。正常流程里节点从预处理开始执行，但从 checkpoint 恢复时已经跑过一部分——不需要再预处理一遍，直接从上次中断的位置继续。这两个字段记录了哪些节点该跳过预处理、哪些要重跑。

⑥ SubGraphs：图可以嵌套——一个图节点里面可能又是一个完整的图（比如一个工具节点的执行路径）。子图的 checkpoint 存在这里，父图自己的检查点存在父级。这个设计让嵌套图的中断恢复天然递归——恢复父图把子图 checkpoint 递进去就好。

⑦ InterruptID2Addr / InterruptID2State：下一节是重点。

三、中断与 checkpoint 的关系

第 4 篇讲 HITL 时说过——Agent 执行到调用工具这一步，框架会暂停等用户确认。暂停的信息存在哪？

就在这个 checkpoint 的这两个 map 里。

InterruptID2Addr 记录了哪个中断 ID 对应图里的哪个节点位置。InterruptID2State 存着该中断的内部状态——框架需要知道这个中断是 PENDING（等着被恢复）、DONE（已处理）、还是 TIMEOUT（过期了）。

恢复流程大致是这样：

用户点击「继续」→ 前端把 checkpointID 和 interruptID 发给后端
框架拿 checkpointID 从 store 里 Get 出来，反序列化成 checkpoint
从 InterruptID2Addr 找到是哪个节点卡住了
从 Channels 和 Inputs 恢复这个节点需要的上下文
从 State 恢复业务数据
从那个节点继续往下跑——前面跑完的不会重新跑一遍

这个流程里最容易被忽略的一步在第 5 步：**State 是你自己的业务对象。**如果你在 WithGenLocalState 里初始化了一堆字段，但中间没更新它们，恢复出来的就是那个初始化的空壳。

checkpoint 不是快照，它不会拍一张你的整个进程内存。 它只存了框架能看见的东西——通道、输入、图中断点、以及你主动放进 State 里的数据。

四、Go 的 typed nil 问题

代码里有一个容易被忽略的细节：normalizeCheckpointTypedNilInputs。

在 checkpoint 的 set 方法里：

func(c *checkPointer) set(ctx context.Context, id string, cp *checkpoint) error {    normalizeCheckpointTypedNilInputs(cp)  // ← 必须先过这一关    data, err := c.serializer.Marshal(cp)    // ...}

这个函数在做什么？

funcisTypedNil(v any) bool {    if v == nil { return false }    rv := reflect.ValueOf(v)    switch rv.Kind() {    case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:        return rv.IsNil()    default:        return false    }}

Go 里有一个著名陷阱：一个非 nil 的 interface 可以包装一个 nil 指针。 比如：

var p *MyType = nilvar v any = pv == nil   // false！因为 interface 本身（类型信息 + 值）不是 nil

如果这个 v 被序列化成一个具体类型的 nil 指针然后存进 checkpoint，反序列化回来之后它就不再是 nil 了——序列化器会把类型信息一起还原，但框架在判断「这个节点有没有输入」时有 nil 检查。一个 typed nil 会让判断误以为是有效输入，走到错误的执行路径。

normalizeCheckpointTypedNilInputs 在序列化之前把这些 typed nil 全部替换成真正的 nil。

保险起见：如果你的 State 里有 interface 字段，存之前自己先检查一遍 typed nil。normalize 只处理 Inputs，不处理你塞在 State 里的自定义数据。

五、流式值的转换：展开再恢复

checkpoint 里存的都是普通数据，但执行中的节点之间可能传的是 streamReader（第 20 篇讲过流式管道）。streamReader 不能直接序列化——它背后有 goroutine 在跑、有 channel 在收数据。

存 checkpoint 时必须先把流读光：

func(c *checkPointer) convertCheckPoint(cp *checkpoint, isStream bool) error {    for _, ch := range cp.Channels {        err = ch.convertValues(func(m map[string]any) error {            return c.sc.convertOutputs(isStream, m)        })    }    return c.sc.convertInputs(isStream, cp.Inputs)}

恢复时反过来——把普通数据包回 streamReader，让下游节点以为自己还是在读一个流：

func(c *checkPointer) restoreCheckPoint(cp *checkpoint, isStream bool) error {    // 同样遍历 Channels 和 Inputs，但做的是 restoreOutputs / restoreInputs}

这意味着存 checkpoint 是有代价的——必须等当前 super-step 的所有流处理完才能序列化。如果你的图里有一个长时间运行的工具调用（比如调用一个图片生成服务等 30 秒），那 checkpoint 就会卡在它完成之后才写。

如果你的 checkpoint store 是一个远程服务（Redis/S3），而 Set 的调用在请求上下文里（没有独立超时），那么这个远程写会成为请求延迟的一部分。解决方式是异步写——但框架现在没帮你做这个，得自己在 Set 实现里异步化。

六、`StateModifier`：读写前后的钩子

type StateModifier func(ctx context.Context, path NodePath, state any) error

每次 checkpoint 被写入或读出时，框架会调这个函数。path 是当前节点的路径（嵌套图里有多个层级），state 是图级的 State。

这个钩子用来做什么？

脱敏
：写入前把手机号、身份证号从 State 里清掉
版本迁移
：读出旧格式 State 时就地转换成新格式
审计日志
：读写前后记录谁在什么时间操作了这个 checkpoint

它在 context 里传递：

ctx = context.WithValue(ctx, stateModifierKey{}, modifier)

所以每个请求可以带不同的 modifier，不像 WithGenLocalState 是编译时固定的。

七、几个不那么明显的东西

① `WithWriteToCheckPointID`：从 A 恢复，存到 B

funcWithWriteToCheckPointID(checkPointID string) Option

没用过的人容易忽略这个字段。它让你可以从一个已有的 checkpoint 恢复执行，但把新的状态存到一个新的 checkpoint ID 里。

应用场景：**回溯。**用户对某个工具调用的结果不满意，想回到调用之前换个问法。这时候你需要从旧 checkpoint 恢复（CheckPointID = old），但不能覆盖它（覆盖了就回不去了），所以写到一个新地方（WriteToCheckPointID = new）。

② `WithForceNewRun`：完全无视已有 checkpoint

func WithForceNewRun() Option

就算 checkpointID 对应的地方有数据，也全部忽略，从头开始跑。相当于给用户一个「重新开始」的按钮。

③ `MigrateCheckpointState`：架构升级

funcMigrateCheckpointState(data []byte, serializer Serializer,    migrate func(state any) (any, bool, error)) ([]byte, error)

你的图 State 改字段了（比如老版本叫 History，新版本改名叫 Messages），但用户还有几十个存量的 checkpoint 没跑完。这个函数就是用来批量迁移的：拿到旧的序列化数据、把所有 State 跑一遍你的 migrate 函数、返回新的序列化数据。

注释里专门提到 changed 返回值：如果 changed=false，说明这条 checkpoint 里的 State 不需要改，函数返回完全不修改的原始字节——没有拷贝、没有重新序列化、没有浪费 CPU。

④ 序列化器默认是内部的 gob

如果不传 WithSerializer，用的就是 eino 自己的 serialization.InternalSerializer（本质上是 gob + 类型注册表）。如果你在 State 里存了 gob 不认识的自定义类型，需要调用 schema.RegisterName[T] 注册。

func init() {    schema.RegisterName[*checkpoint]("_eino_checkpoint")    schema.RegisterName[*dagChannel]("_eino_dag_channel")    // ...}

eino 自己的内部类型是在 init() 里自动注册的，所以默认能用。但你的类型得自己注册。

如果不想注册也不想换序列化器，那就别在 State 里放复杂类型，放 map[string]any。

小结

接口极其简单
：Get(id) → []byte、Set(id, []byte)，框架不关心你存哪
checkpoint 里不是整堆内存
，是 7 种框架能追踪的东西：通道状态、节点输入、图 State、子图、中断信息——你的业务数据只有 State 这一个入口
中断恢复全靠这两个 map
：InterruptID2Addr（哪个节点） + InterruptID2State（什么状态）。第 4 篇讲的上层 HITL，底层就是这两个 map
Go 的 typed nil
可能绕过序列化器的 nil 检查：eino 只在存 checkpoint 前处理 Inputs，不检查你放 State 里的自定义数据
存 checkpoint 之前要先读光所有流
：streamReader 不能直接序列化，得展开成普通值，恢复时再包回去。代价是必须等当前 super-step 跑完
WriteToCheckPointID 是回溯的基础
：从一个旧 checkpoint 恢复，存到新地方，原来的不动
加 CheckPointDeleter 或 TTL
：不然后面全是死 checkpoint

下一篇讲用户画像——Agent 怎么从对话里自动提炼出你是谁、你喜欢什么：异步提取、结构化 Profile、以及为什么不在对话主路径上做。

代码状态说明：
**本篇是源码解读，没有跑 demo。**所有 Go 代码、接口定义均引自 eino v0.9.13 的 compose/checkpoint.go 和 internal/core/interrupt.go；参考 inmem 实现来自 eino-examples/compose/graph/react_with_interrupt/main.go。是我从源码里读的行为，不是我自己写的。
没有实测的部分：typed nil 在序列化前后的行为差异是 Go 语言层面的已知特征，不是我在特定版本上触发的 bug；streamReader 的展开/恢复流程我只做了源码阅读，没有构造一个故意卡流的图来验证行为。