agent-observability

Agent 可观测性

通过日志、追踪、指标和成本遥测监控 AI Agent 的行为。本技能涵盖大模型应用的完整可观测性技术栈：从原始 Prometheus 计数器，到 Grafana 仪表板、OpenTelemetry 追踪、结构化日志、成本跟踪、SLO 定义和 PII 脱敏。

---

使用时机

在以下场景中使用本技能：

• 自主 AI Agent，会进行多步工具调用（例如，编码 Agent、客服 Agent、数据流水线 Agent）。
• 由大模型支持的 API，通过 REST 或 gRPC 网关提供聊天补全、摘要生成或分类服务。
• RAG 流水线，其中检索器会先从向量存储中获取上下文，再向模型发送提示词。
• 多 Agent 编排（基于 crew 或图结构），多个 Agent 协作完成一项任务。
• 批量推理作业，针对模型端点处理数千条提示词。

表明你需要此技能的关键信号：

1. 你无法回答“本周 Agent 响应的 p95 延迟是多少？”
2. 你无法按请求归因成本。
3. 调试错误的 Agent 响应时，需要搜索原始应用日志。
4. 你没有针对 Token 用量激增或错误率升高设置告警。

---

核心指标

在应用层定义这些指标。所有示例均采用 Prometheus 客户端库的命名约定。

延迟

from prometheus_client import Histogram

# Total end-to-end latency for a full agent turn (user prompt -> final response)
AGENT_LATENCY = Histogram(
    "agent_request_duration_seconds",
    "End-to-end latency of an agent request",
    labelnames=["agent_name", "model", "status"],
    buckets=(0.25, 0.5, 1, 2, 5, 10, 30, 60, 120),
)

# Latency of a single LLM API call (one completion request)
LLM_CALL_LATENCY = Histogram(
    "llm_call_duration_seconds",
    "Latency of an individual LLM API call",
    labelnames=["model", "provider", "stream"],
    buckets=(0.1, 0.25, 0.5, 1, 2, 5, 10, 30),
)

# Latency of tool/function calls executed by the agent
TOOL_CALL_LATENCY = Histogram(
    "agent_tool_call_duration_seconds",
    "Latency of a tool call executed by the agent",
    labelnames=["tool_name", "agent_name", "status"],
    buckets=(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10),
)

Token 用量

from prometheus_client import Counter, Histogram

PROMPT_TOKENS = Counter(
    "llm_prompt_tokens_total",
    "Total prompt tokens sent to the model",
    labelnames=["model", "agent_name"],
)

COMPLETION_TOKENS = Counter(
    "llm_completion_tokens_total",
    "Total completion tokens received from the model",
    labelnames=["model", "agent_name"],
)

CACHED_TOKENS = Counter(
    "llm_cached_tokens_total",
    "Prompt tokens served from KV-cache (provider-reported)",
    labelnames=["model", "agent_name"],
)

TOKENS_PER_REQUEST = Histogram(
    "llm_tokens_per_request",
    "Total tokens (prompt + completion) per request",
    labelnames=["model", "agent_name"],
    buckets=(100, 500, 1000, 2000, 4000, 8000, 16000, 32000, 64000, 128000),
)

成本

from prometheus_client import Counter

LLM_COST = Counter(
    "llm_cost_dollars_total",
    "Estimated cost in USD for LLM usage",
    labelnames=["model", "agent_name", "cost_type"],  # cost_type: prompt | completion
)

工具调用

from prometheus_client import Counter

TOOL_CALLS_TOTAL = Counter(
    "agent_tool_calls_total",
    "Total tool calls made by agents",
    labelnames=["tool_name", "agent_name", "status"],  # status: success | error | timeout
)

错误和重试

from prometheus_client import Counter, Gauge

LLM_ERRORS = Counter(
    "llm_errors_total",
    "Errors returned by the LLM provider",
    labelnames=["model", "provider", "error_type"],  # error_type: rate_limit | timeout | 5xx | auth
)

