CamTalk/docs/Eino/quick_start/chapter_06_callback_and_trace.md

---
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 接口方法是右侧所示：

<table>
<tr><td>时机常量</td><td>对应 Handler 方法</td><td>触发点</td><td>输入/输出</td></tr>
<tr><td>TimingOnStart</td><td>OnStart</td><td>组件开始处理前</td><td>CallbackInput</td></tr>
<tr><td>TimingOnEnd</td><td>OnEnd</td><td>组件成功返回后</td><td>CallbackOutput</td></tr>
<tr><td>TimingOnError</td><td>OnError</td><td>组件返回错误时</td><td>error</td></tr>
<tr><td>TimingOnStartWithStreamInput</td><td>OnStartWithStreamInput</td><td>组件接收流式输入时</td><td>StreamReader[CallbackInput]</td></tr>
<tr><td>TimingOnEndWithStreamOutput</td><td>OnEndWithStreamOutput</td><td>组件返回流式输出时</td><td>StreamReader[CallbackOutput]</td></tr>
</table>

**非流式调用时序：**

```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]]