中断与恢复（Interrupt / Resume）

Eino

Eino 的中断恢复机制实现 Human-in-the-Loop：Agent 可在运行中主动暂停，等待外部输入后从断点继续。

整体流程

┌───────────┐     ┌───────────┐     ┌───────────────┐     ┌──────────┐
│  Agent    │────→│  Runner   │────→│ CheckPointStore│     │  调用方   │
│  运行中   │     │  捕获中断  │     │  持久化状态     │     │  收到事件  │
└─────┬─────┘     └─────┬─────┘     └───────┬───────┘     └────┬─────┘
      │                 │                    │                   │
      │  Interrupted    │  gob 序列化        │  Set(cpID, data)  │
      │  Action         │  运行状态           │                   │
      │                 │                    │                   ▼
      │                 │               ┌───────────┐    用户/系统处理
      │                 │               │  外部处理  │    （审批、输入…）
      │                 │               └─────┬─────┘
      │                 │                    │
      │                 │  ← Resume(cpID) ───┘
      │                 │  gob 反序列化
      │  从断点继续 ←───┘  恢复状态

中断：触发暂停

Agent 通过产出 Interrupted Action 触发中断。两种方式：

方式一：Agent 代码直接发送事件

gen.Send(&adk.AgentEvent{
    Action: &adk.AgentAction{
        Interrupted: &adk.InterruptInfo{
            Data: map[string]any{
                "question":  "请确认是否继续？",
                "step":      3,
                "timestamp": time.Now(),
            },
        },
    },
})
// 发送后 goroutine 可以 return，Runner 会接管

方式二：Tool 中返回中断错误（更常用）

func (t *ApprovalTool) InvokableRun(ctx context.Context, argsJSON string, opts ...tool.Option) (string, error) {
    // 需要人工审批 → 返回中断错误
    return "", compose.NewInterruptAndRerunErr(&adk.InterruptInfo{
        Data: ApprovalRequest{
            RequestID: "req-001",
            Content:   "需要管理员审批",
        },
    })
}

💡 compose.NewInterruptAndRerunErr 会触发中断，Resume 时重新执行该 Tool（带上新的参数）。

调用方捕获中断

for {
    event, ok := iter.Next()
    if !ok { break }

    if event.Action != nil && event.Action.Interrupted != nil {
        interruptData := event.Action.Interrupted.Data

        // 向用户展示中断信息，收集输入
        fmt.Printf("Agent 需要您的输入: %v\n", interruptData)
        userInput := getUserInput()

        // 稍后通过 Resume 恢复
        break
    }
}

CheckPointStore — 状态持久化接口

Runner 捕获中断后，将完整运行状态序列化存入 CheckPointStore：

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

框架使用 gob 作为序列化格式。每个 Run 通过 CheckPointID 标识：

runner := adk.NewRunner(ctx, adk.RunnerConfig{
    Agent:           agent,
    CheckPointStore: store,  // ← 必须配置
})

iter := runner.Query(ctx, "给我推荐一本书",
    adk.WithCheckPointID("session-001"),
)

内置实现

实现	说明
`adk.NewInMemoryStore()`	内存存储，适合开发/测试
自定义实现 `CheckPointStore` 接口	可接入 Redis/MySQL/etcd 等

// 自定义 Redis Store 示例
type RedisStore struct {
    client *redis.Client
}

func (s *RedisStore) Set(ctx context.Context, key string, value []byte) error {
    return s.client.Set(ctx, key, value, 24*time.Hour).Err()
}

func (s *RedisStore) Get(ctx context.Context, key string) ([]byte, bool, error) {
    data, err := s.client.Get(ctx, key).Bytes()
    if err == redis.Nil {
        return nil, false, nil
    }
    return data, true, err
}

自定义类型注册（gob）

这是最容易被忽略但最关键的一步。Eino 使用 Go 标准库 encoding/gob 序列化 checkpoint 数据。

为什么需要注册

