XEngineers/CamTalk

Fork 0

Files

hhs 16302af7d2 docs: 添加 Eino 框架参考文档

2026-06-19 14:35:06 +08:00

11 KiB

Raw Permalink Blame History

tags, create time

Eino 快速入门 · 第六章：Callback 与 Trace（可观测性）

概述

在构建 Agent 应用时，我们常常面临一个核心问题：Agent 内部到底发生了什么？ 本章将介绍 Eino 的 Callback 机制——一套非侵入式的旁路钩子系统，让你能在不改动业务代码的前提下，获取组件生命周期的每一个关键信息。通过 Callback 配合 CozeLoop，你将获得完整的链路追踪、性能指标和错误定位能力。

代码位置

入口代码：cmd/ch06/main.go

前置条件

与第一章一致：需要配置一个可用的 ChatModel（OpenAI 或 Ark）。同时，需要与第四章一样设置 PROJECT_ROOT：

export PROJECT_ROOT=/path/to/eino  # Eino 核心库根目录（不设置则默认使用当前目录）

可选：配置 CozeLoop 实现链路追踪：

export COZELOOP_WORKSPACE_ID=your_workspace_id
export COZELOOP_API_TOKEN=your_token

运行

在 examples/quickstart/chatwitheino 目录下执行：

# 设置项目根目录
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 中定义回调处理器的核心接口：

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 携带了组件运行时身份，是日志和追踪中最重要的标识信息。

type RunInfo struct {
    Name      string // 业务名称（节点名或用户指定）
    Type      string // 实现类型（如 "OpenAI"）
    Component string // 组件类型（如 "ChatModel"）
}

[!IMPORTANT] 流式回调注意事项

流式回调必须关闭 StreamReader，否则会导致 goroutine 泄漏

不要修改 Input/Output，它们被所有下游共享

RunInfo 可能为 nil，使用前需要检查

CozeLoop

CozeLoop 是字节跳动开源的 AI 应用可观测性平台，提供了：

链路追踪：完整的调用链路可视化
指标监控：延迟、Token 消耗、错误率等
日志聚合：集中管理所有日志
调试支持：在线查看和调试

集成方式：

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 方法	触发点	输入/输出
TimingOnStart	OnStart	组件开始处理前	CallbackInput
TimingOnEnd	OnEnd	组件成功返回后	CallbackOutput
TimingOnError	OnError	组件返回错误时	error
TimingOnStartWithStreamInput	OnStartWithStreamInput	组件接收流式输入时	StreamReader[CallbackInput]
TimingOnEndWithStreamOutput	OnEndWithStreamOutput	组件返回流式输出时	StreamReader[CallbackOutput]

非流式调用时序：

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

流式调用时序：

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 接口来跳过不必要的调用开销：

func (h *MyHandler) TimingChecker(timing callbacks.Timing) bool {
    // 只启用 Error 检测，其余跳过
    return timing == callbacks.TimingOnError
}

Callback 实战

实现自定义 Callback Handler

直接实现全部 5 个方法比较繁琐。Eino 提供了 callbacks.HandlerHelper 链式构建器，只需注册感兴趣的回调：

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。核心流程如下：

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 收集的数据，我们可以实现三个层面的可观测性：

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 可实现性能最优的按需检测

11 KiB Raw Permalink Blame History Unescape Escape