企业级 Agent 平台的可观测性

原创灵阙教研团队

S 精选进阶教程 | 约 10 分钟阅读更新于 2026-02-28

AI 导读

企业级 Agent 平台的可观测性 Logging、Tracing、Metrics：构建 Agent 系统的全栈可观测性体系为什么 Agent 可观测性不同于传统系统传统微服务的可观测性关注"请求在服务间如何流转"。Agent 系统的可观测性则需要额外回答： Agent 为什么做了这个决策？（推理链可追溯）工具调用的效果如何？（工具级别的成功率和延迟） Token...

企业级 Agent 平台的可观测性

Logging、Tracing、Metrics：构建 Agent 系统的全栈可观测性体系

为什么 Agent 可观测性不同于传统系统

传统微服务的可观测性关注"请求在服务间如何流转"。Agent 系统的可观测性则需要额外回答：

Agent 为什么做了这个决策？（推理链可追溯）
工具调用的效果如何？（工具级别的成功率和延迟）
Token 消耗了多少？成本是多少？（LLM 特有的成本可观测性）
多 Agent 之间如何协作？（Agent 间通信可追踪）

传统微服务可观测性：
  Request ──→ Service A ──→ Service B ──→ Response
  关注：延迟、错误率、吞吐量

Agent 可观测性（额外维度）：
  Task ──→ Agent ──→ [Think → Tool Call → Observe]×N ──→ Result
  关注：推理质量、工具效率、Token成本、决策链路、Agent协作

可观测性三支柱在 Agent 系统中的映射

┌──────────────────────────────────────────────────────────────────┐
│                    Agent 可观测性三支柱                           │
├─────────────────┬─────────────────┬──────────────────────────────┤
│     Logging     │     Tracing     │      Metrics                 │
│     日志        │     链路追踪     │      指标                    │
├─────────────────┼─────────────────┼──────────────────────────────┤
│ 结构化事件记录   │ 端到端执行链路   │ 聚合统计数据                 │
│ - 推理过程       │ - 任务→Agent    │ - 成功率                     │
│ - 工具调用详情   │ - Agent→Tool    │ - 延迟分布                   │
│ - 错误堆栈       │ - Agent→Agent   │ - Token消耗                  │
│ - 决策依据       │ - 多步骤关联     │ - 成本                      │
├─────────────────┼─────────────────┼──────────────────────────────┤
│ 用途：           │ 用途：           │ 用途：                      │
│ 调试单个执行     │ 理解全链路       │ 监控系统健康                 │
│ 审计合规         │ 定位瓶颈         │ 趋势分析                    │
│ 知识沉淀         │ 根因分析         │ 告警触发                    │
└─────────────────┴─────────────────┴──────────────────────────────┘

一、结构化日志（Structured Logging）

日志事件类型

class AgentLogEvent:
    """Agent 日志事件的标准结构"""

    # 事件类型枚举
    class EventType(Enum):
        TASK_START = "task.start"
        TASK_END = "task.end"
        AGENT_THINK = "agent.think"
        AGENT_DECIDE = "agent.decide"
        TOOL_CALL_START = "tool.call.start"
        TOOL_CALL_END = "tool.call.end"
        TOOL_CALL_ERROR = "tool.call.error"
        LLM_REQUEST = "llm.request"
        LLM_RESPONSE = "llm.response"
        MEMORY_READ = "memory.read"
        MEMORY_WRITE = "memory.write"
        HITL_REQUEST = "hitl.request"
        HITL_RESPONSE = "hitl.response"
        ESCALATION = "escalation"
        HANDOFF = "agent.handoff"
        SECURITY_EVENT = "security.event"

    def __init__(self,
                 event_type: EventType,
                 trace_id: str,
                 span_id: str,
                 agent_id: str,
                 data: dict,
                 timestamp: datetime = None):
        self.event_type = event_type
        self.trace_id = trace_id
        self.span_id = span_id
        self.agent_id = agent_id
        self.data = data
        self.timestamp = timestamp or datetime.now()
        self.level = self._infer_level()

    def to_dict(self) -> dict:
        return {
            "timestamp": self.timestamp.isoformat(),
            "level": self.level,
            "event": self.event_type.value,
            "trace_id": self.trace_id,
            "span_id": self.span_id,
            "agent_id": self.agent_id,
            **self.data
        }

