Eino

ADK 核心：Agent / Runner / Tur...

ADK 核心：Agent / Runner / TurnLoop

ADK（Agent Development Kit）在组件与编排之上提供Agent 应用层抽象。Component 是积木，Agent 是成品——自带运行时、事件流、中断恢复。

层级关系

┌──────────────────────────────────────────────────┐
│  ADK（Agent 层）                                  │
│  Agent 接口 · Runner · TurnLoop · Checkpoint      │
│  ChatModelAgent / DeepAgent / 自定义 Agent        │
├──────────────────────────────────────────────────┤
│  compose（编排层）                                │
│  Graph / Chain / Workflow                         │
├──────────────────────────────────────────────────┤
│  components（组件层）                              │
│  ChatModel / Tool / Retriever / Embedding ...     │
└──────────────────────────────────────────────────┘

Agent 接口

Eino 中一切 Agent 的根：

type Agent interface {
    Name(ctx context.Context) string
    Description(ctx context.Context) string
    Run(ctx context.Context, input *AgentInput, options ...AgentRunOption) *AsyncIterator[*AgentEvent]
}

方法	语义
`Name`	唯一标识。作 AgentTool 时被父 Agent 用来区分不同的子 Agent
`Description`	能力自述。作 AgentTool 时会被注入父 Agent 的 prompt，帮助 LLM 判断何时调用
`Run`	核心执行。Future 模式：立即返回迭代器，内部 goroutine 异步产出事件

任何实现了这三个方法的 struct 就是 Eino Agent。

AgentInput — 标准化输入

type AgentInput struct {
    Messages        []*schema.Message  // 与 ChatModel 输入格式一致
    EnableStreaming bool               // 调用方建议 Agent 使用流式
}

Messages：用户指令、对话历史、背景知识——与 ChatModel.Generate/Stream 参数一致
EnableStreaming：建议值，Agent 可自行决定是否忽略。支持流式的组件逐步返回，不支持的走同步

💡 这个设计让 Agent 的"输入边界"与 ChatModel 的"输入边界"保持一致，降低理解成本。

AgentEvent — 事件驱动输出

Agent 不直接返回结果，而是产出事件流。这是一种比"请求→响应"更灵活的模式：

Run() 返回 AsyncIterator
     │
     ├── Event 1: AgentEvent{Output: 文本消息（流式 chunk）}
     ├── Event 2: AgentEvent{Output: 文本消息（流式 chunk）}
     ├── Event 3: AgentEvent{Output: 工具调用结果}
     ├── Event 4: AgentEvent{Action: Interrupted}     ← 中断
     └── Event N: AgentEvent{Output: 最终回复}         ← 结束

结构拆解

type AgentEvent struct {
    AgentName string      // 框架自动填充，标识哪个 Agent 产出
    RunPath   []RunStep   // 从最外层到当前 Agent 的调用轨迹

    Output *AgentOutput   // 内容：消息/流/自定义
    Action *AgentAction   // 控制：退出/中断/跳转/终止循环
    Err    error          // 错误
}

一个 Event 要么有 Output（产出内容），要么有 Action（控制信号），要么有 Err。

AgentOutput

type AgentOutput struct {
    MessageOutput   *MessageVariant  // 模型消息（文本 & 流式）
    CustomizedOutput any             // 任意自定义数据
}

MessageVariant 统一处理流式与非流式：

type MessageVariant struct {
    IsStreaming   bool        // true → 从 MessageStream 逐帧读
    Message       *schema.Message
    MessageStream *schema.StreamReader[*schema.Message]
    Role          schema.RoleType
    ToolName      string      // 工具角色时有效
}

AgentAction

type AgentAction struct {
    Exit            bool                // 立即退出多 Agent 系统
    Interrupted     *InterruptInfo      // 请求中断（触发 Runner 保存 Checkpoint）
    TransferToAgent *TransferToAgentAction // 任务转让（不推荐）
    BreakLoop       *BreakLoopAction    // 终止 LoopAgent 循环
    CustomizedAction any                // 自定义控制信号
}

Action	使用场景
`Interrupted`	需要人工审批、需要用户补充信息
`BreakLoop`	LoopAgent 中到达终止条件
`Exit`	多 Agent 系统中当前分支不再需要执行
`TransferToAgent`	不推荐，用 AgentAsTool 替代

消费事件流 — 完整示例

