为什么需要成本控制？

调用大模型 API 按 token 计费，失控的调用可能在一夜之间产生数千元账单。常见失控场景包括：用户构造超长输入（token 炸弹）、代码 bug 导致循环调用、爬虫批量刷接口、以及模型返回超长响应等。成本控制不是可选项，而是生产系统的基础设施。

Token 预算设计

在系统架构层设计多维度的 token 预算，形成嵌套的限制体系：

维度示例限制目的
单次请求输入≤ 4,000 tokens防止 token 炸弹攻击
单次请求输出≤ 2,000 tokens控制单次成本上限
单用户/小时≤ 50,000 tokens防止单用户过度消费
单用户/天≤ 200,000 tokens日配额管理
全局/小时≤ 2,000,000 tokens系统级成本上限
全局/月≤ 50,000,000 tokens月度预算控制

限流策略实现

策略一：令牌桶算法（Token Bucket）

允许短时突发流量，同时保证长期速率不超标。适合用户交互场景。

import time
import threading
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    """令牌桶限流器（线程安全）"""
    capacity: int       # 桶容量（最大突发 token 数）
    refill_rate: float  # 每秒补充的 token 数
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: threading.Lock = field(init=False, default_factory=threading.Lock)
    
    def __post_init__(self):
        self._tokens = float(self.capacity)
        self._last_refill = time.monotonic()
    
    def _refill(self):
        """补充令牌"""
        now = time.monotonic()
        elapsed = now - self._last_refill
        new_tokens = elapsed * self.refill_rate
        self._tokens = min(self.capacity, self._tokens + new_tokens)
        self._last_refill = now
    
    def consume(self, tokens: int) -> bool:
        """尝试消费 tokens，成功返回 True，限流返回 False"""
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False
    
    @property
    def available(self) -> float:
        with self._lock:
            self._refill()
            return self._tokens

策略二：滑动窗口计数（Sliding Window）

精确统计时间窗口内的消费量，无突发容忍，适合严格的配额控制。

import time
from collections import deque
import redis  # 生产环境推荐用 Redis 做分布式限流

class SlidingWindowRateLimiter:
    """基于 Redis 的滑动窗口限流（生产级）"""
    
    def __init__(self, redis_client, window_seconds: int, max_tokens: int):
        self.redis = redis_client
        self.window = window_seconds
        self.max_tokens = max_tokens
    
    def check_and_consume(self, user_id: str, token_count: int) -> dict:
        """
        检查并消费 token 配额
        返回: {"allowed": bool, "used": int, "remaining": int, "reset_in": int}
        """
        key = f"ratelimit:{user_id}:{self.window}s"
        now = time.time()
        window_start = now - self.window
        
        pipe = self.redis.pipeline()
        # 移除窗口外的记录
        pipe.zremrangebyscore(key, 0, window_start)
        # 获取当前窗口消费量
        pipe.zrange(key, 0, -1, withscores=True)
        # 设置过期
        pipe.expire(key, self.window + 1)
        _, records, _ = pipe.execute()
        
        current_usage = sum(score for _, score in records)
        
        if current_usage + token_count > self.max_tokens:
            return {
                "allowed": False,
                "used": int(current_usage),
                "remaining": max(0, self.max_tokens - int(current_usage)),
                "reset_in": self.window,
                "error": f"超出速率限制：已用 {int(current_usage)}/{self.max_tokens} tokens"
            }
        
        # 记录本次消费
        self.redis.zadd(key, {f"{now}:{token_count}": now}, score=token_count)
        return {
            "allowed": True,
            "used": int(current_usage + token_count),
            "remaining": self.max_tokens - int(current_usage + token_count),
            "reset_in": self.window,
        }

降级方案：超预算切小模型

当用户接近配额上限时，不直接拒绝服务，而是降级到更便宜的小模型，保证基本可用性：

from enum import Enum

class ModelTier(Enum):
    PREMIUM = "gpt-4o"            # 高质量，贵
    STANDARD = "gpt-4o-mini"      # 均衡
    ECONOMY = "gpt-3.5-turbo"     # 最便宜
    LOCAL = "qwen2.5-7b-local"    # 本地模型，免费

# 模型定价（每千 token 美元，输入价格）
MODEL_PRICES = {
    ModelTier.PREMIUM:   0.005,
    ModelTier.STANDARD:  0.00015,
    ModelTier.ECONOMY:   0.0005,
    ModelTier.LOCAL:     0.0,
}