日志输出示例

{"timestamp":"2026-02-28T10:15:30.123Z","level":"INFO","event":"task.start","trace_id":"tr_abc123","agent_id":"supervisor","task":"分析Q4财报","user_id":"u_789"}
{"timestamp":"2026-02-28T10:15:30.456Z","level":"DEBUG","event":"agent.think","trace_id":"tr_abc123","agent_id":"supervisor","reasoning":"需要先获取财报数据，再进行分析","confidence":0.92}
{"timestamp":"2026-02-28T10:15:30.789Z","level":"INFO","event":"tool.call.start","trace_id":"tr_abc123","span_id":"sp_001","agent_id":"supervisor","tool":"file_read","params":{"path":"/data/q4_report.xlsx"}}
{"timestamp":"2026-02-28T10:15:31.234Z","level":"INFO","event":"tool.call.end","trace_id":"tr_abc123","span_id":"sp_001","agent_id":"supervisor","tool":"file_read","status":"success","duration_ms":445,"result_size_bytes":15234}
{"timestamp":"2026-02-28T10:15:31.567Z","level":"INFO","event":"llm.request","trace_id":"tr_abc123","span_id":"sp_002","agent_id":"supervisor","model":"claude-opus-4-6","prompt_tokens":2340,"temperature":0.0}
{"timestamp":"2026-02-28T10:15:35.890Z","level":"INFO","event":"llm.response","trace_id":"tr_abc123","span_id":"sp_002","agent_id":"supervisor","model":"claude-opus-4-6","completion_tokens":890,"duration_ms":4323,"cost_usd":0.038}

敏感信息脱敏

class LogSanitizer:
    """日志脱敏处理"""

    REDACT_PATTERNS = {
        "api_key": r"(sk-|ghp_|gho_|Bearer\s+)[a-zA-Z0-9_-]{20,}",
        "password": r"(?i)(password|passwd|secret|token)\s*[:=]\s*\S+",
        "email": r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b",
        "credit_card": r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b",
    }

    # 不脱敏的字段白名单
    SAFE_FIELDS = {"trace_id", "span_id", "agent_id", "event",
                   "timestamp", "level", "duration_ms", "status"}

    def sanitize(self, log_entry: dict) -> dict:
        """脱敏日志条目"""
        sanitized = {}
        for key, value in log_entry.items():
            if key in self.SAFE_FIELDS:
                sanitized[key] = value
            elif isinstance(value, str):
                sanitized[key] = self._redact_string(value)
            elif isinstance(value, dict):
                sanitized[key] = self.sanitize(value)
            else:
                sanitized[key] = value
        return sanitized

    def _redact_string(self, text: str) -> str:
        result = text
        for name, pattern in self.REDACT_PATTERNS.items():
            result = re.sub(pattern, f"[REDACTED:{name}]", result)
        return result

二、分布式链路追踪（Distributed Tracing）

OpenTelemetry 集成

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
)

# 初始化 Tracer
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent-platform")


