LLM 推理 Token 预算控制不无限制烧钱用硬限制兜底一、你的月付 Token 账单对不上因为你从来没设硬上限打开 OpenAI 的 billing 页面看到四位数的账单是种什么样的体验排查一圈发现一个循环调用 Agent 的 max_tokens 设了 4096但外层还有一个重试逻辑——重试 3 次、每次重新生成、上下文翻倍。最后跑成了一个 28K tokens 的巨型请求一笔就烧了 0.5 美元。Token 成本不可控的根因不是模型定价上涨而是缺乏一套硬限制机制。软限制比如在 Prompt 里说请简短回答是建议性的硬限制在 API 调用层面强制执行 max_tokens 计数器检查才是强制性的。两者之间的差距就是你账单和预算之间的差距。真正有效的 Token 预算控制不是设一个 max_tokens 就完了而是需要三层防护单次调用上限、单次会话累计上限、每分钟/每小时速率限制。每一层都需要硬编码检查不依赖模型的自我约束。二、底层机制与原理剖析Token 消耗的失控通常发生在三种场景循环调用中上下文不断膨胀、Function Calling 的工具返回嵌入了大量数据、多轮对话中输入 tokens 线性增长。每种场景需要不同的限制策略flowchart TD A[Agent 请求] -- B{令牌预算检查} B -- C[单次调用令牌检查] B -- D[会话累计令牌检查] B -- E[速率限制检查] C --|超限| F[截断或拒绝请求] C --|通过| G[执行 API 调用] D --|累计超限| H[触发会话超限策略] H -- I[压缩上下文] H -- J[终止旧会话] E --|速率超限| K[排队等待或降级模型] G -- L[更新令牌计数器] L -- M[记录账单日志] M -- N{预算预警} N --|日预算 80%| O[发送告警] N --|日预算 90%| P[限制非核心调用] N --|日预算 100%| Q[硬拒绝所有调用]核心机制分三层第一层API 级硬限制。在调用 API 之前做本地计数检查。虽然 OpenAI 的 API 本身有 max_tokens 参数但 API 的 max_tokens 只管 completion输出不管 prompt输入。需要在本地累计输入 tokens 预期输出 tokens确保不超过模型上下文窗口的同时也不超过预算上限。第二层会话级累计限制。一个多轮对话的总 tokens 消耗 每轮输入 tokens 的累加。如果不设限制一个长会话可以轻松消耗上万 tokens。策略是在累计接近预算时触发对话历史压缩或主动结束会话。第三层速率限制Rate Limiting。用令牌桶算法控制每分钟的调用次数。不是为了防止 API 限流OpenAI 自己有而是为了在预算维度上做平滑消耗——避免某个突发流量在 1 分钟内消耗掉一小时的预算。三、生产级代码实现 Token 预算控制器 三层防护单次限额、会话累计、速率限制 设计决策所有限制在调用前做本地检查不依赖 API 返回后再处理 import time import threading from dataclasses import dataclass, field from typing import Dict, Optional, Tuple from enum import Enum import tiktoken class BudgetExceededAction(Enum): 超预算后的动作 REJECT reject # 直接拒绝 TRUNCATE truncate # 截断输入 DOWNGRADE downgrade # 降级到更便宜的模型 COMPRESS compress # 压缩对话历史 dataclass class TokenBudget: 令牌预算配置 设计决策日预算与月预算分别独立管理。 日预算用于每日限额控制月预算用于账单预警。 daily_limit: int # 日总 Token 上限 per_call_input_limit: int # 单次调用输入上限 per_call_output_limit: int # 单次调用输出上限 per_session_limit: int # 单次会话累计上限 rpm_limit: int # 每分钟最大请求数 # 预算预警阈值日预算的百分比 warn_threshold: float 0.8 critical_threshold: float 0.9 dataclass class SessionTokenTracker: 会话 Token 追踪器 session_id: str total_input_tokens: int 0 total_output_tokens: int 0 message_count: int 0 created_at: float field(default_factorytime.time) last_active: float field(default_factorytime.time) class TokenRateLimiter: 基于令牌桶的速率限制器 def __init__(self, rpm_limit: int): # 令牌桶容量 rpm_limit每 60/rpm_limit 秒补充一个令牌 self.rate rpm_limit / 60.0 # 每秒补充速率 self.bucket float(rpm_limit) self.capacity rpm_limit self.last_refill time.time() self._lock threading.Lock() def acquire(self) - Tuple[bool, float]: 尝试获取一个令牌 返回 (是否获取成功, 需要等待的秒数) with self._lock: now time.time() # 补充令牌经过的时间 * 补充速率 elapsed now - self.last_refill self.bucket min(self.capacity, self.bucket elapsed * self.rate) self.last_refill now if self.bucket 1.0: self.bucket - 1.0 return True, 0.0 else: # 计算需要等待的时间 wait_time (1.0 - self.bucket) / self.rate return False, wait_time class TokenBudgetController: Token 预算控制器主类 def __init__(self, budget: TokenBudget, model: str gpt-4): self.budget budget self.model model self.encoding tiktoken.encoding_for_model(model) # 日消耗追踪 self._daily_consumed 0 self._daily_reset_time self._next_midnight() self._lock threading.Lock() # 会话追踪 self._sessions: Dict[str, SessionTokenTracker] {} # 速率限制 self._rate_limiter TokenRateLimiter(budget.rpm_limit) def _next_midnight(self) - float: 计算下一个午夜的 Unix 时间戳 now time.time() return now (86400 - now % 86400) def count_tokens(self, text: str) - int: 计算文本的 Token 数量 try: return len(self.encoding.encode(text)) except Exception: # 编码失败时的降级估算按字符数的 1/4 估算 return len(text) // 4 def check_call( self, session_id: str, input_text: str, expected_output_tokens: int 1024 ) - Tuple[bool, Optional[str], BudgetExceededAction]: 检查是否可以发起 API 调用 返回(是否允许, 拒绝原因, 建议动作) input_tokens self.count_tokens(input_text) with self._lock: # 检查 1日预算 if self._daily_consumed input_tokens expected_output_tokens self.budget.daily_limit: reason f日预算超标已用 {self._daily_consumed}/{self.budget.daily_limit} return False, reason, BudgetExceededAction.REJECT # 检查 2单次输入上限 if input_tokens self.budget.per_call_input_limit: reason f单次输入超标{input_tokens} {self.budget.per_call_input_limit} return False, reason, BudgetExceededAction.TRUNCATE # 检查 3会话累计上限 session self._sessions.get(session_id) if session: session_consumed session.total_input_tokens session.total_output_tokens if session_consumed input_tokens expected_output_tokens self.budget.per_session_limit: reason f会话累计超标{session_consumed}/{self.budget.per_session_limit} return False, reason, BudgetExceededAction.COMPRESS # 检查 4速率限制 allowed, wait_time self._rate_limiter.acquire() if not allowed: reason f速率限制需等待 {wait_time:.1f} 秒 return False, reason, BudgetExceededAction.REJECT return True, None, BudgetExceededAction.REJECT def record_usage( self, session_id: str, input_text: str, output_text: str ): 记录一次成功调用的 Token 消耗 input_tokens self.count_tokens(input_text) output_tokens self.count_tokens(output_text) total input_tokens output_tokens with self._lock: # 检查是否需要重置日计数器 if time.time() self._daily_reset_time: self._daily_consumed 0 self._daily_reset_time self._next_midnight() self._daily_consumed total # 更新会话追踪 if session_id not in self._sessions: self._sessions[session_id] SessionTokenTracker( session_idsession_id ) session self._sessions[session_id] session.total_input_tokens input_tokens session.total_output_tokens output_tokens session.message_count 1 session.last_active time.time() def get_budget_status(self) - Dict: 获取当前预算状态 with self._lock: used_pct self._daily_consumed / self.budget.daily_limit status normal if used_pct self.budget.critical_threshold: status critical elif used_pct self.budget.warn_threshold: status warning return { status: status, daily_used: self._daily_consumed, daily_limit: self.budget.daily_limit, used_percentage: round(used_pct * 100, 1), active_sessions: len(self._sessions), } def force_compress_session(self, session_id: str) - Optional[str]: 压缩指定会话——保留最近 N 轮对话旧消息用摘要替代 设计决策这个方法只做计数层面的压缩标记 实际的内容压缩由调用方负责如调用 LLM 生成摘要后替换历史 session self._sessions.get(session_id) if not session: return None # 标记会话需要压缩 return f会话 {session_id} 已使用 {session.total_input_tokens session.total_output_tokens} tokens建议压缩 def cleanup_idle_sessions(self, max_idle_seconds: float 3600): 清理超时空闲会话 now time.time() with self._lock: idle_sessions [ sid for sid, s in self._sessions.items() if now - s.last_active max_idle_seconds ] for sid in idle_sessions: del self._sessions[sid]四、边界分析与架构权衡Token 预算硬限制的缺点硬限制的本质是宁可拒绝服务也要控制成本。如果限制值设置不合理比如日预算过低可能在业务高峰期过早触发导致合法用户请求被拒绝。另外tiktoken 的编码器在不同模型间有差异需要与使用的模型严格对齐否则计数误差会导致过早或过晚触发限制。上下文压缩会话累计超限时的处理是一个复杂的工程问题。简单的保留最近 N 轮会丢失早期的关键信息而用 LLM 生成摘要再替代历史又引入了额外的 Token 消耗和延迟。适用边界最适合有明确预算上限的团队或项目——比如 Startup 的每月 API 预算是 500 美元必须遵守。也适合多租户 SaaS 平台需要按租户分别跟踪和控制 Token 消耗。API 调用量稳定可预测的场景预算管理效果最好。禁用场景不适合对可用性要求极高的关键业务场景——如果 Token 限制可能导致用户无法完成核心操作就应该优先保证可用性把成本控制放在第二优先级。也不适合 API 调用量极低日均 100 次的场景管理开销比实际价值高。五、总结Token 预算控制不是设个 max_tokens就完了。它需要三层硬限制单次调用、会话累计、速率限制。每一层都在 API 调用前做本地检查不依赖模型的自我约束。核心设计决策是在成本和可用性之间选择硬限制兜底。宁可让非核心请求被拒绝也不让账单失控。