gob 序列化时需要知道类型的"蓝图"。内置类型（string、int、map[string]any、*schema.Message 等）已被框架预注册。但你定义的结构体没有——gob 不认识它就会 panic：

panic: gob: type not registered for interface: mypkg.MyData

注册方式

import "github.com/cloudwego/eino/schema"

// 方式一：schema.RegisterName[T]("自定义名称") — 自定义注册名
func init() {
    schema.RegisterName[*ApprovalData]("myapp.ApprovalData")
}

// 方式二：schema.RegisterT[T]() — 自动用 package path + type name
func init() {
    schema.RegisterT[*ApprovalData]()
    // 自动注册名为 "github.com/myorg/myapp.ApprovalData"
}

API	名称	适用
`schema.RegisterName[T](name)`	自定义	需要简洁名称或向后兼容
`schema.RegisterT[T]()`	自动生成	新类型，无需关心名称

哪些类型需要注册？

所有可能存入以下位置的非内置类型都必须注册：

InterruptInfo.Data — 中断携带的数据
adk.SetRunLocalValue(ctx, key, val) — 运行时本地存储
adk.AddSessionValue(ctx, key, val) — 跨 Agent Session（如果值需要被 checkpoint 持久化）

完整示例：从定义到恢复

package main

import (
    "context"
    "encoding/json"
    "fmt"

    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/schema"
)

// ── 1. 定义需要持久化的类型 ──
type ReviewRequest struct {
    DocumentID string `json:"document_id"`
    Title      string `json:"title"`
    Content    string `json:"content"`
}

type ReviewDecision struct {
    Approved bool   `json:"approved"`
    Comment  string `json:"comment"`
}

// ── 2. init() 中注册 ──
func init() {
    schema.RegisterName[*ReviewRequest]("myapp.ReviewRequest")
    schema.RegisterName[*ReviewDecision]("myapp.ReviewDecision")
}

// ── 3. 中断时携带数据 ──
func (a *MyAgent) Run(ctx context.Context, input *adk.AgentInput, opts ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
    iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()

    go func() {
        defer gen.Close()

        // ... 处理逻辑 ...

        // 中断：需要人工审核
        gen.Send(&adk.AgentEvent{
            Action: &adk.AgentAction{
                Interrupted: &adk.InterruptInfo{
                    Data: &ReviewRequest{           // ← 自定义类型
                        DocumentID: "doc-123",
                        Title:      "Q2 财务报告",
                        Content:    "...",
                    },
                },
            },
        })
    }()

    return iter
}

func (a *MyAgent) Resume(ctx context.Context, info *adk.ResumeInfo, opts ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
    iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()

    go func() {
        defer gen.Close()

        // info.ResumeData 是调用方传入的恢复数据
        var decision ReviewDecision
        _ = json.Unmarshal(info.ResumeData, &decision)

        if decision.Approved {
            gen.Send(&adk.AgentEvent{
                Output: &adk.AgentOutput{
                    MessageOutput: &adk.MessageVariant{
                        IsStreaming: false,
                        Message:     schema.AssistantMessage("审核通过，继续处理...", nil),
                    },
                },
            })
        } else {
            gen.Send(&adk.AgentEvent{
                Output: &adk.AgentOutput{
                    MessageOutput: &adk.MessageVariant{
                        IsStreaming: false,
                        Message:     schema.AssistantMessage(fmt.Sprintf("审核驳回: %s", decision.Comment), nil),
                    },
                },
            })
        }
    }()

    return iter
}

// ── 4. 调用方：中断 → 处理 → 恢复 ──
func main() {
    ctx := context.Background()

    agent := &MyAgent{}
    runner := adk.NewRunner(ctx, adk.RunnerConfig{
        Agent:           agent,
        CheckPointStore: adk.NewInMemoryStore(),
    })

    // 首次运行
    iter := runner.Query(ctx, "审核文档 doc-123", adk.WithCheckPointID("cp-1"))
    for {
        event, ok := iter.Next()
        if !ok { break }

        if event.Action != nil && event.Action.Interrupted != nil {
            // 提取中断数据
            req := event.Action.Interrupted.Data.(*ReviewRequest)
            fmt.Printf("需要审核: %s\n", req.Title)

            // 模拟人工决策
            decision, _ := json.Marshal(&ReviewDecision{Approved: true})

            // 恢复
            iter2, _ := runner.ResumeWithParams(ctx, "cp-1", &adk.ResumeParams{
                Targets: map[string]any{
                    agent.Name(ctx): decision,
                },
            })
            consumeEvents(iter2)
            break
        }
    }
}

