💡 LLM 成本优化 / Token 预算

💰 Token 预算

你无法控制你没有预算的东西。Token 预算 = 成本的天花板——按用户、按会话、按天设限，超预算自动降级。

预算层级

预算层级（从细到粗）：

请求级 → 会话级 → 用户级 → 租户级 → 全局级

┌───────────────────────────────────────────────────────┐
│ 全局预算: $5000/月                                      │
│ ┌───────────────────────────────────────────────────┐ │
│ │ 租户A: $2000/月  │  租户B: $1000/月  │  ...       │ │
│ │ ┌──────────────┐ │ ┌──────────────┐              │ │
│ │ │用户1: $50/月 │ │ │用户1: $30/月 │              │ │
│ │ │ ┌──────────┐ │ │ │ ┌──────────┐ │              │ │
│ │ │ │会话1:    │ │ │ │ │会话1:    │ │              │ │
│ │ │ │1000tok/轮│ │ │ │ │800tok/轮 │ │              │ │
│ │ │ └──────────┘ │ │ │ └──────────┘ │              │ │
│ │ └──────────────┘ │ └──────────────┘              │ │
│ └───────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────┘

Token 预算系统实现

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import time

class BudgetLevel(Enum):
    REQUEST = "request"     # 单次请求
    SESSION = "session"     # 单个会话
    USER_DAILY = "user_daily"
    USER_MONTHLY = "user_monthly"
    TENANT_MONTHLY = "tenant_monthly"
    GLOBAL_MONTHLY = "global_monthly"

class DegradeAction(Enum):
    CHEAPER_MODEL = "cheaper_model"     # 降级到更便宜的模型
    REDUCE_CONTEXT = "reduce_context"   # 减少上下文长度
    QUEUE = "queue"                     # 排队等待
    REJECT = "reject"                   # 拒绝请求
    NOTIFY = "notify"                   # 仅通知，不限制

@dataclass
class BudgetLimit:
    level: BudgetLevel
    max_tokens: int
    max_cost: float          # USD
    degrade_action: DegradeAction
    warning_threshold: float = 0.8  # 80% 时告警

@dataclass
class BudgetUsage:
    level: BudgetLevel
    key: str                 # 标识（user_id, session_id 等）
    tokens_used: int = 0
    cost_used: float = 0.0
    period_start: datetime = None
    period_end: datetime = None

