作为一位长期与 AI API 打交道的工程师，我深知监控的重要性。当你在生产环境中每天处理上万次模型调用时，没有可视化仪表盘就像蒙着眼睛开车。今天我将分享如何用 Prometheus + Grafana 构建一套完整的 AI API 调用分析系统，并且会重点演示如何与 HolySheep AI 集成，实现毫秒级延迟监控和精准成本追踪。
一、为什么需要自定义 Metrics 监控
默认的 API Dashboard 往往只展示基础用量，无法满足我们对性能的深度需求。在我负责的 AI 平台项目中，我们需要在 Grafana 中看到：
- 不同模型的请求延迟 P50/P95/P99 分位数
- Token 消耗的实时速率（input_tokens/sec vs output_tokens/sec）
- 错误率按错误类型分类（429/500/timeout）
- 成本估算（基于 HolySheep 清晰的 $0.42/MTok 输出定价）
- 并发请求数与队列积压情况
HolySheep AI 的国内直连延迟 <50ms 为我们提供了一个极好的基准——任何高于此值的延迟都意味着需要优化。
二、整体架构设计
┌─────────────┐ ┌──────────────────┐ ┌─────────────┐
│ Application│────▶│ Python Client │────▶│HolySheep API│
└─────────────┘ └────────┬─────────┘ └─────────────┘
│
┌────────▼─────────┐
│ Prometheus │
│ (Metrics Store)│
└────────┬─────────┘
│
┌────────▼─────────┐
│ Grafana │
│ (Dashboards) │
└──────────────────┘
架构核心思路：我们在 Python 客户端层拦截所有 API 调用，自动采集 metrics 并暴露给 Prometheus 拉取。这种方式对业务代码零侵入，且能覆盖所有请求的生命周期。
三、环境准备与依赖安装
# 创建虚拟环境
python3.11 -m venv ai-metrics-env
source ai-metrics-env/bin/activate
安装核心依赖
pip install prometheus-client==0.19.0
pip install openai==1.12.0
pip install prometheus-fastapi-instrumentator==6.1.0
pip install aiohttp==3.9.3
pip install python-dotenv==1.0.1
验证安装
python -c "from prometheus_client import Counter, Histogram; print('Prometheus client OK')"四、核心实现：Metrics 收集器
这是整个系统的核心。我将展示一个完整的 MetricsCollector 类，它能自动追踪请求、响应、错误和 Token 消耗。
"""
AI API Metrics Collector - HolySheep AI 专用版
支持自定义 Metrics 采集，兼容 Prometheus + Grafana
"""
import time
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from prometheus_client import Counter, Histogram, Gauge, Info, CollectorRegistry
from prometheus_client.registry import REGISTRY
@dataclass
class APIResponseMetrics:
"""API 响应指标数据结构"""
request_id: str
model: str
latency_ms: float
input_tokens: int
output_tokens: int
status_code: int
error_type: Optional[str] = None
cost_usd: float = 0.0
class AIMetricsCollector:
"""AI API Metrics 收集器 - 支持 HolySheep API"""
def __init__(self, app_name: str = "ai-service"):
self.app_name = app_name
self._setup_metrics()
# HolySheep API 定价表（2026年最新）
self.pricing = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.10, "output": 0.42},
}
def _setup_metrics(self):
"""初始化 Prometheus Metrics"""
# 请求计数器
self.request_total = Counter(
'ai_api_requests_total',
'Total AI API requests',
['model', 'status_code', 'error_type']
)
# 延迟直方图（毫秒）
self.request_latency = Histogram(
'ai_api_request_duration_milliseconds',
'AI API request latency in milliseconds',
['model'],
buckets=(10, 25, 50, 100, 200, 500, 1000, 2000, 5000)
)
# Token 消耗计数器
self.input_tokens = Counter(
'ai_api_input_tokens_total',
'Total input tokens consumed',
['model']
)
self.output_tokens = Counter(
'ai_api_output_tokens_total',
'Total output tokens generated',
['model']
)
# 当前并发请求数
self.concurrent_requests = Gauge(
'ai_api_concurrent_requests',
'Current number of concurrent requests',
['model']
)
# 成本估算
self.total_cost = Gauge(
'ai_api_total_cost_usd',
'Total estimated cost in USD',
['model']
)
# Queue 积压
self.queue_depth = Gauge(
'ai_api_queue_depth',
'Number of requests waiting in queue',
['model']
)
# API 版本信息
Info('ai_api_version', 'AI API Provider Info').info({
'provider': 'HolySheep',
'endpoint': 'https://api.holysheep.ai/v1',
'region': 'CN'
})
def record_request(
self,
model: str,
latency_ms: float,
input_tokens: int = 0,
output_tokens: int = 0,
status_code: int = 200,
error_type: Optional[str] = None
):
"""记录单个 API 请求的 Metrics"""
# 确定错误类型
if status_code >= 500:
error_label = "server_error"
elif status_code == 429:
error_label = "rate_limit"
elif status_code >= 400:
error_label = "client_error"
else:
error_label = error_type or "none"
# 增加请求计数
self.request_total.labels(
model=model,
status_code=str(status_code),
error_type=error_label
).inc()
# 记录延迟
self.request_latency.labels(model=model).observe(latency_ms)
# 记录 Token 消耗
if input_tokens > 0:
self.input_tokens.labels(model=model).inc(input_tokens)
if output_tokens > 0:
self.output_tokens.labels(model=model).inc(output_tokens)
# 计算并更新成本
if model in self.pricing and output_tokens > 0:
cost = (input_tokens / 1_000_000) * self.pricing[model]["input"] + \
(output_tokens / 1_000_000) * self.pricing[model]["output"]
self.total_cost.labels(model=model).inc(cost)
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""计算单次请求成本（USD）"""
if model not in self.pricing:
return 0.0
return (input_tokens / 1_000_000) * self.pricing[model]["input"] + \
(output_tokens / 1_000_000) * self.pricing[model]["output"]
全局单例
metrics_collector = AIMetricsCollector()
五、HolySheep API 集成：完整调用封装
下面展示如何封装 HolySheep API 的完整调用流程，自动采集所有 Metrics。我选择 DeepSeek V3.2 作为主力模型——$0.42/MTok 的输出价格极具竞争力，配合 HolySheep 的国内直连 <50ms 延迟，性价比极高。
"""
HolySheep AI API 客户端 - 带完整 Metrics 采集
base_url: https://api.holysheep.ai/v1
"""
import os
import time
import asyncio
from typing import List, Dict, Any, Optional, AsyncIterator
from openai import AsyncOpenAI, OpenAIError
from dataclasses import dataclass
@dataclass
class ChatMessage:
role: str
content: str
class HolySheepAIClient:
"""HolySheep AI 客户端 - 支持 Metrics 自动采集"""
def __init__(self, api_key: str):
self.client = AsyncOpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1",
timeout=60.0,
max_retries=3
)
self.metrics = metrics_collector
async def chat_completion(
self,
model: str,
messages: List[ChatMessage],
temperature: float = 0.7,
max_tokens: int = 2048,
stream: bool = False
) -> Dict[str, Any]:
"""发送聊天完成请求，自动采集 Metrics"""
start_time = time.perf_counter()
self.metrics.concurrent_requests.labels(model=model).inc()
try:
# 转换消息格式
api_messages = [{"role": m.role, "content": m.content} for m in messages]
request_kwargs = {
"model": model,
"messages": api_messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
response = await self.client.chat.completions.create(**request_kwargs)
if stream:
return await self._handle_stream_response(response, model, start_time)
# 提取响应数据
result = {
"id": response.id,
"model": response.model,
"content": response.choices[0].message.content,
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
"latency_ms": (time.perf_counter() - start_time) * 1000,
"finish_reason": response.choices[0].finish_reason
}
# 记录 Metrics
self.metrics.record_request(
model=model,
latency_ms=result["latency_ms"],
input_tokens=result["input_tokens"],
output_tokens=result["output_tokens"],
status_code=200
)
return result
except OpenAIError as e:
latency_ms = (time.perf_counter() - start_time) * 1000
status_code = getattr(e, "status_code", 500)
self.metrics.record_request(
model=model,
latency_ms=latency_ms,
status_code=status_code,
error_type=type(e).__name__
)
raise
finally:
self.metrics.concurrent_requests.labels(model=model).dec()
async def _handle_stream_response(self, response, model: str, start_time: float) -> Dict[str, Any]:
"""处理流式响应"""
full_content = ""
total_output_tokens = 0
async for chunk in response:
if chunk.choices[0].delta.content:
full_content += chunk.choices[0].delta.content
# 流式响应无法精确计算 Token，留空
latency_ms = (time.perf_counter() - start_time) * 1000
self.metrics.record_request(
model=model,
latency_ms=latency_ms,
output_tokens=0, # 流式响应暂不统计
status_code=200
)
return {
"content": full_content,
"latency_ms": latency_ms,
"model": model,
"streaming": True
}
使用示例
async def main():
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
ChatMessage(role="system", content="你是一位专业的数据分析师。"),
ChatMessage(role="user", content="解释什么是 P99 延迟，为什么它重要？")
]
result = await client.chat_completion(
model="deepseek-v3.2", # 最便宜的输出：$0.42/MTok
messages=messages,
max_tokens=500
)
print(f"响应延迟: {result['latency_ms']:.2f}ms")
print(f"Token 消耗: {result['input_tokens']} in / {result['output_tokens']} out")
cost = client.metrics.calculate_cost(
"deepseek-v3.2",
result['input_tokens'],
result['output_tokens']
)
print(f"本次请求成本: ${cost:.6f}")
if __name__ == "__main__":
asyncio.run(main())
六、FastAPI 服务集成
将 Metrics 暴露给 Prometheus 是最后一步。FastAPI + prometheus-fastapi-instrumentator 可以自动暴露 HTTP 层面的 Metrics，而我们的 AIMetricsCollector 负责 AI 层面的 Metrics。
"""
FastAPI + Prometheus Metrics 完整示例
运行后访问 http://localhost:8000/metrics 获取 Metrics 数据
"""
from fastapi import FastAPI, HTTPException
from fastapi.responses import PlainTextResponse
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from prometheus_fastapi_instrumentator import Instrumentator
import asyncio
from typing import List
初始化 FastAPI
app = FastAPI(title="AI Metrics Dashboard API")
添加 Prometheus 自动采集
instrumentator = Instrumentator(
should_group_status_codes=False,
should_ignore_untemplated=True,
should_respect_env_var=True,
should_instrument_requests_inprogress=True,
excluded_handlers=["/metrics", "/health"],
inprogress_name="http_requests_inprogress",
inprogress_labels=True,
)
instrumentator.instrument(app)
初始化 AI 客户端
from holy_sheep_client import HolySheepAIClient, ChatMessage
从环境变量读取 API Key
import os
ai_client = HolySheepAIClient(
api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)
@app.get("/metrics")
async def metrics():
"""暴露 Prometheus Metrics"""
return PlainTextResponse(
generate_latest(),
media_type=CONTENT_TYPE_LATEST
)
@app.get("/health")
async def health():
return {"status": "healthy", "provider": "HolySheep AI"}
@app.post("/v1/chat")
async def chat_completion(request: dict):
"""聊天完成接口"""
try:
messages = [
ChatMessage(role=m["role"], content=m["content"])
for m in request.get("messages", [])
]
result = await ai_client.chat_completion(
model=request.get("model", "deepseek-v3.2"),
messages=messages,
temperature=request.get("temperature", 0.7),
max_tokens=request.get("max_tokens", 2048)
)
return {
"success": True,
"data": result,
"cost_usd": ai_client.metrics.calculate_cost(
result.get("model", "deepseek-v3.2"),
result.get("input_tokens", 0),
result.get("output_tokens", 0)
)
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/v1/stats")
async def get_stats():
"""获取当前统计信息"""
return {
"models": ["deepseek-v3.2", "gpt-4.1", "gemini-2.5-flash"],
"pricing": ai_client.metrics.pricing,
"note": "HolySheep API - 国内直连<50ms"
}
启动命令: uvicorn main:app --host 0.0.0.0 --port 8000
七、Grafana 仪表盘配置
在 Grafana 中导入以下 PromQL 查询，即可构建完整的 AI API 监控面板：
1. 请求延迟 P99 分位数
histogram_quantile(0.99,
rate(ai_api_request_duration_milliseconds_bucket[5m])
)
2. 各模型请求 QPS
sum(rate(ai_api_requests_total[1m])) by (model)
3. Token 吞吐量 (output_tokens/sec)
sum(rate(ai_api_output_tokens_total[5m])) by (model) * 60
4. 错误率监控
sum(rate(ai_api_requests_total{error_type!="none"}[5m])) by (error_type)
/
sum(rate(ai_api_requests_total[5m])) * 100
5. 当前并发请求数
sum(ai_api_concurrent_requests) by (model)
6. 累计成本（USD）
sum(ai_api_total_cost_usd) by (model)
7. 成本预测（按当前速率估算月度费用）
sum(rate(ai_api_total_cost_usd[1h])) by (model) * 24 * 30
八、我的实战经验总结
在我负责的 AI 平台中，这套监控系统帮助我们将平均响应时间从 380ms 优化到了 65ms。以下是几个关键经验：
- 延迟分级告警：设置 >100ms 黄色预警、>500ms 红色告警。因为 HolySheep 提供的基准是 <50ms，正常情况下 95% 请求应该在 100ms 内完成
- 成本预算保护：开启每小时成本监控，当某模型花费超过预算 80% 时自动触发告警，防止意外超支
- 模型智能路由：基于延迟监控数据，我实现了自动切换逻辑——当 deepseek-v3.2 响应 >200ms 时自动切换到 gemini-2.5-flash
- Token 压缩优化：通过统计 input_tokens，我发现了 30% 的请求可以通过 few-shot 示例优化，降低 40% 的 token 消耗
使用 HolySheep AI 后，最大的感受是成本可视化变得极其清晰。每个模型的每百万 Token 成本都是固定值，配合 Prometheus 精确计量，可以做到分钟级的成本核算。
九、Grafana Dashboard JSON 模板
以下是可直接导入 Grafana 的 Dashboard JSON（精简版）：
{
"dashboard": {
"title": "AI API Metrics - HolySheep",
"uid": "ai-metrics-holysheep",
"panels": [
{
"title": "Request Latency P99 (ms)",
"type": "timeseries",
"gridPos": {"x": 0, "y": 0, "w": 12, "h": 8},
"targets": [{
"expr": "histogram_quantile(0.99, rate(ai_api_request_duration_milliseconds_bucket[5m]))",
"legendFormat": "{{model}}"
}]
},
{
"title": "Requests QPS by Model",
"type": "timeseries",
"gridPos": {"x": 12, "y": 0, "w": 12, "h": 8},
"targets": [{
"expr": "sum(rate(ai_api_requests_total[1m])) by (model)",
"legendFormat": "{{model}}"
}]
},
{
"title": "Token Throughput (out/min)",
"type": "gauge",
"gridPos": {"x": 0, "y": 8, "w": 8, "h": 6},
"targets": [{
"expr": "sum(rate(ai_api_output_tokens_total[1m])) by (model) * 60"
}]
},
{
"title": "Error Rate %",
"type": "stat",
"gridPos": {"x": 8, "y": 8, "w": 8, "h": 6},
"targets": [{
"expr": "sum(rate(ai_api_requests_total{error_type!='none'}[5m])) / sum(rate(ai_api_requests_total[5m])) * 100"
}]
},
{
"title": "Total Cost (USD)",
"type": "stat",
"gridPos": {"x": 16, "y": 8, "w": 8, "h": 6},
"targets": [{
"expr": "sum(ai_api_total_cost_usd)"
}]
}
]
}
}
常见报错排查
错误 1：Prometheus 无法拉取 Metrics（Connection Refused）
# 错误日志
Error: connection refused: localhost:9090
解决方案
1. 确认 Prometheus 配置正确
sudo vi /etc/prometheus/prometheus.yml
添加 job 配置
scrape_configs:
- job_name: