Callback 系统：Eino 的事件总线（第52篇-E38）

读完这篇你会知道

全景

Eino 的 callback 系统本质上是把 handler 列表藏在 context 里，然后在 context 流经每个节点时触发。没有全局事件总线、没有注册表、没有反射——只有 context。

注册时：callbacks.CtxWithHandlers(ctx, handlers)└── 把 manager{handlers, globalHandlers, runInfo} 塞进 ctx执行时（组件内部）：EnsureRunInfo(ctx, ...) ← 确保 ctx 有 RunInfocallbacks.OnStart(ctx, in)← 从 ctx 读 manager，逆序触发所有 handler.OnStart... 组件业务逻辑 ...callbacks.OnEnd(ctx, out) ← 正序触发 handler.OnEnd

Context 作为 carrier：manager struct

// internal/callbacks/manager.gotype 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// 全局 slicefunc 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 → 补充 RunInforeturn 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 *RunInfoif start {// OnStart：消费 manager 里的 RunInfo，存到 CtxRunInfoKeyinfo = nMgr.runInfonMgr.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 Tctx, out = handle(ctx, inOut, info, hs)return ctxWithManager(ctx, &nMgr), out}

执行顺序：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，自动实现 TimingCheckerhandler := callbacks.NewHandlerBuilder().OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {// 只关心 OnEndreturn ctx}).Build()// handler.Needed(ctx, info, TimingOnEndWithStreamOutput) 自动返回 false// → 不会为这个 handler 复制 Stream

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，只注册你需要的 timinghandler := 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

小结

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