一次惨痛的生产事故
去年11月的一个周五晚上，我们产品的AI摘要功能突然开始返回乱码。用户群里炸了锅，客服那边一单接一单地转过来。
我第一时间登录服务器查日志，结果发现：日志里只有一行 response.ok，没有任何请求细节、没有模型版本、没有token用量、没有响应时间。排查了两个小时，最后才发现是OpenAI那边悄悄更新了API返回格式，我们的解析逻辑直接炸了。
那天晚上我加班到凌晨三点，发誓一定要把AI API的可观测性搞起来。这次事故让我深刻认识到：调用AI API和调用普通HTTP接口完全不一样，你需要一套专门的监控体系。
AI API可观测性的三大支柱
可观测性（Observability）这个词在云原生领域已经喊了好几年了，但放到AI API场景下，很多人还是一头雾水。说白了就三件事：
1. 日志（Logs）：记录"发生了什么"
日志是最基础的信息。对AI API来说，你需要记录的不只是HTTP状态码，还包括：
- 请求参数：prompt内容、模型名称、temperature、max_tokens
- 响应数据：返回文本、finish_reason（是正常结束还是被截断）
- Token用量：prompt_tokens、completion_tokens、total_tokens
- 耗时：首Token延迟（TTFT）、总响应时间
- 错误信息：错误码、错误类型、重试次数
2. 指标（Metrics）：告诉你"整体状况如何"
日志是细粒度的，指标是聚合后的。你需要关注的核心指标包括：
- P50/P95/P99延迟：分别代表50%、95%、99%的请求在多少毫秒内完成
- 错误率：4xx和5xx错误的占比，按错误码细分
- Token消耗速率：每分钟/每小时消耗多少token
- 成本趋势：按模型、按接口维度的花费变化
- 请求量（QPS）：每秒请求数，帮助做容量规划
3. 追踪（Traces）：帮你还原"一次调用的完整链路"
在AI应用中，一次用户请求可能触发多次API调用——先做意图识别，再检索知识库，最后生成回答。Traces把这些调用串成一条链路，让你一眼看出瓶颈在哪。
比如你发现某次请求耗时3秒，通过Trace可以拆解出：意图识别用了200ms、向量检索用了800ms、LLM生成用了2000ms。问题一目了然。
开源监控工具选型
市面上已经有不少针对AI API的监控工具，我实际用过三个，说说各自的优缺点。
| 工具 | 类型 | 核心优势 | 适用场景 |
|---|---|---|---|
| Helicone | SaaS + 开源 | 一行代码接入，OpenAI兼容性好 | 快速上手，小团队 |
| LangFuse | 开源自托管 | 支持Prompt版本管理、评估打分 | 需要数据隐私的中大团队 |
| OpenTelemetry | 标准框架 | 厂商中立，生态最丰富 | 已有可观测性基建的团队 |
如果你的团队刚开始搞AI API监控，我建议先用Helicone快速跑起来，后面有需要再迁移到LangFuse或自建OTel方案。别一上来就搞太复杂的架构，先把数据采到再说。
Python日志中间件：从零搭建
下面这段代码是我实际在用的日志中间件，支持记录请求/响应/延迟/token用量/模型/错误码，直接复制就能用。
import time
import json
import logging
from functools import wraps
logger = logging.getLogger("ai_api")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
'%(asctime)s | %(levelname)s | %(message)s'
))
logger.addHandler(handler)
def ai_api_logger(func):
"""AI API调用日志装饰器，记录完整的请求/响应信息"""
@wraps(func)
def wrapper(*args, **kwargs):
request_id = kwargs.get("request_id", "unknown")
model = kwargs.get("model", "unknown")
start_time = time.time()
# 记录请求信息
logger.info(
f"[{request_id}] REQUEST | model={model} | "
f"prompt_length={len(str(kwargs.get('messages', '')))}"
)
try:
response = func(*args, **kwargs)
elapsed_ms = (time.time() - start_time) * 1000
# 提取token用量
usage = getattr(response, 'usage', None)
token_info = ""
if usage:
token_info = (
f"prompt_tokens={usage.prompt_tokens} | "
f"completion_tokens={usage.completion_tokens} | "
f"total_tokens={usage.total_tokens}"
)
# 提取finish_reason
choices = getattr(response, 'choices', [])
finish_reason = choices[0].finish_reason if choices else "unknown"
logger.info(
f"[{request_id}] RESPONSE | status=ok | "
f"latency_ms={elapsed_ms:.0f} | {token_info} | "
f"finish_reason={finish_reason}"
)
# 发送指标到Prometheus（可选）
record_metrics(
model=model,
latency_ms=elapsed_ms,
prompt_tokens=usage.prompt_tokens if usage else 0,
completion_tokens=usage.completion_tokens if usage else 0,
status="success"
)
return response
except Exception as e:
elapsed_ms = (time.time() - start_time) * 1000
error_code = getattr(e, 'status_code', 'unknown')
error_type = type(e).__name__
logger.error(
f"[{request_id}] ERROR | latency_ms={elapsed_ms:.0f} | "
f"error_code={error_code} | error_type={error_type} | "
f"message={str(e)[:200]}"
)
record_metrics(
model=model,
latency_ms=elapsed_ms,
prompt_tokens=0,
completion_tokens=0,
status=f"error_{error_code}"
)
raise
return wrapper
def record_metrics(model, latency_ms, prompt_tokens,
completion_tokens, status):
"""将指标推送到Prometheus Pushgateway"""
try:
from prometheus_client import (
Counter, Histogram, CollectorRegistry, push_to_gateway
)
registry = CollectorRegistry()
REQUEST_COUNT = Counter(
'ai_api_requests_total', 'Total requests',
['model', 'status'], registry=registry
)
LATENCY = Histogram(
'ai_api_latency_seconds', 'Request latency',
['model'], registry=registry
)
TOKEN_USAGE = Counter(
'ai_api_tokens_total', 'Token usage',
['model', 'type'], registry=registry
)
REQUEST_COUNT.labels(model=model, status=status).inc()
LATENCY.labels(model=model).observe(latency_ms / 1000)
TOKEN_USAGE.labels(model=model, type='prompt').inc(prompt_tokens)
TOKEN_USAGE.labels(model=model, type='completion').inc(
completion_tokens
)
push_to_gateway(
'localhost:9091', job='ai_api', registry=registry
)
except ImportError:
pass # prometheus_client未安装时静默跳过
# 使用示例
@ai_api_logger
def call_openai(messages, model="gpt-4o", request_id=None, **kwargs):
from openai import OpenAI
client = OpenAI()
return client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
这段代码的核心设计思路是装饰器模式，你只需要在现有的API调用函数上加一个 @ai_api_logger 就能自动采集所有关键数据。不需要改业务逻辑，零侵入。
核心监控指标详解
数据采到了，接下来要知道看什么。我总结了几个最关键的指标和它们的"危险阈值"：
| 指标 | 正常范围 | 警告阈值 | 说明 |
|---|---|---|---|
| P50延迟 | < 500ms | > 1s | 一半请求的响应时间 |
| P95延迟 | < 2s | > 5s | 95%请求的响应时间 |
| P99延迟 | < 5s | > 10s | 长尾请求的响应时间 |
| 错误率 | < 1% | > 3% | 4xx+5xx占总请求比例 |
| Token消耗速率 | 稳定 | 突增50%+ | 可能存在异常循环调用 |
| 单次成本 | < $0.05 | > $0.10 | 单次请求平均花费 |
这里特别提醒一下P99延迟。很多人只看平均值，但平均值会被大量正常请求拉低，掩盖掉少数极端慢的请求。P99才是用户体验的真正底线——如果1%的用户等了10秒以上，这些人大概率已经关掉页面了。
Grafana看板配置示例
有了Prometheus采集数据，下一步就是用Grafana做可视化。下面是一个实用的看板配置JSON片段，你可以直接导入到Grafana中：
{
"dashboard": {
"title": "AI API 监控看板",
"panels": [
{
"title": "请求量 & 错误率",
"type": "timeseries",
"targets": [
{
"expr": "sum(rate(ai_api_requests_total[5m])) by (model)",
"legendFormat": "{{model}} - QPS"
},
{
"expr": "sum(rate(ai_api_requests_total{status=~\"error.*\"}[5m])) / sum(rate(ai_api_requests_total[5m])) * 100",
"legendFormat": "错误率 %"
}
]
},
{
"title": "P50/P95/P99 延迟",
"type": "timeseries",
"targets": [
{
"expr": "histogram_quantile(0.50, sum(rate(ai_api_latency_seconds_bucket[5m])) by (le, model))",
"legendFormat": "P50"
},
{
"expr": "histogram_quantile(0.95, sum(rate(ai_api_latency_seconds_bucket[5m])) by (le, model))",
"legendFormat": "P95"
},
{
"expr": "histogram_quantile(0.99, sum(rate(ai_api_latency_seconds_bucket[5m])) by (le, model))",
"legendFormat": "P99"
}
]
},
{
"title": "Token 消耗趋势",
"type": "timeseries",
"targets": [
{
"expr": "sum(rate(ai_api_tokens_total[1h])) by (model, type)",
"legendFormat": "{{model}} - {{type}}"
}
]
}
]
}
}
这个看板包含三个核心面板：请求量与错误率趋势、延迟分位数分布、Token消耗趋势。实际使用中我还加了一个"按模型维度的成本排行"面板，每天早上看一眼就能知道钱花在哪了。
告警策略：别等用户投诉了才知道
光有看板不够，你得让系统主动告诉你出问题了。我配置了三类告警规则：
延迟突增告警
# P95延迟连续5分钟超过5秒
- alert: AI_API_HIGH_LATENCY
expr: histogram_quantile(0.95, sum(rate(ai_api_latency_seconds_bucket[5m])) by (le)) > 5
for: 5m
labels:
severity: warning
annotations:
summary: "AI API P95延迟超过5秒"
description: "当前P95延迟: {{ $value }}s"
错误率飙升告警
# 错误率连续3分钟超过5%
- alert: AI_API_HIGH_ERROR_RATE
expr: sum(rate(ai_api_requests_total{status=~"error.*"}[3m])) / sum(rate(ai_api_requests_total[3m])) > 0.05
for: 3m
labels:
severity: critical
annotations:
summary: "AI API错误率超过5%"
description: "可能存在上游服务异常或API Key失效"
成本异常告警
# 小时级Token消耗超过平时均值的200%
- alert: AI_API_COST_ANOMALY
expr: sum(increase(ai_api_tokens_total[1h])) > 2 * avg_over_time(sum(increase(ai_api_tokens_total[1h]))[7d:])
for: 1h
labels:
severity: warning
annotations:
summary: "Token消耗异常增长"
description: "当前小时消耗超过7天均值的2倍，请检查是否存在异常调用"
告警渠道我用的企业微信机器人 + 邮件双通道。延迟和错误率走企业微信（需要快速响应），成本异常走邮件（不紧急但需要关注）。千万别把所有告警都发到同一个群，否则就是"狼来了"，时间长了没人看。
踩坑提醒：告警阈值不要一上来就设得太敏感。我第一次把延迟告警设在P95 > 2s就触发，结果每天收到几十条告警，全是OpenAI高峰期的正常波动。后来调到5s，误报率直接降到接近零。
从监控到优化：数据驱动决策
可观测性体系跑起来之后，我们发现了很多之前完全不知道的问题：
- 有15%的请求使用了GPT-4o，但其实用GPT-4o-mini就能搞定，每月白白多花$2000+
- 凌晨2-4点的请求延迟比白天高3倍，因为我们的重试逻辑在高峰期会堆积
- 某个客户的API调用频率异常高，单个用户消耗了我们8%的Token配额
这些发现直接帮我们优化了模型路由策略、调整了限流配置、重新设计了缓存层。一个月下来，API成本降了35%，用户侧的P95延迟从4.2s降到了1.8s。
如果你想了解更多AI API平台的选型和对比，可以看看 TokenNexus 上收录的330+国内外AI API平台，里面有详细的定价、功能、性能数据，帮你快速找到最适合的服务。