28｜Callbacks 与可观测性：定位慢调用和失败节点 - eino教程

28｜Callbacks 与可观测性：定位慢调用和失败节点

《Eino 实战：用 Go 构建 AI 应用与智能体》· 第 28 篇 关键词：Callbacks、OnStart/OnEnd/OnError、节点耗时、脱敏、追踪 前置条件：已完成 27｜HTTP 与 SSE 与 10｜状态与运行选项。

应用上线后，用户说“回答很慢”，或者“有时候会失败”。这时你需要回答两个问题：

没有观测数据，只能猜。Callbacks 就是 Eino 提供的观测入口。

一、回调的三种时机

callbacks.Handler 关心的主要时机：

还有两个流式专用时机：OnStartWithStreamInput、OnEndWithStreamOutput，用于处理流式输入输出的场景。

关键特性：回调是横切的。 挂到整个图上，就能观测到所有节点，不需要修改任何节点内部代码——这正是第 10 篇讲“统一治理能力”的含义。

二、写一个统计回调

用 NewHandlerBuilder 构造，链式注册各时机的处理函数：

handler := callbacks.NewHandlerBuilder().
OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackInput) context.Context {
m.start[key(info)] = time.Now()
return ctx
}).
OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackOutput) context.Context {
m.record[key(info)] = append(m.record[key(info)], time.Since(m.start[key(info)]))
return ctx
}).
OnErrorFn(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
m.failed[key(info)]++
return ctx
}).
Build()

两个必须遵守的规则

1. 必须返回传入的 ctx。

第 10 篇实测过：返回 context.Background() 会导致框架注入的状态丢失，后续节点读状态直接报错。要从 ctx 派生新 context，不要新建。

2. 节点名需要显式设置。

info.Name 来自 WithNodeName。不设置时，回调里只能拿到组件类型名（如 ChatModel、Lambda），多个同类节点会混在一起，无法区分：

_ = g.AddLambdaNode("clean", tpl, compose.WithNodeName("clean"))
_ = g.AddChatModelNode("summarize", model, compose.WithNodeName("summarize"))

RunInfo 的三个字段（第 10 篇也提到过）：

三、完整示例

下面这段程序统计每个节点的耗时与失败次数，并做了一次“正常 vs 超时”的对照实验（已实机运行）：

package main

import (
"context"
"fmt"
"log"
"strings"
"sync"
"time"

"github.com/cloudwego/eino/callbacks"
"github.com/cloudwego/eino/components/model"
"github.com/cloudwego/eino/compose"
"github.com/cloudwego/eino/schema"
)

// slowModel 模拟一个耗时调用，便于观察耗时统计。
type slowModel struct{ delay time.Duration }

func (m *slowModel) Generate(ctx context.Context, in []*schema.Message, _ …model.Option) (*schema.Message, error) {
select {
case <-time.After(m.delay):
case <-ctx.Done():
return nil, ctx.Err()
}
return schema.AssistantMessage("摘要："+in[len(in)–1].Content, nil), nil
}

func (m *slowModel) Stream(ctx context.Context, in []*schema.Message, opts …model.Option) (*schema.StreamReader[*schema.Message], error) {
msg, err := m.Generate(ctx, in, opts…)
if err != nil {
return nil, err
}
sr, sw := schema.Pipe[*schema.Message](1)
go func() { defer sw.Close(); sw.Send(msg, nil) }()
return sr, nil
}

// metrics 收集每个节点的耗时与成败，是“可观测性”的最小实现。
type metrics struct {
mu sync.Mutex
start map[string]time.Time
record map[string][]time.Duration
failed map[string]int
}

func newMetrics() *metrics {
return &metrics{start: map[string]time.Time{}, record: map[string][]time.Duration{}, failed: map[string]int{}}
}

func (m *metrics) key(info *callbacks.RunInfo) string {
name := info.Name
if name == "" {
name = string(info.Component)
}
return name
}

func (m *metrics) report() {
m.mu.Lock()
defer m.mu.Unlock()
fmt.Println("— 节点耗时统计 —")
for name, ds := range m.record {
var total time.Duration
for _, d := range ds {
total += d
}
fmt.Printf(" %-14s 调用=%d 总耗时=%v 失败=%d\\n", name, len(ds), total.Round(time.Millisecond), m.failed[name])
}
}

func main() {
if err := run(); err != nil {
log.Fatal(err)
}
}

func run() error {
ctx := context.Background()
m := newMetrics()

// 1. 回调：记录每个节点的开始与结束。
handler := callbacks.NewHandlerBuilder().
OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackInput) context.Context {
m.mu.Lock()
m.start[m.key(info)] = time.Now()
m.mu.Unlock()
return ctx
}).
OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackOutput) context.Context {
key := m.key(info)
m.mu.Lock()
if t, ok := m.start[key]; ok {
m.record[key] = append(m.record[key], time.Since(t))
}
m.mu.Unlock()
return ctx
}).
OnErrorFn(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
key := m.key(info)
m.mu.Lock()
m.failed[key]++
m.mu.Unlock()
// 只记录错误类型，不打印完整错误原文，避免泄露内部信息。
fmt.Printf(" [错误] 节点=%s 类型=%T\\n", key, err)
return ctx
}).
Build()