func consumeEvents(iter *adk.AsyncIterator[*adk.AgentEvent]) {
    for {
        event, ok := iter.Next()
        if !ok {
            break // 迭代结束
        }

        // 处理错误
        if event.Err != nil {
            log.Printf("Agent error: %v", event.Err)
            continue
        }

        // 处理控制信号
        if event.Action != nil {
            switch {
            case event.Action.Interrupted != nil:
                log.Printf("Agent 中断: %v", event.Action.Interrupted.Data)
                return // 等待 Resume
            case event.Action.Exit:
                log.Println("Agent 退出")
                return
            case event.Action.BreakLoop != nil:
                log.Println("循环终止")
            }
            continue
        }

        // 处理输出内容
        if event.Output != nil {
            if event.Output.CustomizedOutput != nil {
                // 自定义输出
                log.Printf("Custom: %v", event.Output.CustomizedOutput)
                continue
            }

            msgOut := event.Output.MessageOutput
            if msgOut == nil {
                continue
            }

            if msgOut.IsStreaming {
                // 流式：逐帧消费
                stream := msgOut.MessageStream
                for {
                    chunk, err := stream.Recv()
                    if err == io.EOF {
                        break
                    }
                    if err != nil {
                        log.Printf("Stream error: %v", err)
                        break
                    }
                    fmt.Print(chunk.Content) // 打字机效果
                }
                stream.Close()
            } else {
                // 非流式：直接读
                fmt.Print(msgOut.Message.Content)
            }
        }
    }
}

Runner — Agent 执行入口

Runner 是 Agent 的唯一正确调用方式。它负责：

用户请求
    │
    ▼
  Runner ─── 初始化 Context（SessionValues、CheckPointID）
    │        注入 Callback
    │        调用 Agent.Run()
    │        捕获中断 → 保存 Checkpoint
    │        转发事件到 AsyncIterator
    ▼
 调用方（消费事件流）

创建与使用

// 创建
runner := adk.NewRunner(ctx, adk.RunnerConfig{
    Agent:           agent,           // 必填
    EnableStreaming: true,            // 默认 false
    CheckPointStore: store,           // 可选，启用中断恢复时必填
})

// Query：字符串自动包装为 UserMessage
iter := runner.Query(ctx, "帮我搜索今天的新闻")

// Run：完整 Messages + Options
iter := runner.Run(ctx, []*schema.Message{
    schema.SystemMessage("你是助手"),
    schema.UserMessage("你好"),
},
    adk.WithCheckPointID("cp-001"),
    adk.WithSessionValues(map[string]any{"user": "alice"}),
)

// Resume：从断点继续
iter, err := runner.Resume(ctx, "cp-001")
iter, err := runner.ResumeWithParams(ctx, "cp-001", &adk.ResumeParams{
    Targets: map[string]any{"agent-name": resumeData},
})

RunnerConfig

type RunnerConfig struct {
    Agent           Agent             // 必填
    EnableStreaming bool              // 默认流式输出
    CheckPointStore CheckPointStore   // 启用中断/恢复时必填
}

AsyncIterator — 异步事件流

type AsyncIterator[T any] struct { ... }

func (ai *AsyncIterator[T]) Next() (T, bool)

Next() 阻塞直到有新事件或迭代结束
返回 (event, true) 有新事件；(zero, false) 迭代结束

Generator 模式实现 Agent

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()   // ← 必须：确保调用方 Next() 能正常退出

        // 1. 发送首条输出
        gen.Send(&adk.AgentEvent{
            Output: &adk.AgentOutput{
                MessageOutput: &adk.MessageVariant{
                    IsStreaming: false,
                    Message:     schema.AssistantMessage("正在处理...", nil),
                },
            },
        })

        // 2. 调用模型（流式）
        stream, _ := chatModel.Stream(ctx, input.Messages)
        gen.Send(&adk.AgentEvent{
            Output: &adk.AgentOutput{
                MessageOutput: &adk.MessageVariant{
                    IsStreaming:   true,
                    MessageStream: stream, // 框架将 stream 交给调用方消费
                },
            },
        })

        // 3. 需要中断时
        gen.Send(&adk.AgentEvent{
            Action: &adk.AgentAction{
                Interrupted: &adk.InterruptInfo{
                    Data: map[string]any{"step": "awaiting_review"},
                },
            },
        })
    }()

    return iter
}

🚨 gen.Close() 必须在 goroutine 退出前调用（通常用 defer），否则调用方的 Next() 永远阻塞。

TurnLoop — 多轮对话运行时

解决"聊天应用"的典型需求：持续对话、打断、自动记账。

        用户消息队列
             │
    ┌────────▼────────┐
    │    TurnLoop     │
    │  ┌───────────┐  │
    │  │  Runner   │  │ ← Push 触发
    │  │  Agent    │  │ ← Preempt 打断
    │  └───────────┘  │
    │  Checkpoint     │ ← 自动管理
    └─────────────────┘

turnLoop := adk.NewTurnLoop(runner)

// 用户发消息 → 触发 Agent 运行
turnLoop.Push(ctx, "告诉我一个笑话")

// 用户在 Agent 运行中发了新消息 → 抢断当前运行，开始新的
turnLoop.Preempt(ctx, "等等，换一个关于程序员的")

// 停止事件循环
turnLoop.Stop()

内部机制：TurnLoop 在每次 Push/Preempt 时自动管理 CheckPoint，应用层只需声明恢复策略。

AgentRunOption — 请求维度配置

每次 Run() 可以带 Option 定制行为：

内置 Option	作用
`WithSessionValues(map[string]any)`	注入跨 Agent 共享 KV
`WithCallbacks(...callbacks.Handler)`	添加 Callback
`WithCheckPointID(string)`	指定 Checkpoint ID
`WithCancel()`	启用运行时取消