class TracedAgent:
    """带链路追踪的 Agent"""

    def __init__(self, name: str, llm, tools: list):
        self.name = name
        self.llm = llm
        self.tools = tools

    def run(self, task: str) -> str:
        with tracer.start_as_current_span(
            "agent.run",
            attributes={
                "agent.name": self.name,
                "agent.task": task[:200],  # 截断避免过长
            }
        ) as span:
            try:
                result = self._execute(task)
                span.set_attribute("agent.status", "success")
                return result
            except Exception as e:
                span.set_status(trace.StatusCode.ERROR, str(e))
                span.record_exception(e)
                raise

    def _call_llm(self, messages: list) -> str:
        with tracer.start_as_current_span(
            "llm.call",
            attributes={
                "llm.model": self.llm.model,
                "llm.prompt_tokens": self._count_tokens(messages),
            }
        ) as span:
            start = time.time()
            response = self.llm.chat(messages)
            duration = time.time() - start

            span.set_attribute("llm.completion_tokens",
                             response.usage.completion_tokens)
            span.set_attribute("llm.total_tokens",
                             response.usage.total_tokens)
            span.set_attribute("llm.duration_ms", int(duration * 1000))
            span.set_attribute("llm.cost_usd",
                             self._calc_cost(response.usage))

            return response.content

    def _call_tool(self, tool_name: str, params: dict) -> str:
        with tracer.start_as_current_span(
            "tool.call",
            attributes={
                "tool.name": tool_name,
                "tool.params": json.dumps(params)[:500],
            }
        ) as span:
            tool = self._get_tool(tool_name)
            start = time.time()
            try:
                result = tool.execute(params)
                span.set_attribute("tool.status", "success")
                span.set_attribute("tool.result_size",
                                 len(str(result)))
                return result
            except Exception as e:
                span.set_attribute("tool.status", "error")
                span.set_attribute("tool.error", str(e))
                raise
            finally:
                span.set_attribute("tool.duration_ms",
                                 int((time.time() - start) * 1000))

Trace 结构示例

Trace: tr_abc123 (分析Q4财报)
│
├── Span: agent.run [supervisor] (12.5s)
│   │
│   ├── Span: llm.call [claude-opus-4-6] (2.1s)
│   │   └── tokens: 2340 in / 150 out / $0.012
│   │
│   ├── Span: tool.call [file_read] (0.4s)
│   │   └── path: /data/q4_report.xlsx, status: success
│   │
│   ├── Span: agent.handoff [supervisor -> analyst] (0.1s)
│   │
│   ├── Span: agent.run [analyst] (8.2s)
│   │   │
│   │   ├── Span: llm.call [claude-opus-4-6] (3.5s)
│   │   │   └── tokens: 5600 in / 1200 out / $0.048
│   │   │
│   │   ├── Span: tool.call [python_execute] (2.1s)
│   │   │   └── code: "import pandas...", status: success
│   │   │
│   │   └── Span: llm.call [claude-opus-4-6] (2.6s)
│   │       └── tokens: 3800 in / 900 out / $0.034
│   │
│   └── Span: llm.call [claude-opus-4-6] (1.7s)
│       └── tokens: 2100 in / 500 out / $0.018
│
└── Total: 12.5s, 14840 tokens, $0.112

多 Agent 追踪

class MultiAgentTracer:
    """多 Agent 协作的追踪"""

    def trace_handoff(self, source_agent: str,
                       target_agent: str,
                       context: dict):
        """追踪 Agent 间的控制权转移"""
        current_span = trace.get_current_span()

        with tracer.start_as_current_span(
            "agent.handoff",
            attributes={
                "handoff.source": source_agent,
                "handoff.target": target_agent,
                "handoff.reason": context.get("reason", ""),
                "handoff.context_size": len(json.dumps(context)),
            }
        ):
            pass  # Handoff span 用于标记转移点

    def trace_parallel_agents(self, agent_tasks: list[tuple]):
        """追踪并行执行的 Agent"""
        with tracer.start_as_current_span("agents.parallel") as parent:
            parent.set_attribute(
                "parallel.count", len(agent_tasks)
            )
            # 每个并行 Agent 创建子 Span
            # 子 Span 的 parent 都是 parallel span
            pass

三、指标体系（Metrics）

Agent 核心指标定义

from prometheus_client import (
    Counter, Histogram, Gauge, Summary
)

