企业官网建设流程全解析

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_factory=time.time) last_active: float = field(default_factory=time.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_id=session_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 调用前做本地检查，不依赖模型的自我约束。核心设计决策是：在成本和可用性之间，选择硬限制兜底。宁可让非核心请求被拒绝，也不让账单失控。