自定义 Option

// 1. 定义 Option 载体（不导出）
type myOptions struct {
    modelName string
    maxTokens int
}

// 2. 提供 Option 构造函数
func WithModelName(name string) adk.AgentRunOption {
    return adk.WrapImplSpecificOptFn(func(t *myOptions) {
        t.modelName = name
    })
}

func WithMaxTokens(n int) adk.AgentRunOption {
    return adk.WrapImplSpecificOptFn(func(t *myOptions) {
        t.maxTokens = n
    })
}

// 3. 在 Agent.Run 中读取
func (a *MyAgent) Run(ctx context.Context, input *adk.AgentInput, opts ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
    o := adk.GetImplSpecificOptions(&myOptions{}, opts...)
    // 使用 o.modelName, o.maxTokens
}

DesignateAgent — 限定 Option 生效范围

// 仅对 agent_1 生效
opt := adk.WithSessionValues(map[string]any{"key": "val"}).DesignateAgent("agent_1")
iter := runner.Run(ctx, messages, opt)

自定义 Agent 完整实现

// 自定义数据类型 —— 需 gob 注册才能用于中断
type MyAgentState struct {
    StepsCompleted int
    LastQuery      string
}

func init() {
    schema.RegisterName[*MyAgentState]("myagent.MyAgentState")
}

// MyAgent 实现 Agent + ResumableAgent
type MyAgent struct {
    model  model.BaseChatModel
    tools  []tool.BaseTool
}

func (a *MyAgent) Name(ctx context.Context) string {
    return "MyCustomAgent"
}

func (a *MyAgent) Description(ctx context.Context) string {
    return "一个自定义 Agent，支持工具调用和中断恢复"
}

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()

        state := &MyAgentState{LastQuery: input.Messages[len(input.Messages)-1].Content}
        msgs := input.Messages

        for state.StepsCompleted < 10 {
            // 调模型
            resp, err := a.model.Generate(ctx, msgs)
            if err != nil {
                gen.Send(&adk.AgentEvent{Err: err})
                return
            }

            // 检查是否需要工具
            if len(resp.ToolCalls) > 0 {
                // 产出一个要求中断的事件（假设需要人工审批）
                gen.Send(&adk.AgentEvent{
                    Action: &adk.AgentAction{
                        Interrupted: &adk.InterruptInfo{
                            Data: state, // gob 注册后才能序列化
                        },
                    },
                })
                return // 等待 Resume
            }

            // 发送回复
            gen.Send(&adk.AgentEvent{
                Output: &adk.AgentOutput{
                    MessageOutput: &adk.MessageVariant{
                        IsStreaming: false,
                        Message:     resp,
                    },
                },
            })
            state.StepsCompleted++
            return
        }
    }()

    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 包含调用方在 ResumeWithParams 中传入的数据
        userDecision := info.ResumeData // 如 "approved" 或 "rejected"

        gen.Send(&adk.AgentEvent{
            Output: &adk.AgentOutput{
                MessageOutput: &adk.MessageVariant{
                    IsStreaming: false,
                    Message:     schema.AssistantMessage(fmt.Sprintf("收到您的决定: %v，继续执行", userDecision), nil),
                },
            },
        })
    }()

    return iter
}

使用自定义 Agent

agent := &MyAgent{model: chatModel, tools: myTools}

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

// 首次运行
iter := runner.Query(ctx, "帮我执行任务", adk.WithCheckPointID("task-1"))
for {
    event, ok := iter.Next()
    if !ok { break }
    if event.Action != nil && event.Action.Interrupted != nil {
        // 处理中断...
        break
    }
}

// 恢复
iter, _ = runner.ResumeWithParams(ctx, "task-1", &adk.ResumeParams{
    Targets: map[string]any{agent.Name(ctx): "approved"},
})

语言设置

func init() {
    adk.SetLanguage(adk.LanguageChinese) // 或 adk.LanguageEnglish（默认）
}

影响范围：ADK 内置 Prompt（FileSystem、Reduction、Skill、ChatModelAgent 等组件）。自定义 Instruction 需自行处理国际化。

常见陷阱

陷阱	说明
🚨 直接调用 Agent.Run()	绕过 Runner 会丢失 Checkpoint、Callback、SessionValues 等能力
🚨 gen.Close() 未 defer	goroutine panic 后未 Close，调用方 Next() 永久阻塞
🚨 AsyncIterator 只消费一半	剩余 goroutine 会阻塞，确保消费完或主动 return
🚨 遗忘 CheckPointStore	Resume 需要初始化 Runner 时就配置，运行中无法追加
🚨 自定义类型未 gob 注册	存入 InterruptInfo.Data 或 RunLocalValue 会 panic
🚨 `Description` 写得太随意	当 Agent 作为 AgentTool 时，LLM 靠 Description 判断何时调用，写不清楚就不会被调用

编排框架（Compose）ChatModelAgent（ReAct Agent）