AI API 监控与可观测性最佳实践：生产系统稳定运行指南

AI API 监控与可观测性最佳实践：生产系统稳定运行的保障

AI API 已经成为很多应用的核心依赖，但很多团队的监控还停留在「看看能不能调通」的阶段。当 API 出问题时，往往是用户先发现，而不是我们自己。本文介绍 AI API 的监控与可观测性最佳实践，帮你构建可靠的生产系统。

为什么需要监控？

---

关键监控指标

#### 1. 成功率（Success Rate）

定义： 成功请求数 / 总请求数

告警阈值：

• 正常：> 99%
• 警告：< 99% 且 > 95%
• 严重：< 95%

常见失败原因：

• 429 限速错误
• 5xx 服务端错误
• 网络超时
• API Key 失效

---

#### 2. 延迟（Latency）

定义： API 响应时间

关键统计量：

• P50（中位数）：典型响应时间
• P95：95% 的请求在这个时间内完成
• P99：99% 的请求在这个时间内完成
• Max：最大响应时间

告警阈值（以 GPT-4o 为例）：

• P50 > 3s → 警告
• P95 > 10s → 严重
• Max > 30s → 严重

---

#### 3. 速率限制（Rate Limiting）

定义： 429 错误的比例和次数

监控内容：

• RPM（每分钟请求数）使用率
• TPM（每分钟 tokens）使用率
• 429 错误数/分钟
• 触发限速的用户/IP

告警阈值：

• RPM/TPM 使用率 > 80% → 警告
• 429 错误率 > 5% → 严重

---

#### 4. 成本（Cost）

定义： Token 消耗和费用

监控内容：

• 每小时/每天 Token 消耗
• 每小时/每天费用
• 按模型、按用户、按功能的成本分布
• 成本趋势（是否异常增长）

告警阈值：

• 单日费用 > 预算的 120% → 警告
• 成本突然增长 50%+ → 严重

---

#### 5. Token 效率（Token Efficiency）

定义： Output tokens / Input tokens

目标：

• 对话类应用：0.3-0.5（输出是输入的 30-50%）
• 生成类应用：0.8-1.5（输出接近或超过输入）

意义：

• 比例过低 → 可能提示词太长，可以优化
• 比例过高 → 可能输出太长，可以限制 max_tokens

---

监控系统设计

#### 架构概览

```

应用代码 →

埋点（Log/Metrics/Trace）→

采集器（Prometheus/OpenTelemetry）→ 
  存储（Prometheus/Elasticsearch）→ 
    可视化（Grafana）→ 
      告警（Alertmanager/PagerDuty）

```

#### 技术选型（2026 推荐）

---

实战：Prometheus + Grafana 监控

#### 1. 埋点（Metrics）

用 Python 的 `prometheus_client` 库：

```python

from prometheus_client import Counter, Histogram, Gauge, start_http_server

import time

# 定义指标

REQUESTS = Counter(

'ai_api_requests_total',
'Total AI API requests',
['model', 'endpoint', 'status']

)

LATENCY = Histogram(

'ai_api_request_duration_seconds',
'AI API request duration',
['model', 'endpoint'],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30]

)

TOKENS = Counter(

'ai_api_tokens_total',
'Total AI API tokens',
['model', 'type']  # type: input / output

)

COST = Gauge(

'ai_api_cost_dollars',
'AI API cost in dollars',
['model']

)

# 启动 HTTP 服务器（供 Prometheus 拉取）

start_http_server(8000)

# 在 API 调用处埋点

def call_ai_api(model, messages):

start_time = time.time()

try:
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    
    # 成功
    REQUESTS.labels(model=model, endpoint="chat", status="success").inc()
    latency = time.time() - start_time
    LATENCY.labels(model=model, endpoint="chat").observe(latency)
    
    input_tokens = response.usage.prompt_tokens
    output_tokens = response.usage.completion_tokens
    TOKENS.labels(model=model, type="input").inc(input_tokens)
    TOKENS.labels(model=model, type="output").inc(output_tokens)
    
    return response
    
except Exception as e:
    # 失败
    status = "error"
    if hasattr(e, 'status_code'):
        if e.status_code == 429:
            status = "rate_limited"
        elif 500 <= e.status_code < 600:
            status = "server_error"
    
    REQUESTS.labels(model=model, endpoint="chat", status=status).inc()
    raise

```

#### 2. Prometheus 配置

`prometheus.yml`:

```yaml

global:

scrape_interval: 15s

scrape_configs:

- job_name: 'ai-api-monitor'

static_configs:
  - targets: ['localhost:8000']

```

#### 3. Grafana 仪表盘

关键面板：

---