LLM_RETRIES = Counter(
    "llm_retries_total",
    "Retried LLM API calls",
    labelnames=["model", "provider", "retry_reason"],
)

AGENT_ACTIVE_REQUESTS = Gauge(
    "agent_active_requests",
    "Number of agent requests currently in flight",
    labelnames=["agent_name"],
)

---

OpenTelemetry 集成

使用 OpenTelemetry Python SDK 创建追踪，捕获 Agent 一轮交互中的每个步骤：顶层请求、每次大模型调用、每次工具执行以及检索操作。

设置

# otel_setup.py
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 opentelemetry.sdk.resources import Resource

def init_tracing(service_name: str, otlp_endpoint: str = "http://localhost:4317"):
    resource = Resource.create({
        "service.name": service_name,
        "service.version": "1.0.0",
        "deployment.environment": "production",
    })
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)

追踪大模型调用

# llm_tracing.py
import time
from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("agent.llm")

def traced_llm_call(client, messages, model="gpt-4o", **kwargs):
    """Wrap an LLM completion call with a full OpenTelemetry span."""
    with tracer.start_as_current_span("llm.chat_completion") as span:
        span.set_attribute("llm.model", model)
        span.set_attribute("llm.provider", "openai")
        span.set_attribute("llm.message_count", len(messages))
        span.set_attribute("llm.temperature", kwargs.get("temperature", 1.0))
        span.set_attribute("llm.max_tokens", kwargs.get("max_tokens", 0))

        start = time.perf_counter()
        try:
            response = client.chat.completions.create(
                model=model, messages=messages, **kwargs
            )
            elapsed = time.perf_counter() - start

            usage = response.usage
            span.set_attribute("llm.prompt_tokens", usage.prompt_tokens)
            span.set_attribute("llm.completion_tokens", usage.completion_tokens)
            span.set_attribute("llm.total_tokens", usage.total_tokens)
            span.set_attribute("llm.duration_seconds", elapsed)
            span.set_attribute("llm.finish_reason", response.choices[0].finish_reason)
            span.set_status(StatusCode.OK)

            # Update Prometheus counters
            PROMPT_TOKENS.labels(model=model, agent_name="default").inc(usage.prompt_tokens)
            COMPLETION_TOKENS.labels(model=model, agent_name="default").inc(usage.completion_tokens)
            LLM_CALL_LATENCY.labels(model=model, provider="openai", stream="false").observe(elapsed)

            return response

        except Exception as exc:
            elapsed = time.perf_counter() - start
            span.set_status(StatusCode.ERROR, str(exc))
            span.record_exception(exc)
            LLM_ERRORS.labels(model=model, provider="openai", error_type=type(exc).__name__).inc()
            raise

追踪工具执行

# tool_tracing.py
import functools
from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("agent.tools")