class TokenBudgetManager:
    """Token 预算管理器"""
    
    # 默认预算策略
    DEFAULT_LIMITS = {
        BudgetLevel.REQUEST: BudgetLimit(
            level=BudgetLevel.REQUEST,
            max_tokens=8000,
            max_cost=0.50,
            degrade_action=DegradeAction.REDUCE_CONTEXT
        ),
        BudgetLevel.SESSION: BudgetLimit(
            level=BudgetLevel.SESSION,
            max_tokens=100000,
            max_cost=5.00,
            degrade_action=DegradeAction.CHEAPER_MODEL
        ),
        BudgetLevel.USER_DAILY: BudgetLimit(
            level=BudgetLevel.USER_DAILY,
            max_tokens=500000,
            max_cost=10.00,
            degrade_action=DegradeAction.CHEAPER_MODEL
        ),
        BudgetLevel.USER_MONTHLY: BudgetLimit(
            level=BudgetLevel.USER_MONTHLY,
            max_tokens=10000000,
            max_cost=50.00,
            degrade_action=DegradeAction.REJECT
        ),
    }
    
    def __init__(self, redis_client, custom_limits: dict = None):
        self.redis = redis_client
        self.limits = custom_limits or self.DEFAULT_LIMITS
        self.alerts = []
    
    async def check_budget(self, user_id: str, session_id: str,
                           estimated_tokens: int = 0) -> dict:
        """检查预算是否充足"""
        checks = {}
        
        # 检查各层级预算
        for level in [BudgetLevel.REQUEST, BudgetLevel.SESSION, 
                      BudgetLevel.USER_DAILY, BudgetLevel.USER_MONTHLY]:
            limit = self.limits[level]
            usage = await self._get_usage(level, user_id, session_id)
            
            # 检查 token 预算
            token_pct = (usage.tokens_used + estimated_tokens) / limit.max_tokens
            # 检查成本预算
            cost_pct = usage.cost_used / limit.max_cost if limit.max_cost > 0 else 0
            
            max_pct = max(token_pct, cost_pct)
            over_budget = max_pct > 1.0
            near_limit = max_pct > limit.warning_threshold
            
            checks[level.value] = {
                'over_budget': over_budget,
                'usage_pct': max_pct,
                'near_limit': near_limit,
                'degrade_action': limit.degrade_action.value if over_budget else None
            }
            
            # 告警
            if near_limit and not over_budget:
                await self._send_alert(level, user_id, max_pct)
        
        # 汇总决策
        over_levels = [k for k, v in checks.items() if v['over_budget']]
        if over_levels:
            # 取最严格的降级策略
            most_restrictive = self._most_restrictive_action(over_levels)
            return {
                'allowed': most_restrictive != DegradeAction.REJECT,
                'action': most_restrictive.value,
                'checks': checks
            }
        
        return {'allowed': True, 'action': None, 'checks': checks}
    
    async def record_usage(self, user_id: str, session_id: str,
                           tokens_used: int, cost: float):
        """记录使用量"""
        for level in [BudgetLevel.REQUEST, BudgetLevel.SESSION,
                      BudgetLevel.USER_DAILY, BudgetLevel.USER_MONTHLY]:
            key = self._usage_key(level, user_id, session_id)
            # Redis INCRBY + 设置过期时间
            pipe = self.redis.pipeline()
            pipe.incrby(f"{key}:tokens", tokens_used)
            pipe.incrbyfloat(f"{key}:cost", cost)
            
            # 设置过期时间
            if level == BudgetLevel.USER_DAILY:
                pipe.expire(f"{key}:tokens", 86400)
                pipe.expire(f"{key}:cost", 86400)
            elif level == BudgetLevel.USER_MONTHLY:
                pipe.expire(f"{key}:tokens", 30*86400)
                pipe.expire(f"{key}:cost", 30*86400)
            
            await pipe.execute()
    
    async def _get_usage(self, level: BudgetLevel, user_id: str, 
                          session_id: str) -> BudgetUsage:
        """获取当前使用量"""
        key = self._usage_key(level, user_id, session_id)
        tokens = int(await self.redis.get(f"{key}:tokens") or 0)
        cost = float(await self.redis.get(f"{key}:cost") or 0)
        
        return BudgetUsage(
            level=level,
            key=key,
            tokens_used=tokens,
            cost_used=cost
        )
    
    def _usage_key(self, level: BudgetLevel, user_id: str, 
                    session_id: str) -> str:
        """生成 Redis key"""
        today = datetime.now().strftime('%Y-%m-%d')
        month = datetime.now().strftime('%Y-%m')
        
        if level == BudgetLevel.REQUEST:
            return f"budget:rq:{session_id}"
        elif level == BudgetLevel.SESSION:
            return f"budget:ss:{session_id}"
        elif level == BudgetLevel.USER_DAILY:
            return f"budget:ud:{user_id}:{today}"
        elif level == BudgetLevel.USER_MONTHLY:
            return f"budget:um:{user_id}:{month}"
    
    def _most_restrictive_action(self, levels: list) -> DegradeAction:
        """选择最严格的降级策略"""
        priority = [DegradeAction.REJECT, DegradeAction.CHEAPER_MODEL,
                    DegradeAction.REDUCE_CONTEXT, DegradeAction.QUEUE]
        for action in priority:
            for level in levels:
                limit = self.limits.get(BudgetLevel(level))
                if limit and limit.degrade_action == action:
                    return action
        return DegradeAction.NOTIFY
    
    async def _send_alert(self, level: BudgetLevel, user_id: str, pct: float):
        """发送预算告警"""
        alert = {
            'level': level.value,
            'user_id': user_id,
            'usage_pct': pct,
            'timestamp': datetime.now().isoformat()
        }
        self.alerts.append(alert)
        # 实际实现：发到 Slack/Email/Webhook
        print(f"⚠️ 预算告警: {level.value} - 用户 {user_id} 已使用 {pct*100:.0f}%")

降级策略实现