# ---- 任务级指标 ----
task_total = Counter(
    "agent_tasks_total",
    "Total number of tasks",
    ["agent_name", "status"]  # status: success/failure/timeout
)

task_duration = Histogram(
    "agent_task_duration_seconds",
    "Task execution duration",
    ["agent_name"],
    buckets=[1, 5, 10, 30, 60, 120, 300, 600]
)

# ---- LLM 调用指标 ----
llm_calls = Counter(
    "agent_llm_calls_total",
    "Total LLM API calls",
    ["agent_name", "model", "status"]
)

llm_tokens = Counter(
    "agent_llm_tokens_total",
    "Total tokens consumed",
    ["agent_name", "model", "direction"]  # direction: input/output
)

llm_cost = Counter(
    "agent_llm_cost_usd_total",
    "Total LLM cost in USD",
    ["agent_name", "model"]
)

llm_latency = Histogram(
    "agent_llm_latency_seconds",
    "LLM API call latency",
    ["model"],
    buckets=[0.5, 1, 2, 5, 10, 30]
)

# ---- 工具调用指标 ----
tool_calls = Counter(
    "agent_tool_calls_total",
    "Total tool calls",
    ["agent_name", "tool_name", "status"]
)

tool_latency = Histogram(
    "agent_tool_latency_seconds",
    "Tool call latency",
    ["tool_name"],
    buckets=[0.1, 0.5, 1, 5, 10, 30]
)

tool_fallback = Counter(
    "agent_tool_fallback_total",
    "Tool fallback triggered",
    ["tool_name", "fallback_tool"]
)

# ---- 系统健康指标 ----
active_agents = Gauge(
    "agent_active_count",
    "Currently active agents",
    ["agent_name"]
)

memory_usage = Gauge(
    "agent_memory_usage_bytes",
    "Agent memory usage",
    ["agent_name", "memory_type"]  # buffer/vector/episodic
)

queue_depth = Gauge(
    "agent_task_queue_depth",
    "Pending tasks in queue",
    ["priority"]
)

指标仪表盘布局

┌──────────────────────────────────────────────────────────┐
│                  Agent Platform Dashboard                │
├──────────────────────────────────────────────────────────┤
│                                                          │
│  [实时概览]                                              │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │
│  │ 活跃Agent │ │ 任务成功率 │ │ 平均延迟  │ │ 今日成本  │  │
│  │    12     │ │  94.2%   │ │  8.3s    │ │  $47.20  │  │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │
│                                                          │
│  [任务趋势 - 24h]                                        │
│  成功 ████████████████████████████████░░░░  94.2%       │
│  失败 ██░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░   4.1%       │
│  超时 █░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░   1.7%       │
│                                                          │
│  [LLM 调用分布]          [工具调用 Top 5]               │
│  claude-opus-4-6  45%    file_read      234             │
│  gpt-4o           30%    web_search     189             │
│  gemini-2.5-pro   25%    code_execute   156             │
│                          database_query  98             │
│                          file_write      67             │
│                                                          │
│  [成本明细 - 按模型]      [错误率 - 按工具]               │
│  claude: $28.50          shell_exec  8.2%               │
│  gpt-4o: $12.30          web_search  3.1%               │
│  gemini: $6.40           db_query    1.5%               │
│                                                          │
└──────────────────────────────────────────────────────────┘

四、告警规则设计

告警层级

