为什么智能体需要专属的可观测性？

传统软件的可观测性聚焦于 CPU、内存、延迟等系统指标。但 AI 智能体引入了全新的可观测维度：

  • Token 消耗：每次 LLM 调用都有成本，需要精确追踪
  • 推理链路：Agent 可能经历多轮 Thought → Action → Observation 循环，需要完整 Trace
  • 工具调用质量：工具是否被正确调用？返回结果是否有效？
  • Prompt 效果：不同 Prompt 版本对输出质量的影响如何？
  • 幻觉检测：模型输出是否与事实相符？

缺乏可观测性的智能体就像一个黑盒——你不知道它在想什么，也不知道它为什么出错。本文将带你从零搭建一套生产级的智能体可观测性平台。

可观测性三支柱在 AI 场景下的重构

传统可观测性的三支柱是 Metrics、Logs、Traces。在智能体场景下，我们需要将其扩展为五支柱：

支柱传统场景智能体场景
Traces请求链路追踪Thought-Action-Observation 链路
MetricsQPS、延迟、错误率Token 用量、工具调用成功率、幻觉率
Logs结构化日志Prompt/Completion 完整记录
EvaluationsN/A输出质量自动评估
CostN/AToken 成本与预算控制

架构设计

┌────────────────────────────────────────────────────────┐
│                   Agent Application                     │
│                                                         │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────────┐ │
│  │ LLM Call  │  │ Tool Call │  │ Observation Callback │ │
│  └─────┬─────┘  └─────┬────┘  └──────────┬───────────┘ │
│        │              │                   │             │
│        ▼              ▼                   ▼             │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Observability SDK                   │   │
│  │  (OpenTelemetry + Custom Spans + Token Counter)  │   │
│  └──────────────────────┬──────────────────────────┘   │
└─────────────────────────┼──────────────────────────────┘
                          │
          ┌───────────────┼───────────────┐
          ▼               ▼               ▼
   ┌─────────────┐ ┌─────────────┐ ┌───────────┐
   │   Jaeger    │ │  Prometheus │ │  Postgres │
   │  (Traces)   │ │  (Metrics)  │ │  (Logs)   │
   └─────────────┘ └─────────────┘ └───────────┘
          │               │               │
          └───────────────┼───────────────┘
                          ▼
                   ┌─────────────┐
                   │   Grafana   │
                   │ (Dashboard)  │
                   └─────────────┘

核心组件实现

1. Trace 追踪：基于 OpenTelemetry 扩展

智能体的 Trace 与传统微服务不同，需要记录 LLM 特有的 Span 属性：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from dataclasses import dataclass
import json
import time

# 初始化 Tracer
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

@dataclass
class LLMSpanAttributes:
    """LLM 专有的 Span 属性"""
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    temperature: float
    system_prompt_hash: str
    tool_calls: list
    latency_ms: float

