--- tags: [Eino, Callback, Trace, 可观测性, CozeLoop] create time: 2026-04-29 15:30 --- # Eino 快速入门 · 第六章：Callback 与 Trace（可观测性） ## 概述 在构建 Agent 应用时，我们常常面临一个核心问题：**Agent 内部到底发生了什么？** 本章将介绍 Eino 的 Callback 机制——一套非侵入式的旁路钩子系统，让你能在不改动业务代码的前提下，获取组件生命周期的每一个关键信息。通过 Callback 配合 CozeLoop，你将获得完整的链路追踪、性能指标和错误定位能力。 ## 代码位置 - 入口代码：[cmd/ch06/main.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/cmd/ch06/main.go) ## 前置条件 与第一章一致：需要配置一个可用的 ChatModel（OpenAI 或 Ark）。同时，需要与第四章一样设置 `PROJECT_ROOT`： ```bash export PROJECT_ROOT=/path/to/eino # Eino 核心库根目录（不设置则默认使用当前目录） ``` 可选：配置 CozeLoop 实现链路追踪： ```bash export COZELOOP_WORKSPACE_ID=your_workspace_id export COZELOOP_API_TOKEN=your_token ``` ## 运行 在 `examples/quickstart/chatwitheino` 目录下执行： ```bash # 设置项目根目录 export PROJECT_ROOT=/path/to/your/project # 可选：配置 CozeLoop export COZELOOP_WORKSPACE_ID=your_workspace_id export COZELOOP_API_TOKEN=your_token go run ./cmd/ch06 ``` 输出示例： ``` [trace] starting session: 083d16da-6b13-4fe6-afb0-c45d8f490ce1 you> 你好 [trace] chat_model_generate: model=gpt-4.1-mini tokens=150 [trace] tool_call: name=list_files duration=23ms [assistant] 你好！有什么我可以帮助你的吗？ ``` ## 从黑盒到白盒：为什么需要 Callback 前几章我们实现的 Agent 是一个"黑盒"：输入问题，输出答案，但中间发生了什么我们并不清楚。 **黑盒的问题：** - 不知道模型调用了多少次 - 不知道 Tool 执行了多长时间 - 不知道 Token 消耗了多少 - 出问题时难以定位原因 > [!NOTE] Callback 定位 > Callback 是 Eino 的**旁路机制**——从 component 到 compose，一以贯之。它在固定点位触发，可抽取实时信息（输入、输出、错误、流式数据），用途覆盖观测、日志、指标、追踪、调试、审计等场景。 **类比理解：** - **Agent** = "业务逻辑"（主路） - **Callback** = "旁路钩子"（在固定点位抽取信息） ## 关键概念 ### Handler 接口 `Handler` 是 Eino 中定义回调处理器的核心接口： ```go 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 } ``` **设计理念：** - **旁路机制**：不干扰主流程，在固定点位抽取信息 - **全流程覆盖**：从 component 到 compose 到 adk，所有组件都支持 - **状态传递**：同一 Handler 的 OnStart→OnEnd 可通过 context 传递状态 - **性能优化**：实现 `TimingChecker` 接口可跳过不需要的时机 > [!TIP] RunInfo 结构 > `RunInfo` 携带了组件运行时身份，是日志和追踪中最重要的标识信息。 ```go type RunInfo struct { Name string // 业务名称（节点名或用户指定） Type string // 实现类型（如 "OpenAI"） Component string // 组件类型（如 "ChatModel"） } ``` > [!IMPORTANT] 流式回调注意事项 > - 流式回调必须关闭 StreamReader，否则会导致 goroutine 泄漏 > - 不要修改 Input/Output，它们被所有下游共享 > - RunInfo 可能为 nil，使用前需要检查 ### CozeLoop CozeLoop 是字节跳动开源的 AI 应用可观测性平台，提供了： - **链路追踪**：完整的调用链路可视化 - **指标监控**：延迟、Token 消耗、错误率等 - **日志聚合**：集中管理所有日志 - **调试支持**：在线查看和调试 **集成方式：** ```go import ( clc "github.com/cloudwego/eino-ext/callbacks/cozeloop" "github.com/cloudwego/eino/callbacks" "github.com/coze-dev/cozeloop-go" ) // 创建 CozeLoop 客户端 client, err := cozeloop.NewClient( cozeloop.WithAPIToken(apiToken), cozeloop.WithWorkspaceID(workspaceID), ) // 注册为全局 Callback callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client)) ``` ### Callback 的触发时机 Callback 在组件生命周期的 5 个关键时机触发。下表中 `Timing*` 是 Eino 内部常量名（用于 `TimingChecker` 接口），对应的 Handler 接口方法是右侧所示：
时机常量对应 Handler 方法触发点输入/输出
TimingOnStartOnStart组件开始处理前CallbackInput
TimingOnEndOnEnd组件成功返回后CallbackOutput
TimingOnErrorOnError组件返回错误时error
TimingOnStartWithStreamInputOnStartWithStreamInput组件接收流式输入时StreamReader[CallbackInput]
TimingOnEndWithStreamOutputOnEndWithStreamOutput组件返回流式输出时StreamReader[CallbackOutput]
**非流式调用时序：** ```mermaid sequenceDiagram participant Client as 业务代码 participant CM as ChatModel participant CB as Callback Handler Client->>CM: Generate(ctx, messages) CM->>CB: OnStart(messages) Note over CB: "记录输入，启动计时" CM->>CM: 模型处理 CM->>CB: OnEnd(response) Note over CB: "记录输出，计算耗时" CM-->>Client: response ``` **流式调用时序：** ```mermaid sequenceDiagram participant Client as 业务代码 participant CM as ChatModel participant CB as Callback Handler Client->>CM: Stream(ctx, messages) CM->>CB: OnStart(messages) Note over CB: "记录输入，启动计时" CM->>CM: 模型处理（流式） CM->>CB: OnEndWithStreamOutput(reader) Note over CB: "返回 StreamReader，逐 chunk 消费" loop 逐块消费 CB->>CB: reader.Read() CB->>CB: 处理 chunk end CM-->>Client: stream chunks ``` > [!WARNING] 流式错误处理 > 流式错误（stream 中途出错）**不会触发 OnError**，而是在 StreamReader 中返回。消费时务必检查 `reader.Err()`。 ### TimingChecker 优化 如果你的 Handler 不需要某些时机（比如只关心错误），可以实现 `TimingChecker` 接口来跳过不必要的调用开销： ```go func (h *MyHandler) TimingChecker(timing callbacks.Timing) bool { // 只启用 Error 检测，其余跳过 return timing == callbacks.TimingOnError } ``` ## Callback 实战 ### 实现自定义 Callback Handler 直接实现全部 5 个方法比较繁琐。Eino 提供了 `callbacks.HandlerHelper` 链式构建器，只需注册感兴趣的回调： ```go import "github.com/cloudwego/eino/callbacks" // 使用 NewHandlerHelper 注册感兴趣的回调 handler := callbacks.NewHandlerHelper(). OnStart(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context { log.Printf("[trace] %s/%s start", info.Component, info.Name) return ctx }). OnEnd(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context { log.Printf("[trace] %s/%s end", info.Component, info.Name) return ctx }). OnError(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context { log.Printf("[trace] %s/%s error: %v", info.Component, info.Name, err) return ctx }). Handler() // 注册为全局 Callback callbacks.AppendGlobalHandlers(handler) ``` **注意**：`RunInfo` 可能为 `nil`（如顶层调用），使用前务必检查。 ### 集成与注册 完整代码见 [cmd/ch06/main.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/cmd/ch06/main.go)。核心流程如下： ```go func main() { ctx := context.Background() // 1. 可选：注册自定义日志 Callback handler := callbacks.NewHandlerHelper(). OnStart(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context { log.Printf("[trace] %s/%s start", info.Component, info.Name) return ctx }). OnEnd(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context { log.Printf("[trace] %s/%s end", info.Component, info.Name) return ctx }). Handler() callbacks.AppendGlobalHandlers(handler) // 2. 可选：启用 CozeLoop 链路追踪 apiToken := os.Getenv("COZELOOP_API_TOKEN") workspaceID := os.Getenv("COZELOOP_WORKSPACE_ID") if apiToken != "" && workspaceID != "" { client, _ := cozeloop.NewClient( cozeloop.WithAPIToken(apiToken), cozeloop.WithWorkspaceID(workspaceID), ) defer func() { time.Sleep(5 * time.Second) // 等待数据上报 client.Close(ctx) }() callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client)) } // 3. 正常创建并运行 Agent... } ``` ## 可观测性的三大价值 通过 Callback 收集的数据，我们可以实现三个层面的可观测性： ```mermaid quadrantChart title Observability Dimensions x-axis Low Impact --> High Impact y-axis Low Cost --> High Value "错误追踪": [0.8, 0.9] "成本优化": [0.6, 0.7] "性能分析": [0.4, 0.5] "审计合规": [0.9, 0.8] ``` | 维度 | 关键指标 | 典型场景 | |------|----------|----------| | **错误追踪** | 错误类型、堆栈、出错节点 | Agent 响应异常时快速定位是模型侧还是 Tool 侧的问题 | | **成本优化** | Token 消耗、每轮对话花费 | 识别高消耗对话，优化 Prompt 或切换更经济的模型 | | **性能分析** | 延迟分布、耗时 Top N | 发现慢查询——某个 Tool 执行时间过长影响整体体验 | ## 本章小结 > [!SUMMARY] 要点回顾 > - **Callback** 是 Eino 的非侵入式观测钩子，在组件生命周期的 5 个时机触发 > - 使用 `callbacks.HandlerHelper` 可链式构建 Handler，只注册感兴趣的回调 > - 通过 `callbacks.AppendGlobalHandlers` 注册全局 Callback，业务代码零修改 > - **CozeLoop** 提供开箱即用的链路追踪和可视化 > - 结合 `TimingChecker` 可实现性能最优的按需检测 ## 关联笔记 - [[Eino/quick_start/chapter_01_hello_eino.md]] - [[Eino/quick_start/chapter_04_tool.md]] - [[Eino/quick_start/chapter_05_agent.md]]