Eino Callback 系统详解：从入门到实战

Eino 的 callback 系统是一个轻量、高效的事件通知机制。

核心设计是“把 handler 列表藏在 context 里”，通过 context 在组件调用链中传递，没有全局事件总线、注册表或反射。

本文将从设计原理、操作步骤、关键特性到常见问题，带你全面掌握这个系统的使用。

一、核心设计：Context 作为 carrier

Callback 系统的一切都围绕一个 manager 结构体展开。它被塞进 context 的 key-value 中，随调用链流动。

每个节点从 context 中读取 manager 并触发 handler。

1.1 manager 结构体

// internal/callbacks/manager.go
type manager struct {
    globalHandlers []Handler        // 进程级 handlers（AppendGlobalHandlers 设置）
    handlers       []Handler        // 此次调用注册的 handlers（CtxWithHandlers 设置）
    runInfo       *RunInfo          // 当前组件身份（类型、名字、组件种类）
}

type CtxManagerKey struct{}         // context 的 key（空结构体，避免 string key 冲突）

func ctxWithManager(ctx context.Context, m *manager) context.Context {
    return context.WithValue(ctx, CtxManagerKey{}, m)
}

func managerFromCtx(ctx context.Context) (*manager, bool) {
    v := ctx.Value(CtxManagerKey{})
    m, ok := v.(*manager)
    if ok && m != nil {
        n := *m          // ← 注意：浅拷贝，不共享同一指针
        return &n, true
    }
    return nil, false
}

关键设计： managerFromCtx 返回的是 *m 的一个副本（n := *m），不是指向原结构体的指针。

这意味着每个节点读到的是独立的 manager 实例。对 manager 的修改不会影响其他节点——Graph 并发执行时不会有数据竞争。

二、两条注册路径

路径一：全局注册（进程级，对所有节点生效）

// 进程启动时调用一次
callbacks.AppendGlobalHandlers(myTracingHandler, myMetricsHandler)

// 内部实现
var GlobalHandlers []Handler       // 全局 slice
func AppendGlobalHandlers(handlers ...Handler) {
    GlobalHandlers = append(GlobalHandlers, handlers...)
}

适合：链路追踪（OpenTelemetry）、全局 Token 计量、Langfuse 集成。

任何需要观测所有节点的场景都适用。

路径二：每次调用注册（通过 context，对此次调用生效）

// 在发起 Agent 调用之前
ctx = callbacks.CtxWithHandlers(ctx, []callbacks.Handler{
    &SessionTokenTracker{sessionID: sid},
    &AuditLogger{requestID: rid},
})
sr, err := agent.Stream(ctx, messages)

CtxWithHandlers 把 handlers 塞入 manager。context 带着 manager 流经整个调用链。

Graph 里每个节点都能从 ctx 里读到它们。

三、EnsureRunInfo：组件方法开头的必要调用

每个组件方法（Generate、Stream、Invoke……）开头都会调：

ctx = callbacks.EnsureRunInfo(ctx, cm.GetType(), components.ComponentOfChatModel)

做了什么：

func EnsureRunInfo(ctx context.Context, typ string, comp components.Component) context.Context {
    cbm, ok := managerFromCtx(ctx)
    if !ok {
        // ctx 里没有 manager → 用全局 handlers 创建一个新的
        return InitCallbacks(ctx, &RunInfo{Type: typ, Component: comp})
    }
    if cbm.runInfo == nil {
        // manager 存在但没有 RunInfo → 补充 RunInfo
        return ReuseHandlers(ctx, &RunInfo{Type: typ, Component: comp})
    }
    return ctx   // 已经有 RunInfo，直接返回
}

为什么需要？ callback handler 需要知道“触发这次事件的是哪个组件”（类型、名字、种类）。这就是 RunInfo。

Graph 在调度节点时会提前注入 RunInfo。但如果你直接调用 chatModel.Generate()（不经过 Graph），就没有人提前注入。

EnsureRunInfo 兜底处理这种情况。

四、On[T] 核心函数

所有 OnStart/OnEnd/OnError 最终都走同一个泛型函数：