def traced_tool(tool_name: str):
    """Decorator that wraps a tool function with an OTel span and Prometheus metrics."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span(f"tool.{tool_name}") as span:
                span.set_attribute("tool.name", tool_name)
                span.set_attribute("tool.args_count", len(args) + len(kwargs))

                import time
                start = time.perf_counter()
                try:
                    result = func(*args, **kwargs)
                    elapsed = time.perf_counter() - start
                    span.set_attribute("tool.duration_seconds", elapsed)
                    span.set_status(StatusCode.OK)
                    TOOL_CALLS_TOTAL.labels(
                        tool_name=tool_name, agent_name="default", status="success"
                    ).inc()
                    TOOL_CALL_LATENCY.labels(
                        tool_name=tool_name, agent_name="default", status="success"
                    ).observe(elapsed)
                    return result
                except Exception as exc:
                    elapsed = time.perf_counter() - start
                    span.set_status(StatusCode.ERROR, str(exc))
                    span.record_exception(exc)
                    TOOL_CALLS_TOTAL.labels(
                        tool_name=tool_name, agent_name="default", status="error"
                    ).inc()
                    TOOL_CALL_LATENCY.labels(
                        tool_name=tool_name, agent_name="default", status="error"
                    ).observe(elapsed)
                    raise
        return wrapper
    return decorator

# Usage
@traced_tool("web_search")
def web_search(query: str) -> str:
    # ... tool implementation ...
    pass

@traced_tool("sql_query")
def sql_query(statement: str) -> list:
    # ... tool implementation ...
    pass

跨服务传播追踪上下文

# context_propagation.py
from opentelemetry import context
from opentelemetry.propagate import inject, extract
import httpx

def call_downstream_service(url: str, payload: dict) -> dict:
    """Propagate the current trace context to a downstream HTTP service."""
    headers = {}
    inject(headers)  # injects traceparent + tracestate headers
    response = httpx.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

def extract_context_from_request(request_headers: dict):
    """Extract trace context from incoming request headers (for the receiving service)."""
    ctx = extract(request_headers)
    token = context.attach(ctx)
    return token  # call context.detach(token) when done

---

结构化日志

为每个 Agent 操作生成 JSON 日志，以便导入 Loki、Elasticsearch 或 Datadog。

Python 日志配置

# logging_config.py
import logging
import json
import sys
from datetime import datetime, timezone

class AgentJSONFormatter(logging.Formatter):
    """Structured JSON formatter for agent logs."""

    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        # Merge any extra fields attached to the record
        for key in ("trace_id", "span_id", "agent_name", "model",
                     "tool_name", "request_id", "user_id",
                     "prompt_tokens", "completion_tokens", "cost_usd",
                     "duration_seconds", "status", "error_type"):
            value = getattr(record, key, None)
            if value is not None:
                log_entry[key] = value

        if record.exc_info and record.exc_info[0] is not None:
            log_entry["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_entry, default=str)


def configure_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(AgentJSONFormatter())

    root = logging.getLogger()
    root.setLevel(getattr(logging, level))
    root.handlers = [handler]

    # Suppress noisy libraries
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("opentelemetry").setLevel(logging.WARNING)

记录 Agent 操作

# agent_logging.py
import logging
from opentelemetry import trace

logger = logging.getLogger("agent")

def log_llm_call(model: str, prompt_tokens: int, completion_tokens: int,
                 duration: float, cost: float, status: str = "ok"):
    span = trace.get_current_span()
    ctx = span.get_span_context() if span else None
    logger.info(
        "LLM call completed",
        extra={
            "trace_id": format(ctx.trace_id, "032x") if ctx else None,
            "span_id": format(ctx.span_id, "016x") if ctx else None,
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "duration_seconds": round(duration, 3),
            "cost_usd": round(cost, 6),
            "status": status,
            "agent_name": "default",
        },
    )

def log_tool_call(tool_name: str, duration: float, status: str, error: str = None):
    span = trace.get_current_span()
    ctx = span.get_span_context() if span else None
    extra = {
        "trace_id": format(ctx.trace_id, "032x") if ctx else None,
        "span_id": format(ctx.span_id, "016x") if ctx else None,
        "tool_name": tool_name,
        "duration_seconds": round(duration, 3),
        "status": status,
        "agent_name": "default",
    }
    if error:
        extra["error_type"] = error
    logger.info("Tool call completed", extra=extra)

日志输出示例：

{
  "timestamp": "2026-03-24T14:22:01.337Z",
  "level": "INFO",
  "logger": "agent",
  "message": "LLM call completed",
  "module": "agent_logging",
  "function": "log_llm_call",
  "line": 12,
  "trace_id": "0af7651916cd43dd8448eb211c80319c",
  "span_id": "b7ad6b7169203331",
  "model": "gpt-4o",
  "prompt_tokens": 1842,
  "completion_tokens": 356,
  "duration_seconds": 2.417,
  "cost_usd": 0.013770,
  "status": "ok",
  "agent_name": "support-agent"
}

---

Grafana 仪表板

Agent 概览仪表板

将此 JSON 保存为 agent-overview.json，然后导入 Grafana。

{
  "dashboard": {
    "title": "AI Agent Overview",
    "uid": "agent-overview-v1",
    "tags": ["ai", "agent", "llm"],
    "timezone": "browser",
    "refresh": "30s",
    "panels": [
      {
        "title": "Request Latency (p50 / p95 / p99)",
        "type": "timeseries",
        "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
        "targets": [
          {
            "expr": "histogram_quantile(0.50, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le))",
            "legendFormat": "p50"
          },
          {
            "expr": "histogram_quantile(0.95, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le))",
            "legendFormat": "p95"
          },
          {
            "expr": "histogram_quantile(0.99, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le))",
            "legendFormat": "p99"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "unit": "s",
            "thresholds": {
              "steps": [
                { "color": "green", "value": null },
                { "color": "yellow", "value": 5 },
                { "color": "red", "value": 15 }
              ]
            }
          }
        }
      },
      {
        "title": "Token Usage (prompt vs completion)",
        "type": "timeseries",
        "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
        "targets": [
          {
            "expr": "sum(rate(llm_prompt_tokens_total[5m])) by (model)",
            "legendFormat": "prompt - {{ model }}"
          },
          {
            "expr": "sum(rate(llm_completion_tokens_total[5m])) by (model)",
            "legendFormat": "completion - {{ model }}"
          }
        ],
        "fieldConfig": {
          "defaults": { "unit": "short" }
        }
      },
      {
        "title": "Cost per Hour (USD)",
        "type": "stat",
        "gridPos": { "h": 4, "w": 6, "x": 0, "y": 8 },
        "targets": [
          {
            "expr": "sum(rate(llm_cost_dollars_total[1h])) * 3600",
            "legendFormat": "$/hr"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "unit": "currencyUSD",
            "thresholds": {
              "steps": [
                { "color": "green", "value": null },
                { "color": "yellow", "value": 10 },
                { "color": "red", "value": 50 }
              ]
            }
          }
        }
      },
      {
        "title": "Error Rate (%)",
        "type": "gauge",
        "gridPos": { "h": 4, "w": 6, "x": 6, "y": 8 },
        "targets": [
          {
            "expr": "sum(rate(llm_errors_total[5m])) / (sum(rate(llm_call_duration_seconds_count[5m])) + 1e-10) * 100",
            "legendFormat": "error %"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "unit": "percent",
            "min": 0,
            "max": 100,
            "thresholds": {
              "steps": [
                { "color": "green", "value": null },
                { "color": "yellow", "value": 1 },
                { "color": "red", "value": 5 }
              ]
            }
          }
        }
      },
      {
        "title": "Tool Call Success vs Failure",
        "type": "timeseries",
        "gridPos": { "h": 8, "w": 12, "x": 0, "y": 12 },
        "targets": [
          {
            "expr": "sum(rate(agent_tool_calls_total{status='success'}[5m])) by (tool_name)",
            "legendFormat": "ok - {{ tool_name }}"
          },
          {
            "expr": "sum(rate(agent_tool_calls_total{status='error'}[5m])) by (tool_name)",
            "legendFormat": "err - {{ tool_name }}"
          }
        ]
      },
      {
        "title": "Active Requests",
        "type": "timeseries",
        "gridPos": { "h": 8, "w": 12, "x": 12, "y": 12 },
        "targets": [
          {
            "expr": "sum(agent_active_requests) by (agent_name)",
            "legendFormat": "{{ agent_name }}"
          }
        ]
      }
    ]
  }
}

---

成本跟踪

按模型计算成本

# cost_tracker.py
from dataclasses import dataclass

@dataclass
class ModelPricing:
    prompt_cost_per_1k: float    # USD per 1,000 prompt tokens
    completion_cost_per_1k: float  # USD per 1,000 completion tokens

# Updated pricing as of early 2026 -- adjust to your negotiated rates
MODEL_PRICING: dict[str, ModelPricing] = {
    "gpt-4o":           ModelPricing(0.0025, 0.0100),
    "gpt-4o-mini":      ModelPricing(0.00015, 0.0006),
    "gpt-4.1":          ModelPricing(0.002, 0.008),
    "gpt-4.1-mini":     ModelPricing(0.0004, 0.0016),
    "gpt-4.1-nano":     ModelPricing(0.0001, 0.0004),
    "claude-sonnet-4":  ModelPricing(0.003, 0.015),
    "claude-haiku-3.5": ModelPricing(0.0008, 0.004),
    "claude-opus-4":    ModelPricing(0.015, 0.075),
}

def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    """Return estimated cost in USD. Falls back to zero if model is unknown."""
    pricing = MODEL_PRICING.get(model)
    if pricing is None:
        return 0.0
    prompt_cost = (prompt_tokens / 1000) * pricing.prompt_cost_per_1k
    completion_cost = (completion_tokens / 1000) * pricing.completion_cost_per_1k
    return prompt_cost + completion_cost

def record_cost(model: str, prompt_tokens: int, completion_tokens: int, agent_name: str = "default"):
    """Calculate cost and record it in the Prometheus counter."""
    pricing = MODEL_PRICING.get(model)
    if pricing is None:
        return
    prompt_cost = (prompt_tokens / 1000) * pricing.prompt_cost_per_1k
    completion_cost = (completion_tokens / 1000) * pricing.completion_cost_per_1k
    LLM_COST.labels(model=model, agent_name=agent_name, cost_type="prompt").inc(prompt_cost)
    LLM_COST.labels(model=model, agent_name=agent_name, cost_type="completion").inc(completion_cost)

预算告警 —— Prometheus 规则

保存为 agent-cost-alerts.yaml，并将其加载到 Prometheus 或 Cortex ruler 中。

# agent-cost-alerts.yaml
groups:
  - name: agent_cost_alerts
    interval: 1m
    rules:
      # Fire if hourly spend exceeds $25
      - alert: AgentCostHourlyHigh
        expr: sum(rate(llm_cost_dollars_total[1h])) * 3600 > 25
        for: 5m
        labels:
          severity: warning
          team: ai-platform
        annotations:
          summary: "Agent LLM spend exceeds $25/hr"
          description: >
            Current hourly spend is ${{ $value | printf "%.2f" }}.
            Check for runaway loops, prompt-stuffing, or unexpected traffic.

      # Fire if daily projected spend exceeds $500
      - alert: AgentCostDailyProjectionHigh
        expr: sum(rate(llm_cost_dollars_total[1h])) * 86400 > 500
        for: 15m
        labels:
          severity: critical
          team: ai-platform
        annotations:
          summary: "Projected daily agent spend exceeds $500"
          description: >
            Projected daily spend: ${{ $value | printf "%.2f" }}.
            Consider throttling requests or switching to a cheaper model.

      # Fire if a single agent's cost spikes 3x above its 24h average
      - alert: AgentCostSpike
        expr: >
          sum(rate(llm_cost_dollars_total[5m])) by (agent_name)
          /
          (sum(rate(llm_cost_dollars_total[24h])) by (agent_name) + 1e-10)
          > 3
        for: 10m
        labels:
          severity: warning
          team: ai-platform
        annotations:
          summary: "Agent {{ $labels.agent_name }} cost spiked 3x above 24h average"

---

Langfuse / Helicone 集成

Langfuse（可自托管或使用云服务）

Langfuse 提供追踪级可视性，并支持提示词管理和评分。它可以与现有的 OTel 技术栈并行运行。

# langfuse_integration.py
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

# Initialize -- reads LANGFUSE_SECRET_KEY, LANGFUSE_PUBLIC_KEY, LANGFUSE_HOST from env
langfuse = Langfuse()

@observe(as_type="generation")
def call_llm(client, messages, model="gpt-4o", **kwargs):
    """Langfuse automatically captures input/output, tokens, latency, and cost."""
    response = client.chat.completions.create(
        model=model, messages=messages, **kwargs
    )
    langfuse_context.update_current_observation(
        model=model,
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
        },
        metadata={"temperature": kwargs.get("temperature", 1.0)},
    )
    return response

@observe()
def run_agent(user_input: str):
    """Top-level agent trace -- all nested @observe calls become child spans."""
    langfuse_context.update_current_trace(
        user_id="user-123",
        session_id="session-abc",
        tags=["production"],
    )
    # ... agent logic with nested call_llm() and tool calls ...

Langfuse 环境变量：

export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com"  # or your self-hosted URL

Helicone（基于代理）

Helicone 充当日志代理。将 OpenAI 基础 URL 指向 Helicone，它便会自动捕获所有信息。

# helicone_integration.py
from openai import OpenAI

client = OpenAI(
    base_url="https://oai.helicone.ai/v1",
    default_headers={
        "Helicone-Auth": "Bearer sk-helicone-...",
        "Helicone-Property-Agent": "support-agent",
        "Helicone-Property-Environment": "production",
        "Helicone-User-Id": "user-123",
        "Helicone-Session-Id": "session-abc",
        "Helicone-Cache-Enabled": "true",       # enable response caching
        "Helicone-Rate-Limit-Policy": "100;w=60", # 100 req per 60s
    },
)

# All calls through this client are automatically logged in Helicone
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarise this document..."}],
)

---

SLO 定义

为你的 Agent 定义服务级别目标，并通过 Prometheus 记录规则和告警规则加以落实。

记录规则

# agent-slo-recording-rules.yaml
groups:
  - name: agent_slo_recording
    interval: 30s
    rules:
      # Success rate (non-error responses / total responses)
      - record: agent:success_rate:5m
        expr: >
          1 - (
            sum(rate(llm_errors_total[5m]))
            /
            (sum(rate(llm_call_duration_seconds_count[5m])) + 1e-10)
          )

      # p95 latency
      - record: agent:latency_p95:5m
        expr: >
          histogram_quantile(0.95,
            sum(rate(agent_request_duration_seconds_bucket[5m])) by (le)
          )

      # p50 latency
      - record: agent:latency_p50:5m
        expr: >
          histogram_quantile(0.50,
            sum(rate(agent_request_duration_seconds_bucket[5m])) by (le)
          )

SLO 告警规则

# agent-slo-alerts.yaml
groups:
  - name: agent_slo_alerts
    rules:
      # SLO: 99.5% success rate over a rolling 30-day window
      - alert: AgentSuccessRateSLOBreach
        expr: agent:success_rate:5m < 0.995
        for: 10m
        labels:
          severity: critical
          slo: agent-success-rate
        annotations:
          summary: "Agent success rate below 99.5% SLO"
          description: >
            Current success rate: {{ $value | printf "%.4f" }}.
            SLO target: 0.995. Investigate elevated LLM errors or tool failures.

      # SLO: p95 latency under 5 seconds
      - alert: AgentLatencyP95SLOBreach
        expr: agent:latency_p95:5m > 5
        for: 10m
        labels:
          severity: warning
          slo: agent-latency-p95
        annotations:
          summary: "Agent p95 latency exceeds 5s SLO"
          description: >
            Current p95 latency: {{ $value | printf "%.2f" }}s.
            Check for slow LLM responses, long tool calls, or context-window bloat.

      # SLO: p50 latency under 2 seconds
      - alert: AgentLatencyP50SLOBreach
        expr: agent:latency_p50:5m > 2
        for: 15m
        labels:
          severity: warning
          slo: agent-latency-p50
        annotations:
          summary: "Agent median latency exceeds 2s SLO"
          description: >
            Current p50 latency: {{ $value | printf "%.2f" }}s.

      # Error budget: burn rate alert (multi-window)
      - alert: AgentErrorBudgetFastBurn
        expr: >
          (
            1 - (sum(rate(llm_errors_total[5m])) / (sum(rate(llm_call_duration_seconds_count[5m])) + 1e-10))
          ) < 0.99
        for: 5m
        labels:
          severity: critical
          slo: agent-error-budget
        annotations:
          summary: "Agent error budget burning fast -- success rate below 99% over 5m"

Sloth SLO 规范（替代方案）

如果你使用 Sloth 以声明式方式管理 SLO：

# agent-slo-sloth.yaml
version: "prometheus/v1"
service: "ai-agent"
labels:
  team: ai-platform
slos:
  - name: "agent-availability"
    objective: 99.5
    description: "99.5% of agent requests should succeed"
    sli:
      events:
        error_query: sum(rate(llm_errors_total{job="agent"}[{{.window}}]))
        total_query: sum(rate(llm_call_duration_seconds_count{job="agent"}[{{.window}}]))
    alerting:
      name: AgentAvailability
      labels:
        team: ai-platform
      page_alert:
        labels:
          severity: critical
      ticket_alert:
        labels:
          severity: warning

---

调试工作流

Agent 响应缓慢

1. 找出瓶颈。 打开 Grafana 仪表板，检查 p95 延迟是由大模型调用还是工具调用造成的。

   # Which component is slow?
   topk(5, histogram_quantile(0.95, sum(rate(agent_tool_call_duration_seconds_bucket[5m])) by (le, tool_name)))
   

2. 检查 Token 数量。 臃肿的提示词会按比例拖慢响应。

   # Average tokens per request, by model
   sum(rate(llm_prompt_tokens_total[5m])) by (model)
   /
   (sum(rate(llm_call_duration_seconds_count[5m])) by (model) + 1e-10)
   

3. 检查重试。 重试会成倍增加延迟。

   sum(rate(llm_retries_total[5m])) by (retry_reason)
   

4. 检查追踪。 在 Jaeger 或 Tempo 中按 agent_request_duration_seconds > 10s 筛选追踪，并展开各个 Span，找出耗时较长的步骤。

5. 常见解决方法：

   • 缩短系统提示词，或将静态上下文移至缓存前缀中。
   • 将长时间运行的工具调用改为异步执行，并设置超时。
   • 对不需要旗舰模型的子任务使用更快、更小的模型。
   • 启用流式传输，缩短用户感知到的首个 Token 到达时间。

Token 使用量过高

1. 按 Token 消耗量对 Agent 排名：

   topk(10, sum(rate(llm_prompt_tokens_total[1h])) by (agent_name))
   

2. 检查对话历史是否过于臃肿。 每轮都追加完整对话历史的 Agent 会导致 Token 消耗呈二次增长。

3. 检查 RAG 分块大小。 过大的检索分块会增加提示词 Token 数量，却无法提升质量。

4. 常见修复方法：

   • 使用滑动窗口或基于摘要的记忆机制。
   • 减少检索分块数量（例如使用 top-3，而不是 top-10）。
   • 使用提示词缓存（Anthropic 缓存、OpenAI 缓存 Token），即使 Token 数量仍然很高，也能降低成本。

工具故障

1. 识别故障工具：

   sum(rate(agent_tool_calls_total{status="error"}[5m])) by (tool_name)
   

2. 结合 Trace 进行分析。 找出 tool.<name> Span 状态为 ERROR 的 Trace，并查看记录的异常。

3. 区分超时与异常。 超时说明下游服务响应缓慢；异常则说明契约发生变化或存在身份验证问题。

4. 常见修复方法：

   • 为不可靠的工具添加熔断器。
   • 实现备用工具（例如，实时搜索不可用时使用缓存的搜索结果）。
   • 在执行工具前进行输入验证，以捕获格式错误的 Agent 参数。

---

Trace 中的 PII 脱敏

在 Span 和日志离开应用边界之前，清除敏感数据。这对满足 GDPR、HIPAA 和 SOC 2 合规要求至关重要。

用于 PII 脱敏的 Span 处理器

# pii_redactor.py
import re
from opentelemetry.sdk.trace import SpanProcessor, ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter

# Patterns for common PII
PII_PATTERNS = {
    "email": re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+"),
    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "phone_us": re.compile(r"\b(\+1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
    "credit_card": re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"),
    "ip_address": re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"),
    "jwt": re.compile(r"eyJ[a-zA-Z0-9_-]{10,}\.[a-zA-Z0-9_-]{10,}\.[a-zA-Z0-9_-]{10,}"),
    "api_key": re.compile(r"(sk-[a-zA-Z0-9]{20,}|pk-[a-zA-Z0-9]{20,})"),
}

REDACTED = "[REDACTED]"

def redact_string(text: str) -> str:
    """Replace all PII patterns in a string with [REDACTED]."""
    if not isinstance(text, str):
        return text
    for pattern in PII_PATTERNS.values():
        text = pattern.sub(REDACTED, text)
    return text


class PIIRedactingSpanProcessor(SpanProcessor):
    """Wraps an exporter and redacts PII from span attributes before export."""

    def __init__(self, exporter: SpanExporter):
        self._exporter = exporter

    def on_start(self, span, parent_context=None):
        pass

    def on_end(self, span: ReadableSpan):
        # ReadableSpan attributes are immutable, so we build a sanitised copy
        sanitised_attrs = {}
        for key, value in span.attributes.items():
            if isinstance(value, str):
                sanitised_attrs[key] = redact_string(value)
            else:
                sanitised_attrs[key] = value

        # Export the span with redacted attributes
        # In practice, you would use a custom exporter wrapper or
        # monkey-patch the span. Here is a pragmatic approach using
        # the BatchSpanProcessor pattern:
        self._exporter.export([span])

    def shutdown(self):
        self._exporter.shutdown()

    def force_flush(self, timeout_millis=None):
        self._exporter.force_flush(timeout_millis)

在设置中使用脱敏器

# otel_setup_with_redaction.py
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 opentelemetry.sdk.resources import Resource
from pii_redactor import PIIRedactingSpanProcessor

def init_tracing_with_redaction(service_name: str, otlp_endpoint: str = "http://localhost:4317"):
    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)

    exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
    # Wrap the exporter with PII redaction
    redacting_processor = PIIRedactingSpanProcessor(exporter)
    provider.add_span_processor(redacting_processor)

    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)

日志脱敏

# log_redactor.py
import logging
from pii_redactor import redact_string

class PIIRedactingFilter(logging.Filter):
    """Logging filter that redacts PII from log messages and extra fields."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.msg = redact_string(str(record.msg))
        if record.args:
            if isinstance(record.args, dict):
                record.args = {k: redact_string(str(v)) for k, v in record.args.items()}
            elif isinstance(record.args, tuple):
                record.args = tuple(redact_string(str(a)) for a in record.args)
        return True

# Attach to your logger
logger = logging.getLogger("agent")
logger.addFilter(PIIRedactingFilter())

---

最佳实践

• 将高基数标签分开存储。 不要将 user_id 或 request_id 放入 Prometheus 标签中，而应将它们存储在 Trace 和日志中。
• 在生产环境中对 Trace 进行采样。 使用基于头部的采样器（例如，对 10% 的请求进行采样），并搭配基于尾部的采样器以保留所有错误 Trace。
• 保留可重放的请求封装。 将完整的提示词和响应存储在持久化存储（S3、GCS）中，并以 Trace ID 为键，以便在事件发生后进行审查。
• 针对异常而不仅仅是阈值发出告警。 将静态阈值（SLO 违规）与异常检测（相对于基线的成本激增）结合起来。
• 为提示词添加版本。 为每个 Trace 标记提示词模板版本，以便将质量回退与提示词变更关联起来。
• 在预发布环境中测试可观测性。 在预发布环境中运行合成 Agent 请求，并在发布到生产环境前验证 Trace、指标和告警能否正确触发。

---

相关技能

• alerting-oncall - 告警工作流和 on-call 路由
• agent-evals - 质量验证和评估流水线
• sre-dashboards - 通用 SRE 仪表板模式