注册要点速查

场景	需要的注册
`InterruptInfo.Data = map[string]any{"k": v}`	v 的类型需要注册（如果 v 是自定义 struct）
`InterruptInfo.Data = &MyStruct{}`	`MyStruct` 需要注册
`SetRunLocalValue(ctx, k, &MyStruct{})`	`MyStruct` 需要注册
`AddSessionValue(ctx, k, 42)`	不需要（int 是内置类型）
`InterruptInfo.Data = "hello"`	不需要（string 是内置类型）
struct 的字段中包含 `map[string]any`	不需要（interface{} 被 gob 原生支持）

🔬 框架已预注册：*schema.Message、[]*schema.Message、*schema.Document、map[string]any 等核心类型。只要你的自定义类型中没有嵌套新的自定义 struct pointer，就不需要注册。

恢复

简单恢复

目标 Agent 的 Resume 被调用，不需要额外参数：

iter, err := runner.Resume(ctx, "cp-1")

精确恢复（推荐）

指定要恢复哪个 Agent + 传入恢复数据：

iter, err := runner.ResumeWithParams(ctx, "cp-1", &adk.ResumeParams{
    Targets: map[string]any{
        "MyAgent": decision,           // 目标 Agent Name → 恢复数据
        "SubAgent": otherDecision,
    },
    // 也可以指定只恢复特定中断点
    TargetInterrupts: map[string]bool{"MyAgent": true},
})

ResumableAgent 接口

支持 Resume 的 Agent 需实现：

type ResumableAgent interface {
    Agent
    Resume(ctx context.Context, info *ResumeInfo, opts ...AgentRunOption) *AsyncIterator[*AgentEvent]
}

type ResumeInfo struct {
    ResumeData  any    // ResumeWithParams 传入的恢复数据
    // ... 其他内部字段
}

💡 ChatModelAgent、DeepAgent、所有内置 Workflow Agent 已自动实现了 ResumableAgent。

TurnLoop + 中断

TurnLoop 自动管理多轮中断恢复的记账（bookkeeping），应用层只需声明策略：

Push("用户消息") → Agent 运行 → 中断
    │  TurnLoop 自动保存 Checkpoint
    ▼
Resume("cp-id", params) → Agent 继续 → 完成
    │  TurnLoop 自动更新 Checkpoint
    ▼
Push("下一条消息") → 新轮次

常见陷阱

陷阱	说明
🚨 忘记 gob 注册	序列化时 panic：`gob: type not registered for interface`。排查很困难，因为 panic 发生在 Runner 内部
🚨 `schema.RegisterName` vs `gob.RegisterName`	必须用 `schema.RegisterName[T]`，底层会自动调用 `gob.RegisterName`。直接用 `gob.RegisterName` 可能和框架的类型管理不兼容
🚨 结构体字段不是导出的	gob 只序列化导出字段（首字母大写）。私有字段在 Resume 后值为零值
🚨 注册名称不一致	同一个类型在序列化和反序列化时注册名不同会导致 `gob: name not registered`
🚨 CheckPointStore 未配置	必须在 `NewRunner` 时就配置；运行时无法追加，中断状态会丢失
🚨 CheckPointID 复用	Resume 后状态更新，旧状态被覆盖。如需保留旧快照，使用新的 CheckPointID
🚨 未实现 ResumableAgent	自定义 Agent 需显式实现 `Resume()`，否则 Runner 恢复时会报错

多 Agent 协作 Callback 与可观测性