ALERT_RULES = [
    # P0: 立即处理
    AlertRule(
        name="agent_task_success_rate_critical",
        condition="rate(agent_tasks_total{status='failure'}[5m]) "
                  "> 0.2",
        severity="critical",
        message="Agent 任务失败率超过 20%",
        channels=["pagerduty", "slack-oncall"],
        runbook="https://wiki/runbooks/agent-high-failure-rate"
    ),

    AlertRule(
        name="agent_security_violation",
        condition="increase(agent_security_events_total"
                  "{severity='critical'}[1m]) > 0",
        severity="critical",
        message="检测到 Agent 安全违规事件",
        channels=["pagerduty", "slack-security"],
    ),

    # P1: 1 小时内处理
    AlertRule(
        name="agent_cost_anomaly",
        condition="rate(agent_llm_cost_usd_total[1h]) "
                  "> 2 * avg_over_time("
                  "rate(agent_llm_cost_usd_total[1h])[7d:1h])",
        severity="warning",
        message="Agent LLM 成本异常，当前速率是过去 7 天均值的 2 倍",
        channels=["slack-ops"],
    ),

    AlertRule(
        name="agent_latency_high",
        condition="histogram_quantile(0.99, "
                  "agent_task_duration_seconds) > 120",
        severity="warning",
        message="Agent 任务 P99 延迟超过 2 分钟",
        channels=["slack-ops"],
    ),

    # P2: 下一工作日处理
    AlertRule(
        name="agent_tool_fallback_rate",
        condition="rate(agent_tool_fallback_total[1h]) "
                  "/ rate(agent_tool_calls_total[1h]) > 0.1",
        severity="info",
        message="工具降级率超过 10%",
        channels=["slack-dev"],
    ),
]

五、成本可观测性

Token 消耗追踪

class CostTracker:
    """LLM 成本追踪"""

    # 模型定价（每百万 Token）
    PRICING = {
        "claude-opus-4-6": {"input": 15.0, "output": 75.0},
        "claude-sonnet-4": {"input": 3.0, "output": 15.0},
        "gpt-4o": {"input": 2.5, "output": 10.0},
        "gemini-2.5-pro": {"input": 1.25, "output": 10.0},
    }

    def calculate_cost(self, model: str,
                       input_tokens: int,
                       output_tokens: int) -> float:
        """计算单次调用成本"""
        pricing = self.PRICING.get(model, {"input": 5.0, "output": 15.0})
        cost = (
            input_tokens * pricing["input"] / 1_000_000 +
            output_tokens * pricing["output"] / 1_000_000
        )
        return round(cost, 6)

    def get_daily_report(self) -> dict:
        """生成每日成本报告"""
        return {
            "date": date.today().isoformat(),
            "total_cost_usd": self._query_total_cost(),
            "by_model": self._query_cost_by_model(),
            "by_agent": self._query_cost_by_agent(),
            "by_task_type": self._query_cost_by_task_type(),
            "token_efficiency": self._calc_efficiency(),
            "budget_remaining": self._budget_remaining(),
            "forecast_eom": self._forecast_end_of_month(),
        }

成本报告示例

Daily Cost Report - 2026-02-28
================================

Total: $47.20 / Budget: $100.00 (47.2%)
Forecast EOM: $1,378 / Monthly Budget: $3,000 (46.0%)

By Model:
  claude-opus-4-6    $28.50  (60.4%)  ██████████████████░░░░
  gpt-4o             $12.30  (26.1%)  ████████░░░░░░░░░░░░░
  gemini-2.5-pro     $6.40   (13.5%)  ████░░░░░░░░░░░░░░░░░

By Agent:
  code-reviewer       $18.20  384 tasks  $0.047/task
  research-agent      $15.30  156 tasks  $0.098/task
  data-analyst        $8.70   89 tasks   $0.098/task
  support-agent       $5.00   412 tasks  $0.012/task

Token Efficiency:
  Useful output / Total tokens: 34.2%
  Cache hit rate: 67.8%

六、可观测性基础设施

部署配置示例

# docker-compose.observability.yml
services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.96.0
    ports:
      - "4317:4317"   # gRPC
      - "4318:4318"   # HTTP
    volumes:
      - ./otel-config.yaml:/etc/otelcol/config.yaml

  prometheus:
    image: prom/prometheus:v2.51.0
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:10.4.0
    ports:
      - "3000:3000"
    volumes:
      - ./grafana/dashboards:/var/lib/grafana/dashboards

  jaeger:
    image: jaegertracing/all-in-one:1.55
    ports:
      - "16686:16686"  # UI
      - "14268:14268"  # Collector

  loki:
    image: grafana/loki:2.9.5
    ports:
      - "3100:3100"