func On[T any](ctx context.Context, inOut T, handle Handle[T], timing CallbackTiming, start bool) (context.Context, T) {
    mgr, ok := managerFromCtx(ctx)
    if !ok {
        return ctx, inOut   // 没有注册任何 handler，直接返回
    }
    nMgr := *mgr

    // RunInfo 的生命周期管理
    var info *RunInfo
    if start {
        // OnStart：消费 manager 里的 RunInfo，存到 CtxRunInfoKey
        info = nMgr.runInfo
        nMgr.runInfo = nil   // ← 清空，防止被多次使用
        ctx = context.WithValue(ctx, CtxRunInfoKey{}, info)
    } else {
        // OnEnd/OnError：RunInfo 已经被 OnStart 存到 ctx 了
        if nMgr.runInfo != nil {
            info = nMgr.runInfo
        } else {
            info, _ = ctx.Value(CtxRunInfoKey{}).(*RunInfo)   // ← 从 ctx 读回
        }
    }

    // 收集有效 handler（支持 TimingChecker 过滤）
    hs := make([]Handler, 0, len(nMgr.handlers)+len(nMgr.globalHandlers))
    for _, handler := range append(nMgr.handlers, nMgr.globalHandlers...) {
        timingChecker, ok_ := handler.(TimingChecker)
        if !ok_ || timingChecker.Needed(ctx, info, timing) {
            hs = append(hs, handler)   // 没实现 TimingChecker，或者 Needed 返回 true
        }
    }

    // 分发给具体的 handle 函数（OnStartHandle、OnEndHandle...）
    var out T
    ctx, out = handle(ctx, inOut, info, hs)
    return ctxWithManager(ctx, &nMgr), out
}

这个函数是整个系统的核心入口。它负责从 context 中提取 manager，管理 RunInfo 的生命周期。

同时，它会收集所有有效的 handler，并通过 TimingChecker 过滤掉不需要的 timing 回调。

最后，它将控制权分发给具体的 handle 函数，如 OnStartHandle 或 OnEndHandle。

五、执行顺序：OnStart 逆序，OnEnd 正序

这是一个刻意的设计，对应中间件/洋葱模型：

// OnStart：逆序（最后注册的 handler 最先执行）
func OnStartHandle[T any](...) {
    for i := len(handlers) - 1; i >= 0; i-- {
        ctx = handlers[i].OnStart(ctx, runInfo, input)
    }
}

// OnEnd：正序（最先注册的 handler 最先执行）
func OnEndHandle[T any](...) {
    for _, handler := range handlers {
        ctx = handler.OnEnd(ctx, runInfo, output)
    }
}

假设注册了 [A, B, C]：

• OnStart 顺序：C → B → A（逆序，像中间件 wrap）
• OnEnd 顺序：A → B → C（正序，unwrap）

实践意义： 如果你有一个计时 handler 想测量“某个内层 handler 的执行耗时”，可以利用这个顺序。

外层 handler 在 OnStart 记开始时间，OnEnd 记结束时间，内层 handler 在中间执行业务逻辑。

六、TimingChecker：避免不必要的开销

如果一个 handler 只关心 OnEnd（比如用量统计），但 Stream 场景会触发 OnEndWithStreamOutput——这需要复制 StreamReader，很贵。

type TimingChecker interface {
    Needed(ctx context.Context, info *RunInfo, timing CallbackTiming) bool
}

实现 Needed 返回 false 的 timing，框架会跳过该 handler：

// 用 HandlerBuilder，自动实现 TimingChecker
handler := callbacks.NewHandlerBuilder().
    OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        // 只关心 OnEnd
        return ctx
    }).
    Build()

// handler.Needed(ctx, info, TimingOnEndWithStreamOutput) 自动返回 false
// → 不会为这个 handler 复制 Stream

小提示： 使用 NewHandlerBuilder 构建 handler 时，它会自动根据你注册的 OnStartFn、OnEndFn 等方法来生成 TimingChecker。

你无需手动实现 Needed 方法，框架会自动过滤掉不需要的 timing 回调。

七、Stream 的 Copy 机制

Stream handler（OnEndWithStreamOutput）拿到的是 StreamReader 的一份独立副本：

// 伪代码
cpy := output.Copy    // output.Copy 是 func(n int) []*StreamReader[T]
inOuts := cpy(len(handlers) + 1)   // 复制 len+1 份：len 份给 handlers，1 份给主流程
for i, handler := range handlers {
    ctx = handler.OnEndWithStreamOutput(ctx, runInfo, inOuts[i])   // 第 i 份
}
return ctx, inOuts[len(inOuts)-1]   // 最后一份还给主流程

