Callback 与可观测性

Eino

Callback 与可观测性

Eino 的 Callback 机制实现横切面关注点（cross-cutting concerns） 的非侵入式注入：日志、链路追踪、指标采集等旁路功能，不影响业务主流程。

核心理念

业务主路（Agent / Graph / Component）
    ↓ OnStart      — 切入点
    ↓ OnEnd        — 切入点
    ↓ OnError      — 切入点
旁路钩子（Callback Handler）— 抽取信息但不改变流程

Handler 接口 — 5 个触发时机

type Handler interface {
    OnStart(ctx context.Context, info *RunInfo, input CallbackInput) context.Context
    OnEnd(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context
    OnError(ctx context.Context, info *RunInfo, err error) context.Context
    OnStartWithStreamInput(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context
    OnEndWithStreamOutput(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context
}

RunInfo — 触发实体元信息

type RunInfo struct {
    Name      string     // 节点名称
    Type      string     // 实现类型，如 "OpenAI"
    Component Component  // 组件类型，如 "ChatModel"
}

三种注入方式

方式	API	适用场景
全局注入	`callbacks.AppendGlobalHandlers(h)`	全局 tracing / logging，服务初始化时注入
Graph 运行时注入	`compose.WithCallbacks(h).DesignateNode("name")`	特定图或特定节点的 callback
手动注入	`callbacks.InitCallbacks(ctx, info, handlers...)`	单独使用 Component 时

HandlerHelper — 快速构建

handler := callbacks.NewHandlerHelper().
    OnStart(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        log.Printf("[%s/%s] start", info.Component, info.Name)
        return ctx
    }).
    OnEnd(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        log.Printf("[%s/%s] end", info.Component, info.Name)
        return ctx
    }).
    OnError(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
        log.Printf("[%s/%s] error: %v", info.Component, info.Name, err)
        return ctx
    }).
    Handler()

// 全局注册
callbacks.AppendGlobalHandlers(handler)

// 或按 Graph 注册
runnable.Invoke(ctx, input, compose.WithCallbacks(handler))

// 指定节点
runnable.Invoke(ctx, input, compose.WithCallbacks(handler).DesignateNode("llm"))

OpenTelemetry Tracing 集成

tracer := otel.Tracer("eino-service")

handler := callbacks.NewHandlerHelper().
    OnStart(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        ctx, span := tracer.Start(ctx, info.Name,
            trace.WithAttributes(
                attribute.String("component", string(info.Component)),
                attribute.String("type", info.Type),
            ),
        )
        return ctx
    }).
    OnEnd(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        span := trace.SpanFromContext(ctx)
        span.End()
        return ctx
    }).
    OnError(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
        span := trace.SpanFromContext(ctx)
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        span.End()
        return ctx
    }).
    Handler()

执行顺序

在 Eino 内部：

阶段	顺序
`OnStart`	逆序触发（last-to-first）
`OnEnd` / `OnError`	正序触发（first-to-last）

Graph 编排层面，框架自动为 ChatModel、Tool 等节点注入 Callback 切面。

Agent Callback（ADK 层）

ADK 也支持在 Agent 级别消费事件流：

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

iter := runner.Query(ctx, "hello",
    adk.WithCallbacks(handler), // 注入 callback
)

也可以在 Runner 初始化时全局配置，所有 Run 请求共享。

CozeLoop — 字节跳动内部可观测性平台

import "github.com/cloudwego/eino-ext/callbacks/cozeloop"

handler := cozeloop.NewHandler(/* config */)
callbacks.AppendGlobalHandlers(handler)

提供可视化链路追踪、Agent 调用拓扑图、性能分析等功能。

常见陷阱

陷阱	说明
🚨 StreamReader 未 Close	流式 Callback 中收到 StreamReader 必须 Close
🚨 OnStart 修改 input	Callback 的 input/output 是值拷贝，修改不影响实际流程
🚨 `AppendGlobalHandlers` 并发调用	非并发安全，仅应在初始化阶段调用一次
🚨 Context 传递	OnStart 返回的 ctx 会传入 OnEnd / OnError，可用于传递 span 等

中断与恢复（Interrupt / Resume）最佳实践与陷阱汇总