工程实践建议

从 Trace 开始：Trace 是最有价值的可观测性数据，能串联整个执行链路
成本可观测性是必需品：Agent 系统的成本可以在几分钟内失控，必须有实时监控
日志脱敏是硬性要求：Agent 日志中可能包含用户数据和 API 密钥
采样策略：生产环境使用尾部采样（tail-based sampling），错误和慢请求 100% 采集
推理链路保留：Agent 的 thinking/reasoning 是调试的关键信息，即使压缩也要保留摘要
告警疲劳防治：告警规则要分级，低优先级告警聚合批量发送

参考资料

OpenTelemetry 官方文档：Agent 可观测性的行业标准
LangSmith / LangFuse：LLM 应用专用的可观测性平台
Arize Phoenix：开源的 LLM 可观测性工具
Braintrust：Agent 评估 + 可观测性一体化平台
Datadog LLM Observability

Maurice | [email protected]

深度加工（NotebookLM 生成）

基于本文内容生成的 PPT 大纲、博客摘要、短视频脚本与 Deep Dive 播客，用于多场景复用

PPT 大纲（5-8 张幻灯片）点击展开

企业级 Agent 平台的可观测性 — ppt

基于您上传的关于“企业级 Agent 平台的可观测性”的文章，我为您生成了一份 8 张幻灯片的 PPT 大纲。已按照您的要求采用 Markdown 格式输出。

企业级 Agent 平台可观测性概述

传统系统的局限性：传统微服务主要关注请求在服务间的流转（延迟、错误率、吞吐量），无法完全覆盖 Agent 系统的需求 [1]。
Agent 系统的独特关注点：需要额外追踪 Agent 的决策推理链、工具调用成功率与延迟、LLM 的 Token 成本消耗，以及多 Agent 之间的协作情况 [1]。
全栈可观测性目标：将传统的 Logging、Tracing、Metrics 三支柱体系映射并扩展到 Agent 的工作流中 [1]。
核心基调：成本监控和链路追踪不仅是运维需求，更是 Agent 知识沉淀与系统健康的核心保障 [1, 2]。

可观测性三支柱在 Agent 中的映射

结构化日志（Logging）：记录推理过程、工具调用详情及错误堆栈，主要用于单次调试、审计合规与知识沉淀 [1]。
分布式链路追踪（Tracing）：捕捉任务到 Agent、Agent 到工具或 LLM 的端到端执行链路，用于理解全链路与定位系统瓶颈 [1]。
指标聚合（Metrics）：统计算法成功率、延迟分布与 Token 消耗成本，用于系统宏观监控、趋势分析及触发告警 [1]。

结构化日志（Logging）与安全合规

标准化事件生命周期：涵盖任务启停（task.start/end）、Agent 思考决策（agent.think）、LLM 请求及工具调用等标准日志事件结构 [1]。
丰富的上下文关联：每条日志需绑定 trace_id、span_id 与 agent_id，确保不同日志事件能被准确串联 [1, 3]。
强制敏感信息脱敏：因日志可能包含用户数据和 API 密钥，需通过正则匹配过滤 API Key、密码、邮箱及信用卡等隐私信息 [2, 3]。
保留推理核心：Agent 的 reasoning/thinking 是调试的关键信息，即使需要压缩日志，也必须保留推理摘要 [2]。

多维度分布式链路追踪（Tracing）