日志管理

#### 结构化日志

用 JSON 格式记录日志：

```python

import json

import logging

from datetime import datetime

class StructuredFormatter(logging.Formatter):

def format(self, record):
    log_data = {
        "timestamp": datetime.utcnow().isoformat(),
        "level": record.levelname,
        "message": record.getMessage(),
        "logger": record.name,
        "request_id": getattr(record, "request_id", None),
        "model": getattr(record, "model", None),
        "latency_ms": getattr(record, "latency_ms", None),
        "tokens": getattr(record, "tokens", None),
        "error": getattr(record, "error", None),
    }
    # 移除 None 值
    log_data = {k: v for k, v in log_data.items() if v is not None}
    return json.dumps(log_data, ensure_ascii=False)

# 配置日志

logger = logging.getLogger("ai-api")

handler = logging.StreamHandler()

handler.setFormatter(StructuredFormatter())

logger.addHandler(handler)

logger.setLevel(logging.INFO)

# 使用

logger.info(

"AI API call completed",
extra={
    "request_id": "req-123",
    "model": "gpt-4o-mini",
    "latency_ms": 850,
    "tokens": {"input": 500, "output": 200}
}

)

```

#### 关键日志字段

---

链路追踪（Tracing）

用 OpenTelemetry 实现全链路追踪：

```python

from opentelemetry import trace

from opentelemetry.sdk.trace import TracerProvider

from opentelemetry.sdk.trace.export import BatchSpanProcessor

from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 初始化

provider = TracerProvider()

jaeger_exporter = JaegerExporter(

agent_host_name="localhost",
agent_port=6831,

)

provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))

trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# 使用

with tracer.start_as_current_span("ai_api_call") as span:

span.set_attribute("model", "gpt-4o-mini")
span.set_attribute("tokens.input", 500)

response = client.chat.completions.create(...)

span.set_attribute("tokens.output", response.usage.completion_tokens)
span.set_attribute("latency_ms", latency_ms)

```

---

告警规则

#### Prometheus 告警规则示例

`alerts.yml`:

```yaml

groups:

• name: ai_api_alerts

rules:

# 成功率告警

- alert: AIAPIHighErrorRate

expr: |
  sum(rate(ai_api_requests_total{status!="success"}[5m])) 
  / sum(rate(ai_api_requests_total[5m])) > 0.05
for: 5m
labels:
  severity: warning
annotations:
  summary: "AI API 错误率超过 5%"
  description: "当前错误率: {{ $value | humanizePercentage }}"

# 延迟告警

- alert: AIAPIHighLatency

expr: histogram_quantile(0.95, sum(rate(ai_api_request_duration_seconds_bucket[5m])) by (le)) > 10
for: 5m
labels:
  severity: warning
annotations:
  summary: "AI API P95 延迟超过 10 秒"
  description: "当前 P95 延迟: {{ $value }}s"

# 限速告警

- alert: AIAPIRateLimited

expr: sum(rate(ai_api_requests_total{status="rate_limited"}[1m])) > 10
for: 1m
labels:
  severity: critical
annotations:
  summary: "AI API 正在被限速"
  description: "过去 1 分钟限速次数: {{ $value }}"

# 成本告警

- alert: AIAPIHighCost

expr: sum(ai_api_cost_dollars) > 100
for: 1h
labels:
  severity: warning
annotations:
  summary: "AI API 成本超过 $100/小时"
  description: "当前成本: ${{ $value }}"

```

---

最佳实践

#### 1. 黄金信号（Four Golden Signals）

对于 AI API，监控这四个信号就够了：

#### 2. 告警分级

#### 3. 避免告警疲劳

• 只对真正重要的事情告警
• 聚合相似告警
• 设置适当的持续时间（for: 5m）
• 定期清理无用的告警规则

#### 4. 成本监控

• 按模型、按用户、按功能拆分成本
• 设置每日/每周预算告警
• 监控成本趋势，发现异常增长
• 定期 review 成本，优化不必要的调用

---

总结

AI API 监控是生产系统稳定运行的保障：

• ✅ 监控四个黄金信号：延迟、流量、错误、饱和度
• ✅ Metrics + Logs + Tracing 三者结合
• ✅ Prometheus + Grafana 是成熟的开源方案
• ✅ 告警分级，避免告警疲劳
• ✅ 成本监控同样重要

建议：

1. 从简单的 Metrics 开始，逐步完善

2. 先监控成功率和延迟，再添加成本和其他

3. 用 Grafana 做可视化，直观看到趋势

4. 告警规则要持续优化，避免误报和漏报

可在本站查看更多 AI API 中转平台，找到更稳定、更易监控的选择。