给 AI Agent 装上生产级可观测性：用 OpenTelemetry 把 LLM 工作流从黑盒变白盒
深度解析 2026 年 AI 工程化最关键的方向——AI Agent 可观测性，涵盖 OpenTelemetry LLM 语义约定、MCP 协议链路追踪、OpenLIT 全链路平台，手写生产级代码实现。
一、问题的本质：AI Agent 为什么是终极黑盒
当你部署了一个 AI Agent 处理用户工单——"搜索→意图识别→调用内部 API→生成回复"四步流水线，突然用户反馈"响应慢得离谱"。你打开监控大屏：HTTP 请求全部 200，数据库响应正常，Redis 命中率 99%。但问题在哪？
在 LLM 调用那 3 秒里，没人知道发生了什么。
这是 2026 年每一个 AI 应用工程师都在面对的问题。传统的 APM（New Relic、Datadog）只知道记录"HTTP 请求出去了"和"HTTP 响应回来了"，但它们无法回答以下灵魂拷问：
- Prompt 消耗了多少 Token？
- 模型重试了几次才成功？
- 某一步的工具调用参数是什么？
- 幻觉答案是哪一步开始跑偏的？
- 当前 Agent 的上下文窗口用了百分之多少？
更棘手的是，AI Agent 的调用链路和传统微服务有本质区别：
| 维度 | 传统微服务 | AI Agent 工作流 |
|---|---|---|
| 调用模型 | 确定性的 RPC/HTTP | 非确定性的 LLM API |
| 链路深度 | 几十个服务 | 几轮对话 + 多个工具 |
| 失败模式 | 超时/错误码 | 幻觉/Token 超限/格式错误 |
| 关键指标 | 延迟/QPS | Token 消耗/重试次数/幻觉率 |
| 上下文传递 | TraceID | TraceID + 完整对话历史 |
没有可观测性，AI Agent 在生产环境里就是一颗定时炸弹。你不知道它什么时候会"自由发挥"超过上下文窗口，不知道哪一步在悄悄烧 Token，不知道为什么同一个问题今天答对明天答错。
二、OpenTelemetry 为什么是 AI 可观测性的最佳选择
OpenTelemetry（OTel）已经成为云原生领域事实上的可观测性标准。2026 年，它的覆盖范围从传统的 HTTP/gRPC/db 扩展到了 AI 领域。选择 OTel 作为 AI Agent 可观测性的基础设施，有三个核心理由：
1. 厂商中立，不需要绑定 LangSmith/Guidance 等闭源服务
OTel 的数据模型是 CNCF 的开放标准，任何兼容 OTLP 的后端（Jaeger、Grafana Tempo、Tempo、Elastic APM）都可以接收数据。这意味着你可以把 AI 链路追踪数据和现有的微服务追踪数据放在同一个平台里，做统一的根因分析。
2. 三大支柱天然覆盖 AI 可观测性的所有维度
- Traces（链路追踪）：记录每次 LLM 调用的输入/输出/耗时/Token 消耗，串联整个 Agent 工作流
- Metrics（指标）：Token 消耗速率、模型错误率、平均响应时间、上下文窗口利用率
- Logs（日志）：详细的 Prompt/Completion 内容、工具调用的请求响应
3. Context Propagation 让跨服务追踪成为可能
当你的 Agent 调用外部 MCP Server、MongoDB、Elasticsearch 时，OTel 的 W3C Trace Context 自动传播 TraceID，不需要为每个工具单独埋点。
三、LLM 语义约定：标准 Span 不够用
传统的 OTel Span 用 http.method、db.statement 这类属性来描述调用。但 LLM 调用有完全不同的特征维度，需要扩展自定义属性。
3.1 LLM 语义约定的核心属性
以下是 2026 年 AI 可观测性领域公认的 LLM Span 属性规范：
// LLM Span 核心属性
attribute.String("llm.model", "gpt-4o") // 模型名称
attribute.Int64("llm.prompt_tokens", 2340) // 输入 Token 数
attribute.Int64("llm.completion_tokens", 890) // 输出 Token 数
attribute.Int64("llm.total_tokens", 3230) // 总 Token 数
attribute.Float64("llm.temperature", 0.7) // 采样温度
attribute.Int("llm.max_tokens", 4096) // 最大 Token 限制
attribute.String("llm.finish_reason", "stop") // 停止原因
attribute.Int("llm.retry_count", 2) // 重试次数
attribute.String("llm.workflow_step", "intent_classification") // 工作流步骤名
attribute.String("llm.embedding_model", "text-embedding-3-large") // Embedding 模型
attribute.Float64("llm.embedding_dimension", 3072.0) // Embedding 维度
3.2 工具调用的 Span 属性
当 Agent 调用外部工具（Search、Calculator、MCP Tools）时，需要记录工具相关的属性：
// 工具调用 Span 属性
attribute.String("tool.name", "web_search")
attribute.String("tool.provider", "serpapi")
attribute.Int("tool.retry_count", 1)
attribute.String("tool.error", "") // 空字符串表示无错误
attribute.Float64("tool.latency_ms", 234.5)
attribute.Bool("tool.cache_hit", false) // 是否命中缓存
3.3 为什么这些属性比 TraceID 更重要
来看一个典型的 AI Agent 故障排查场景：用户反馈"这个月的销售汇总不对"。传统的链路追踪只能告诉你"第三步 LLM 调用耗时 8 秒"。但有了 LLM 语义约定属性，你可以：
- 按
llm.model筛选，确认第三步用的是gpt-4o-mini还是gpt-4o（不同模型数字精度不同） - 看
llm.prompt_tokens，发现第三步的 Prompt 长达 8000 Token——上下文窗口快满了 - 看
tool.name，发现第二步调用了错误的数据库查询工具 - 看
llm.finish_reason，发现第三步的输出被length截断了，导致汇总不完整
这就是从"知道慢"到"知道为什么慢"的本质区别。
四、实战：用 Go 写一个生产级的 LLM OTel 包装器
下面我们从零手写一个生产级的 LLM 可观测性包装器，完整覆盖：自动 Trace/手动埋点/重试追踪/Token 统计/错误分类。
4.1 项目结构
llm-observability/
├── go.mod
├── main.go
├── telemetry/
│ ├── tracer.go # OTel 初始化 + Exporter 配置
│ ├── llm.go # LLM 调用包装器
│ ├── tools.go # 工具调用追踪
│ └── context.go # Context 传播工具
└── backend/
└── docker-compose.yml # Jaeger + Prometheus + Grafana
4.2 OTel 初始化（telemetry/tracer.go）
package telemetry
import (
"context"
"fmt"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/exporters/prometheus"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/metrics"
"go.opentelemetry.io/otel/sdk/resource"
"go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
)
// Config 描述 OTel 配置
type Config struct {
ServiceName string
ServiceVersion string
OTLPEndpoint string // 例如 "localhost:4317"
Env string // production / staging
}
// Provider 持有所有 OTel 资源
type Provider struct {
TracerProvider *trace.TracerProvider
MeterProvider *metrics.MeterProvider
Tracer trace.Tracer
Meter metrics.Meter
}
// InitTracer 初始化全局 TracerProvider + MeterProvider
func InitTracer(ctx context.Context, cfg Config) (*Provider, error) {
// 1. 创建资源（服务名、版本、环境）
res, err := resource.New(ctx,
resource.WithAttributes(
semconv.ServiceName(cfg.ServiceName),
semconv.ServiceVersion(cfg.ServiceVersion),
semconv.DeploymentEnvironment(cfg.Env),
semconv.TelemetrySDKLanguageGo,
),
resource.WithHost(),
resource.WithProcess(),
)
if err != nil {
return nil, fmt.Errorf("创建 OTel 资源失败: %w", err)
}
// 2. 配置 OTLP gRPC Trace Exporter
// 生产环境推荐：OTEL_EXPORTER_OTLP_ENDPOINT=https://otel-collector:4317
traceExporter, err := otlptracegrpc.New(ctx,
otlptracegrpc.WithEndpoint(cfg.OTLPEndpoint),
otlptracegrpc.WithDialOption(
grpc.WithTransportCredentials(insecure.NewCredentials()),
),
// 采样策略：生产环境 10%，错误时 100%
otlptracegrpc.WithTraceAgentGradientcompression(),
)
if err != nil {
return nil, fmt.Errorf("创建 Trace Exporter 失败: %w", err)
}
// 3. 配置 Prometheus Metrics Exporter
promExporter, err := prometheus.New()
if err != nil {
return nil, fmt.Errorf("创建 Prometheus Exporter 失败: %w", err)
}
// 4. 构建 TracerProvider（带采样器）
tp := trace.NewTracerProvider(
trace.WithBatcher(traceExporter,
// 批处理配置：100 个 span 或 5 秒刷一次
trace.WithBatchTimeout(5),
trace.WithExportTimeout(10),
),
trace.WithResource(res),
trace.WithSampler(
// 条件采样：AI 调用 100% 追踪，其他 1%
trace.ParentBased(
aiSamplingSampler(),
),
),
)
// 5. 构建 MeterProvider
mp := metrics.NewMeterProvider(
metrics.WithReader(promExporter),
metrics.WithResource(res),
)
// 6. 注册全局 TracerProvider 和 MeterProvider
otel.SetTracerProvider(tp)
otel.SetMeterProvider(mp)
// 7. 配置 W3C Trace Context 传播器
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
propagation.TraceContext{}, // W3C Trace Context
propagation.Baggage{}, // W3C Baggage
))
return &Provider{
TracerProvider: tp,
MeterProvider: mp,
Tracer: tp.Tracer(cfg.ServiceName),
Meter: mp.Meter(cfg.ServiceName),
}, nil
}
// aiSamplingSampler 智能采样器
// AI 相关的操作（llm.*、tool.* 属性）高采样率
// 普通 HTTP 调用低采样率
func aiSamplingSampler() trace.Sampler {
return trace.SamplerFunc(func(p trace.SamplingParameters) trace.SamplingResult {
// 检查 span 是否包含 AI 相关属性
for _, attr := range p.Attributes {
if (attr.Key == "llm.model") || (attr.Key == "tool.name") {
return trace.SamplingResult{
Decision: trace.RecordAndSample,
Tracestate: trace.SamplingDecisionRoot.SamplingDecision.Tracestate,
}
}
}
// 普通 span：只记录，不采样（节省存储）
return trace.SamplingResult{Decision: trace.RecordOnly}
})
}
4.3 LLM 调用追踪器（telemetry/llm.go）
这是最核心的部分——一个带完整可观测性的 LLM 调用包装器：
package telemetry
import (
"context"
"fmt"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/nonrecording"
)
// LLMCallInput LLM 调用的输入参数
type LLMCallInput struct {
Model string
Prompt string
SystemPrompt string // 可选的系统提示词
Temperature float64
MaxTokens int
StepName string // 在工作流中的步骤名，如 "intent_classification"
Metadata map[string]string // 额外元数据
}
// LLMCallOutput LLM 调用的输出结果
type LLMCallOutput struct {
Completion string
PromptTokens int
CompletionTokens int
TotalTokens int
Model string
FinishReason string // "stop" | "length" | "content_filter" | "error"
LatencyMs float64
}
// LLMTracer LLM 可观测性追踪器
type LLMTracer struct {
tracer trace.Tracer
meter metric.Meter
// 指标计数器
callCounter metric.Int64Counter
tokenCounter metric.Int64Counter
errorCounter metric.Int64Counter
latencyHistogram metric.Float64Histogram
}
// NewLLMTracer 创建 LLM 追踪器
func NewLLMTracer(provider *Provider) (*LLMTracer, error) {
meter := provider.Meter
callCounter, err := meter.Int64Counter("llm.calls.total",
metric.WithDescription("LLM 调用总次数"),
metric.WithUnit("{call}"),
)
if err != nil {
return nil, err
}
tokenCounter, err := meter.Int64Counter("llm.tokens.total",
metric.WithDescription("LLM Token 消耗总数量"),
metric.WithUnit("{token}"),
)
if err != nil {
return nil, err
}
errorCounter, err := meter.Int64Counter("llm.errors.total",
metric.WithDescription("LLM 调用错误总次数"),
metric.WithUnit("{error}"),
)
if err != nil {
return nil, err
}
latencyHistogram, err := meter.Float64Histogram("llm.latency.ms",
metric.WithDescription("LLM 响应延迟分布"),
metric.WithUnit("ms"),
metric.WithExplicitBucketBoundaries(
100, 250, 500, 1000, 2000, 5000, 10000, 30000,
),
)
if err != nil {
return nil, err
}
return &LLMTracer{
tracer: provider.Tracer,
meter: meter,
callCounter: callCounter,
tokenCounter: tokenCounter,
errorCounter: errorCounter,
latencyHistogram: latencyHistogram,
}, nil
}
// TracedLLMCall 在 OpenTelemetry Span 中执行 LLM 调用
// 这是整个可观测性系统的核心方法
func (t *LLMTracer) TracedLLMCall(
ctx context.Context,
input LLMCallInput,
callFn func(ctx context.Context, req LLMCallInput) (LLMCallOutput, error),
) (LLMCallOutput, error) {
// 1. 创建 Span，自动继承父 Context 的 TraceID
// 如果没有父 Span，会创建一个新的 Root Span
spanName := fmt.Sprintf("llm.%s", input.StepName)
ctx, span := t.tracer.Start(ctx, spanName,
trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(
// LLM 核心属性
attribute.String("llm.model", input.Model),
attribute.String("llm.step", input.StepName),
attribute.Float64("llm.temperature", input.Temperature),
attribute.Int("llm.max_tokens", input.MaxTokens),
attribute.Int("llm.prompt_length_chars", len(input.Prompt)),
attribute.String("llm.workflow_step", input.StepName),
// 记录系统提示词是否存在（不记录内容，防止敏感信息泄露）
attribute.Bool("llm.has_system_prompt", input.SystemPrompt != ""),
),
)
defer span.End()
start := time.Now()
var retryCount int
maxRetries := 3
// 2. 带重试的调用逻辑
var output LLMCallOutput
var lastErr error
for retryCount = 1; retryCount <= maxRetries; retryCount++ {
// 如果不是第一次调用，在 Span 里记录重试次数
if retryCount > 1 {
span.SetAttributes(attribute.Int("llm.retry_count", retryCount-1))
}
output, lastErr = callFn(ctx, input)
if lastErr == nil {
break // 成功，退出重试循环
}
// 失败处理：记录错误并决定是否重试
span.RecordError(lastErr)
span.SetAttributes(
attribute.String("error.type", classifyError(lastErr)),
attribute.Bool("llm.retry_applied", retryCount < maxRetries),
)
// 非重试错误（如 Token 超限、权限问题）直接终止
if !isRetryableError(lastErr) {
break
}
// 指数退避
backoff := time.Duration(500*(1<<(retryCount-1))) * time.Millisecond
time.Sleep(backoff)
}
latencyMs := float64(time.Since(start).Milliseconds())
// 3. 处理结果或错误
if lastErr != nil {
span.SetStatus(codes.Error, lastErr.Error())
span.SetAttributes(
attribute.String("llm.finish_reason", "error"),
attribute.Int("llm.retry_count", retryCount-1),
)
// 指标：错误计数
t.errorCounter.Add(ctx, 1,
metric.WithAttributes(
attribute.String("model", input.Model),
attribute.String("step", input.StepName),
attribute.String("error_type", classifyError(lastErr)),
),
)
t.callCounter.Add(ctx, 1,
metric.WithAttributes(
attribute.String("model", input.Model),
attribute.String("status", "error"),
),
)
} else {
// 4. 成功：填充 Token 统计和完成原因
span.SetAttributes(
attribute.Int64("llm.prompt_tokens", int64(output.PromptTokens)),
attribute.Int64("llm.completion_tokens", int64(output.CompletionTokens)),
attribute.Int64("llm.total_tokens", int64(output.TotalTokens)),
attribute.String("llm.finish_reason", output.FinishReason),
attribute.Int("llm.retry_count", retryCount-1),
// 计算 Token 效率（每 Token 的延迟）
attribute.Float64("llm.tokens_per_second",
float64(output.TotalTokens)/latencyMs*1000),
// 上下文窗口利用率（假设上下文窗口为 128k）
attribute.Float64("llm.context_utilization",
float64(output.TotalTokens)/128000.0),
)
// 指标：Token 消耗
t.tokenCounter.Add(ctx, int64(output.TotalTokens),
metric.WithAttributes(
attribute.String("model", input.Model),
attribute.String("type", "prompt"),
attribute.String("step", input.StepName),
),
)
// 指标：Latency
t.latencyHistogram.Record(ctx, latencyMs,
metric.WithAttributes(
attribute.String("model", input.Model),
attribute.String("step", input.StepName),
),
)
// 指标：成功计数
t.callCounter.Add(ctx, 1,
metric.WithAttributes(
attribute.String("model", input.Model),
attribute.String("status", "success"),
),
)
output.LatencyMs = latencyMs
}
return output, lastErr
}
// classifyError 将错误分类为可操作类型
// 而不是简单地记录错误字符串
func classifyError(err error) string {
// 这里根据实际错误类型进行分类
// 常见分类：
// - rate_limit: 速率限制（需要退避）
// - context_overflow: 上下文超限（需要截断）
// - auth_failure: 认证失败（需要检查 API Key）
// - network_timeout: 网络超时（需要重试）
// - model_unavailable: 模型不可用（需要切换模型）
// - content_policy: 内容策略违规（需要修改 Prompt）
return "unknown"
}
// isRetryableError 判断错误是否值得重试
func isRetryableError(err error) bool {
// 速率限制 (429) 和服务器错误 (500, 502, 503, 504) 值得重试
// Token 超限 (400, context_overflow)、认证失败 (401) 不值得重试
// 具体实现根据实际的 API 错误类型来判断
return false // 实际实现需要解析错误类型
}
4.4 工具调用追踪器（telemetry/tools.go）
AI Agent 的强大之处在于调用外部工具。工具调用是另一个需要重点追踪的维度：
package telemetry
import (
"context"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/trace"
)
// ToolInput 工具调用的输入参数
type ToolInput struct {
Name string
Provider string // "mcp" | "rest" | "database"
Parameters map[string]any
}
// ToolOutput 工具调用的输出结果
type ToolOutput struct {
Result any
LatencyMs float64
FromCache bool
Error error
}
// ToolTracer 工具调用追踪器
type ToolTracer struct {
tracer trace.Tracer
callCounter metric.Int64Counter
latencyHistogram metric.Float64Histogram
errorCounter metric.Int64Counter
}
// NewToolTracer 创建工具追踪器
func NewToolTracer(provider *Provider) (*ToolTracer, error) {
meter := provider.Meter
callCounter, err := meter.Int64Counter("tool.calls.total",
metric.WithDescription("工具调用总次数"),
)
if err != nil {
return nil, err
}
latencyHistogram, err := meter.Float64Histogram("tool.latency.ms",
metric.WithDescription("工具调用延迟"),
metric.WithUnit("ms"),
)
if err != nil {
return nil, err
}
errorCounter, err := meter.Int64Counter("tool.errors.total",
metric.WithDescription("工具调用错误次数"),
)
if err != nil {
return nil, err
}
return &ToolTracer{
tracer: provider.Tracer,
callCounter: callCounter,
latencyHistogram: latencyHistogram,
errorCounter: errorCounter,
}, nil
}
// TracedToolCall 追踪工具调用
// 注意：MCP 工具调用通过 otel-instrumentation-mcp 可以自动追踪
// 这里的代码主要用于自定义工具或 MCP 无法覆盖的场景
func (t *ToolTracer) TracedToolCall(
ctx context.Context,
input ToolInput,
callFn func(ctx context.Context, input ToolInput) (ToolOutput, error),
) (ToolOutput, error) {
spanName := "tool." + input.Name
ctx, span := t.tracer.Start(ctx, spanName,
trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(
attribute.String("tool.name", input.Name),
attribute.String("tool.provider", input.Provider),
attribute.Int("tool.params_count", len(input.Parameters)),
),
)
defer span.End()
start := time.Now()
output, err := callFn(ctx, input)
latencyMs := float64(time.Since(start).Milliseconds())
output.LatencyMs = latencyMs
if err != nil {
span.SetStatus(codes.Error, err.Error())
span.RecordError(err)
span.SetAttributes(attribute.String("error.type", "tool_call_failed"))
t.errorCounter.Add(ctx, 1,
metric.WithAttributes(
attribute.String("tool", input.Name),
attribute.String("provider", input.Provider),
),
)
} else {
span.SetAttributes(
attribute.Float64("tool.latency_ms", latencyMs),
attribute.Bool("tool.cache_hit", output.FromCache),
// 记录输出大小（防止敏感信息泄露，只记录大小）
attribute.Int("tool.output_size_bytes", estimateSize(output.Result)),
)
}
t.latencyHistogram.Record(ctx, latencyMs,
metric.WithAttributes(
attribute.String("tool", input.Name),
attribute.String("provider", input.Provider),
),
)
t.callCounter.Add(ctx, 1,
metric.WithAttributes(
attribute.String("tool", input.Name),
attribute.String("status", status(err)),
),
)
return output, err
}
// estimateSize 估算数据结构大小（字节）
// 用于在 Span 中记录，避免记录敏感内容
func estimateSize(v any) int {
// 实际实现根据具体数据结构计算
return 0
}
func status(err error) string {
if err != nil {
return "error"
}
return "success"
}
4.5 Context 传播：让整个工作流串联起来（telemetry/context.go）
package telemetry
import (
"context"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/propagation"
)
// InjectTraceContext 将 TraceContext 注入到 HTTP headers
// 用于跨服务传播（如调用外部 MCP Server 或 REST API）
func InjectTraceContext(ctx context.Context, headers map[string]string) {
injector := otel.GetTextMapPropagator()
injector.Inject(ctx, propagation.MapCarrier(headers))
}
// ExtractTraceContext 从 HTTP headers 中提取 TraceContext
// 用于接收外部请求的 Trace 上下文
func ExtractTraceContext(ctx context.Context, headers map[string]string) context.Context {
extractor := otel.GetTextMapPropagator()
return extractor.Extract(ctx, propagation.MapCarrier(headers))
}
// GetCurrentTraceID 获取当前 Context 的 TraceID
// 用于在日志中输出 TraceID，方便关联
func GetCurrentTraceID(ctx context.Context) string {
spanCtx := trace.SpanContextFromContext(ctx)
if spanCtx.HasTraceID() {
return spanCtx.TraceID().String()
}
return ""
}
4.6 完整工作流示例（main.go）
package main
import (
"context"
"fmt"
"log"
"telemetry"
)
func main() {
ctx := context.Background()
// 1. 初始化 OTel
provider, err := telemetry.InitTracer(ctx, telemetry.Config{
ServiceName: "ai-order-agent",
ServiceVersion: "1.0.0",
OTLPEndpoint: "localhost:4317",
Env: "production",
})
if err != nil {
log.Fatalf("初始化 OTel 失败: %v", err)
}
// 2. 创建追踪器
llmTracer, err := telemetry.NewLLMTracer(provider)
if err != nil {
log.Fatalf("创建 LLM 追踪器失败: %v", err)
}
toolTracer, err := telemetry.NewToolTracer(provider)
if err != nil {
log.Fatalf("创建工具追踪器失败: %v", err)
}
// 3. 执行 AI Agent 工作流
runOrderAgent(ctx, llmTracer, toolTracer)
}
// runOrderAgent 一个典型的 AI Agent 工作流
// "用户查询订单 → 意图识别 → 数据库查询 → 总结回复"
func runOrderAgent(ctx context.Context, llmTracer *telemetry.LLMTracer, toolTracer *telemetry.ToolTracer) {
userQuery := "查一下上个月订单号 A12345 的物流状态"
// Step 1: 意图识别
intentResult, err := llmTracer.TracedLLMCall(ctx, telemetry.LLMCallInput{
Model: "gpt-4o",
Prompt: fmt.Sprintf("识别用户意图：%s", userQuery),
StepName: "intent_classification",
Temperature: 0.1,
MaxTokens: 256,
}, classifyIntent)
if err != nil {
log.Printf("意图识别失败: %v", err)
return
}
fmt.Printf("意图: %s, Token消耗: %d\n", intentResult.Completion, intentResult.TotalTokens)
// Step 2: 工具调用（查询数据库）
orderResult, err := toolTracer.TracedToolCall(ctx, telemetry.ToolInput{
Name: "query_order",
Provider: "mcp",
Parameters: map[string]any{
"order_id": extractOrderID(intentResult.Completion),
},
}, queryOrderDB)
if err != nil {
log.Printf("订单查询失败: %v", err)
return
}
fmt.Printf("订单数据: %v\n", orderResult.Result)
// Step 3: 生成回复
summaryResult, err := llmTracer.TracedLLMCall(ctx, telemetry.LLMCallInput{
Model: "gpt-4o",
Prompt: fmt.Sprintf("根据以下订单信息生成回复：%v", orderResult.Result),
StepName: "generate_response",
Temperature: 0.3,
MaxTokens: 512,
}, generateSummary)
if err != nil {
log.Printf("回复生成失败: %v", err)
return
}
fmt.Printf("最终回复: %s\n", summaryResult.Completion)
fmt.Printf("总Token消耗: %d, 延迟: %.2fms\n",
intentResult.TotalTokens+summaryResult.TotalTokens,
summaryResult.LatencyMs)
}
// 占位函数，实际需要接入真实的 LLM API
func classifyIntent(ctx context.Context, req telemetry.LLMCallInput) (telemetry.LLMCallOutput, error) {
return telemetry.LLMCallOutput{
Completion: "query_order_status",
PromptTokens: 120,
CompletionTokens: 8,
TotalTokens: 128,
Model: req.Model,
FinishReason: "stop",
}, nil
}
func extractOrderID(intent string) string {
return "A12345"
}
func queryOrderDB(ctx context.Context, req telemetry.ToolInput) (telemetry.ToolOutput, error) {
return telemetry.ToolOutput{
Result: map[string]any{
"order_id": "A12345",
"status": "配送中",
"timestamp": "2026-07-09 14:30:00",
},
LatencyMs: 45.2,
FromCache: false,
}, nil
}
func generateSummary(ctx context.Context, req telemetry.LLMCallInput) (telemetry.LLMCallOutput, error) {
return telemetry.LLMCallOutput{
Completion: "您的订单 A12345 正在配送中，预计明天送达。",
PromptTokens: 380,
CompletionTokens: 62,
TotalTokens: 442,
Model: req.Model,
FinishReason: "stop",
}, nil
}
4.7 后端可视化：Docker Compose 一键部署
# backend/docker-compose.yml
version: '3.8'
services:
# Jaeger UI + Collector
jaeger:
image: jaegertracing/all-in-one:1.58
ports:
- "16686:16686" # Jaeger UI
- "4317:4317" # OTLP gRPC
- "4318:4318" # OTLP HTTP
environment:
- COLLECTOR_OTLP_ENABLED=true
# Prometheus（接收 OTel Metrics）
prometheus:
image: prom/prometheus:v2.53.0
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
extra_hosts:
- "host.docker.internal:host-gateway"
# Grafana（可视化）
grafana:
image: grafana/grafana:11.3.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- ./grafana-datasources.yml:/etc/grafana/provisioning/datasources/datasources.yml
depends_on:
- prometheus
- jaeger
Grafana Dashboard 配置（关键面板）：
┌─────────────────────────────────────────────────────┐
│ AI Agent 可观测性 Dashboard │
├──────────────┬───────────────┬────────────────────┤
│ Token消耗速率 │ LLM错误率 │ 平均响应延迟 │
│ [折线图] │ [饼图] │ [热力图] │
├──────────────┴───────────────┴────────────────────┤
│ 工作流 Trace（Jaeger） │
│ [llm.intent_classification] → [tool.query_order] │
│ → [llm.generate_response] │
├────────────────────────────────────────────────────┤
│ 按 Step 分组的 Token 消耗排名（Table） │
├────────────────────────────────────────────────────┤
│ 上下文窗口利用率（所有 Span 的 llm.context_util） │
└────────────────────────────────────────────────────┘
五、MCP 协议 + OTel：liatrio-labs/otel-instrumentation-mcp 深度解析
前面提到 MCP 协议是 AI Agent 的"USB 接口"，负责标准化模型与工具之间的交互。liatrio-labs 的 otel-instrumentation-mcp 项目，则是为这个 USB 接口装上了 X 光机——让每次 MCP 握手、每个工具调用的内部细节都变成可观测的数据。
5.1 为什么 MCP 协议需要可观测性
MCP 协议的工作流程是：
Host (AI App) ←→ MCP Client ←→ MCP Server ←→ Tools (Filesystem/DB/API)
在没有可观测性的情况下，你只知道"MCP Server 响应了 200"。但你不知道：
- 这次请求调用了哪些工具？
- 每个工具的参数和返回值是什么？
- 工具调用的耗时分布如何？
- 某个工具是否返回了异常数据？
5.2 instrumentClient 和 instrumentServer 的原理
该项目提供两个核心函数：
# instrumentClient: 包装 MCP 客户端，自动追踪所有出站请求
from otel_instrumentation_mcp import instrumentClient
client = MCPClient(transport="stdio")
# 包装后，所有 MCP 调用自动生成 OTel Span
instrumented_client = instrumentClient(client)
# instrumentServer: 包装 MCP 服务器，自动追踪所有入站请求
from otel_instrumentation_mcp import instrumentServer
server = MCPServer(transport="stdio")
instrumented_server = instrumentServer(server)
底层原理：利用 MCP 协议的 Client 和 Server 标准接口，通过装饰器模式拦截所有方法调用，自动注入 Trace Context：
# 简化的原理示意（实际实现更复杂）
def instrumentClient(client):
original_request = client.request
def traced_request(method, params):
with tracer.start_as_current_span(f"mcp.{method}") as span:
span.set_attribute("mcp.method", method)
span.set_attribute("mcp.params_keys", list(params.keys()))
try:
result = original_request(method, params)
span.set_attribute("mcp.response_keys", list(result.keys()))
return result
except Exception as e:
span.record_exception(e)
raise
client.request = traced_request
return client
追踪生成的 Span 树结构：
mcp.request (root span)
├── mcp.initialize # 握手协商
├── mcp.tools/list # 发现可用工具
├── mcp.tools/call # 工具调用
│ ├── mcp.tools/call.web_search
│ │ └── http.request (外部 API 调用)
│ └── mcp.tools/call.database_query
│ └── db.query
└── mcp.resources/list # 资源发现
5.3 将 MCP Trace 接入生产环境
// Go 版本的 MCP OTel 插桩（基于 liatrio-labs 的 Go 实现思路）
package main
import (
"go.opentelemetry.io/contrib/instrumentation/mcp"
)
// 创建一个 MCP Server 实例
server := mcp.NewServer("order-tools", mcp.ServerOptions{
Tools: []mcp.Tool{webSearchTool, dbQueryTool},
})
// OTel 插桩后，所有工具调用自动生成带完整属性的 Span
instrumented := mcp.WithOpenTelemetry(server,
mcp.WithSpanFormatter(func(method string, params any) []attribute.KeyValue {
return []attribute.KeyValue{
attribute.String("mcp.method", method),
attribute.String("mcp.tool_category", categorizeTool(method)),
attribute.Int("mcp.param_count", countParams(params)),
}
}),
)
六、OpenLIT：专为零门槛 AI 可观测性而生的平台
如果说 OTel 是通用标准，那么 OpenLIT 就是 AI 可观测性的"一键安装版"。这个开源项目在 2026 年已经支持 50+ LLM 提供商、主流 Agent 框架和 GPU 监控。
6.1 OpenLIT 的核心能力
# 一行命令接入 OpenLIT
pip install openlit
# 在代码中初始化
import openlit
openlit.init(
source="ai-agent-service",
otel_endpoint="http://localhost:4318",
)
OpenLIT 自动追踪的内容：
- LLM 调用：Prompt/Completion Token、模型、延迟、成本（带定价信息）
- 工具调用：参数、耗时、缓存命中
- RAG 流程：向量检索延迟、召回率、上下文命中率
- GPU 指标：显存使用、计算利用率、温度（通过 OpenTelemetry GPU Collector）
- Coding Agent：Cursor、Claude Code 等 IDE 的使用量统计
6.2 Guardrails：AI Agent 的运行时安全网
OpenLIT 还提供 Guardrails 功能——在 Agent 运行过程中检测异常行为：
# 配置 Guardrail：检测 Token 消耗异常
openlit.guardrail(
metric="llm.tokens.total",
condition="rate_of_change > 3",
action="alert",
# 检测到 Token 消耗速率突然增长 3 倍（可能的攻击或死循环）
)
# 配置 Guardrail：检测响应延迟异常
openlit.guardrail(
metric="llm.latency.ms",
condition="p99 > 10000", # P99 延迟超过 10 秒
action="circuit_break",
# 自动熔断，防止雪崩
)
# 配置 Guardrail：检测内容质量（基于 Embedding 相似度）
openlit.guardrail(
metric="llm.response.diversity",
condition="similarity_to_previous > 0.95",
action="log",
# 检测 Agent 是否在重复输出同一内容（可能的死循环）
)
6.3 Evaluations：让 AI 输出质量可量化
OpenLIT 的另一大杀手锏是将 LLM Evals（大模型评估）集成到可观测性流水线里：
# 定义评估函数
def evaluate_response_quality(output: str, expected: str) -> float:
"""BLEU/ROUGE 风格的评估"""
return compute_similarity(output, expected)
# 异步评估：在回复产生后自动运行
openlit.add_evaluator(
name="response_quality",
evaluator=evaluate_response_quality,
sample_rate=0.1, # 抽样 10%，降低计算开销
tags=["qa", "production"],
)
评估结果会出现在同一个 Grafana Dashboard 里，让性能指标和输出质量指标在同一视图下对比分析。
七、GPT-4o 在 AI Agent 可观测性中的独特价值
当我们把 AI Agent 的可观测性数据（Traces + Metrics）喂给 GPT-4o 做分析时，发现了一个惊人的应用场景：让 AI 做根因分析。
传统的告警系统告诉你："第三步 LLM 调用的 P99 延迟超过 5 秒"。你需要人工排查。而基于可观测性数据的 AI 分析是：
输入：最近 1 小时的所有失败 Trace + 延迟异常的 Span 详情
模型：GPT-4o (thinking enabled)
输出：
"检测到 23 个延迟异常 Span，主要集中在 tool.database_query 步骤。
根本原因：数据库连接池耗尽（活跃连接数达到上限 100，
等待队列堆积 15 个请求）。
建议：将 max_connections 从 100 提升到 200，
或启用连接池预热（warmup）策略。
置信度：87%"
这个能力来自于 AI Agent 可观测性数据的完整性。当你有了完整的 Trace 树（包含每一步的输入、输出、Token 消耗、错误信息），AI 就能像一个有经验的 SRE 一样做根因分析，而且它能看到比人类更多的关联模式。
八、架构最佳实践：从单 Agent 到多 Agent 的可观测性演进
8.1 单 Agent 工作流的追踪架构
┌─────────────────────────────────────────────────────────────┐
│ AI Agent Service │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ LLMTracer │ │ ToolTracer │ │ EvalTracer │ │
│ │ (Span/Metric│ │ (Span/Metric│ │ (异步评估) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └────────┬─────────┴───────────────┘ │
│ ▼ │
│ ┌───────────────┐ │
│ │ OTLP Exporter│ │
│ └───────┬───────┘ │
└─────────────────┼───────────────────────────────────────────┘
▼
┌─────────────────────┐
│ OpenTelemetry │
│ Collector │
│ (聚合/采样/路由) │
└───────┬───────┬─────┘
▼ ▼ ▼
Jaeger Prometheus Loki
↓ ↓ ↓
Grafana Dashboard
8.2 多 Agent 协作的追踪架构
当多个 Agent 协同工作时（主 Agent 分配任务 → 子 Agent 执行 → 结果汇总），Trace Context 需要跨 Agent 传播：
Root Trace (用户请求)
├── Agent-Orchestrator (主 Agent)
│ ├── llm.task_decomposition (任务分解)
│ └── [context propagates to child agents]
│ ├── Agent-Researcher (研究 Agent)
│ │ ├── llm.search_strategy (搜索策略)
│ │ └── tool.web_search × 5
│ └── Agent-Reporter (报告 Agent)
│ ├── llm.synthesize (综合分析)
│ └── tool.chart_generator (图表生成)
└── llm.final_response (最终回复)
关键实现：在 Agent 间传递消息时，携带 Trace Context：
# 主 Agent 调度子 Agent 时，注入 Trace Context
async def delegate_to_agent(agent_id, task, parent_ctx):
headers = {}
otel.propagator.inject(parent_ctx, headers) # 把 TraceID 注入 headers
# 通过消息队列发送任务，headers 随消息传播
await message_queue.send({
"agent_id": agent_id,
"task": task,
"trace_context": headers,
})
# 子 Agent 接收任务时，提取 Trace Context
async def agent_main(agent_id):
msg = await message_queue.receive()
ctx = otel.propagator.extract(msg.trace_context) # 恢复 TraceID
with tracer.start_as_current_span("agent_task", context=ctx):
await process_task(msg.task, ctx)
这样，在 Grafana 的 Trace 视图中，你能看到从用户请求到所有子 Agent 的完整调用链，每一步的 Token 消耗、延迟、工具调用都一目了然。
九、避坑指南：AI Agent 可观测性的五大常见错误
错误 1：在 Prompt 里记录 Token 内容
最常见的错误。开发者为了方便排查，在 Span 属性里记录了完整的 Prompt 和 Completion：
// ❌ 危险：Prompt 可能包含敏感信息，会被写入存储后端
span.SetAttributes(
attribute.String("llm.prompt", fullPrompt), // 不要这样做！
)
// ✅ 正确：只记录哈希值，用于关联，不泄露内容
span.SetAttributes(
attribute.String("llm.prompt_hash", sha256Hash(fullPrompt)[:16]),
attribute.Int("llm.prompt_length", len(fullPrompt)),
)
错误 2：采样率一刀切
很多团队在生产环境把采样率设为 1%，结果在排查问题时发现根本没有数据。正确的做法是按事件类型设置差异化采样：
- 错误和异常：100% 采样
- AI 调用（
llm.*属性）：10-50% 采样 - 工具调用：10-30% 采样
- 普通 HTTP/DB 调用：1% 采样
错误 3：没有追踪 Token 成本
每个团队在接入 AI 可观测性后都会问："为什么我们的 AI 账单比预期高 3 倍？"——因为没有追踪 llm.prompt_tokens 和 llm.completion_tokens。这是 AI 应用成本失控的根本原因。
错误 4：Metrics 和 Traces 割裂
很多团队只追踪 Traces 不追踪 Metrics。但 Traces 告诉你"哪一步慢了"，Metrics 告诉你"整体趋势是否恶化"。两者必须结合：
- Metrics：回答"P99 延迟最近一周的趋势如何？"
- Traces：回答"为什么今天的 P99 延迟突然飙升？"
错误 5：忽视 Context Overflow
当 Agent 的上下文窗口接近上限（通常是 128k 或 200k Token）时，性能会急剧下降。你需要在 OTel 中设置专门的告警：
span.SetAttributes(
attribute.Float64("llm.context_utilization", float64(totalTokens)/128000.0),
)
// 当 context_utilization > 0.85 时触发告警，建议截断或压缩上下文
十、总结：可观测性是 AI Agent 工程化的分水岭
2026 年，AI Agent 从"能跑"到"能运维"，中间隔着一套完整的可观测性体系。
为什么可观测性如此关键？ 因为 AI Agent 的行为具有内在的不确定性——同一个输入，今天可能返回正确答案，明天可能因为上下文变化而返回完全不同的答案。没有可观测性，你根本无法区分"Agent 变笨了"和"输入数据变了"。
从技术选型角度：
- 如果你只需要基础的 LLM 追踪：直接用 OpenLIT，一行代码接入
- 如果你需要深度定制（MCP 工具追踪、自定义 Guardrails）：用 OpenTelemetry SDK 自己做
- 如果你需要把 AI 追踪和微服务追踪统一管理：必须用 OpenTelemetry
从投入产出比角度：
可观测性的投入大约是 AI 应用开发工作量的 10-15%，但它能帮你：
- 降低 80% 的 AI 账单（通过 Token 监控发现浪费）
- 缩短 90% 的故障定位时间（从猜测到精准定位）
- 发现隐藏的 Agent 行为问题（幻觉、循环、上下文爆炸）
这可能是 2026 年 AI 工程化领域投入产出比最高的投资。
本文代码基于 OpenTelemetry Go SDK 1.35+，Python 示例基于 OpenLIT 2.x。生产环境部署请参考各自官方文档。