在大模型应用开发中，API 调用成本往往是最大的开销之一。一个设计良好的计费与监控系统，不仅能帮你精准掌控成本，还能为模型路由优化提供数据支撑。本文将手把手教你实现一套完整的大模型 API 网关计费系统。
一、为什么大模型 API 网关需要特殊的计费设计？
1.1 传统 API 网关 vs 大模型 API 网关
传统 API 网关的计费逻辑相对简单：
┌─────────────────────────────────────────────────────────────┐
│ 传统 API 网关 │
├─────────────────────────────────────────────────────────────┤
│ 请求次数 × 单价 = 费用 │
│ │
│ 特点： │
│ ✓ 请求大小固定（URL + Header + Body） │
│ ✓ 响应大小可预估 │
│ ✓ 无特殊计量维度 │
└─────────────────────────────────────────────────────────────┘大模型 API 网关则复杂得多：
┌─────────────────────────────────────────────────────────────┐
│ 大模型 API 网关 │
├─────────────────────────────────────────────────────────────┤
│ (输入Token + 输出Token) × 模型单价 = 费用 │
│ │
│ 特点： │
│ ⚠ 输入 Token 数量差异巨大（从几十到几万） │
│ ⚠ 输出 Token 需要实时统计（流式响应） │
│ ⚠ 不同模型单价差异可达 100 倍 │
│ ⚠ 缓存命中可大幅降低实际成本 │
│ ⚠ 并发请求下计量需要原子操作 │
└─────────────────────────────────────────────────────────────┘1.2 计费维度分析
大模型 API 的计费通常包含以下几个维度：
1.3 为什么需要一个专门的网关？
在生产环境中，我们通常需要：
统一入口：聚合多个模型提供商的 API
成本监控：实时掌握各项目、各模型的调用成本
限流保护：防止单个用户耗尽配额
路由优化：自动选择性价比最高的模型
审计追溯：记录每笔调用的详细成本
二、Token 计费模型设计
2.1 核心数据模型
# models/billing.py
"""
大模型 API 网关计费模块
核心数据模型定义
"""
from enum import Enum
from datetime import datetime
from typing import Optional
from dataclasses import dataclass, field
from decimal import Decimal
class ModelProvider(Enum):
"""模型提供商枚举"""
OPENAI = "openai"
ANTHROPIC = "anthropic"
AZURE_OPENAI = "azure_openai"
HUGGING_FACE = "hugging_face"
LOCAL = "local" # 本地部署模型
@dataclass
class ModelPricing:
"""模型定价配置"""
model_id: str
provider: ModelProvider
input_price_per_1k: Decimal # 每1000输入Token价格（美元）
output_price_per_1k: Decimal # 每1000输出Token价格（美元）
cache_hit_price_per_1k: Decimal = Decimal("0") # 缓存命中价格
@property
def total_price_per_1k(self) -> Decimal:
"""总价格（输入+输出）"""
return self.input_price_per_1k + self.output_price_per_1k
@dataclass
class ProjectQuota:
"""项目配额配置"""
project_id: str
project_name: str
monthly_budget_usd: Decimal # 月度预算（美元）
daily_request_limit: int = 10000 # 日请求上限
rate_limit_rpm: int = 60 # 每分钟请求数限制
# 不同模型的配额限制（可选）
model_quota_override: dict[str, int] = field(default_factory=dict)
@dataclass
class TokenUsage:
"""Token使用记录"""
request_id: str
project_id: str
model_id: str
provider: ModelProvider
# Token统计
input_tokens: int
output_tokens: int
cache_hit_tokens: int = 0
# 时间戳
timestamp: datetime = field(default_factory=datetime.utcnow)
# 成本计算
@property
def input_cost(self) -> Decimal:
"""输入Token成本"""
return Decimal(self.input_tokens) / 1000 * self._get_input_price()
@property
def output_cost(self) -> Decimal:
"""输出Token成本"""
return Decimal(self.output_tokens) / 1000 * self._get_output_price()
@property
def cache_savings(self) -> Decimal:
"""缓存节省的成本"""
if self.cache_hit_tokens > 0:
return Decimal(self.cache_hit_tokens) / 1000 * self._get_cache_price()
return Decimal("0")
@property
def total_cost(self) -> Decimal:
"""总成本"""
return self.input_cost + self.output_cost
def _get_input_price(self) -> Decimal:
# 实际实现中从配置获取
return Decimal("0.01") # 默认值
def _get_output_price(self) -> Decimal:
return Decimal("0.03") # 默认值
def _get_cache_price(self) -> Decimal:
return Decimal("0") # 缓存免费
@dataclass
class CostReport:
"""成本报告"""
project_id: str
period_start: datetime
period_end: datetime
# 汇总统计
total_requests: int = 0
total_input_tokens: int = 0
total_output_tokens: int = 0
total_cache_hit_tokens: int = 0
# 成本汇总
total_cost: Decimal = Decimal("0")
input_cost: Decimal = Decimal("0")
output_cost: Decimal = Decimal("0")
cache_savings: Decimal = Decimal("0")
# 按模型分解
cost_by_model: dict[str, Decimal] = field(default_factory=dict)
# 按日期分解
cost_by_day: dict[str, Decimal] = field(default_factory=dict)2.2 定价配置文件
# config/pricing.yaml
# 模型定价配置 - 按提供商分类
providers:
openai:
models:
- model_id: gpt-4o
input_price_per_1k: 0.005 # $5/1M tokens
output_price_per_1k: 0.015 # $15/1M tokens
description: "最新GPT-4模型，支持视觉"
- model_id: gpt-4o-mini
input_price_per_1k: 0.00015 # $0.15/1M tokens
output_price_per_1k: 0.0006 # $0.60/1M tokens
description: "轻量级GPT-4，性价比高"
- model_id: gpt-4-turbo
input_price_per_1k: 0.01 # $10/1M tokens
output_price_per_1k: 0.03 # $30/1M tokens
description: "GPT-4 Turbo，更快更便宜"
- model_id: gpt-3.5-turbo
input_price_per_1k: 0.0005 # $0.50/1M tokens
output_price_per_1k: 0.0015 # $1.50/1M tokens
description: "最便宜的GPT-3.5"
anthropic:
models:
- model_id: claude-3-5-sonnet-20241022
input_price_per_1k: 0.003 # $3/1M tokens
output_price_per_1k: 0.015 # $15/1M tokens
description: "Claude 3.5 Sonnet，性能出色"
- model_id: claude-3-opus
input_price_per_1k: 0.015 # $15/1M tokens
output_price_per_1k: 0.075 # $75/1M tokens
description: "最强大的Claude模型"
- model_id: claude-3-haiku
input_price_per_1k: 0.00025 # $0.25/1M tokens
output_price_per_1k: 0.00125 # $1.25/1M tokens
description: "轻量级Claude，速度快"
azure_openai:
endpoint_template: "https://{resource}.openai.azure.com/"
models:
- model_id: gpt-4o
input_price_per_1k: 0.004 # Azure定价
output_price_per_1k: 0.012
deployment_name: "gpt-4o"
- model_id: gpt-4-turbo
input_price_per_1k: 0.008
output_price_per_1k: 0.024
# 默认配额配置
default_quotas:
free_tier:
monthly_budget_usd: 5
daily_request_limit: 100
rate_limit_rpm: 10
pro_tier:
monthly_budget_usd: 100
daily_request_limit: 10000
rate_limit_rpm: 60
enterprise_tier:
monthly_budget_usd: 1000
daily_request_limit: 100000
rate_limit_rpm: 300
# 成本告警阈值
cost_alerts:
daily_warning_percent: 80 # 日预算80%告警
daily_critical_percent: 95 # 日预算95%严重告警
monthly_warning_percent: 70
monthly_critical_percent: 902.3 计费服务实现
# services/billing_service.py
"""
计费服务 - 核心计费逻辑实现
"""
import asyncio
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Optional
import json
import redis.asyncio as redis
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_
from models.billing import TokenUsage, CostReport, ModelPricing, ProjectQuota
from models.database import UsageRecord, ProjectBilling
class BillingService:
"""计费服务"""
def __init__(
self,
redis_client: redis.Redis,
db_session: AsyncSession,
pricing_config: dict
):
self.redis = redis_client
self.db = db_session
self.pricing = self._load_pricing(pricing_config)
# Redis键前缀
self.KEY_USAGE = "billing:usage:{project_id}:{date}"
self.KEY_QUOTA = "billing:quota:{project_id}"
self.KEY_RATE_LIMIT = "billing:ratelimit:{project_id}"
def _load_pricing(self, config: dict) -> dict[str, ModelPricing]:
"""加载定价配置"""
pricing = {}
for provider, provider_config in config.get("providers", {}).items():
for model_config in provider_config.get("models", []):
model_id = model_config["model_id"]
pricing[model_id] = ModelPricing(
model_id=model_id,
provider=ModelProvider(provider),
input_price_per_1k=Decimal(str(model_config["input_price_per_1k"])),
output_price_per_1k=Decimal(str(model_config["output_price_per_1k"])),
cache_hit_price_per_1k=Decimal(
str(model_config.get("cache_hit_price_per_1k", 0))
)
)
return pricing
def get_model_pricing(self, model_id: str) -> Optional[ModelPricing]:
"""获取模型定价"""
return self.pricing.get(model_id)
async def record_usage(
self,
request_id: str,
project_id: str,
model_id: str,
input_tokens: int,
output_tokens: int,
cache_hit_tokens: int = 0
) -> TokenUsage:
"""
记录一次Token使用
Args:
request_id: 请求ID
project_id: 项目ID
model_id: 模型ID
input_tokens: 输入Token数
output_tokens: 输出Token数
cache_hit_tokens: 缓存命中Token数
Returns:
TokenUsage对象
"""
pricing = self.get_model_pricing(model_id)
if not pricing:
raise ValueError(f"Unknown model: {model_id}")
# 计算成本
input_cost = Decimal(input_tokens) / 1000 * pricing.input_price_per_1k
output_cost = Decimal(output_tokens) / 1000 * pricing.output_price_per_1k
total_cost = input_cost + output_cost
# 创建使用记录
usage = TokenUsage(
request_id=request_id,
project_id=project_id,
model_id=model_id,
provider=pricing.provider,
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_hit_tokens=cache_hit_tokens
)
# 原子性增加Redis计数器
date_str = datetime.utcnow().strftime("%Y%m%d")
hour_str = datetime.utcnow().strftime("%Y%m%d%H")
pipeline = self.redis.pipeline()
# 1. 项目当日使用量
key_daily = f"billing:daily:{project_id}:{date_str}"
pipeline.hincrby(key_daily, "requests", 1)
pipeline.hincrby(key_daily, "input_tokens", input_tokens)
pipeline.hincrby(key_daily, "output_tokens", output_tokens)
pipeline.hincrby(key_daily, "cost", float(total_cost * 1000)) # 存毫精度
pipeline.expire(key_daily, 86400 * 7) # 保留7天
# 2. 项目当小时使用量（用于实时监控）
key_hourly = f"billing:hourly:{project_id}:{hour_str}"
pipeline.hincrby(key_hourly, "requests", 1)
pipeline.hincrby(key_hourly, "input_tokens", input_tokens)
pipeline.hincrby(key_hourly, "output_tokens", output_tokens)
pipeline.expire(key_hourly, 86400 * 2)
# 3. 模型维度统计
key_model = f"billing:model:{model_id}:{date_str}"
pipeline.hincrby(key_model, "requests", 1)
pipeline.hincrby(key_model, "input_tokens", input_tokens)
pipeline.hincrby(key_model, "output_tokens", output_tokens)
pipeline.hincrby(key_model, "cost", float(total_cost * 1000))
pipeline.expire(key_model, 86400 * 7)
await pipeline.execute()
# 4. 异步写入数据库（不阻塞主流程）
asyncio.create_task(self._persist_usage_record(usage))
return usage
async def _persist_usage_record(self, usage: TokenUsage):
"""持久化使用记录到数据库"""
try:
record = UsageRecord(
request_id=usage.request_id,
project_id=usage.project_id,
model_id=usage.model_id,
provider=usage.provider.value,
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
cache_hit_tokens=usage.cache_hit_tokens,
cost=usage.total_cost,
timestamp=usage.timestamp
)
self.db.add(record)
await self.db.commit()
except Exception as e:
# 记录失败不影响主流程
print(f"Failed to persist usage record: {e}")
async def check_quota(self, project_id: str, model_id: Optional[str] = None) -> bool:
"""
检查项目配额
Args:
project_id: 项目ID
model_id: 模型ID（可选，用于检查模型级别配额）
Returns:
是否允许请求
"""
# 1. 检查速率限制（分钟级）
rate_key = f"ratelimit:{project_id}"
current = await self.redis.incr(rate_key)
if current == 1:
await self.redis.expire(rate_key, 60)
if current > 60: # 默认RPM限制
return False
# 2. 检查日预算
date_str = datetime.utcnow().strftime("%Y%m%d")
daily_key = f"billing:daily:{project_id}:{date_str}"
daily_cost_milli = await self.redis.hget(daily_key, "cost")
if daily_cost_milli:
daily_cost = Decimal(daily_cost_milli) / 1000
# 获取项目日预算配置
daily_budget = await self._get_project_daily_budget(project_id)
if daily_cost >= daily_budget:
return False
# 3. 检查月预算
month_str = datetime.utcnow().strftime("%Y%m")
month_key = f"billing:monthly:{project_id}:{month_str}"
month_cost_milli = await self.redis.get(month_key)
if month_cost_milli:
month_cost = Decimal(month_cost_milli) / 1000
monthly_budget = await self._get_project_monthly_budget(project_id)
if month_cost >= monthly_budget:
return False
return True
async def _get_project_daily_budget(self, project_id: str) -> Decimal:
"""获取项目日预算（从数据库或缓存获取）"""
# 简化实现，实际应从数据库读取
cache_key = f"project:budget:{project_id}"
cached = await self.redis.get(cache_key)
if cached:
return Decimal(cached)
return Decimal("10") # 默认$10/天
async def _get_project_monthly_budget(self, project_id: str) -> Decimal:
"""获取项目月预算"""
cache_key = f"project:budget:monthly:{project_id}"
cached = await self.redis.get(cache_key)
if cached:
return Decimal(cached)
return Decimal("100") # 默认$100/月
async def get_project_daily_usage(self, project_id: str) -> dict:
"""获取项目当日使用统计"""
date_str = datetime.utcnow().strftime("%Y%m%d")
key = f"billing:daily:{project_id}:{date_str}"
data = await self.redis.hgetall(key)
if not data:
return {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"cost_usd": 0
}
return {
"requests": int(data.get(b"requests", 0)),
"input_tokens": int(data.get(b"input_tokens", 0)),
"output_tokens": int(data.get(b"output_tokens", 0)),
"cost_usd": float(data.get(b"cost", 0)) / 1000
}
async def generate_cost_report(
self,
project_id: str,
start_date: datetime,
end_date: datetime
) -> CostReport:
"""生成成本报告"""
report = CostReport(
project_id=project_id,
period_start=start_date,
period_end=end_date
)
# 从数据库查询汇总数据
query = select(
func.sum(UsageRecord.input_tokens).label("total_input"),
func.sum(UsageRecord.output_tokens).label("total_output"),
func.sum(UsageRecord.cache_hit_tokens).label("total_cache"),
func.sum(UsageRecord.cost).label("total_cost"),
func.count(UsageRecord.id).label("total_requests"),
UsageRecord.model_id
).where(
and_(
UsageRecord.project_id == project_id,
UsageRecord.timestamp >= start_date,
UsageRecord.timestamp <= end_date
)
).group_by(UsageRecord.model_id)
result = await self.db.execute(query)
rows = result.all()
for row in rows:
report.total_input_tokens += row.total_input or 0
report.total_output_tokens += row.total_output or 0
report.total_cache_hit_tokens += row.total_cache or 0
report.total_cost += row.total_cost or Decimal("0")
report.total_requests += row.total_requests or 0
if row.model_id:
report.cost_by_model[row.model_id] = row.total_cost or Decimal("0")
return report三、用量监控架构设计
3.1 整体架构图
┌─────────────────────────────────────────────────────────────────────────────┐
│ 大模型 API 网关监控系统 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │
│ │ Client A │ │ Client B │ │ Client C │ │ 更多... │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └─────┬─────┘ │
│ │ │ │ │ │
│ └───────────────────┼───────────────────┼──────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌───────────────────────────────────────────────────────────────────────┐ │
│ │ FastAPI Gateway │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Auth Middle │ │ Rate Limit │ │ Billing │ │ Router │ │ │
│ │ │ 认证鉴权 │ │ 限流控制 │ │ 计费记录 │ │ 路由选择 │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ │ │ │ │ │ │ │
│ │ └────────────────┼────────────────┼────────────────┘ │ │
│ │ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ Prometheus Metrics │ │ │
│ │ │ • llm_requests_total • llm_tokens_input_total │ │ │
│ │ │ • llm_tokens_output_total • llm_cost_usd_total │ │ │
│ │ │ • llm_request_duration_seconds • llm_cache_hit_ratio │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌───────────────────────────────┼───────────────────────────────┐ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Prometheus Server │ │ │
│ │ │ :9090 /metrics │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Grafana Dashboards │ │ │
│ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │
│ │ │ │ 成本概览 │ │ Token统计 │ │ 模型对比 │ │ │ │
│ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │
│ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │
│ │ │ │ 项目排行 │ │ 趋势分析 │ │ 告警中心 │ │ │ │
│ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘3.2 Prometheus 指标定义
# metrics/prometheus_metrics.py
"""
Prometheus 指标定义
大模型 API 网关专用的指标集合
"""
from prometheus_client import Counter, Histogram, Gauge, Info
from prometheus_client.exposition import generate_latest, CONTENT_TYPE_LATEST
# =============================================================================
# 计数器 (Counter) - 只会增加的值
# =============================================================================
# 请求相关计数器
llm_requests_total = Counter(
"llm_requests_total",
"LLM API 总请求数",
["project_id", "model_id", "provider", "status"]
)
llm_requests_by_endpoint = Counter(
"llm_requests_by_endpoint_total",
"按端点统计的请求数",
["endpoint", "method", "status"]
)
# Token计数器
llm_tokens_input_total = Counter(
"llm_tokens_input_total",
"LLM 输入 Token 总数",
["project_id", "model_id", "provider"]
)
llm_tokens_output_total = Counter(
"llm_tokens_output_total",
"LLM 输出 Token 总数",
["project_id", "model_id", "provider"]
)
llm_tokens_cached_total = Counter(
"llm_tokens_cached_total",
"缓存命中的 Token 总数",
["project_id", "model_id"]
)
# 成本计数器
llm_cost_input_usd = Counter(
"llm_cost_input_usd_total",
"输入 Token 成本（美元）",
["project_id", "model_id", "provider"]
)
llm_cost_output_usd = Counter(
"llm_cost_output_usd_total",
"输出 Token 成本（美元）",
["project_id", "model_id", "provider"]
)
llm_cost_total_usd = Counter(
"llm_cost_total_usd_total",
"LLM 调用总成本（美元）",
["project_id", "model_id", "provider"]
)
# 错误计数器
llm_errors_total = Counter(
"llm_errors_total",
"LLM 调用错误总数",
["project_id", "model_id", "error_type"]
)
llm_rate_limit_hits_total = Counter(
"llm_rate_limit_hits_total",
"限流触发次数",
["project_id"]
)
# =============================================================================
# 直方图 (Histogram) - 用于计算分位数和分布
# =============================================================================
# 请求延迟
llm_request_duration_seconds = Histogram(
"llm_request_duration_seconds",
"LLM API 请求耗时（秒）",
["project_id", "model_id", "provider", "endpoint"],
buckets=(0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0)
)
# Token 分布
llm_tokens_per_request = Histogram(
"llm_tokens_per_request",
"每次请求的 Token 数量分布",
["model_id", "type"], # type: input 或 output
buckets=(10, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 50000)
)
# 成本分布
llm_cost_per_request_usd = Histogram(
"llm_cost_per_request_usd",
"每次请求的成本分布（美元）",
["project_id", "model_id"],
buckets=(0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0)
)
# =============================================================================
# 仪表盘 (Gauge) - 可以增加或减少的值
# =============================================================================
# 当前配额使用
quota_usage_percent = Gauge(
"quota_usage_percent",
"当前配额使用百分比",
["project_id", "quota_type"] # quota_type: daily, monthly
)
# 活跃请求数
active_requests = Gauge(
"llm_active_requests",
"当前正在处理的请求数",
["model_id"]
)
# 模型健康状态
model_health_status = Gauge(
"model_health_status",
"模型健康状态 (1=健康, 0=异常)",
["model_id", "provider"]
)
# 缓存命中率
cache_hit_ratio = Gauge(
"cache_hit_ratio",
"缓存命中率",
["cache_type"] # cache_type: semantic, exact
)
# =============================================================================
# 信息指标 (Info) - 静态信息
# =============================================================================
model_info = Info(
"llm_model",
"LLM 模型信息"
)
gateway_info = Info(
"llm_gateway",
"API 网关信息"
)
# =============================================================================
# 指标收集器类
# =============================================================================
class MetricsCollector:
"""指标收集器 - 提供便捷的指标记录方法"""
@staticmethod
def record_request(
project_id: str,
model_id: str,
provider: str,
status: str,
duration: float
):
"""记录一次请求"""
llm_requests_total.labels(
project_id=project_id,
model_id=model_id,
provider=provider,
status=status
).inc()
llm_request_duration_seconds.labels(
project_id=project_id,
model_id=model_id,
provider=provider,
endpoint="chat/completions"
).observe(duration)
@staticmethod
def record_tokens(
project_id: str,
model_id: str,
provider: str,
input_tokens: int,
output_tokens: int,
cached_tokens: int = 0,
cost_usd: float = 0
):
"""记录 Token 使用"""
llm_tokens_input_total.labels(
project_id=project_id,
model_id=model_id,
provider=provider
).inc(input_tokens)
llm_tokens_output_total.labels(
project_id=project_id,
model_id=model_id,
provider=provider
).inc(output_tokens)
if cached_tokens > 0:
llm_tokens_cached_total.labels(
project_id=project_id,
model_id=model_id
).inc(cached_tokens)
if cost_usd > 0:
llm_cost_total_usd.labels(
project_id=project_id,
model_id=model_id,
provider=provider
).inc(cost_usd)
@staticmethod
def record_error(
project_id: str,
model_id: str,
error_type: str
):
"""记录错误"""
llm_errors_total.labels(
project_id=project_id,
model_id=model_id,
error_type=error_type
).inc()
@staticmethod
def record_rate_limit(project_id: str):
"""记录限流触发"""
llm_rate_limit_hits_total.labels(
project_id=project_id
).inc()3.3 FastAPI 中间件集成
# middleware/billing_middleware.py
"""
计费中间件 - FastAPI 集成
"""
import time
import uuid
from typing import Callable
from decimal import Decimal
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from metrics.prometheus_metrics import MetricsCollector
from services.billing_service import BillingService
class BillingMiddleware(BaseHTTPMiddleware):
"""
计费中间件
功能：
1. 生成请求ID
2. 统计请求时间
3. 提取 Token 使用信息
4. 记录 Prometheus 指标
5. 上报到计费服务
"""
async def dispatch(
self,
request: Request,
call_next: Callable
) -> Response:
# 生成请求ID
request_id = str(uuid.uuid4())
request.state.request_id = request_id
# 记录开始时间
start_time = time.time()
# 获取项目ID（从header或认证信息）
project_id = request.headers.get("X-Project-ID", "default")
request.state.project_id = project_id
# 尝试获取模型ID
model_id = "unknown"
input_tokens = 0
output_tokens = 0
# 处理请求
try:
response = await call_next(request)
# 计算耗时
duration = time.time() - start_time
# 从响应header获取使用信息
# 实际实现中，LLM响应会携带这些信息
input_tokens = int(response.headers.get("X-Usage-Input-Tokens", 0))
output_tokens = int(response.headers.get("X-Usage-Output-Tokens", 0))
model_id = response.headers.get("X-Model-ID", model_id)
# 记录Prometheus指标
MetricsCollector.record_request(
project_id=project_id,
model_id=model_id,
provider="openai", # 从响应获取更准确
status="success",
duration=duration
)
if input_tokens or output_tokens:
MetricsCollector.record_tokens(
project_id=project_id,
model_id=model_id,
provider="openai",
input_tokens=input_tokens,
output_tokens=output_tokens
)
# 添加请求ID到响应头
response.headers["X-Request-ID"] = request_id
return response
except Exception as e:
duration = time.time() - start_time
# 记录错误
MetricsCollector.record_request(
project_id=project_id,
model_id=model_id,
provider="openai",
status="error",
duration=duration
)
MetricsCollector.record_error(
project_id=project_id,
model_id=model_id,
error_type=type(e).__name__
)
return JSONResponse(
status_code=500,
content={
"error": "Internal server error",
"request_id": request_id,
"message": str(e)
}
)
class TokenUsageMiddleware(BaseHTTPMiddleware):
"""
Token 使用量统计中间件
专门用于解析 LLM 响应中的 usage 信息
"""
def __init__(self, app, billing_service: BillingService):
super().__init__(app)
self.billing_service = billing_service
async def dispatch(
self,
request: Request,
call_next: Callable
) -> Response:
response = await call_next(request)
# 如果是流式响应，需要特殊处理
if response.headers.get("transfer-encoding") == "chunked":
# 流式响应暂不统计（需要在流结束后统计）
return response
# 从响应body或header提取usage
# 这里简化处理，实际应解析JSON响应
project_id = getattr(request.state, "project_id", "default")
request_id = getattr(request.state, "request_id", "")
# 模拟获取使用量
# 实际实现中应该解析LLM响应
usage = {
"input_tokens": int(response.headers.get("X-Input-Tokens", 0)),
"output_tokens": int(response.headers.get("X-Output-Tokens", 0)),
"cache_hit_tokens": int(response.headers.get("X-Cache-Hit-Tokens", 0))
}
if usage["input_tokens"] > 0 or usage["output_tokens"] > 0:
model_id = response.headers.get("X-Model-ID", "unknown")
# 异步记录到计费服务（不阻塞响应）
asyncio.create_task(
self.billing_service.record_usage(
request_id=request_id,
project_id=project_id,
model_id=model_id,
input_tokens=usage["input_tokens"],
output_tokens=usage["output_tokens"],
cache_hit_tokens=usage["cache_hit_tokens"]
)
)
return response四、配额管理与限流策略
4.1 多级限流架构
┌─────────────────────────────────────────────────────────────────────────────┐
│ 多级限流架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 请求入口 │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ Level 1: 全局限流 (Global) │ │
│ │ ┌──────────────────────────────────────────────────────────────────┐ │ │
│ │ │ • 限制总QPS: 10000 req/s │ │ │
│ │ │ • 使用Redis滑动窗口算法 │ │ │
│ │ │ • 防止系统过载 │ │ │
│ │ └──────────────────────────────────────────────────────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ Level 2: 项目限流 (Project) │ │
│ │ ┌──────────────────────────────────────────────────────────────────┐ │ │
│ │ │ • RPM (Requests Per Minute): 60 │ │ │
│ │ │ • TPM (Tokens Per Minute): 100000 │ │ │
│ │ │ • 日预算: $10 │ │ │
│ │ │ • 月预算: $100 │ │ │
│ │ └──────────────────────────────────────────────────────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ Level 3: 模型限流 (Model) │ │
│ │ ┌──────────────────────────────────────────────────────────────────┐ │ │
│ │ │ • GPT-4o: 100 req/min (昂贵模型) │ │ │
│ │ │ • GPT-3.5: 1000 req/min (便宜模型) │ │ │
│ │ │ • Claude-3.5: 200 req/min │ │ │
│ │ └──────────────────────────────────────────────────────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 到达 LLM 提供商 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘4.2 限流配置
# config/rate_limits.yaml
# 限流配置
global:
# 全局QPS限制
max_qps: 10000
# 全局并发限制
max_concurrent_requests: 500
# 限流算法
algorithm: "sliding_window" # sliding_window | token_bucket | fixed_window
# 滑动窗口大小（秒）
window_size_seconds: 60
tiers:
free:
rpm: 10 # 每分钟请求数
tpm: 10000 # 每分钟Token数
rpd: 100 # 每天请求数
daily_budget_usd: 1 # 日预算
monthly_budget_usd: 5 # 月预算
standard:
rpm: 60
tpm: 100000
rpd: 10000
daily_budget_usd: 10
monthly_budget_usd: 100
pro:
rpm: 300
tpm: 500000
rpd: 100000
daily_budget_usd: 50
monthly_budget_usd: 500
enterprise:
rpm: 1000
tpm: 2000000
rpd: 1000000
daily_budget_usd: 500
monthly_budget_usd: 5000
# 模型级别限流
model_limits:
gpt-4o:
rpm: 100
tpm: 50000
gpt-4-turbo:
rpm: 200
tpm: 100000
gpt-4o-mini:
rpm: 1000
tpm: 500000
gpt-3.5-turbo:
rpm: 2000
tpm: 1000000
claude-3-5-sonnet-20241022:
rpm: 200
tpm: 100000
claude-3-opus:
rpm: 50
tpm: 20000
# 限流响应配置
rate_limit_response:
status_code: 429
headers:
X-RateLimit-Limit: true # 显示限制
X-RateLimit-Remaining: true # 剩余请求数
X-RateLimit-Reset: true # 重置时间
Retry-After: true # 等待秒数4.3 限流器实现
# services/rate_limiter.py
"""
限流器实现 - 多级限流
"""
import time
import asyncio
from typing import Optional
from enum import Enum
import redis.asyncio as redis
import yaml
class RateLimitType(Enum):
"""限流类型"""
GLOBAL = "global"
PROJECT = "project"
MODEL = "model"
class SlidingWindowRateLimiter:
"""
滑动窗口限流器
算法说明：
使用Redis有序集合实现滑动窗口算法
每个请求作为集合中的一个元素，时间戳为分数
统计窗口内的元素数量来判断是否超限
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
async def is_allowed(
self,
key: str,
limit: int,
window_seconds: int
) -> tuple[bool, int, int]:
"""
检查是否允许请求
Args:
key: 限流键（project_id, model_id等）
limit: 限制数量
window_seconds: 窗口大小（秒）
Returns:
(是否允许, 剩余请求数, 重置时间戳)
"""
now = time.time()
window_start = now - window_seconds
# 使用Redis事务保证原子性
pipe = self.redis.pipeline()
# 1. 删除窗口外的旧记录
pipe.zremrangebyscore(key, 0, window_start)
# 2. 统计当前窗口内请求数
pipe.zcard(key)
# 3. 添加当前请求
pipe.zadd(key, {f"{now}::{asyncio.get_event_loop().time()}": now})
# 4. 设置过期时间
pipe.expire(key, window_seconds + 1)
results = await pipe.execute()
current_count = results[1]
# 计算剩余请求数和重置时间
remaining = max(0, limit - current_count - 1)
reset_time = int(now + window_seconds)
return current_count < limit, remaining, reset_time
class RateLimiter:
"""多级限流器"""
def __init__(
self,
redis_client: redis.Redis,
config_path: str = "config/rate_limits.yaml"
):
self.redis = redis_client
self.sliding_limiter = SlidingWindowRateLimiter(redis_client)
# 加载配置
with open(config_path) as f:
self.config = yaml.safe_load(f)
# 配置缓存
self._project_tier_cache = {}
self._cache_ttl = 300 # 5分钟
async def check_rate_limit(
self,
project_id: str,
model_id: Optional[str] = None,
token_count: int = 0
) -> tuple[bool, dict]:
"""
检查限流
Args:
project_id: 项目ID
model_id: 模型ID
token_count: 请求的Token数量
Returns:
(是否允许, 限流信息)
"""
result = {
"allowed": True,
"limit_type": None,
"remaining": 0,
"reset": 0,
"retry_after": 0
}
# 1. 检查全局限流
global_allowed, remaining, reset = await self._check_global_limit()
if not global_allowed:
return False, {
"allowed": False,
"limit_type": "global",
"remaining": remaining,
"reset": reset,
"retry_after": reset - int(time.time())
}
# 2. 检查项目RPM
rpm_allowed, rpm_remaining, rpm_reset = await self._check_project_rpm(
project_id
)
if not rpm_allowed:
return False, {
"allowed": False,
"limit_type": "project_rpm",
"remaining": rpm_remaining,
"reset": rpm_reset,
"retry_after": rpm_reset - int(time.time())
}
# 3. 检查项目TPM
tpm_allowed, tpm_remaining, tpm_reset = await self._check_project_tpm(
project_id,
token_count
)
if not tpm_allowed:
return False, {
"allowed": False,
"limit_type": "project_tpm",
"remaining": tpm_remaining,
"reset": tpm_reset,
"retry_after": tpm_reset - int(time.time())
}
# 4. 检查模型限流
if model_id:
model_allowed, model_remaining, model_reset = await self._check_model_limit(
model_id,
token_count
)
if not model_allowed:
return False, {
"allowed": False,
"limit_type": f"model_{model_id}",
"remaining": model_remaining,
"reset": model_reset,
"retry_after": model_reset - int(time.time())
}
return True, result
async def _check_global_limit(self) -> tuple[bool, int, int]:
"""检查全局限流"""
global_config = self.config.get("global", {})
max_qps = global_config.get("max_qps", 10000)
window_size = global_config.get("window_size_seconds", 60)
return await self.sliding_limiter.is_allowed(
key="ratelimit:global",
limit=max_qps,
window_seconds=window_size
)
async def _check_project_rpm(
self,
project_id: str
) -> tuple[bool, int, int]:
"""检查项目RPM"""
tier = await self._get_project_tier(project_id)
rpm = tier.get("rpm", 60)
return await self.sliding_limiter.is_allowed(
key=f"ratelimit:project:{project_id}:rpm",
limit=rpm,
window_seconds=60
)
async def _check_project_tpm(
self,
project_id: str,
token_count: int
) -> tuple[bool, int, int]:
"""检查项目TPM"""
tier = await self._get_project_tier(project_id)
tpm = tier.get("tpm", 100000)
# 使用Redis HINCRBY来累加Token数
key = f"ratelimit:project:{project_id}:tpm"
now = time.time()
window_start = now - 60
pipe = self.redis.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zadd(key, {f"{now}::{token_count}": now})
pipe.zrangebyscore(key, window_start, now)
pipe.expire(key, 61)
results = await pipe.execute()
current_tokens = sum(
int(item.split("::")[1])
for item in results[2]
)
remaining = max(0, tpm - current_tokens - token_count)
reset_time = int(now + 60)
return current_tokens + token_count <= tpm, remaining, reset_time
async def _check_model_limit(
self,
model_id: str,
token_count: int
) -> tuple[bool, int, int]:
"""检查模型限流"""
model_limits = self.config.get("model_limits", {}).get(model_id, {})
rpm = model_limits.get("rpm", 100)
key = f"ratelimit:model:{model_id}:rpm"
return await self.sliding_limiter.is_allowed(
key=key,
limit=rpm,
window_seconds=60
)
async def _get_project_tier(self, project_id: str) -> dict:
"""获取项目等级配置"""
# 简单的缓存实现
cache_key = f"project:tier:{project_id}"
cached = await self.redis.get(cache_key)
if cached:
import json
return json.loads(cached)
# 从数据库或配置获取
# 这里简化处理
tier = self.config.get("tiers", {}).get("standard", {})
# 缓存
await self.redis.setex(
cache_key,
self._cache_ttl,
json.dumps(tier)
)
return tier五、多模型成本对比与路由优化
5.1 模型成本对比表
# config/model_comparison.yaml
# 模型成本对比分析
models:
- id: gpt-4o
provider: openai
input_cost_per_1k: 0.005
output_cost_per_1k: 0.015
total_cost_per_1k: 0.020
context_window: 128000
capabilities:
- text
- vision
- function_call
best_for: "复杂推理、多模态任务"
performance_rank: 5
- id: gpt-4o-mini
provider: openai
input_cost_per_1k: 0.00015
output_cost_per_1k: 0.0006
total_cost_per_1k: 0.00075
context_window: 128000
capabilities:
- text
- function_call
best_for: "简单问答、批量处理"
performance_rank: 3
- id: claude-3-5-sonnet-20241022
provider: anthropic
input_cost_per_1k: 0.003
output_cost_per_1k: 0.015
total_cost_per_1k: 0.018
context_window: 200000
capabilities:
- text
- vision
- extended_thinking
best_for: "长文本分析、代码生成"
performance_rank: 5
# 路由规则配置
routing_rules:
# 按任务类型路由
task_based:
- task: "simple_qa"
models: ["gpt-4o-mini", "claude-3-haiku"]
fallback: "claude-3-5-sonnet-20241022"
- task: "code_generation"
models: ["claude-3-5-sonnet-20241022", "gpt-4o"]
fallback: "gpt-4-turbo"
- task: "complex_reasoning"
models: ["gpt-4o", "claude-3-5-sonnet-20241022"]
fallback: "gpt-4o"
- task: "long_context"
models: ["claude-3-5-sonnet-20241022"]
fallback: "claude-3-5-sonnet-20241022"
# 按复杂度路由
complexity_based:
threshold_low: 500 # <500 tokens -> 便宜模型
threshold_high: 2000 # >2000 tokens -> 昂贵模型
between: "中等模型"
# 成本优化路由
cost_optimized:
enabled: true
max_cost_per_1k: 0.005
prefer_cache: true
batch_similar_requests: true5.2 成本优化路由服务
# services/cost_router.py
"""
成本优化路由服务
根据成本、性能、可用性选择最优模型
"""
import json
import yaml
from typing import Optional
from decimal import Decimal
import redis.asyncio as redis
class CostAwareRouter:
"""成本感知路由器"""
def __init__(
self,
redis_client: redis.Redis,
config_path: str = "config/model_comparison.yaml"
):
self.redis = redis_client
# 加载配置
with open(config_path) as f:
self.config = yaml.safe_load(f)
self.models = {
m["id"]: m for m in self.config.get("models", [])
}
async def select_model(
self,
task_type: Optional[str] = None,
prompt_tokens: int = 0,
max_cost_per_1k: Optional[float] = None,
prefer_latency: bool = False
) -> tuple[str, dict]:
"""
选择最优模型
Args:
task_type: 任务类型
prompt_tokens: 提示词Token数
max_cost_per_1k: 最大成本限制
prefer_latency: 是否优先考虑延迟
Returns:
(模型ID, 选择原因)
"""
candidates = []
# 1. 根据任务类型筛选
if task_type:
candidates = self._get_models_by_task(task_type)
else:
candidates = list(self.models.keys())
# 2. 过滤成本超限的模型
if max_cost_per_1k:
candidates = [
m for m in candidates
if self.models[m]["total_cost_per_1k"] <= max_cost_per_1k
]
# 3. 检查模型可用性
available = []
for model_id in candidates:
if await self._is_model_available(model_id):
available.append(model_id)
if not available:
# 所有模型都不可用，使用fallback
return "claude-3-5-sonnet-20241022", {"reason": "fallback"}
# 4. 根据prompt长度选择
if prompt_tokens > 100000:
# 超长上下文优先选择支持大窗口的模型
for model_id in available:
if self.models[model_id]["context_window"] >= 128000:
return model_id, {"reason": "long_context"}
# 5. 根据复杂度路由
if prompt_tokens < 500:
# 简单任务选择便宜的模型
candidates = [
m for m in available
if self.models[m]["performance_rank"] <= 3
]
if candidates:
return min(
candidates,
key=lambda m: self.models[m]["total_cost_per_1k"]
), {"reason": "cost_optimized"}
# 6. 综合评分选择
return self._select_by_score(available, prefer_latency)
def _get_models_by_task(self, task_type: str) -> list[str]:
"""根据任务类型获取候选模型"""
rules = self.config.get("routing_rules", {}).get("task_based", [])
for rule in rules:
if rule["task"] == task_type:
return rule["models"]
return list(self.models.keys())
async def _is_model_available(self, model_id: str) -> bool:
"""检查模型是否可用"""
# 从Redis获取健康状态
health_key = f"model:health:{model_id}"
status = await self.redis.get(health_key)
if status is None:
# 默认可用
return True
return status == b"healthy"
def _select_by_score(
self,
candidates: list[str],
prefer_latency: bool
) -> tuple[str, dict]:
"""
综合评分选择模型
评分公式：score = performance_rank * (1 - cost_weight) + cost * cost_weight
如果 prefer_latency 为 true，则考虑延迟因素
"""
cost_weight = 0.6 if prefer_latency else 0.8
scores = {}
for model_id in candidates:
model = self.models[model_id]
# 性能分数（越低越好，rank 1-5）
perf_score = model["performance_rank"] / 5.0
# 成本分数（标准化）
cost_score = model["total_cost_per_1k"] / 0.020 # 以最高成本为基准
# 综合分数
scores[model_id] = (
perf_score * (1 - cost_weight) +
cost_score * cost_weight
)
# 选择分数最低的
selected = min(scores, key=scores.get)
return selected, {"reason": "score_based", "score": scores[selected]}
async def get_cost_comparison(
self,
model_ids: list[str],
input_tokens: int,
output_tokens: int
) -> dict:
"""
获取多模型成本对比
Returns:
各模型的成本对比数据
"""
result = {}
total_tokens = input_tokens + output_tokens
for model_id in model_ids:
if model_id not in self.models:
continue
model = self.models[model_id]
input_cost = (input_tokens / 1000) * model["input_cost_per_1k"]
output_cost = (output_tokens / 1000) * model["output_cost_per_1k"]
total_cost = input_cost + output_cost
result[model_id] = {
"provider": model["provider"],
"input_cost": round(input_cost, 6),
"output_cost": round(output_cost, 6),
"total_cost": round(total_cost, 6),
"cost_per_1k": model["total_cost_per_1k"],
"best_for": model["best_for"]
}
return result
async def get_savings_report(
self,
baseline_model: str,
actual_model: str,
input_tokens: int,
output_tokens: int
) -> dict:
"""
计算节省报告
比较使用实际模型vs基准模型的成本差异
"""
comparison = await self.get_cost_comparison(
[baseline_model, actual_model],
input_tokens,
output_tokens
)
if baseline_model not in comparison or actual_model not in comparison:
return {"error": "Invalid model IDs"}
baseline_cost = comparison[baseline_model]["total_cost"]
actual_cost = comparison[actual_model]["total_cost"]
savings = baseline_cost - actual_cost
savings_percent = (savings / baseline_cost * 100) if baseline_cost > 0 else 0
return {
"baseline_model": baseline_model,
"actual_model": actual_model,
"baseline_cost_usd": baseline_cost,
"actual_cost_usd": actual_cost,
"savings_usd": savings,
"savings_percent": round(savings_percent, 2)
}六、Grafana Dashboard 配置
6.1 Dashboard JSON
{
"annotations": {
"list": []
},
"editable": true,
"fiscalYearStartMonth": 0,
"graphTooltip": 0,
"id": null,
"links": [],
"liveNow": false,
"panels": [
{
"title": "总成本（今日）",
"type": "stat",
"gridPos": {"h": 4, "w": 6, "x": 0, "y": 0},
"targets": [
{
"expr": "sum(increase(llm_cost_total_usd_total[24h]))",
"legendFormat": "Total Cost"
}
],
"fieldConfig": {
"defaults": {
"unit": "currencyUSD",
"thresholds": {
"mode": "absolute",
"steps": [
{"color": "green", "value": null},
{"color": "yellow", "value": 100},
{"color": "red", "value": 500}
]
}
}
}
},
{
"title": "Token使用量（今日）",
"type": "stat",
"gridPos": {"h": 4, "w": 6, "x": 6, "y": 0},
"targets": [
{
"expr": "sum(increase(llm_tokens_input_total[24h])) + sum(increase(llm_tokens_output_total[24h]))",
"legendFormat": "Total Tokens"
}
]
},
{
"title": "请求数（今日）",
"type": "stat",
"gridPos": {"h": 4, "w": 6, "x": 12, "y": 0},
"targets": [
{
"expr": "sum(increase(llm_requests_total[24h]))",
"legendFormat": "Total Requests"
}
]
},
{
"title": "缓存命中率",
"type": "gauge",
"gridPos": {"h": 4, "w": 6, "x": 18, "y": 0},
"targets": [
{
"expr": "sum(rate(llm_tokens_cached_total[1h])) / (sum(rate(llm_tokens_input_total[1h])) + 0.001)"
}
],
"fieldConfig": {
"defaults": {
"unit": "percentunit",
"thresholds": {
"mode": "absolute",
"steps": [
{"color": "red", "value": null},
{"color": "yellow", "value": 0.1},
{"color": "green", "value": 0.3}
]
}
}
}
},
{
"title": "成本趋势",
"type": "timeseries",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 4},
"targets": [
{
"expr": "sum by (project_id) (increase(llm_cost_total_usd_total[1h]))",
"legendFormat": "{{project_id}}"
}
],
"options": {
"legend": {"displayMode": "table"},
"tooltip": {"mode": "multi"}
}
},
{
"title": "各模型成本占比",
"type": "piechart",
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 4},
"targets": [
{
"expr": "sum by (model_id) (increase(llm_cost_total_usd_total[24h]))",
"legendFormat": "{{model_id}}"
}
]
},
{
"title": "Token分布",
"type": "histogram",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 12},
"targets": [
{
"expr": "llm_tokens_per_request_bucket",
"legendFormat": "{{model_id}} - {{type}}"
}
]
},
{
"title": "P99 响应延迟",
"type": "timeseries",
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 12},
"targets": [
{
"expr": "histogram_quantile(0.99, sum(rate(llm_request_duration_seconds_bucket[5m])) by (le, model_id))",
"legendFormat": "P99 - {{model_id}}"
}
],
"options": {
"legend": {"displayMode": "table"},
"yAxis": {"unit": "s"}
}
}
],
"refresh": "30s",
"schemaVersion": 38,
"style": "dark",
"tags": ["llm", "billing", "monitoring"],
"templating": {
"list": [
{
"name": "project",
"type": "query",
"query": "label_values(llm_requests_total, project_id)"
}
]
},
"time": {
"from": "now-24h",
"to": "now"
},
"title": "LLM API Gateway - 成本监控",
"uid": "llm-billing-dashboard",
"version": 1
}七、成本日报生成
# scripts/daily_cost_report.py
"""
成本日报生成脚本
每天定时执行，生成成本报告并发送邮件
"""
import asyncio
import json
from datetime import datetime, timedelta
from typing import List
import asyncpg
import redis.asyncio as redis
from jinja2 import Template
import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class CostReportGenerator:
"""成本日报生成器"""
def __init__(self, config: dict):
self.config = config
# Redis连接
self.redis = redis.Redis(
host=config["redis"]["host"],
port=config["redis"]["port"],
decode_responses=True
)
async def generate_daily_report(self, date: datetime) -> dict:
"""生成日报数据"""
date_str = date.strftime("%Y%m%d")
# 1. 获取所有项目
project_keys = await self.redis.keys(f"billing:daily:*:{date_str}")
projects = []
total_cost = 0
total_requests = 0
total_tokens = 0
for key in project_keys:
# 提取project_id
parts = key.split(":")
project_id = parts[2]
data = await self.redis.hgetall(key)
cost = float(data.get("cost", 0)) / 1000
requests = int(data.get("requests", 0))
input_tokens = int(data.get("input_tokens", 0))
output_tokens = int(data.get("output_tokens", 0))
project_info = {
"project_id": project_id,
"cost_usd": round(cost, 4),
"requests": requests,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens
}
# 获取项目配额使用情况
quota_key = f"project:quota:{project_id}"
quota_data = await self.redis.hgetall(quota_key)
if quota_data:
daily_budget = float(quota_data.get("daily_budget", 100))
project_info["budget_usage_percent"] = round(
cost / daily_budget * 100, 2
)
projects.append(project_info)
total_cost += cost
total_requests += requests
total_tokens += input_tokens + output_tokens
# 按成本排序
projects.sort(key=lambda x: x["cost_usd"], reverse=True)
# 2. 获取各模型使用统计
model_keys = await self.redis.keys(f"billing:model:*:{date_str}")
models = []
for key in model_keys:
parts = key.split(":")
model_id = parts[2]
data = await self.redis.hgetall(key)
models.append({
"model_id": model_id,
"cost_usd": round(float(data.get("cost", 0)) / 1000, 4),
"requests": int(data.get("requests", 0)),
"input_tokens": int(data.get("input_tokens", 0)),
"output_tokens": int(data.get("output_tokens", 0))
})
models.sort(key=lambda x: x["cost_usd"], reverse=True)
# 3. 获取缓存统计
cache_key = f"billing:cache:{date_str}"
cache_data = await self.redis.hgetall(cache_key)
cache_stats = {
"hit_count": int(cache_data.get("hit_count", 0)),
"hit_tokens": int(cache_data.get("hit_tokens", 0)),
"hit_rate": 0.0
}
if total_tokens > 0:
cache_stats["hit_rate"] = round(
cache_stats["hit_tokens"] / total_tokens * 100, 2
)
# 4. 生成报告
report = {
"date": date.strftime("%Y-%m-%d"),
"generated_at": datetime.utcnow().isoformat(),
"summary": {
"total_cost_usd": round(total_cost, 4),
"total_requests": total_requests,
"total_tokens": total_tokens,
"avg_cost_per_request": round(
total_cost / total_requests, 6
) if total_requests > 0 else 0,
"avg_cost_per_1k_tokens": round(
total_cost / (total_tokens / 1000), 4
) if total_tokens > 0 else 0
},
"top_projects": projects[:10],
"models": models,
"cache_stats": cache_stats
}
return report
def render_html(self, report: dict) -> str:
"""渲染HTML报告"""
template = """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>LLM 成本日报 - {{ date }}</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.header { background: #1a73e8; color: white; padding: 20px; border-radius: 8px; }
.summary { display: flex; gap: 20px; margin: 20px 0; }
.card { background: #f5f5f5; padding: 15px; border-radius: 8px; flex: 1; }
.card h3 { margin: 0 0 10px 0; color: #666; font-size: 14px; }
.card .value { font-size: 24px; font-weight: bold; color: #333; }
table { width: 100%; border-collapse: collapse; margin: 20px 0; }
th, td { padding: 12px; text-align: left; border-bottom: 1px solid #ddd; }
th { background: #f5f5f5; font-weight: 600; }
.cost { color: #e91e63; font-weight: bold; }
.warning { background: #fff3cd; }
.alert { background: #f8d7da; }
</style>
</head>
<body>
<div class="header">
<h1>LLM API 成本日报</h1>
<p>日期: {{ date }}</p>
</div>
<div class="summary">
<div class="card">
<h3>总成本</h3>
<div class="value">${{ "%.4f"|format(summary.total_cost_usd) }}</div>
</div>
<div class="card">
<h3>总请求数</h3>
<div class="value">{{ "{:,}".format(summary.total_requests) }}</div>
</div>
<div class="card">
<h3>总 Token 数</h3>
<div class="value">{{ "{:,}".format(summary.total_tokens) }}</div>
</div>
<div class="card">
<h3>缓存命中率</h3>
<div class="value">{{ "%.1f"|format(cache_stats.hit_rate) }}%</div>
</div>
</div>
<h2>Top 10 项目</h2>
<table>
<tr>
<th>项目ID</th>
<th>成本</th>
<th>请求数</th>
<th>Token数</th>
<th>预算使用</th>
</tr>
{% for project in top_projects %}
<tr class="{{ 'warning' if project.budget_usage_percent > 80 else '' }}">
<td>{{ project.project_id }}</td>
<td class="cost">${{ "%.4f"|format(project.cost_usd) }}</td>
<td>{{ "{:,}".format(project.requests) }}</td>
<td>{{ "{:,}".format(project.total_tokens) }}</td>
<td>{{ "%.1f"|format(project.budget_usage_percent|default(0)) }}%</td>
</tr>
{% endfor %}
</table>
<h2>模型使用统计</h2>
<table>
<tr>
<th>模型</th>
<th>成本</th>
<th>请求数</th>
<th>输入Token</th>
<th>输出Token</th>
</tr>
{% for model in models %}
<tr>
<td>{{ model.model_id }}</td>
<td class="cost">${{ "%.4f"|format(model.cost_usd) }}</td>
<td>{{ "{:,}".format(model.requests) }}</td>
<td>{{ "{:,}".format(model.input_tokens) }}</td>
<td>{{ "{:,}".format(model.output_tokens) }}</td>
</tr>
{% endfor %}
</table>
</body>
</html>
"""
return Template(template).render(**report)
async def send_report(self, report: dict, recipients: List[str]):
"""发送报告邮件"""
html = self.render_html(report)
message = MIMEMultipart("alternative")
message["Subject"] = f"LLM 成本日报 - {report['date']}"
message["From"] = self.config["email"]["from"]
message["To"] = ", ".join(recipients)
html_part = MIMEText(html, "html")
message.attach(html_part)
await aiosmtplib.send(
message,
hostname=self.config["email"]["smtp_host"],
port=self.config["email"]["smtp_port"],
username=self.config["email"]["username"],
password=self.config["email"]["password"]
)
async def main():
"""主函数"""
config = {
"redis": {
"host": "localhost",
"port": 6379
},
"email": {
"from": "billing@example.com",
"smtp_host": "smtp.example.com",
"smtp_port": 587,
"username": "billing@example.com",
"password": "password"
}
}
generator = CostReportGenerator(config)
# 生成昨天（昨天）的日报
yesterday = datetime.utcnow() - timedelta(days=1)
report = await generator.generate_daily_report(yesterday)
# 保存JSON
with open(f"cost_report_{yesterday.strftime('%Y%m%d')}.json", "w") as f:
json.dump(report, f, indent=2)
# 发送邮件
await generator.send_report(
report,
["admin@example.com", "finance@example.com"]
)
print(f"Report generated: cost_report_{yesterday.strftime('%Y%m%d')}.json")
if __name__ == "__main__":
asyncio.run(main())八、总结
本文详细介绍了大模型 API 网关的计费与监控系统的设计实现，主要包括：
8.1 核心要点回顾
Token 计费模型
支持按输入/输出 Token 分别计费
支持按模型、项目、时间周期多维度统计
使用 Redis 实现高性能计数器
用量监控架构
Prometheus + Grafana 经典组合
自定义 LLM 专用指标
多视图 Dashboard（成本、Token、延迟、缓存）
配额管理
多级限流（全局 → 项目 → 模型）
滑动窗口算法实现精准限流
支持 RPM/TPM/日预算/月预算多维度限制
成本优化
模型成本对比分析
智能路由选择最优模型
成本节省报告
自动化报告
每日成本日报自动生成
HTML 格式邮件发送
支持自定义模板
8.2 扩展方向
实时告警：接入 AlertManager，实现超预算自动告警
成本预测：基于历史数据预测月度成本
A/B 测试：支持模型对比实验
配额预警：提前通知用户配额使用情况
退款处理：处理因质量问题导致的费用退还
8.3 注意事项
精度问题：金额计算使用
Decimal类型，避免浮点精度丢失原子性：Redis 操作使用 Pipeline 保证原子性
异步处理：数据库写入异步化，不阻塞主流程
缓存失效：监控配置变更时及时刷新缓存
数据保留：合理设置数据保留周期，平衡存储成本
评论区