// 清洗节点：把工单描述整理成模型需要的消息列表。
tpl := compose.InvokableLambda(func(_ context.Context, in map[string]any) ([]*schema.Message, error) {
s, _ := in["ticket"].(string)
return []*schema.Message{schema.UserMessage(strings.TrimSpace(s))}, nil
})

g := compose.NewGraph[map[string]any, *schema.Message]()
// WithNodeName 让回调里的 info.Name 可读；不设置时只能拿到组件类型名。
_ = g.AddLambdaNode("clean", tpl, compose.WithNodeName("clean"))
_ = g.AddChatModelNode("summarize", &slowModel{delay: 30 * time.Millisecond}, compose.WithNodeName("summarize"))
_ = g.AddEdge(compose.START, "clean")
_ = g.AddEdge("clean", "summarize")
_ = g.AddEdge("summarize", compose.END)

runnable, err := g.Compile(ctx)
if err != nil {
return err
}

for i := 0; i < 3; i++ {
if _, err := runnable.Invoke(ctx, map[string]any{"ticket": " 打印机离线 "}, compose.WithCallbacks(handler)); err != nil {
return err
}
}
m.report()

// 2. 对照实验：同一份代码，超短超时会触发失败，验证错误也被记录。
fmt.Println("— 使用极短超时触发失败 —")
shortCtx, cancel := context.WithTimeout(ctx, 1*time.Millisecond)
defer cancel()
if _, err := runnable.Invoke(shortCtx, map[string]any{"ticket": "打印机离线"}, compose.WithCallbacks(handler)); err != nil {
// 对外只给出可读提示，不暴露内部细节。
fmt.Println(" 对外提示: 处理失败，请稍后重试")
}
m.report()
return nil
}

实机输出：

— 节点耗时统计 —
clean 调用=3 总耗时=0s 失败=0
summarize 调用=3 总耗时=92ms 失败=0
Graph 调用=3 总耗时=92ms 失败=0
— 使用极短超时触发失败 —
[错误] 节点=summarize 类型=context.deadlineExceededError
[错误] 节点=Graph 类型=*compose.internalError
对外提示: 处理失败，请稍后重试
— 节点耗时统计 —
Graph 调用=3 总耗时=92ms 失败=1
clean 调用=4 总耗时=0s 失败=0
summarize 调用=3 总耗时=92ms 失败=1

这份输出回答了开头的两个问题：

“慢在哪一步？” summarize 节点耗时 92ms，clean 几乎为 0。问题定位到具体节点，而不是笼统的“整个请求慢”。 如果没设 WithNodeName，这里看到的会是 ChatModel、Lambda 这样的类型名，多个同类节点就分不清了。

“为什么失败？” 错误类型是 context.deadlineExceededError——是超时，不是鉴权失败，也不是模型返回异常。错误类型直接指明了排查方向（对照第 03 篇的超时排查思路）。

注意 Graph 节点也记录了一次失败：它是外层容器，错误会向上冒泡。分析时要看最内层的失败节点。

四、脱敏：观测最容易出事的地方

可观测性有个反直觉的风险：为了“看得清楚”，把不该记的东西记下来了。

建议的粒度：

只记元信息，不记内容。 需要排查内容问题时，用请求 ID 在受控的存储中按需检索，而不是全量落日志。

五、接入追踪平台

人工统计只适合排查，生产环境通常接入追踪系统。Eino 的 eino-ext 提供了回调集成，可以把节点执行上报到可观测平台：

用法上和上面的自定义回调一致：作为 Handler 挂到运行上。好处是能拿到完整的调用链视图，一眼看出哪一层耗时最长。

无论接入哪个平台，脱敏原则不变——上报之前先过滤敏感字段。

六、常见问题

回调里拿不到节点名 没有设置 WithNodeName，只能拿到组件类型名。给节点命名后再看。

多个同类节点数据混在一起 同样是命名问题。RunInfo.Name 只适合展示，不唯一；如果需要精确区分，可结合节点 key 或路径信息。

回调里读状态报错 返回了新的 context 而不是传入的 ctx（第 10 篇）。

统计到的耗时和用户感知差很多 用户感知还包括网络往返、排队、前端渲染。回调只测量服务端节点执行时间，要结合接入层的埋点一起看。

日志里出现了用户数据 脱敏没做。检查是否记录了请求/响应正文、工具参数或错误详情。

只看总耗时看不出问题 按节点拆分，并区分“第一个 token 的时间”和“总时长”——对用户感知影响最大的是前者。

七、小结

• Callbacks 提供 OnStart/OnEnd/OnError 等时机，是横切的，无需改节点代码。
• 回调必须返回传入的 ctx，否则会丢失状态。
• 用 WithNodeName 给节点命名，否则同类节点的统计会混在一起。
• 按节点统计耗时与失败，能把“慢”和“坏”定位到具体环节。
• 只记录元信息，不记录内容：节点名、耗时、状态、错误类型、请求 ID。
• 生产环境可接入 APMPlus、CozeLoop 等平台，但仍需先做脱敏。

下一篇《可靠性设计：超时、重试、降级与幂等》，我们讨论失败处理。

参考资料与版本说明

资料核对日期为 2026 年 10 月 9 日：

本文示例按 Eino v0.9.21 编写，使用模拟耗时模型（不访问外部服务），已实机运行，正文输出（含超时失败对照）为真实结果。