性能监控与日志记录让 Skills 运行透明化

预计完成时间：45 分钟 · 涉及工具：Docker、OpenTelemetry Collector、Prometheus、Grafana

2025 年 11 月的一个深夜，运维团队收到报警：某个部署了三周的 Agent Skills 突然开始间歇性超时。日志翻了三轮，只看到“调用失败”四个字——没有调用链上下文，没有工具级别耗时，没有任何可以定位瓶颈的数据。最终排查花了整整四天。

这不是个例。当 Skills 从单次对话扩展到成百上千的并发执行时，没有可观测性的系统就像在黑屋子里修手表——你知道它在走，但你永远不知道哪一颗齿轮正在磨损。

本章将带你把黑屋子里的灯打开。你会学到如何用统一的结构化日志为每个 Skill 调用打上“身份证”，如何定义那些真正值得报警的指标，以及如何用 Prometheus + Grafana 搭一套开箱即用的监控看板。读完这一章，你将拥有一个能够回答“谁、什么时候、调了什么、花了多久、结果如何”的完整观测体系。

你需要什么

开始之前，请确认你的开发环境满足以下条件：

如果你还没有现成的 Prometheus，本章会在步骤中提供 Docker Compose 一键部署配置，十分钟内即可拥有全套本地监控栈。

最终成果

完成本章所有步骤后，你将得到：

1. 一套标准化的结构化日志格式，每条日志都包含 trace_id、skill_id、tool_name 等关联字段，可以用一条 SQL 语句串联起一个 Skill 调用的完整生命周期。
2. 四类核心指标的采集管线：调用延迟（P50/P95/P99）、错误率、工具级别成功率、Token 消耗量。
3. 一个 Grafana 监控看板，包含指标趋势图、T谱（Trace Waterfall）和异常告警面板，能一眼判断整个 Skills 集群的健康状态。

为什么做这个：可观测性是生产级 Agent 系统的基础设施，而不是锦上添花的装饰。它能让你在用户发现故障之前定位问题，在扩容之前发现瓶颈，在预算失控之前看到成本分布。

步骤一：统一结构化日志并注入关联 ID

为什么结构化日志是第一步

传统日志让人类阅读，结构化日志让机器索引。当数百个 Skill 并发执行时，你不可能用 grep 在几十万行日志中手工追踪一个请求。解决这个问题的核心策略是 关联 ID（Correlation ID）：在 Skill 入口生成一个唯一的 trace_id，并在每一个工具调用、模型调用、外部 API 请求中透传这个 ID。

从当前调研资料看，Claude Code 的遥测体系正是基于这一思路：Skills 的每次会话、每个工具调用都被打上统一的 Tracing 上下文，通过 OpenTelemetry 协议导出。本章的实现方式与之保持一致的语义。

动手实现

首先，在 Skill 的入口函数中初始化 OpenTelemetry 并生成 Trace ID：

# skill_tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import uuid
import logging
import json
import time

# 1. 初始化 TracerProvider，绑定服务名称
resource = Resource(attributes={
    SERVICE_NAME: "my-skill-service"
})
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

# 2. 配置 OTLP 导出器（发送到本地 OpenTelemetry Collector）
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",  # OTLP gRPC 端口
    insecure=True  # 本地开发环境不使用 TLS
)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

tracer = trace.get_tracer(__name__)

class StructuredLogger:
    """统一的结构化日志记录器，自动注入 trace_id 和 skill_id"""

    def __init__(self, skill_id: str):
        self.skill_id = skill_id

    def log(self, level: str, message: str, extra: dict = None):
        # 3. 获取当前 Span 的 trace_id（如果存在）
        current_span = trace.get_current_span()
        trace_id = format(current_span.get_span_context().trace_id, '032x') \
                   if current_span else "no-trace"

        # 4. 构建结构化日志条目
        log_entry = {
            "timestamp": int(time.time() * 1000),
            "level": level,
            "trace_id": trace_id,
            "skill_id": self.skill_id,
            "message": message,
            **(extra or {})
        }

        # 输出为 JSON 行，便于后续收集
        print(json.dumps(log_entry, ensure_ascii=False), flush=True)

    def info(self, msg, **kwargs): self.log("INFO", msg, kwargs)
    def error(self, msg, **kwargs): self.log("ERROR", msg, kwargs)
    def warn(self, msg, **kwargs): self.log("WARN", msg, kwargs)

预期结果：每个 Skill 实例化时获得一个 StructuredLogger，所有后续日志都会自动带上 trace_id 和 skill_id。在终端中你看到的日志是 JSON 行，可以用 jq 按 trace_id 筛选。

踩坑经验

