Eino 生产部署与可观测性工程实战

核心摘要

Eino 应用只有在模型调用、工具、状态和发布行为都有明确工程契约时，才算具备生产条件。应固定 Eino 版本，传播统一 Deadline，在每个依赖边界限制并发，只重试安全操作，通过 Eino Callback 接入可观测性，并在请求链路之外评测答案质量。本文以 Eino v0.9.21 和标准 Go 服务模式为基线，不依赖未公开或不存在的调试、评测包。

目录

1. 先定义生产契约
2. 选择同步执行还是持久任务
3. 约束时间、并发与重试
4. 接入 Eino Callback
5. 可运行的 Callback 示例
6. 设计 Trace 与指标
7. 发布前评测质量
8. 安全发布与回滚
9. 故障排查清单
10. 常见问题
11. 一手资料

先定义生产契约

生产契约要在选择 Kubernetes 参数或 Trace 平台之前，说明服务究竟承诺什么。

至少定义以下不变量：

Eino 提供组合、Agent、Callback、Streaming 和 Checkpoint Hook，但不会替业务决定这些契约。应把 Eino 组件模型当作服务边界内的执行层。

选择同步执行还是持久任务

执行模式取决于任务语义，不能简单套用“Agent 都应该异步”。

同步请求

以下情况适合同步 HTTP 或 RPC：

• 结果能够满足调用方延迟预算
• 取消后应立即停止有用工作
• 没有需要恢复的长生命周期外部副作用
• 响应可以安全地重新生成

Handler 持有根 context.Context。所有模型、Retriever 和 Tool 调用都必须从它派生，避免客户端断开后后台继续消耗资源。

持久任务

以下情况应使用持久队列和状态机：

• 执行可能超过 Gateway Timeout
• 进程重启后必须继续
• Tool 会产生外部可见副作用
• 人工审批或延迟回调会暂停工作

持久化 Job ID、输入版本、Prompt 与 Tool 配置版本、当前步骤、Idempotency Key 和最终结果。队列只提供投递，不会自动提供 Exactly-once 业务效果；每个副作用仍需要自己的去重契约。

约束时间、并发与重试

容量控制必须落在稀缺依赖边界，而不只是 HTTP Server 入口。

建议分别限制：

• 活跃用户请求数
• 每个 Provider 与 Model 的请求数
• 每个下游 Tool 的调用数
• Embedding 与 Rerank 批次数
• 后台评测任务数

全局 Goroutine 上限仍可能压垮单一 Provider，同时让其他资源闲置。按依赖设置 Semaphore，才能观察饱和点并实现隔离。

Deadline 层级

子级 Deadline 不能超过父级剩余时间：

request deadline
  -> orchestration deadline
     -> model attempt deadline
     -> tool attempt deadline

还要为输出校验和序列化预留时间。模型调用如果耗尽整个请求 Deadline，系统就没有时间取消 Tool 或返回可控错误。

重试规则

只有同时满足以下三个条件才重试：

1. 错误已分类且属于瞬态故障
2. 操作本身幂等，或受 Idempotency Key 保护
3. 剩余 Deadline 足以完成 Backoff 和下一次尝试

认证失败、参数错误、策略拒绝和任意副作用都不应盲目重试。监控中需要区分 Attempt Count 与 Logical Operation Count。

接入 Eino Callback

Eino v0.9.21 在组件执行周围提供五个 Callback 时点：

• OnStart
• OnEnd
• OnError
• OnStartWithStreamInput
• OnEndWithStreamOutput

按最小有效范围挂载 Handler：

全局 Handler 注册不是线程安全的，只能在启动阶段执行一次。Callback 输入和输出指针会被下游共享，修改它们可能产生数据竞争。流式 Callback 获得的副本在消费后必须关闭。

Callback 是埋点 Hook，不是完整的可观测后端。你需要根据数据策略把它适配到 OpenTelemetry、Metrics 与结构化日志。

可运行的 Callback 示例