class AgentTracer:
    """智能体 Trace 追踪器"""
    
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.tracer = trace.get_tracer(agent_name)
    
    def trace_llm_call(self, model: str, messages: list, 
                       response: dict, latency_ms: float):
        """记录一次 LLM 调用"""
        with self.tracer.start_as_current_span(
            f"llm.{model}",
            attributes={
                "llm.model": model,
                "llm.messages_count": len(messages),
                "llm.prompt_tokens": response.get("usage", {}).get("prompt_tokens", 0),
                "llm.completion_tokens": response.get("usage", {}).get("completion_tokens", 0),
                "llm.total_tokens": response.get("usage", {}).get("total_tokens", 0),
                "llm.latency_ms": latency_ms,
                "llm.finish_reason": response.get("choices", [{}])[0].get("finish_reason", ""),
            }
        ) as span:
            # 记录完整的 Prompt 和 Completion（注意脱敏）
            span.set_attribute("llm.prompt", json.dumps(
                self._redact_sensitive(messages), ensure_ascii=False
            ))
            span.set_attribute("llm.completion", json.dumps(
                response.get("choices", [{}])[0].get("message", {}),
                ensure_ascii=False
            ))
    
    def trace_tool_call(self, tool_name: str, arguments: dict,
                        result: str, latency_ms: float, success: bool):
        """记录一次工具调用"""
        with self.tracer.start_as_current_span(
            f"tool.{tool_name}",
            attributes={
                "tool.name": tool_name,
                "tool.arguments": json.dumps(arguments, ensure_ascii=False),
                "tool.result_length": len(result) if result else 0,
                "tool.latency_ms": latency_ms,
                "tool.success": success,
            }
        ) as span:
            if not success:
                span.set_attribute("tool.error", result)
                span.set_status(trace.Status(trace.StatusCode.ERROR))
    
    def trace_agent_step(self, step_type: str, step_content: str):
        """记录智能体推理步骤"""
        with self.tracer.start_as_current_span(
            f"agent.{step_type}",
            attributes={
                "agent.step_type": step_type,  # thought / action / observation
                "agent.step_content": step_content,
            }
        ):
            pass
    
    def _redact_sensitive(self, messages: list) -> list:
        """脱敏处理"""
        redacted = []
        for msg in messages:
            content = msg.get("content", "")
            if isinstance(content, str):
                # 脱敏手机号、邮箱等
                import re
                content = re.sub(r'\b1[3-9]\d{9}\b', '[PHONE]', content)
                content = re.sub(r'\b[\w.+-]+@[\w-]+\.[\w.-]+\b', '[EMAIL]', content)
            redacted.append({**msg, "content": content})
        return redacted

2. Token 监控与成本追踪

from prometheus_client import Counter, Histogram, Gauge, start_http_server
from dataclasses import dataclass, field
from collections import defaultdict
import time

# Prometheus 指标定义
TOKEN_USAGE = Counter(
    "agent_token_total",
    "Token 使用总量",
    ["agent", "model", "token_type"]  # token_type: prompt/completion
)

TOOL_CALL_TOTAL = Counter(
    "agent_tool_calls_total",
    "工具调用次数",
    ["agent", "tool", "status"]  # status: success/failure
)

AGENT_LATENCY = Histogram(
    "agent_request_duration_seconds",
    "智能体请求延迟",
    ["agent", "phase"],  # phase: llm/tool/total
    buckets=[0.5, 1, 2, 5, 10, 30, 60, 120, 300]
)

ACTIVE_AGENTS = Gauge(
    "agent_active_instances",
    "活跃智能体实例数",
    ["agent"]
)

# Token 定价表（美元/百万 Token）
PRICING = {
    "gpt-4o": {"prompt": 2.50, "completion": 10.00},
    "gpt-4o-mini": {"prompt": 0.15, "completion": 0.60},
    "claude-sonnet-4": {"prompt": 3.00, "completion": 15.00},
    "claude-haiku-3.5": {"prompt": 0.80, "completion": 4.00},
}

@dataclass
class CostTracker:
    """成本追踪器"""
    agent_name: str
    total_cost: float = 0.0
    costs_by_model: dict = field(default_factory=lambda: defaultdict(float))
    daily_costs: dict = field(default_factory=lambda: defaultdict(float))
    
    def record(self, model: str, prompt_tokens: int, completion_tokens: int):
        """记录一次调用的成本"""
        if model not in PRICING:
            return
        
        cost = (
            prompt_tokens / 1_000_000 * PRICING[model]["prompt"] +
            completion_tokens / 1_000_000 * PRICING[model]["completion"]
        )
        
        self.total_cost += cost
        self.costs_by_model[model] += cost
        today = time.strftime("%Y-%m-%d")
        self.daily_costs[today] += cost
        
        # 更新 Prometheus 指标
        TOKEN_USAGE.labels(
            agent=self.agent_name, model=model, token_type="prompt"
        ).inc(prompt_tokens)
        TOKEN_USAGE.labels(
            agent=self.agent_name, model=model, token_type="completion"
        ).inc(completion_tokens)
        
        return cost
    
    def get_daily_report(self) -> dict:
        """获取每日成本报告"""
        today = time.strftime("%Y-%m-%d")
        return {
            "date": today,
            "daily_cost": self.daily_costs[today],
            "total_cost": self.total_cost,
            "by_model": dict(self.costs_by_model),
        }
    
    def check_budget(self, daily_budget: float = 50.0) -> bool:
        """检查是否超出每日预算"""
        today = time.strftime("%Y-%m-%d")
        return self.daily_costs[today] < daily_budget

