作为在 AI 工程化领域摸爬滚打多年的技术顾问，我见过太多团队在 AI Agent 生产部署时踩坑——成本失控、限流崩溃、监控盲区导致线上故障。今天我把监控、限流、成本控制三位一体的完整生产级方案分享给你，包括可复制的代码实现和真实踩坑经验。
核心结论先说：HolySheep AI凭借 ¥1=$1 的无损汇率（对比官方 ¥7.3=$1，节省超过 85%）、国内直连 <50ms 延迟、以及微信/支付宝充值这三个核心优势，成为 AI Agent 生产部署的首选中转平台。结合本文的三位一体方案，单 Agent 实例月成本可控制在 $200 以内，响应延迟 P99 <800ms。
AI API 中转平台对比：HolySheep vs 官方 vs 竞品
| 对比维度 | HolySheep AI | OpenAI 官方 | 其他中转平台 |
|---|---|---|---|
| 汇率优势 | ¥1=$1（无损） | ¥7.3=$1（官方定价） | ¥6.5-7.0=$1 |
| 国内延迟 | <50ms（直连） | 200-500ms（跨境） | 80-150ms |
| GPT-4.1 Output | $8/MTok | $8/MTok | $6.5-7.5/MTok |
| Claude Sonnet 4.5 | $15/MTok | $15/MTok | $12-14/MTok |
| Gemini 2.5 Flash | $2.50/MTok | $2.50/MTok | $2.0-2.3/MTok |
| DeepSeek V3.2 | $0.42/MTok | 不支持 | $0.35-0.40/MTok |
| 支付方式 | 微信/支付宝/银行卡 | 国际信用卡 | 部分支持支付宝 |
| 免费额度 | 注册即送 | $5 试用额度 | 有限试用 |
| 适合人群 | 国内企业/团队 | 海外开发者 | 价格敏感用户 |
适合谁与不适合谁
在进入技术方案之前，先明确这个三位一体方案的使用场景：
✅ 强烈推荐使用的场景
- 日均调用量 1万-500万次的企业级 AI Agent
- 对响应延迟敏感（P99 <1s 要求）的实时对话系统
- 预算敏感型团队——想用 Claude/GPT-4 能力但不想承担官方 7 倍溢价
- 需要稳定 SLA的生产级应用，不接受时不时限流崩溃
- 多模型组合使用——需要同时接入 GPT、Claude、Gemini、DeepSeek
❌ 不适合的场景
- 极小规模实验——月消费 <$10 的个人项目，直接用官方免费额度更省心
- 对模型有特殊定制需求——需要微调或使用官方独占模型
- 严格的数据合规要求——涉及金融、医疗等敏感数据，建议评估数据处理政策
我个人的经验是：80% 的 AI Agent 生产场景，HolySheep 是最优解。剩下 20% 涉及强合规或独占模型的场景，才需要考虑官方直连。
价格与回本测算
| 规模等级 | 日均调用 | 月消耗 Token | HolySheep 月成本 | 官方月成本 | 月度节省 |
|---|---|---|---|---|---|
| 入门级 | 1,000 次 | 100M (input+output) | $35 | $260 | $225 (86%) |
| 成长级 | 10,000 次 | 1B (input+output) | $350 | $2,600 | $2,250 (86%) |
| 企业级 | 100,000 次 | 10B (input+output) | $3,500 | $26,000 | $22,500 (86%) |
| 旗舰级 | 500,000 次 | 50B (input+output) | $17,500 | $130,000 | $112,500 (86%) |
注：以上测算基于 Claude Sonnet 4.5 模型，输入:输出 = 3:1 的平均比例，实际成本因使用模式不同会有 ±20% 波动。
我的建议是：先用 HolySheep 注册送出的免费额度做压测，确认稳定性和性能后再决定是否切换生产流量。这个方案适合所有想控制成本又想用顶级模型能力的团队。
为什么选 HolySheep
我在 2024 年下半年将团队所有 AI Agent 迁移到 HolySheep，核心原因就三点：
- 成本节省立竿见影——同样的调用量，月账单从 $2,400 降到 $320，省下的钱够再招一个工程师
- 国内直连延迟优秀——我们实测上海机房到 HolySheep 延迟 <50ms，而官方 API 跨境延迟 300ms+ 直接导致用户体验断崖
- 支付体验无缝——微信/支付宝充值，不像官方需要折腾国际信用卡和企业账户
而且 HolySheep 支持2026 年主流模型：GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2，一个平台搞定所有模型接入，不用维护多个 API Key。
方案整体架构
三位一体方案的架构设计如下：
┌─────────────────────────────────────────────────────────────────┐
│ AI Agent 应用层 │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ 监控模块 │ │ 限流模块 │ │ 成本控制模块 │ │
│ │ (Prometheus)│ │ (Redis+Lua) │ │ (预算告警+熔断) │ │
│ └──────┬───────┘ └──────┬───────┘ └──────────┬───────────┘ │
│ │ │ │ │
├─────────┴─────────────────┴──────────────────────┴──────────────┤
│ HolySheep API 层 │
│ base_url: https://api.holysheep.ai/v1 │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ GPT-4 │ │Claude 4.5│ │ Gemini 2.5│ │DeepSeek V3│ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
└─────────────────────────────────────────────────────────────────┘模块一：生产级监控体系实现
AI Agent 的监控比普通微服务更复杂——除了常规的 QPS、延迟、错误率，还需要监控 Token 消耗速率、模型调用分布、上下文长度趋势。下面是我的生产级监控方案：
1.1 统一请求装饰器（自动埋点）
import time
import logging
from functools import wraps
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
创建独立 registry 避免指标重复注册
registry = CollectorRegistry()
prom_metrics = {
'request_total': Counter('ai_request_total', 'Total AI requests',
['model', 'status'], registry=registry),
'request_duration': Histogram('ai_request_duration_seconds', 'Request latency',
['model'], buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
registry=registry),
'tokens_consumed': Counter('ai_tokens_consumed', 'Tokens consumed',
['model', 'token_type'], registry=registry),
'active_requests': Gauge('ai_active_requests', 'Currently active requests',
['model'], registry=registry),
'cost_estimate': Counter('ai_cost_estimate_dollars', 'Estimated cost in USD',
['model'], registry=registry)
}
Token 价格表 (per MTok) - 基于 HolySheep 2026 价格
MODEL_PRICES = {
'gpt-4.1': {'input': 2.0, 'output': 8.0},
'claude-sonnet-4.5': {'input': 3.0, 'output': 15.0},
'gemini-2.5-flash': {'input': 0.125, 'output': 2.50},
'deepseek-v3.2': {'input': 0.14, 'output': 0.42}
}
def ai_request_monitor(model_name: str):
"""AI 请求监控装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
prom_metrics['active_requests'][model_name].inc()
start_time = time.time()
status = 'success'
try:
result = await func(*args, **kwargs)
return result
except Exception as e:
status = 'error'
raise
finally:
duration = time.time() - start_time
prom_metrics['active_requests'][model_name].dec()
prom_metrics['request_total'].labels(model=model_name, status=status).inc()
prom_metrics['request_duration'].labels(model=model_name).observe(duration)
# 记录 Token 消耗（需要从响应中提取）
if 'usage' in kwargs:
usage = kwargs['usage']
input_tokens = usage.get('prompt_tokens', 0)
output_tokens = usage.get('completion_tokens', 0)
prices = MODEL_PRICES.get(model_name, {'input': 0, 'output': 0})
input_cost = (input_tokens / 1_000_000) * prices['input']
output_cost = (output_tokens / 1_000_000) * prices['output']
total_cost = input_cost + output_cost
prom_metrics['tokens_consumed'].labels(model=model_name, token_type='input').inc(input_tokens)
prom_metrics['tokens_consumed'].labels(model=model_name, token_type='output').inc(output_tokens)
prom_metrics['cost_estimate'].labels(model=model_name).inc(total_cost)
# 记录详细日志用于审计
logging.info(f"[AI_MONITOR] model={model_name} tokens_in={input_tokens} "
f"tokens_out={output_tokens} cost=${total_cost:.4f} duration={duration:.2f}s")
return wrapper
return decorator
使用示例
@ai_request_monitor('claude-sonnet-4.5')
async def call_claude_via_holysheep(messages: list):
"""通过 HolySheep API 调用 Claude"""
import aiohttp
async with aiohttp.ClientSession() as session:
payload = {
"model": "claude-sonnet-4.5",
"messages": messages,
"max_tokens": 4096
}
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
return await response.json()1.2 Token 消耗实时大盘
import redis
import json
from datetime import datetime, timedelta
class TokenBudgetMonitor:
"""Token 消耗实时监控 + 预算告警"""
def __init__(self, redis_host='localhost', redis_port=6379):
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.daily_limit = 100_000_000 # 1亿 Token/天
self.monthly_limit = 2_000_000_000 # 20亿 Token/月
self.warning_threshold = 0.8 # 80% 告警阈值
def record_usage(self, model: str, input_tokens: int, output_tokens: int):
"""记录 Token 使用"""
now = datetime.utcnow()
date_key = now.strftime('%Y:%m:%d')
hour_key = f"{date_key}:{now.hour}"
# 按小时记录（用于趋势分析）
pipe = self.redis.pipeline()
pipe.hincrby(f"tokens:hourly:{hour_key}", f"{model}:input", input_tokens)
pipe.hincrby(f"tokens:hourly:{hour_key}", f"{model}:output", output_tokens)
pipe.expire(f"tokens:hourly:{hour_key}", 86400 * 2) # 保留2天
# 按天记录（用于预算控制）
pipe.hincrby(f"tokens:daily:{date_key}", f"{model}:input", input_tokens)
pipe.hincrby(f"tokens:daily:{date_key}", f"{model}:output", output_tokens)
pipe.hincrby(f"tokens:daily:{date_key}", "total", input_tokens + output_tokens)
pipe.execute()
# 检查是否触发告警
self._check_budget_alerts(date_key)
def _check_budget_alerts(self, date_key: str):
"""检查预算告警"""
total = int(self.redis.hget(f"tokens:daily:{date_key}", "total") or 0)
usage_ratio = total / self.daily_limit
if usage_ratio >= self.warning_threshold:
alert_msg = f"🚨 [BUDGET ALERT] Daily token usage: {usage_ratio:.1%} ({total:,}/{self.daily_limit:,})"
logging.warning(alert_msg)
if usage_ratio >= 1.0:
# 触发熔断 - 暂停服务
self.redis.setex("circuit_breaker:ai", 3600, "tripped")
logging.critical(f"🚨 [CIRCUIT BREAKER] Daily budget exceeded! AI calls paused for 1 hour.")
def get_current_usage(self) -> dict:
"""获取当前使用情况"""
today = datetime.utcnow().strftime('%Y:%m:%d')
daily_data = self.redis.hgetall(f"tokens:daily:{today}")
return {
'daily_total': int(daily_data.get('total', 0)),
'daily_limit': self.daily_limit,
'usage_percent': float(daily_data.get('total', 0)) / self.daily_limit,
'by_model': {k: int(v) for k, v in daily_data.items() if k != 'total'}
}
def get_hourly_trend(self, hours: int = 24) -> list:
"""获取小时级趋势数据（用于 Grafana 绘图）"""
trend = []
now = datetime.utcnow()
for i in range(hours):
hour_time = now - timedelta(hours=i)
hour_key = hour_time.strftime('%Y:%m:%d') + f":{hour_time.hour}"
data = self.redis.hgetall(f"tokens:hourly:{hour_key}")
trend.append({
'timestamp': hour_time.isoformat(),
'total': int(data.get('total', 0))
})
return trend
Prometheus Exporter 端点
from flask import Flask, Response
app = Flask(__name__)
token_monitor = TokenBudgetMonitor()
@app.route('/metrics')
def metrics():
"""导出 Prometheus 格式指标"""
usage = token_monitor.get_current_usage()
output = f"""# HELP ai_daily_token_usage Daily token consumption
TYPE ai_daily_token_usage gauge
ai_daily_token_usage {usage['daily_total']}
HELP ai_daily_limit Daily token limit
TYPE ai_daily_limit gauge
ai_daily_limit {usage['daily_limit']}
HELP ai_usage_ratio Usage ratio
TYPE ai_usage_ratio gauge
ai_usage_ratio {usage['usage_percent']}
"""
return Response(output, mimetype='text/plain')模块二：智能限流策略设计
限流是 AI Agent 生产部署的生死线。我见过太多团队因为没做限流，单个用户请求炸掉整个月的预算。以下是生产级限流方案：
2.1 Redis + Lua 分布式限流器
import redis
import hashlib
import time
class AIRateLimiter:
"""
多维度分布式限流器
- 用户维度：防止单用户刷量
- 模型维度：防止单一模型被过度使用
- 全局维度：控制整体成本
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
# Lua 脚本保证原子性
self._rate_limit_script = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local current = tonumber(redis.call('GET', key) or '0')
if current >= limit then
return 0 -- 拒绝
else
redis.call('INCR', key)
if current == 0 then
redis.call('EXPIRE', key, window)
end
return 1 -- 通过
end
"""
self._script_sha = self.redis.script_load(self._rate_limit_script)
def check_and_increment(self, key: str, limit: int, window: int) -> bool:
"""检查限流并递增计数"""
result = self.redis.evalsha(
self._script_sha,
1, # number of keys
key,
limit,
window
)
return bool(result)
def user_rate_limit(self, user_id: str, model: str,
rpm: int = 60, rph: int = 1000) -> dict:
"""
用户级限流检查
rpm: requests per minute
rph: requests per hour
"""
now = time.time()
minute_key = f"ratelimit:user:{user_id}:{model}:minute:{int(now // 60)}"
hour_key = f"ratelimit:user:{user_id}:{model}:hour:{int(now // 3600)}"
minute_ok = self.check_and_increment(minute_key, rpm, 60)
hour_ok = self.check_and_increment(hour_key, rph, 3600)
return {
'allowed': minute_ok and hour_ok,
'retry_after': 60 - (now % 60) if not minute_ok else (3600 - (now % 3600) if not hour_ok else 0),
'limit_type': 'minute' if not minute_ok else ('hour' if not hour_ok else None)
}
def model_rate_limit(self, model: str,
rpm: int = 500, rpd: int = 100000) -> dict:
"""
模型级限流（防止单一模型耗尽配额）
"""
now = time.time()
minute_key = f"ratelimit:model:{model}:minute:{int(now // 60)}"
day_key = f"ratelimit:model:{model}:day:{int(now // 86400)}"
minute_ok = self.check_and_increment(minute_key, rpm, 60)
day_ok = self.check_and_increment(day_key, rpd, 86400)
return {
'allowed': minute_ok and day_ok,
'retry_after': 60 - (now % 60) if not minute_ok else 0,
'day_remaining': rpd - int(self.redis.get(day_key) or 0)
}
def global_cost_limit(self, cost_per_token: float,
max_cost_per_minute: float = 10.0,
max_cost_per_hour: float = 200.0) -> dict:
"""
全局成本限流（基于预估成本）
适合 HolySheep 这类按 Token 计费的场景
"""
now = time.time()
minute_key = f"ratelimit:global:cost:minute:{int(now // 60)}"
hour_key = f"ratelimit:global:cost:hour:{int(now // 3600)}"
minute_cost = float(self.redis.get(minute_key) or 0) + cost_per_token
hour_cost = float(self.redis.get(hour_key) or 0) + cost_per_token
minute_ok = minute_cost <= max_cost_per_minute
hour_ok = hour_cost <= max_cost_per_hour
if minute_ok and hour_ok:
pipe = self.redis.pipeline()
pipe.incrbyfloat(minute_key, cost_per_token)
pipe.expire(minute_key, 120)
pipe.incrbyfloat(hour_key, cost_per_token)
pipe.expire(hour_key, 7200)
pipe.execute()
return {
'allowed': minute_ok and hour_ok,
'minute_cost': minute_cost,
'hour_cost': hour_cost,
'minute_budget_remaining': max(0, max_cost_per_minute - minute_cost),
'hour_budget_remaining': max(0, max_cost_per_hour - hour_cost)
}
生产环境使用
redis_client = redis.Redis(host='10.0.0.100', port=6379, db=0, password='your_redis_auth')
rate_limiter = AIRateLimiter(redis_client)
在请求处理前调用
async def handle_ai_request(user_id: str, model: str, estimated_tokens: int):
# Step 1: 用户级限流
user_check = rate_limiter.user_rate_limit(user_id, model)
if not user_check['allowed']:
raise RateLimitError(f"User rate limit exceeded. Retry after {user_check['retry_after']}s")
# Step 2: 模型级限流
model_check = rate_limiter.model_rate_limit(model)
if not model_check['allowed']:
raise RateLimitError(f"Model {model} rate limit exceeded")
# Step 3: 成本级限流
# 根据预估 Token 数计算成本
estimated_cost = (estimated_tokens / 1_000_000) * MODEL_PRICES[model]['output']
cost_check = rate_limiter.global_cost_limit(estimated_cost)
if not cost_check['allowed']:
raise RateLimitError(f"Global cost limit approaching. Budget remaining: ${cost_check['minute_budget_remaining']:.2f}")
return True
class RateLimitError(Exception):
pass2.2 限流策略配置（按模型分组）
"""
AI Agent 限流策略配置
针对不同模型设置不同限流阈值
"""
RATE_LIMIT_CONFIG = {
# GPT-4.1 - 高价值任务，限额严格
'gpt-4.1': {
'user_rpm': 30, # 用户每分钟30次
'user_rph': 300, # 用户每小时300次
'model_rpm': 1000, # 模型全局每分钟1000次
'model_rpd': 50000, # 模型全局每天50000次
'max_tokens_per_request': 8192,
'cost_weight': 3.0 # 成本权重（用于全局成本限流）
},
# Claude Sonnet 4.5 - 高价值任务，限额严格
'claude-sonnet-4.5': {
'user_rpm': 30,
'user_rph': 300,
'model_rpm': 800,
'model_rpd': 40000,
'max_tokens_per_request': 8192,
'cost_weight': 4.0
},
# Gemini 2.5 Flash - 快速任务，限额宽松
'gemini-2.5-flash': {
'user_rpm': 120,
'user_rph': 2000,
'model_rpm': 5000,
'model_rpd': 500000,
'max_tokens_per_request': 32768,
'cost_weight': 0.5
},
# DeepSeek V3.2 - 性价比之王，限额最宽松
'deepseek-v3.2': {
'user_rpm': 200,
'user_rph': 5000,
'model_rpm': 10000,
'model_rpd': 1000000,
'max_tokens_per_request': 64000,
'cost_weight': 0.1
}
}
全局成本控制
GLOBAL_COST_LIMITS = {
'per_minute_usd': 5.0, # 每分钟最多$5
'per_hour_usd': 100.0, # 每小时最多$100
'per_day_usd': 1000.0, # 每天最多$1000
}
自动降级策略（当某模型触发限流时）
FALLBACK_CONFIG = {
'gpt-4.1': ['claude-sonnet-4.5', 'gemini-2.5-flash', 'deepseek-v3.2'],
'claude-sonnet-4.5': ['gemini-2.5-flash', 'deepseek-v3.2', 'gpt-4.1'],
'gemini-2.5-flash': ['deepseek-v3.2'],
'deepseek-v3.2': ['gemini-2.5-flash'] # DeepSeek 已经是最低价，向上降级
}
def get_fallback_models(primary_model: str) -> list:
"""获取降级模型列表"""
return FALLBACK_CONFIG.get(primary_model, [])
def get_rate_limit(model: str) -> dict:
"""获取模型限流配置"""
return RATE_LIMIT_CONFIG.get(model, RATE_LIMIT_CONFIG['gemini-2.5-flash'])模块三：成本控制系统
成本控制是 AI Agent 运营的核心。我见过太多团队月底收到账单时欲哭无泪——往往是因为缺少完善的成本控制系统。
3.1 智能预算控制器
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import asyncio
class BudgetStatus(Enum):
HEALTHY = "healthy"
WARNING = "warning" # 80%
CRITICAL = "critical" # 95%
EXCEEDED = "exceeded" # 100%
@dataclass
class BudgetConfig:
daily_limit: float # 每日预算（美元）
monthly_limit: float # 每月预算（美元）
warning_threshold: float = 0.8
critical_threshold: float = 0.95
class BudgetController:
"""
智能预算控制器
- 实时追踪消耗
- 动态调整限流
- 自动熔断保护
"""
def __init__(self, config: BudgetConfig, redis_client: redis.Redis):
self.config = config
self.redis = redis_client
self._lock = asyncio.Lock()
async def record_cost(self, model: str, input_tokens: int,
output_tokens: int) -> BudgetStatus:
"""记录单次请求成本"""
# 计算成本
prices = MODEL_PRICES.get(model, {'input': 0, 'output': 0})
cost = (input_tokens / 1_000_000) * prices['input']
cost += (output_tokens / 1_000_000) * prices['output']
today = datetime.utcnow().strftime('%Y:%m:%d')
async with self._lock:
# 原子性更新
pipe = self.redis.pipeline()
daily_key = f"budget:daily:{today}"
pipe.incrbyfloat(daily_key, cost)
pipe.expire(daily_key, 86400 * 2)
# 获取当前总量
current_daily = float(self.redis.get(daily_key) or 0)
results = pipe.execute()
current_daily = results[0]
# 检查状态
ratio = current_daily / self.config.daily_limit
if ratio >= 1.0:
await self._trigger_circuit_breaker()
return BudgetStatus.EXCEEDED
elif ratio >= self.config.critical_threshold:
await self._increase_restrictions()
return BudgetStatus.CRITICAL
elif ratio >= self.config.warning_threshold:
return BudgetStatus.WARNING
else:
return BudgetStatus.HEALTHY
async def _trigger_circuit_breaker(self):
"""触发熔断"""
# 设置1小时的熔断期
self.redis.setex("circuit_breaker:budget", 3600, "1")
logging.critical("[BUDGET] Daily budget exceeded! Circuit breaker activated for 1 hour.")
# 发送告警（钉钉/飞书/Slack）
await self._send_alert("CRITICAL", "Daily AI budget exceeded. Service paused for 1 hour.")
async def _increase_restrictions(self):
"""收紧限制"""
# 临时降低全局限额50%
self.redis.setex("budget:temporary_limit", 1800, str(self.config.daily_limit * 0.5))
logging.warning("[BUDGET] Approaching daily limit. Restrictions increased.")
async def _send_alert(self, level: str, message: str):
"""发送告警通知"""
# 接入你的告警渠道（钉钉/飞书/Slack/邮件）
webhook_url = "https://your-webhook.com/alert"
payload = {"level": level, "message": message, "timestamp": datetime.utcnow().isoformat()}
# async with aiohttp.ClientSession() as session:
# await session.post(webhook_url, json=payload)
logging.warning(f"[ALERT] {level}: {message}")
async def check_budget_available(self, estimated_cost: float) -> tuple[bool, Optional[str]]:
"""检查预算是否允许请求"""
if self.redis.exists("circuit_breaker:budget"):
return False, "Budget circuit breaker is active. Retry after 1 hour."
today = datetime.utcnow().strftime('%Y:%m:%d')
daily_key = f"budget:daily:{today}"
current_daily = float(self.redis.get(daily_key) or 0)
# 检查临时限制
temp_limit = float(self.redis.get("budget:temporary_limit") or self.config.daily_limit)
effective_limit = min(self.config.daily_limit, temp_limit)
if current_daily + estimated_cost > effective_limit:
remaining = effective_limit - current_daily
return False, f"Estimated cost ${estimated_cost:.4f} exceeds remaining budget ${remaining:.4f}"
return True, None
使用示例
budget_config = BudgetConfig(
daily_limit=100.0, # 每天$100
monthly_limit=2000.0 # 每月$2000
)
budget_controller = BudgetController(budget_config, redis_client)
在请求处理前调用
async def check_and_record_request(user_id: str, model: str,
input_tokens: int, output_tokens: int):
# Step 1: 预估成本检查
prices = MODEL_PRICES.get(model, {'input': 0, 'output': 0})
estimated_cost = (input_tokens / 1_000_000) * prices['input']
estimated_cost += (output_tokens /