下面的程序构建一个双节点 Eino Chain，并通过单次调用范围的 Callback 记录每个节点耗时。代码可直接基于 Eino v0.9.21 编译，不依赖外部 API。

package main

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

	"github.com/cloudwego/eino/callbacks"
	"github.com/cloudwego/eino/compose"
)

type startedAtKey struct{}

func main() {
	ctx := context.Background()
	chain := compose.NewChain[string, string]().
		AppendLambda(
			compose.InvokableLambda(func(
				ctx context.Context,
				input string,
			) (string, error) {
				if err := ctx.Err(); err != nil {
					return "", err
				}
				return strings.TrimSpace(input), nil
			}),
			compose.WithNodeName("normalize"),
		).
		AppendLambda(
			compose.InvokableLambda(func(
				ctx context.Context,
				input string,
			) (string, error) {
				if err := ctx.Err(); err != nil {
					return "", err
				}
				return "answer: " + input, nil
			}),
			compose.WithNodeName("answer"),
		)

	runnable, err := chain.Compile(
		ctx,
		compose.WithGraphName("support_request"),
		compose.WithMaxRunSteps(8),
	)
	if err != nil {
		log.Fatal(err)
	}

	handler := callbacks.NewHandlerBuilder().
		OnStartFn(func(
			ctx context.Context,
			info *callbacks.RunInfo,
			_ callbacks.CallbackInput,
		) context.Context {
			return context.WithValue(ctx, startedAtKey{}, time.Now())
		}).
		OnEndFn(func(
			ctx context.Context,
			info *callbacks.RunInfo,
			_ callbacks.CallbackOutput,
		) context.Context {
			startedAt, _ := ctx.Value(startedAtKey{}).(time.Time)
			if info != nil {
				log.Printf(
					"component=%s name=%s duration=%s",
					info.Component,
					info.Name,
					time.Since(startedAt),
				)
			}
			return ctx
		}).
		OnErrorFn(func(
			ctx context.Context,
			info *callbacks.RunInfo,
			runErr error,
		) context.Context {
			if info != nil {
				log.Printf(
					"component=%s name=%s error=%q",
					info.Component,
					info.Name,
					runErr,
				)
			}
			return ctx
		}).
		Build()

	runCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	output, err := runnable.Invoke(
		runCtx,
		"  diagnose checkout timeout  ",
		compose.WithCallbacks(handler),
	)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(output)
	// Output: answer: diagnose checkout timeout
}

模块依赖固定为：

require github.com/cloudwego/eino v0.9.21

接入 OpenTelemetry 时，可以在 Callback Start 创建 Span，把它放入返回的 Context，在 End 或 Error 中记录有界 Attribute，最后结束 Span。Span Name 使用 eino.chat_model、eino.tool 这类低基数字段，Model、Tool 身份放在 Attribute 中，不要把 User ID 或请求文本拼进 Span Name。

设计 Trace 与指标

OpenTelemetry 把 Trace 定义为通过 Context 连接的嵌套 Span。一个实用的 Eino 请求链路如下：

agent.request
  eino.agent
    eino.chat_model
    eino.tool
      downstream.rpc
    eino.chat_model
  output.validate

遥测需要足以诊断行为，但不能变成 Prompt 归档库。

推荐记录

• Application、Environment、Release 与 Prompt Version
• Eino Component、Node、Agent、Model 与 Tool Name
• Attempt Number 与 Result Class
• 输入输出字节数分桶
• Provider 返回的 Prompt、Completion 与 Total Token
• Queue Time、Model Latency、Tool Latency 与首 Token 延迟
• Policy Decision、Timeout、Cancellation 与 Fallback Reason

默认避免记录

• 原始 Prompt 与模型输出
• Credential、Authorization Header、Cookie 或 Signed URL
• 未经限制的 Tool Argument 和 Result
• 个人信息或高基数 Tenant 标识

对敏感值做 Hash 不代表自动安全。采集前必须先定义留存、访问、删除、采样和地域存储规则。

服务级指标

系统健康与答案质量必须分开：