⚠️ 注意：OTLP 环境变量不会自动传递给子进程

调研资料明确指出，Claude Code 设置的 OTEL_* 环境变量不会传递给 Skills 调用的子进程（如 Bash 工具启动的外部程序）。如果你的 Skill 内部又调用了外部服务，需要在这些子进程的环境中显式设置相同的遥测变量，否则会导致 Tracing 链路断裂。一个可行的做法是在 Skill 代码中通过 os.environ 主动注入，或在 Docker 镜像中预设。

从当前调研资料看，这种隔离设计是为了防止 Skills 运行环境被意外污染，但也意味着你需要自行维护跨进程的上下文传播。

步骤二：关键指标定义与告警规则

哪些指标真正值得关心

Agent Skills 的运行特征与传统微服务不同。一个 Skill 可能在一次调用中触发十几个工具，模型调用本身就是重操作，而错误有可能是部分降级而非整体失败。因此，指标设计必须反映“组合调用”的特性。

以下是经生产验证的四类核心指标：

这四类指标覆盖了“快不快”“对不对”“稳不稳”“贵不贵”四个维度。

动手实现：在 Skill 代码中暴露 Prometheus 指标

使用 prometheus_client 库定义上报点：

# skill_metrics.py
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry, push_to_gateway
import time
import functools

# 全局注册表（避免与默认注册表冲突）
registry = CollectorRegistry()

# 1. 定义指标
skill_calls = Counter(
    'skill_calls_total', 
    'Total number of skill invocations',
    ['skill_id', 'status'],  # status: success / error / partial_error
    registry=registry
)

skill_duration = Histogram(
    'skill_duration_seconds',
    'Skill execution duration in seconds',
    ['skill_id'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],  # 根据实际 Skill 耗时调整分桶
    registry=registry
)

tool_calls = Counter(
    'tool_calls_total',
    'Total tool calls made by skills',
    ['skill_id', 'tool_name', 'status'],
    registry=registry
)

token_usage = Counter(
    'token_usage_total',
    'Total token consumption',
    ['skill_id', 'model', 'type'],  # type: input / output
    registry=registry
)

# 2. 装饰器：自动记录每次 Skill 调用的耗时和状态
def instrument_skill(skill_id: str):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                duration = time.time() - start

                # 记录成功调用
                skill_calls.labels(skill_id=skill_id, status='success').inc()
                skill_duration.labels(skill_id=skill_id).observe(duration)

                return result
            except Exception as e:
                duration = time.time() - start

                # 记录失败调用
                skill_calls.labels(skill_id=skill_id, status='error').inc()
                skill_duration.labels(skill_id=skill_id).observe(duration)

                raise  # 重新抛出异常，不影响上层处理
        return wrapper
    return decorator

# 3. 工具调用记录函数
def record_tool_call(skill_id: str, tool_name: str, success: bool):
    status = 'success' if success else 'error'
    tool_calls.labels(skill_id=skill_id, tool_name=tool_name, status=status).inc()

# 4. Token 消耗记录函数
def record_token_usage(skill_id: str, model: str, input_tokens: int, output_tokens: int):
    token_usage.labels(skill_id=skill_id, model=model, type='input').inc(input_tokens)
    token_usage.labels(skill_id=skill_id, model=model, type='output').inc(output_tokens)

# 5. 定期推送至 Pushgateway（适用于短期/批处理 Skill）
def push_metrics(pushgateway_url: str = "http://localhost:9091"):
    push_to_gateway(pushgateway_url, job='skill_metrics', registry=registry)

预期结果：当你在 Skill 函数上使用 @instrument_skill("my-skill") 装饰器，并在工具调用后调用 record_tool_call()，Prometheus 将能采集到这些指标的时序数据。在 Prometheus 表达式浏览器中输入 rate(skill_calls_total[5m])，你应该看到实时的调用速率曲线。

告警规则示例

在 Prometheus 配置中添加以下规则文件：

# prometheus_rules.yml
groups:
  - name: skill_alerts
    rules:
      - alert: HighSkillErrorRate
        expr: |
          sum(rate(skill_calls_total{status="error"}[5m])) 
          / 
          sum(rate(skill_calls_total[5m])) > 0.01
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Skill error rate exceeds 1%"
          description: "Current error rate is {{ $value | humanizePercentage }}"

      - alert: HighP99Latency
        expr: histogram_quantile(0.99, 
          sum(rate(skill_duration_seconds_bucket[5m])) by (le, skill_id)
        ) > 30
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P99 latency for skill {{ $labels.skill_id }} exceeds 30s"

步骤三：集成 Prometheus + Grafana 看板

快速搭建监控栈