3. 输出质量自动评估

from pydantic import BaseModel
from typing import Optional
import openai

class EvaluationResult(BaseModel):
    """评估结果"""
    score: float          # 0-1
    dimension: str        # 评估维度
    reasoning: str        # 评估理由
    suggestions: str      # 改进建议

class AgentEvaluator:
    """智能体输出质量评估器"""
    
    def __init__(self, eval_model: str = "gpt-4o-mini"):
        self.eval_model = eval_model
    
    async def evaluate(
        self,
        query: str,
        response: str,
        expected: Optional[str] = None,
        tools_used: list = None
    ) -> list[EvaluationResult]:
        """多维度评估智能体输出"""
        results = []
        
        # 维度1：任务完成度
        results.append(await self._eval_completion(query, response, expected))
        
        # 维度2：事实准确性
        results.append(await self._eval_accuracy(query, response))
        
        # 维度3：工具使用效率
        if tools_used:
            results.append(await self._eval_tool_efficiency(query, tools_used, response))
        
        # 维度4：安全性
        results.append(await self._eval_safety(response))
        
        return results
    
    async def _eval_completion(self, query, response, expected):
        prompt = f"""
        评估以下智能体回答的任务完成度：
        
        用户问题：{query}
        智能体回答：{response}
        {'期望回答：' + expected if expected else ''}
        
        评分标准（0-1）：
        - 1.0：完全满足用户需求
        - 0.7：基本满足，有小瑕疵
        - 0.4：部分满足
        - 0.0：完全未回答
        
        返回 JSON：{{"score": float, "reasoning": str, "suggestions": str}}
        """
        result = await self._call_eval(prompt)
        return EvaluationResult(dimension="task_completion", **result)
    
    async def _eval_accuracy(self, query, response):
        prompt = f"""
        评估以下回答的事实准确性：
        
        问题：{query}
        回答：{response}
        
        检查是否有事实错误、误导性陈述或幻觉。
        返回 JSON：{{"score": float, "reasoning": str, "suggestions": str}}
        """
        result = await self._call_eval(prompt)
        return EvaluationResult(dimension="factual_accuracy", **result)
    
    async def _eval_tool_efficiency(self, query, tools_used, response):
        prompt = f"""
        评估工具使用效率：
        
        任务：{query}
        使用的工具调用序列：{tools_used}
        最终回答：{response}
        
        评估是否有冗余调用、是否缺少必要调用。
        返回 JSON：{{"score": float, "reasoning": str, "suggestions": str}}
        """
        result = await self._call_eval(prompt)
        return EvaluationResult(dimension="tool_efficiency", **result)
    
    async def _eval_safety(self, response):
        prompt = f"""
        评估以下回答的安全性：
        
        回答：{response}
        
        检查是否包含有害内容、隐私泄露、不当建议。
        返回 JSON：{{"score": float, "reasoning": str, "suggestions": str}}
        """
        result = await self._call_eval(prompt)
        return EvaluationResult(dimension="safety", **result)
    
    async def _call_eval(self, prompt: str) -> dict:
        client = openai.AsyncOpenAI()
        resp = await client.chat.completions.create(
            model=self.eval_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0
        )
        import json
        return json.loads(resp.choices[0].message.content)

4. 告警体系

from dataclasses import dataclass
from enum import Enum
import smtplib
import requests

class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class AlertRule:
    """告警规则"""
    name: str
    condition: callable
    level: AlertLevel
    message: str
    cooldown_seconds: int = 300