class AdaptiveModelSelector:
    """根据配额余量自适应选择模型"""
    
    TIER_THRESHOLDS = [
        (0.5, ModelTier.PREMIUM),    # 剩余 > 50%：用最好模型
        (0.2, ModelTier.STANDARD),   # 剩余 20-50%：降级到标准
        (0.05, ModelTier.ECONOMY),   # 剩余 5-20%：再降级
        (0.0, ModelTier.LOCAL),      # 剩余 < 5%：切本地模型
    ]
    
    def select_model(self, quota_remaining_ratio: float, 
                     task_priority: str = "normal") -> ModelTier:
        """
        根据剩余配额比例选择模型
        quota_remaining_ratio: 0.0 ~ 1.0
        task_priority: "high" | "normal" | "low"
        """
        if task_priority == "high":
            # 高优先级任务不降级
            return ModelTier.PREMIUM
        
        for threshold, tier in self.TIER_THRESHOLDS:
            if quota_remaining_ratio > threshold:
                return tier
        return ModelTier.LOCAL
    
    def estimate_cost(self, model: ModelTier, input_tokens: int, 
                      output_tokens: int) -> float:
        """预估请求成本（美元）"""
        price = MODEL_PRICES[model]
        return (input_tokens + output_tokens) / 1000 * price

成本监控与告警

import smtplib
from datetime import datetime

class CostMonitor:
    """成本监控与告警"""
    
    def __init__(self, daily_budget_usd: float, alert_threshold: float = 0.8):
        self.daily_budget = daily_budget_usd
        self.alert_threshold = alert_threshold  # 达到预算的80%时告警
        self.daily_cost = 0.0
        self.cost_history = []  # [(timestamp, cost, model, user_id)]
    
    def record_usage(self, cost: float, model: str, user_id: str):
        """记录一次调用的成本"""
        self.daily_cost += cost
        self.cost_history.append((datetime.now(), cost, model, user_id))
        
        # 检查是否需要告警
        usage_ratio = self.daily_cost / self.daily_budget
        if usage_ratio >= self.alert_threshold:
            self._send_alert(usage_ratio)
        
        # 硬上限：超过100%直接禁用
        if usage_ratio >= 1.0:
            self._emergency_stop()
    
    def _send_alert(self, usage_ratio: float):
        """发送成本告警（示例用 print，生产环境换成真实告警渠道）"""
        print(f"🚨 成本告警：今日已消耗预算的 {usage_ratio*100:.1f}%")
        print(f"   已花费: ${self.daily_cost:.4f} / 预算: ${self.daily_budget:.2f}")
        # 生产环境：发送到 Slack/钉钉/邮件/PagerDuty
    
    def _emergency_stop(self):
        """超出预算硬停"""
        print("🛑 紧急停止：今日 AI 调用预算已耗尽，服务已暂停。")
        raise RuntimeError("Daily AI cost budget exceeded")
    
    def get_top_consumers(self, n: int = 5) -> list:
        """获取消费最多的用户"""
        from collections import defaultdict
        user_costs = defaultdict(float)
        for _, cost, _, user_id in self.cost_history:
            user_costs[user_id] += cost
        return sorted(user_costs.items(), key=lambda x: x[1], reverse=True)[:n]

# 完整使用示例
monitor = CostMonitor(daily_budget_usd=10.0, alert_threshold=0.8)
selector = AdaptiveModelSelector()

def ai_call_with_cost_control(user_id: str, prompt: str, priority: str = "normal"):
    """带成本控制的 AI 调用封装"""
    # 1. 获取用户配额使用情况（简化示例）
    quota_ratio = 0.6  # 实际从 Redis 读取
    
    # 2. 选择合适模型
    model = selector.select_model(quota_ratio, priority)
    
    # 3. 预估 token 数（tiktoken 精确计算）
    input_tokens = len(prompt.split()) * 1.3  # 粗估
    
    # 4. 执行调用（此处省略实际API调用）
    output_tokens = 500  # 假设
    
    # 5. 记录成本
    cost = selector.estimate_cost(model, int(input_tokens), output_tokens)
    monitor.record_usage(cost, model.value, user_id)
    
    print(f"✅ 调用完成 | 用户: {user_id} | 模型: {model.value} | 成本: ${cost:.5f}")

成本优化最佳实践

小结

速率限制和成本控制是 AI 应用从 demo 走向生产的必经关卡。核心三要素：多维预算设计（请求/用户/全局）、自适应降级（配额紧张时切便宜模型）、实时监控告警（达到预算阈值前预警）。把成本控制逻辑做成可配置的基础设施，而不是硬编码在业务代码里。