wonder/all-in-kingsoft
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

vault backup: 2026-04-29 18:54:02

Browse Source
This commit is contained in:
wonder
2026-04-29 18:54:02 +08:00
parent 7bb4b3b498
commit 84f40fc9d9
6 changed files with 1081 additions and 916 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Eino/quick_start/chapter_06_callback_and_trace.md
+127 -171
View File
@@ -1,13 +1,13 @@
---
Description: ""
date: "2026-03-12"
lastmod: ""
tags: []
title: 第六章：Callback 与 Trace（可观测性）
weight: 6
tags: [Eino, Callback, Trace, 可观测性, CozeLoop]
create time: 2026-04-29 15:30
---
本章目标：理解 Callback 机制，集成 CozeLoop 实现链路追踪和可观测性。
# Eino 快速入门 · 第六章：Callback 与 Trace（可观测性）
## 概述
在构建 Agent 应用时，我们常常面临一个核心问题：**Agent 内部到底发生了什么？** 本章将介绍 Eino 的 Callback 机制——一套非侵入式的旁路钩子系统，让你能在不改动业务代码的前提下，获取组件生命周期的每一个关键信息。通过 Callback 配合 CozeLoop，你将获得完整的链路追踪、性能指标和错误定位能力。
## 代码位置
@@ -64,14 +64,10 @@ you> 你好
- 不知道 Token 消耗了多少
- 出问题时难以定位原因
**Callback 的定位：**
> [!NOTE] Callback 定位
> Callback 是 Eino 的**旁路机制**——从 component 到 compose，一以贯之。它在固定点位触发，可抽取实时信息（输入、输出、错误、流式数据），用途覆盖观测、日志、指标、追踪、调试、审计等场景。
- **Callback 是 Eino 的旁路机制**：从 component 到 compose（下文详谈）到 adk，一以贯之
- **Callback 在固定点位触发**：组件生命周期的 5 个关键时机
- **Callback 可抽取实时信息**：输入、输出、错误、流式数据等
- **Callback 用途广泛**：观测、日志、指标、追踪、调试、审计等
**简单类比：**
**类比理解：**
- **Agent** = "业务逻辑"（主路）
- **Callback** = "旁路钩子"（在固定点位抽取信息）
@@ -110,21 +106,21 @@ type Handler interface {
- **状态传递**：同一 Handler 的 OnStart→OnEnd 可通过 context 传递状态
- **性能优化**：实现 `TimingChecker` 接口可跳过不需要的时机
**RunInfo 结构：**
> [!TIP] RunInfo 结构
> `RunInfo` 携带了组件运行时身份，是日志和追踪中最重要的标识信息。
```go
type RunInfo struct {
Name string // 业务名称（节点名或用户指定）
Type string // 实现类型（如 "OpenAI"）
Component string // 组件类型（如 "ChatModel"）
Name string // 业务名称（节点名或用户指定）
Type string // 实现类型（如 "OpenAI"）
Component string // 组件类型（如 "ChatModel"）
}
```
**重要提示：**
- 流式回调必须关闭 StreamReader，否则会导致 goroutine 泄漏
- 不要修改 Input/Output，它们被所有下游共享
- RunInfo 可能为 nil，使用前需要检查
> [!IMPORTANT] 流式回调注意事项
> - 流式回调必须关闭 StreamReader，否则会导致 goroutine 泄漏
> - 不要修改 Input/Output，它们被所有下游共享
> - RunInfo 可能为 nil，使用前需要检查
### CozeLoop
@@ -160,68 +156,70 @@ Callback 在组件生命周期的 5 个关键时机触发。下表中 `Timing*`
<table>
<tr><td>时机常量</td><td>对应 Handler 方法</td><td>触发点</td><td>输入/输出</td></tr>
<tr><td><pre>TimingOnStart</pre></td><td><pre>OnStart</pre></td><td>组件开始处理前</td><td>CallbackInput</td></tr>
<tr><td><pre>TimingOnEnd</pre></td><td><pre>OnEnd</pre></td><td>组件成功返回后</td><td>CallbackOutput</td></tr>
<tr><td><pre>TimingOnError</pre></td><td><pre>OnError</pre></td><td>组件返回错误时</td><td>error</td></tr>
<tr><td><pre>TimingOnStartWithStreamInput</pre></td><td><pre>OnStartWithStreamInput</pre></td><td>组件接收流式输入时</td><td>StreamReader[CallbackInput]</td></tr>
<tr><td><pre>TimingOnEndWithStreamOutput</pre></td><td><pre>OnEndWithStreamOutput</pre></td><td>组件返回流式输出时</td><td>StreamReader[CallbackOutput]</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>
**示例：ChatModel 调用流程**
**非流式调用时序：**
```
┌─────────────────────────────────────────┐
│ ChatModel.Generate(ctx, messages) │
└─────────────────────────────────────────┘
↓
┌──────────────────────┐
│ OnStart │ ← 输入: CallbackInput (messages)
└──────────────────────┘
↓
┌──────────────────────┐
│ 模型处理 │
└──────────────────────┘
↓
┌──────────────────────┐
│ OnEnd │ ← 输出: CallbackOutput (response)
└──────────────────────┘
```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
```
**示例：流式输出流程**
**流式调用时序：**
```
┌─────────────────────────────────────────┐
│ ChatModel.Stream(ctx, messages) │
└─────────────────────────────────────────┘
↓
┌──────────────────────┐
│ OnStart │ ← 输入: CallbackInput (messages)
└──────────────────────┘
↓
┌──────────────────────┐
│ 模型处理（流式） │
└──────────────────────┘
↓
┌──────────────────────┐
│ OnEndWithStreamOutput │ ← 输出: StreamReader[CallbackOutput]
└──────────────────────┘
↓
┌──────────────────────┐
│ 逐个 chunk 返回 │
└──────────────────────┘
```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()`。
- 流式错误（stream 中途出错）不会触发 OnError，而是在 StreamReader 中返回
- 同一 Handler 的 OnStart→OnEnd 可通过 context 传递状态
- 不同 Handler 之间没有执行顺序保证
### TimingChecker 优化
## Callback 的实现
如果你的 Handler 不需要某些时机（比如只关心错误），可以实现 `TimingChecker` 接口来跳过不必要的调用开销：
### 1. 实现自定义 Callback Handler
```go
func (h *MyHandler) TimingChecker(timing callbacks.Timing) bool {
// 只启用 Error 检测，其余跳过
return timing == callbacks.TimingOnError
}
```
完整实现 `Handler` 接口需要实现所有 5 个方法，较为繁琐。Eino 提供了 `callbacks.HandlerHelper` 帮助类来简化实现：
## Callback 实战
### 实现自定义 Callback Handler
直接实现全部 5 个方法比较繁琐。Eino 提供了 `callbacks.HandlerHelper` 链式构建器，只需注册感兴趣的回调：
```go
import "github.com/cloudwego/eino/callbacks"
@@ -246,122 +244,80 @@ handler := callbacks.NewHandlerHelper().
callbacks.AppendGlobalHandlers(handler)
```
**注意**：`RunInfo` 可能为 `nil`（如顶层调用没有 RunInfo），使用前请检查。
**注意**：`RunInfo` 可能为 `nil`（如顶层调用），使用前务必检查。
### 2. 集成 CozeLoop
### 集成与注册
```go
func setupCozeLoop(ctx context.Context) (*cozeloop.Client, error) {
apiToken := os.Getenv("COZELOOP_API_TOKEN")
workspaceID := os.Getenv("COZELOOP_WORKSPACE_ID")
if apiToken == "" || workspaceID == "" {
return nil, nil // 未配置则跳过
}
client, err := cozeloop.NewClient(
cozeloop.WithAPIToken(apiToken),
cozeloop.WithWorkspaceID(workspaceID),
)
if err != nil {
return nil, err
}
// 注册为全局 Callback
callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
return client, nil
}
```
### 3. 在 main 中使用
完整代码见 [cmd/ch06/main.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/cmd/ch06/main.go)。核心流程如下：
```go
func main() {
ctx := context.Background()
// 设置 CozeLoop（可选）
client, err := setupCozeLoop(ctx)
if err != nil {
log.Printf("cozeloop setup failed: %v", err)
}
if client != nil {
// 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) // 等待数据上报
time.Sleep(5 * time.Second) // 等待数据上报
client.Close(ctx)
}()
callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
}
// 创建 Agent 并运行...
// 3. 正常创建并运行 Agent...
}
```
**关键代码片段（**注意：这是简化后的代码片段，不能直接运行，完整代码请参考** [cmd/ch06/main.go](https://github.com/cloudwego/eino-examples/blob/main/quickstart/chatwitheino/cmd/ch06/main.go)）：
## 可观测性的三大价值
```go
// 设置 CozeLoop 追踪
cozeloopApiToken := os.Getenv("COZELOOP_API_TOKEN")
cozeloopWorkspaceID := os.Getenv("COZELOOP_WORKSPACE_ID")
if cozeloopApiToken != "" && cozeloopWorkspaceID != "" {
client, err := cozeloop.NewClient(
cozeloop.WithAPIToken(cozeloopApiToken),
cozeloop.WithWorkspaceID(cozeloopWorkspaceID),
)
if err != nil {
log.Fatalf("cozeloop.NewClient failed: %v", err)
}
defer func() {
time.Sleep(5 * time.Second)
client.Close(ctx)
}()
callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
}
通过 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]
```
## 可观测性的价值
### 1. 性能分析
通过 Callback 收集的数据，可以分析：
- 模型调用延迟分布
- Tool 执行时间排行
- Token 消耗趋势
### 2. 错误追踪
当 Agent 出现问题时：
- 查看完整的调用链路
- 定位是哪个环节出错
- 分析错误原因
### 3. 成本优化
通过 Token 消耗数据：
- 识别高消耗的对话
- 优化 Prompt 减少 Token
- 选择更经济的模型
| 维度 | 关键指标 | 典型场景 |
|------|----------|----------|
| **错误追踪** | 错误类型、堆栈、出错节点 | Agent 响应异常时快速定位是模型侧还是 Tool 侧的问题 |
| **成本优化** | Token 消耗、每轮对话花费 | 识别高消耗对话，优化 Prompt 或切换更经济的模型 |
| **性能分析** | 延迟分布、耗时 Top N | 发现慢查询——某个 Tool 执行时间过长影响整体体验 |
## 本章小结
- **Callback**：Eino 的观测钩子，在关键节点触发回调
- **CozeLoop**：字节跳动的 AI 应用可观测性平台
- **全局注册**：通过 `callbacks.AppendGlobalHandlers` 注册全局 Callback
- **非侵入式**：业务代码不需要修改，Callback 自动触发
- **可观测性价值**：性能分析、错误追踪、成本优化
> [!SUMMARY] 要点回顾
> - **Callback** 是 Eino 的非侵入式观测钩子，在组件生命周期的 5 个时机触发
> - 使用 `callbacks.HandlerHelper` 可链式构建 Handler，只注册感兴趣的回调
> - 通过 `callbacks.AppendGlobalHandlers` 注册全局 Callback，业务代码零修改
> - **CozeLoop** 提供开箱即用的链路追踪和可视化
> - 结合 `TimingChecker` 可实现性能最优的按需检测
## 扩展思考
## 关联笔记
**其他 Callback 实现：**
- OpenTelemetry Callback：对接标准可观测性协议
- 自定义日志 Callback：记录到本地文件
- 指标 Callback：对接 Prometheus 等监控系统
**高级用法：**
- 在 Callback 中实现采样（只记录部分请求）
- 在 Callback 中实现限流（根据 Token 消耗）
- 在 Callback 中实现告警（错误率过高时通知）
- [[Eino/quick_start/chapter_01_hello_eino.md]]
- [[Eino/quick_start/chapter_04_tool.md]]
- [[Eino/quick_start/chapter_05_agent.md]]