行业标准集成：基于 OpenTelemetry 构建链路追踪基础，记录完整的父子 Span 层级 [3]。
精细化调用追踪：追踪每次 LLM 调用的所用模型、Token 消耗、耗时与预估成本；同时追踪工具调用的状态与返回结果大小 [4, 5]。
多 Agent 协作追踪：通过专门的 "agent.handoff" 跨度追踪 Agent 间的控制权转移、转移原因及上下文大小 [6]。
并发执行追踪：通过父级并发 Span 关联多个并行执行的子 Agent 任务 [6, 7]。

Agent 核心指标体系（Metrics）

任务级监控：统计不同 Agent 的总任务数及成功/失败/超时状态，并追踪任务执行的延迟分布 [7]。
LLM 与工具监控：统计各类模型的 API 调用量、Token 吞吐度及成本；监测工具调用耗时与降级（Fallback）触发频次 [7, 8]。
系统健康度评估：实时记录当前活跃 Agent 数量、内存占用情况（缓冲/向量/情景内存）及任务队列深度 [8]。
可视化大盘呈现：构建仪表盘展示实时成本明细、工具错误率及模型调用比例（如 Claude/GPT-4o 占比），帮助直观掌控全局 [8, 9]。

细粒度的成本可观测性

成本监控的必要性：Agent 系统的成本可能在几分钟内失控，必须拥有细粒度、实时且准确的成本追踪机制 [2]。
多模型差异化核算：内置各主流大模型（如 Claude、GPT-4o、Gemini）的输入与输出 Token 阶梯定价，自动换算单次调用成本 [10, 11]。
全景每日成本报告：生成包含总耗费、模型维度的消耗占比、具体 Agent/任务类型的均摊成本及月末费用预测的财务报告 [11, 12]。
效能分析优化：统计有效输出 Token 占总 Token 的比例（Token Efficiency）及缓存命中率，以指导降本增效 [12]。

分级告警规则设计

P0 级告警（立即处理）：当 Agent 任务失败率超过 20%，或检测到严重安全违规事件时，直接通过 PagerDuty 等渠道触发紧急告警 [9, 10]。
P1 级告警（1 小时内处理）：监控 LLM 成本速率异常飙升（如达均值 2 倍）或任务 P99 延迟超过 2 分钟的性能瓶颈问题 [10]。
P2 级告警（下一工作日处理）：关注工具降级率超过 10% 等影响服务降级但不致命的问题 [10]。
防告警疲劳机制：对告警规则进行分级管理，对低优先级告警聚合批量发送，避免无效打扰 [2]。

基础设施构建与工程实践建议

推荐技术栈全景图：采用 OTel SDK 采集数据，以 Jaeger 存储 Trace，Prometheus 存储 Metrics，最终通过 Grafana 统一可视化 [2, 12]。
从 Trace 开始建设：Trace 是最有价值的可观测数据，可以完整串联整个 Agent 执行的动态上下文 [2]。
智能采样策略：生产环境建议采用尾部采样（tail-based sampling），对常规请求进行采样，但确保错误和慢请求实现 100% 采集 [2]。
一体化闭环管理：通过集成 LangSmith / Braintrust 等专业 LLM 平台，实现评测与可观测性的一体化 [2]。

博客摘要 + 核心看点点击展开

企业级 Agent 平台的可观测性 — summary

SEO 友好博客摘要

在构建企业级 Agent 平台时，传统的微服务监控已无法满足复杂业务的需求[1]。本文深入探讨如何通过结构化日志、分布式链路追踪与多维指标监控，构建专属的 Agent 全栈可观测性体系[1]。相比传统微服务，Agent 可观测性需要额外揭示大模型的决策推理链、工具调用效率、多 Agent 协作状态及 Token 消耗成本[1]。本指南详细解析了日志脱敏机制、OpenTelemetry 集成方法、监控大盘配置以及成本告警规则等核心工程实践[2-5]。对于希望优化 Agent 系统性能、精准控制 LLM 成本并确保系统安全合规的开发者而言，这是一篇极具实战价值的技术宝典[6]。

3 条核心看点