HTTP 200 但答案没有证据支撑，属于运行成功、产品失败。两层都需要独立门禁。

发布前评测质量

评测应在在线请求链路之外，对版本化数据集执行。

至少维护四组测试：

1. Golden Task：有确定字段或结果的基准任务
2. Adversarial Task：覆盖 Prompt Injection、跨租户访问与破坏性操作
3. Regression Task：真实故障脱敏后沉淀的回归用例
4. Operational Task：模拟 Timeout、Rate Limit、Partial Stream 与 Tool Failure

自由文本答案需要明确定义 Rubric，并用人工标签校准 LLM Judge。每次结果要一起记录 Model Revision、Prompt Version、Tool Catalog、Data Snapshot、Temperature、Retry Policy 与 Judge Version。

发布门禁应把候选版本与当前生产基线比较。质量收益不能以突破延迟、成本或安全预算为代价；工作负载不一致时，也不能声称框架性能更强。

安全发布与回滚

一个 Agent Release 不只是二进制：

• 应用 Commit
• Eino 与 Provider SDK 版本
• Model Identifier 与 Revision
• System Prompt 与 Skill 内容
• Tool Schema 与 Policy
• 检索语料与 Index Version
• Evaluator 与 Threshold 版本

使用 Canary 或按版本路由。持久任务应继续绑定创建它的版本，除非状态格式有显式迁移。回滚必须恢复完整执行包，而不只是 Go Binary。

扩大流量前至少验证：

• Startup、Readiness 与 Dependency Check
• Cancellation 与 Graceful Shutdown
• Provider 与 Tool 限流行为
• HTTP、RPC、Queue 与 MCP 边界上的 Trace Propagation
• Output Validation 与 Policy Denial
• 新旧版本并存

故障排查清单

延迟或失败率上升时：

1. 检查请求量、Queue Time 与 Admission Control 拒绝。
2. 按 Model、Retriever、Tool 拆分延迟，不只看聚合值。
3. 对比 Attempt Count 和 Logical Call，识别重试放大。
4. 检查 Deadline 传播和未关闭的 StreamReader。
5. 检查 Provider Quota、Connection Pool 与下游饱和。
6. 确认受影响 Trace 的 Prompt、Model、Tool Schema 与 Index Version。
7. 抽样脱敏后的失败路径，与发布基线比较。
8. 确认回归与版本相关时，回滚完整执行包。

常见问题

Eino 提供生产可观测后端吗？

Eino 提供 Callback Hook 和执行元数据。Trace、Metric、Log 后端仍需自行选择和运维，同时定义语义 Attribute 与隐私控制。

Callback 应该全局挂载还是按请求挂载？

进程启动时挂载一个强制基础遥测 Handler；按请求 Handler 用于 Tenant-aware Sampling 或临时诊断。执行开始后不要再修改全局 Handler。

如何统计流式错误？

必须统计消费 Stream 时返回的错误，而不只统计创建 Stream 时的错误。Eino Callback 文档指出，流中间错误不会进入 OnError；Consumer 必须记录它并关闭 Reader。

Readiness Check 应该检查什么？

Readiness 要证明进程可以接收任务，但不要发起昂贵的模型调用。检查配置、策略、Key、必要存储和本地队列；Provider 可达性单独监控，避免健康检查形成级联压力。

总结

Eino 生产工程的本质是契约工程。放量前先约束时间、并发、重试、状态、授权与遥测。使用 v0.9.21 真实暴露的 Callback API 接入标准可观测基础设施，并把质量评测与请求成功分开。安全发布必须能够作为一个完整版本包被测量、复现和回滚。

相关资源

• Eino 流式处理与 Callback 系统
• Eino 多智能体协作
• Eino RAG Pipeline 生产实践
• OpenTelemetry 术语

一手资料

• Eino v0.9.21 Callback Package
• Eino v0.9.21 ADK API
• CloudWeGo Callback 与 Trace 指南
• OpenTelemetry Trace API 规范
• OpenTelemetry 生成式 AI 语义约定