class DegradeStrategy:
    """降级策略"""
    
    # 模型降级链
    MODEL_DEGRADE_CHAIN = {
        "claude-opus-4-20250514": "claude-sonnet-4-20250514",
        "claude-sonnet-4-20250514": "claude-3-5-haiku-20241022",
        "gpt-5.5": "gpt-5.4",
        "gpt-5.4": "gpt-5.4-mini",
        "gpt-4o": "gpt-4o-mini",
        "gpt-4o-mini": "deepseek-chat",
    }
    
    def get_cheaper_model(self, current_model: str) -> str:
        """获取下一级更便宜的模型"""
        return self.MODEL_DEGRADE_CHAIN.get(current_model, "deepseek-chat")
    
    def reduce_context(self, messages: list, max_tokens: int) -> list:
        """减少上下文长度"""
        total = sum(len(m['content']) for m in messages)
        if total <= max_tokens * 4:  # 粗估
            return messages
        
        # 保留系统消息 + 最近的对话
        system_msgs = [m for m in messages if m['role'] == 'system']
        other_msgs = [m for m in messages if m['role'] != 'system']
        
        # 从旧到新截断
        budget = max_tokens * 4
        system_size = sum(len(m['content']) for m in system_msgs)
        remaining = budget - system_size
        
        kept = []
        for msg in reversed(other_msgs):
            if remaining <= 0:
                break
            kept.append(msg)
            remaining -= len(msg['content'])
        
        return system_msgs + list(reversed(kept))

# 集成到请求流程
degrade = DegradeStrategy()

async def chat_with_budget(user_id: str, session_id: str, 
                            messages: list, model: str = "gpt-4o"):
    # 1. 检查预算
    budget_check = await budget_manager.check_budget(
        user_id, session_id, estimated_tokens=1000
    )
    
    current_model = model
    current_messages = messages
    
    if not budget_check['allowed']:
        return {"error": "已超出使用预算，请明天再试"}
    
    if budget_check['action'] == 'cheaper_model':
        current_model = degrade.get_cheaper_model(model)
    elif budget_check['action'] == 'reduce_context':
        current_messages = degrade.reduce_context(messages, max_tokens=4000)
    
    # 2. 调用 LLM
    response = await call_llm(model=current_model, messages=current_messages)
    
    # 3. 记录使用量
    tokens = response.usage.total_tokens
    cost = calculate_cost(current_model, response.usage)
    await budget_manager.record_usage(user_id, session_id, tokens, cost)
    
    return {
        "response": response.choices[0].message.content,
        "model_used": current_model,
        "degraded": current_model != model,
        "tokens_used": tokens,
        "cost": cost
    }

按用户类型设置预算

成本报警

class CostAlertManager:
    """成本报警管理"""
    
    ALERT_LEVELS = [
        {'threshold': 0.5, 'level': 'info', 'message': '已使用50%预算'},
        {'threshold': 0.8, 'level': 'warning', 'message': '已使用80%预算'},
        {'threshold': 0.95, 'level': 'critical', 'message': '已使用95%预算'},
        {'threshold': 1.0, 'level': 'emergency', 'message': '已超出预算'},
    ]
    
    async def check_and_alert(self, scope: str, usage_pct: float):
        """检查并发送报警"""
        for alert in self.ALERT_LEVELS:
            if usage_pct >= alert['threshold']:
                await self._send_alert(scope, alert, usage_pct)
    
    async def _send_alert(self, scope: str, alert: dict, pct: float):
        """发送报警（多渠道）"""
        message = f"[{alert['level'].upper()}] {scope}: {alert['message']} ({pct*100:.1f}%)"
        
        # Slack 报警
        if alert['level'] in ['critical', 'emergency']:
            await send_slack("#cost-alerts", message)
        
        # Email 报警
        if alert['level'] == 'emergency':
            await send_email("admin@company.com", 
                           "🚨 LLM 成本超预算", message)
        
        # 自动降级
        if alert['level'] == 'emergency':
            await self._auto_degrade(scope)
    
    async def _auto_degrade(self, scope: str):
        """自动降级——紧急保护"""
        # 全局降级到最便宜的模型
        await set_global_default_model("deepseek-chat")
        await send_slack("#cost-alerts", 
                        f"🚨 已自动降级到 deepseek-chat 以控制成本")

← 上一篇：Batch API | 下一篇：Embedding 成本 → | 返回首页