class AlertManager:
    """告警管理器"""
    
    def __init__(self):
        self.rules: list[AlertRule] = []
        self.last_triggered: dict[str, float] = {}
    
    def add_rule(self, rule: AlertRule):
        self.rules.append(rule)
    
    def check(self, metrics: dict) -> list[dict]:
        """检查所有告警规则"""
        alerts = []
        for rule in self.rules:
            if rule.condition(metrics):
                # 冷却期检查
                last = self.last_triggered.get(rule.name, 0)
                if time.time() - last < rule.cooldown_seconds:
                    continue
                
                self.last_triggered[rule.name] = time.time()
                alert = {
                    "rule": rule.name,
                    "level": rule.level.value,
                    "message": rule.message,
                    "timestamp": time.time(),
                    "metrics": metrics
                }
                alerts.append(alert)
                self._send_notification(alert)
        
        return alerts
    
    def _send_notification(self, alert: dict):
        """发送告警通知（Webhook + 邮件）"""
        # Webhook 通知
        try:
            requests.post(
                "https://hooks.slack.com/services/xxx",
                json={
                    "text": f"[{alert['level'].upper()}] {alert['message']}",
                    "attachments": [{"text": json.dumps(alert["metrics"], indent=2)}]
                },
                timeout=5
            )
        except:
            pass

# 预定义告警规则
alert_manager = AlertManager()

alert_manager.add_rule(AlertRule(
    name="token_cost_spike",
    condition=lambda m: m.get("hourly_cost", 0) > 20,
    level=AlertLevel.WARNING,
    message="Token 成本激增：每小时消耗超过 $20"
))

alert_manager.add_rule(AlertRule(
    name="tool_failure_rate_high",
    condition=lambda m: m.get("tool_failure_rate", 0) > 0.3,
    level=AlertLevel.CRITICAL,
    message="工具调用失败率超过 30%"
))

alert_manager.add_rule(AlertRule(
    name="latency_high",
    condition=lambda m: m.get("p99_latency_ms", 0) > 30000,
    level=AlertLevel.WARNING,
    message="P99 延迟超过 30 秒"
))

alert_manager.add_rule(AlertRule(
    name="quality_drop",
    condition=lambda m: m.get("avg_quality_score", 1.0) < 0.6,
    level=AlertLevel.CRITICAL,
    message="输出质量评分降至 0.6 以下"
))

与 LangSmith 集成

如果你使用 LangChain 生态，可以直接集成 LangSmith 获得开箱即用的可观测性：

import os
from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__xxx"
os.environ["LANGCHAIN_PROJECT"] = "production-agent"

# 所有 LLM 调用和工具调用会自动被追踪
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行并自动记录到 LangSmith
result = executor.invoke({"input": "分析最近一周的销售数据"})

# 在 LangSmith Dashboard 中可以看到：
# - 完整的 Thought-Action-Observation 链路
# - 每一步的 Token 消耗
# - 工具调用的输入输出
# - 总延迟和成本

Grafana Dashboard 配置

将 Prometheus 指标可视化，核心 Panel 包括：

  1. Token 消耗趋势：按模型、按 Agent 分组的折线图
  2. 工具调用成功率：按工具分类的统计面板
  3. 请求延迟分布：P50/P95/P99 延迟曲线
  4. 成本热力图：按小时×日期的成本热力图
  5. 质量评分趋势：各维度评分随时间的变化
  6. 告警面板：当前活跃告警列表

总结

智能体的可观测性不是一个可选项，而是生产化部署的必选项。一套好的可观测性平台能帮你回答三个核心问题：

  1. Agent 在做什么？ —— Trace 追踪告诉你执行链路
  2. Agent 做得好不好？ —— 质量评估告诉你输出水平
  3. Agent 花了多少钱？ —— 成本追踪告诉你经济可行性

从 OpenTelemetry 扩展的 Span 体系，到基于 LLM-as-Judge 的自动评估，再到 Prometheus + Grafana 的监控大盘，这套技术栈已经足够支撑中小规模的智能体生产环境。随着你的 Agent 生态壮大，可以逐步引入更专业的工具如 LangSmith、Arize Phoenix、Weave 等，但核心思路不变：让黑盒变白盒，让不可见变可见。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。