完整部署文件如下，保存为 docker-compose.observability.yml：

# docker-compose.observability.yml
version: '3.9'

services:
  # OpenTelemetry Collector：接收 Skills 的 trace 和 log 数据
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.112.0
    command: ["--config=/etc/otel-collector-config.yaml"]
    volumes:
      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
    depends_on:
      - prometheus

  # Prometheus：抓取指标并存储
  prometheus:
    image: prom/prometheus:v2.53.0
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./prometheus_rules.yml:/etc/prometheus/prometheus_rules.yml
      - prometheus_data:/prometheus
    ports:
      - "9090:9090"

  # Grafana：可视化看板
  grafana:
    image: grafana/grafana:11.1.0
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin  # 仅用于本地开发
    volumes:
      - grafana_data:/var/lib/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

volumes:
  prometheus_data:
  grafana_data:

OpenTelemetry Collector 配置（otel-collector-config.yaml）：

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

exporters:
  # 将指标数据发送给 Prometheus
  prometheus:
    endpoint: "0.0.0.0:8889"

  # 将日志数据输出到控制台（生产环境应替换为 Loki 等后端）
  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [logging]  # 简化为日志输出，实际应接入 Jaeger
    metrics:
      receivers: [otlp]
      exporters: [prometheus]

    logs:
      receivers: [otlp]
      exporters: [logging]

启动监控栈：

docker-compose -f docker-compose.observability.yml up -d

预期结果：访问 http://localhost:9090 应看到 Prometheus UI，访问 http://localhost:3000 应看到 Grafana 登录页（用户名 admin，密码 admin）。

从当前调研资料看，这种基于 OpenTelemetry 的统一采集方案是 Claude Code 生态推荐的可观测性模式——使用 OTLP 协议作为标准传输层，后端可按需替换。

搭建 Grafana 看板

登录 Grafana 后，进入 Data Sources → Add data source → Prometheus，输入 http://prometheus:9090 并保存。

然后创建 Dashboard，添加以下面板：

面板 1：Skill 调用 QPS 与错误率

# QPS（每秒查询数）
sum(rate(skill_calls_total[1m]))

# 错误率百分比
sum(rate(skill_calls_total{status="error"}[5m])) 
/ 
sum(rate(skill_calls_total[5m])) * 100

面板 2：P50/P95/P99 延迟折线图

# P99 延迟
histogram_quantile(0.99, 
  sum(rate(skill_duration_seconds_bucket[5m])) by (le, skill_id)
)
# 分别替换分位数为 0.50 和 0.95

面板 3：工具调用成功率热力图

sum(rate(tool_calls_total{status="success"}[5m])) by (tool_name)
/
sum(rate(tool_calls_total[5m])) by (tool_name) * 100

面板 4：Token 消耗趋势

sum(increase(token_usage_total{type="input"}[1h])) by (skill_id)
sum(increase(token_usage_total{type="output"}[1h])) by (skill_id)

预期结果：四个面板实时刷新，展示 Skills 集群的整体健康画像。当任何一个告警规则触发时，你可以在 Grafana 的 Alerting 页面看到活动警报。

回顾

本章中，你完成了三件事：

1. 统一了日志格式：引入结构化日志与 trace_id 关联机制，让每个 Skill 调用的所有日志可被索引和关联。
2. 定义了关键指标：围绕延迟、错误率、工具成功率和 Token 消耗建立了四维度监控体系，并设定了合理的告警阈值。
3. 搭建了生产级监控看板：用 OpenTelemetry + Prometheus + Grafana 构建了端到端的可观测性管线。

截至当前调研资料，Claude Code 官方正在推动以 OpenTelemetry 为统一标准的遥测方案，本章的实现路径与此保持技术栈一致，便于后续无缝接入 Claude Code 的官方遥测体系。

下一步，你需要确保这些“受监控的 Skills”在逻辑上也是正确的——单元测试是让每一个 Skill 的行为可验证、可回归的基础工程。下一章《Skills 单元测试的核心难点是模拟模型行为》将直面如何 Mock 大模型调用这个棘手问题。

行动清单

1. 将 StructuredLogger 类集成到你当前的 Skill 项目中，替换所有 print() 调用。
2. 在 3 个核心 Skill 上使用 @instrument_skill 装饰器，运行后检查 Prometheus 是否有新指标摄入。
3. 根据你的业务场景调整告警阈值（至少调整错误率阈值和 P99 延迟阈值）。
4. 运行一次压测（模拟 50 并发），观察 Grafana 看板的延迟分位数变化。
5. 确认所有外部工具调用都已调用 record_tool_call()，补齐遗漏的埋点。