28Callbacks 与可观测性定位慢调用和失败节点《Eino 实战用 Go 构建 AI 应用与智能体》· 第 28 篇关键词Callbacks、OnStart/OnEnd/OnError、节点耗时、脱敏、追踪前置条件已完成 27HTTP 与 SSE 与 10状态与运行选项。应用上线后用户说“回答很慢”或者“有时候会失败”。这时你需要回答两个问题慢在哪一步是模型调用、检索还是某个业务接口为什么失败是超时、鉴权还是外部服务出错没有观测数据只能猜。Callbacks 就是 Eino 提供的观测入口。一、回调的三种时机callbacks.Handler关心的主要时机时机触发点用途OnStart节点开始执行记录开始时间、输入规模OnEnd节点正常结束记录耗时、输出规模OnError节点出错记录错误类型、定位失败节点还有两个流式专用时机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()returnctx}).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)]))returnctx}).OnErrorFn(func(ctx context.Context,info*callbacks.RunInfo,errerror)context.Context{m.failed[key(info)]returnctx}).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 篇也提到过字段说明Name节点显示名来自WithNodeName不唯一仅用于展示Type运行类型标识Component组件类型如Graph、Lambda、ChatModel三、完整示例下面这段程序统计每个节点的耗时与失败次数并做了一次“正常 vs 超时”的对照实验已实机运行packagemainimport(contextfmtlogstringssynctimegithub.com/cloudwego/eino/callbacksgithub.com/cloudwego/eino/components/modelgithub.com/cloudwego/eino/composegithub.com/cloudwego/eino/schema)// slowModel 模拟一个耗时调用便于观察耗时统计。typeslowModelstruct{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():returnnil,ctx.Err()}returnschema.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...)iferr!nil{returnnil,err}sr,sw:schema.Pipe[*schema.Message](1)gofunc(){defersw.Close();sw.Send(msg,nil)}()returnsr,nil}// metrics 收集每个节点的耗时与成败是“可观测性”的最小实现。typemetricsstruct{mu sync.Mutex startmap[string]time.Time recordmap[string][]time.Duration failedmap[string]int}funcnewMetrics()*metrics{returnmetrics{start:map[string]time.Time{},record:map[string][]time.Duration{},failed:map[string]int{}}}func(m*metrics)key(info*callbacks.RunInfo)string{name:info.Nameifname{namestring(info.Component)}returnname}func(m*metrics)report(){m.mu.Lock()deferm.mu.Unlock()fmt.Println(--- 节点耗时统计 ---)forname,ds:rangem.record{vartotal time.Durationfor_,d:rangeds{totald}fmt.Printf( %-14s 调用%d 总耗时%v 失败%d\n,name,len(ds),total.Round(time.Millisecond),m.failed[name])}}funcmain(){iferr:run();err!nil{log.Fatal(err)}}funcrun()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()returnctx}).OnEndFn(func(ctx context.Context,info*callbacks.RunInfo,_callbacks.CallbackOutput)context.Context{key:m.key(info)m.mu.Lock()ift,ok:m.start[key];ok{m.record[key]append(m.record[key],time.Since(t))}m.mu.Unlock()returnctx}).OnErrorFn(func(ctx context.Context,info*callbacks.RunInfo,errerror)context.Context{key:m.key(info)m.mu.Lock()m.failed[key]m.mu.Unlock()// 只记录错误类型不打印完整错误原文避免泄露内部信息。fmt.Printf( [错误] 节点%s 类型%T\n,key,err)returnctx}).Build()// 清洗节点把工单描述整理成模型需要的消息列表。tpl:compose.InvokableLambda(func(_context.Context,inmap[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)iferr!nil{returnerr}fori:0;i3;i{if_,err:runnable.Invoke(ctx,map[string]any{ticket: 打印机离线 },compose.WithCallbacks(handler));err!nil{returnerr}}m.report()// 2. 对照实验同一份代码超短超时会触发失败验证错误也被记录。fmt.Println(--- 使用极短超时触发失败 ---)shortCtx,cancel:context.WithTimeout(ctx,1*time.Millisecond)defercancel()if_,err:runnable.Invoke(shortCtx,map[string]any{ticket:打印机离线},compose.WithCallbacks(handler));err!nil{// 对外只给出可读提示不暴露内部细节。fmt.Println( 对外提示: 处理失败请稍后重试)}m.report()returnnil}实机输出--- 节点耗时统计 --- 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节点耗时 92msclean几乎为 0。问题定位到具体节点而不是笼统的“整个请求慢”。如果没设WithNodeName这里看到的会是ChatModel、Lambda这样的类型名多个同类节点就分不清了。“为什么失败”错误类型是context.deadlineExceededError——是超时不是鉴权失败也不是模型返回异常。错误类型直接指明了排查方向对照第 03 篇的超时排查思路。注意Graph节点也记录了一次失败它是外层容器错误会向上冒泡。分析时要看最内层的失败节点。四、脱敏观测最容易出事的地方可观测性有个反直觉的风险为了“看得清楚”把不该记的东西记下来了。不要记录原因完整的模型请求与响应可能包含用户隐私、内部资料工具参数与返回值原文可能含账号、地址、业务数据API Key、Token凭据泄露完整错误堆栈到日志外发系统可能含内部路径、服务名建议的粒度记录什么例子节点名与组件类型summarize/ChatModel耗时92ms状态成功 / 失败错误类型context.deadlineExceededError规模指标消息条数、Token 数、结果字符数请求 ID用于串联同一次请求的各条日志只记元信息不记内容。需要排查内容问题时用请求 ID 在受控的存储中按需检索而不是全量落日志。五、接入追踪平台人工统计只适合排查生产环境通常接入追踪系统。Eino 的eino-ext提供了回调集成可以把节点执行上报到可观测平台集成说明APMPlus字节跳动的可观测平台CozeLoopCoze 的观测与评测平台用法上和上面的自定义回调一致作为Handler挂到运行上。好处是能拿到完整的调用链视图一眼看出哪一层耗时最长。无论接入哪个平台脱敏原则不变——上报之前先过滤敏感字段。六、常见问题回调里拿不到节点名没有设置WithNodeName只能拿到组件类型名。给节点命名后再看。多个同类节点数据混在一起同样是命名问题。RunInfo.Name只适合展示不唯一如果需要精确区分可结合节点 key 或路径信息。回调里读状态报错返回了新的context而不是传入的ctx第 10 篇。统计到的耗时和用户感知差很多用户感知还包括网络往返、排队、前端渲染。回调只测量服务端节点执行时间要结合接入层的埋点一起看。日志里出现了用户数据脱敏没做。检查是否记录了请求/响应正文、工具参数或错误详情。只看总耗时看不出问题按节点拆分并区分“第一个 token 的时间”和“总时长”——对用户感知影响最大的是前者。七、小结Callbacks 提供OnStart/OnEnd/OnError等时机是横切的无需改节点代码。回调必须返回传入的ctx否则会丢失状态。用WithNodeName给节点命名否则同类节点的统计会混在一起。按节点统计耗时与失败能把“慢”和“坏”定位到具体环节。只记录元信息不记录内容节点名、耗时、状态、错误类型、请求 ID。生产环境可接入 APMPlus、CozeLoop 等平台但仍需先做脱敏。下一篇《可靠性设计超时、重试、降级与幂等》我们讨论失败处理。参考资料与版本说明资料核对日期为2026 年 10 月 9 日Callback 用户手册ChatModel 使用说明Callback 示例Callbacks 组件集成Eino 主仓库本文示例按Eino v0.9.21编写使用模拟耗时模型不访问外部服务已实机运行正文输出含超时失败对照为真实结果。