必须 Close： 每个 handler 必须调 sr.Close() 关闭自己的那份。否则底层 goroutine 无法退出，造成泄漏。

func (t *MyHandler) OnEndWithStreamOutput(ctx context.Context, info *callbacks.RunInfo,
    output *schema.StreamReader[callbacks.CallbackOutput]) context.Context {
    defer output.Close()   // ← 必须
    for {
        chunk, err := output.Recv()
        if err != nil { break }
        // 处理 chunk...
    }
    return ctx
}

八、写 Handler 的正确姿势

// 1. 用 NewHandlerBuilder：自动实现 TimingChecker，只注册你需要的 timing
handler := callbacks.NewHandlerBuilder().
    OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        // 只处理 ChatModel 节点
        if info == nil || info.Component != components.ComponentOfChatModel {
            return ctx
        }
        return context.WithValue(ctx, startTimeKey{}, time.Now())   // 存开始时间
    }).
    OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        if info == nil || info.Component != components.ComponentOfChatModel {
            return ctx
        }
        start, _ := ctx.Value(startTimeKey{}).(time.Time)
        log.Printf("[%s] latency=%v", info.Name, time.Since(start))   // 读开始时间
        return ctx
    }).
    Build()

// 2. 不要 mutate input/output：多个节点并发执行时共享同一指针
// BAD:
func (h *BadHandler) OnEnd(ctx, info, output) context.Context {
    out := output.(*model.CallbackOutput)
    out.Message.Content = "modified"   // ← 数据竞争！
    return ctx
}

// 3. context 从 OnStart 流到 OnEnd：同一 handler 实例内部传状态用 ctx.WithValue

编写 handler 时，请遵循以下原则：

• 使用 NewHandlerBuilder：自动实现 TimingChecker，只注册你需要的 timing 回调。
• 不要修改 input/output：多个节点并发执行时共享同一指针，修改会导致数据竞争。正确的做法是只读，或者通过 context 传递额外状态。
• 利用 context 传状态：context 从 OnStart 流到 OnEnd，同一 handler 实例内部传状态建议用 ctx.WithValue。

九、常见问题与注意事项

Q1: 为什么必须调用 output.Close()？

Stream 的 Copy 机制会为每个 handler 分配独立的 StreamReader 副本。

如果 handler 不 Close，底层 goroutine 会一直等待读取，无法退出，造成 goroutine 泄漏。

务必使用 defer output.Close()。

Q2: 多个 handler 是否可以修改 input/output？

不可以。 每个节点在并发执行时，input/output 共享同一指针。

修改 input/output 会导致数据竞争。正确的做法是只读，或者通过 context 传递额外状态。

Q3: 如何让 handler 只对特定类型的组件生效？

在 handler 的 OnStart 或 OnEnd 方法中检查 info.Component 或 info.Type，不匹配时直接返回 ctx。

例如：if info.Component != components.ComponentOfChatModel { return ctx }。

Q4: 全局注册和每次调用注册的 handler 执行顺序是什么？

在 On[T] 函数中，handler 的顺序是：先执行每次调用注册的 handlers（nMgr.handlers），再执行全局 handlers（nMgr.globalHandlers）。

但注意：OnStart 是逆序执行，所以最后注册的 handler 最先执行；OnEnd 是正序执行。

Q5: 如果我没有实现 TimingChecker，所有 timing 都会被触发吗？

是的。如果 handler 没有实现 TimingChecker 接口，Needed 方法不存在，框架会默认认为需要所有 timing。

推荐使用 NewHandlerBuilder 自动生成 TimingChecker，只注册你需要的 timing 回调，避免性能浪费。

十、小结

Eino callback 系统的四个核心设计：

1. Context 作为 carrier：manager 塞进 ctx 的 key-value，随调用链流动，无全局状态（GlobalHandlers 除外）
2. EnsureRunInfo 兜底：组件方法开头调用，确保不经 Graph 直接调用时也能触发 callback
3. On[T] 统一分发：TimingChecker 过滤不需要的 timing → 分发给具体 handle 函数 → OnStart 逆序 / OnEnd 正序
4. Stream 独立副本：每个 handler 拿独立 StreamReader，必须 Close，否则 goroutine 泄漏

代码来源：eino/callbacks · eino/internal/callbacks