大模型 API 超额与限额处理:如何防止账单失控
每个对接过大模型 API 的开发者都应该问自己一个问题:如果代码出了 Bug,最坏情况下账单会是多少? 超额事故通常不是因为业务用量太高,而是因为没有在正确的位置设置护栏。
我见过不止一次这种事故:某个 Agent 项目上线第一周,工具调用那段逻辑有个边界条件没处理好,模型返回的结果不满足预期格式,代码判断”失败就重试”,但重试之间没有终止条件,也没有对同一任务设置最大尝试次数。这段逻辑挂在一个夜间定时任务里,跑了一整晚,第二天早上账单金额是平时日均消耗的几十倍。事后复盘,问题根本不在”模型用得太多”,而在于代码里压根没有任何东西会喊停——没有限流、没有重试上限、也没有硬限额兜底,三层防御全部缺失,Bug 才有机会一路狂奔到天亮。这也是为什么这篇文章不打算只讲”厂商有哪些限额”,而是把重点放在你自己代码里应该加的护栏上——厂商的限额是最后一道防线,不是唯一防线。
超额的四大常见原因
1. 无限循环调用
Agent 或对话流程中的逻辑错误(如重试无上限、工具调用死循环),可能在几分钟内产生几千次 API 调用。
2. Prompt 膨胀失控
RAG 系统检索结果过多、历史对话无限累积,导致每次调用的 token 量远超预期。
3. 并发爆炸
任务队列中的任务并发设置过高,或批量任务触发大量并行请求,瞬间拉高用量。
4. 测试/开发环境混用
开发人员在生产账号下进行压力测试或大批量实验,消耗了生产预算。
这四种原因里,前两种是最隐蔽的,因为它们不是”调用次数明显异常”,而是”单次调用的成本被悄悄放大”。举个例子:RAG 场景下如果检索 Top-K 设置得过大(比如把 K 从 5 调到 50 去做效果对比测试,忘了改回来),单次请求的 prompt token 量可能从几百涨到几千甚至上万,调用次数没变,但账单成本可能翻十几倍——这种情况下光盯着”请求数”这个指标是发现不了问题的,你需要同时监控每次请求的平均 token 量,一旦这个数字出现台阶式跳变,基本就是 Prompt 膨胀出问题了。第 3 种”并发爆炸”则常见于批处理脚本:很多人写批量任务时习惯用 asyncio.gather 或线程池一次性把所有任务扔出去,任务数一旦上千,瞬间打出去的并发请求可能远超账户的 RPM 上限,先是触发一大片限流报错,紧接着重试机制又会雪上加霜——如果重试策略本身没设上限,429 触发重试、重试又触发 429,形成恶性循环,这也是超额事故里比较典型的连锁反应。
厂商侧的限额机制
声明:截至 2026-06,以各厂商官方文档为准。
| 限额类型 | 说明 | 各厂商支持情况 |
|---|---|---|
| RPM(每分钟请求数上限) | 超出后 API 返回 429 错误 | 几乎所有厂商均有,级别随账户等级提升 |
| TPM(每分钟 token 上限) | 按 token 速率限制 | OpenAI、Anthropic 等均有 |
| 每日用量上限 | 超出当日停止服务 | 部分厂商支持手动配置 |
| 每月硬限额(Hard Limit) | 超出当月停止服务 | OpenAI、Anthropic 等控制台可配置 |
| 软限额(Soft Limit/告警) | 达到阈值发送告警,不停服 | 大多数厂商支持邮件/webhook 告警 |
重要: 厂商的速率限制(RPM/TPM)主要是为了防止滥用和保护系统稳定性,不等于成本保护——在限额内持续高强度调用仍会产生高额费用。
触发限流和触发硬限额,在代码里收到的报错是两回事,处理方式也应该不一样,很多人会把这两种情况混在一起处理,结果要么该重试的不重试,要么不该重试的一直重试:
- 触发 RPM/TPM 限流:通常对应 HTTP 429,报错体里一般会带类似
rate_limit_exceeded、requests这类字段(具体文案各厂商、各版本会有差异,以你实际收到的原始报错为准)。这种情况说明”你这一秒钟打太猛了”,正确做法是退避后重试——等一会儿再试大概率能成功,前面提到的指数退避重试就是为它准备的。 - 触发每月/每日硬限额:常见对应 HTTP 402 或者一个专门的
insufficient_quota/余额不足类错误码,这种情况说明”钱包已经空了”或”本月配额用完了”,这时候再怎么重试都没用,重试只会让你的日志刷满一堆同样的失败记录,正确做法是立刻停止对该 Key 的调用、走告警通知人工处理,不要让重试逻辑在这种场景下空转。
如果你的重试逻辑不区分这两种错误码,最坏的情况是:账户已经欠费或者已经打满硬限额,代码却还在傻乎乎地按指数退避一遍遍重试,虽然不会产生新的计费(请求本身会被拒绝),但会拖慢整个系统的响应,日志里全是噪音,真正的告警反而被淹没。建议在重试装饰器里对错误类型做判断,只对 429 走退避重试,对 402/配额类错误直接抛出并触发告警。
工程层面的防御策略
第一层:应用代码内部限流
import time
from collections import deque
class RateLimiter:
def __init__(self, max_calls_per_minute: int):
self.max_calls = max_calls_per_minute
self.calls = deque()
def check(self) -> bool:
now = time.time()
# 清除 60 秒之前的记录
while self.calls and now - self.calls[0] > 60:
self.calls.popleft()
if len(self.calls) >= self.max_calls:
return False # 超限,拒绝本次调用
self.calls.append(now)
return True
在最靠近 API 调用的位置加限流,比依赖厂商报错更安全。
这段 RateLimiter 代码看着简单,但几个细节值得展开讲讲。首先是为什么用 deque 而不是普通 list:滑动窗口限流需要频繁地从队首弹出过期记录、从队尾追加新记录,list.pop(0) 是 O(n) 操作,数据量一大会拖慢每次调用的检查开销;deque 的两端操作都是 O(1),这也是 Python 里实现滑动窗口、环形缓冲区这类结构的标准选择。其次,check() 里”先清理过期记录,再判断是否超限,最后才追加”这个顺序不能颠倒——如果先追加再判断,你统计到的就不是”过去 60 秒的调用数”,而是”过去 60 秒加上这一次”,窗口边界会算错,长期跑下去会导致实际放行的调用数比你设定的上限略高。
这个限流器有一个明确的适用边界:它只在单进程内有效。 如果你的服务是多进程部署(比如用 Gunicorn 起了 4 个 worker)或者多副本部署(Kubernetes 里跑了 3 个 Pod),每个进程/副本都会维护自己独立的一份 self.calls,实际总调用量是你设定上限的 N 倍——这也是很多团队”明明设了限流,账单还是超了”的真实原因:限流器本身没错,错在部署形态和限流器的假设不匹配。如果你是这种多实例场景,需要把限流状态放到 Redis 这类外部存储里做集中式限流,思路和上面差不多,只是把 deque 换成 Redis 的有序集合(sorted set),用时间戳做 score:
import time
import redis
r = redis.Redis()
def check_distributed(key: str, max_calls: int, window_seconds: int = 60) -> bool:
now = time.time()
pipe = r.pipeline()
pipe.zremrangebyscore(key, 0, now - window_seconds) # 清理过期记录
pipe.zcard(key) # 统计当前窗口内的调用数
_, current_count = pipe.execute()
if current_count >= max_calls:
return False
r.zadd(key, {str(now): now})
r.expire(key, window_seconds)
return True
这样无论你的服务开多少个进程或副本,大家都读写同一个 Redis key,限流统计口径才是准的。如果你的场景没有多实例部署,单进程内的 deque 版本足够用,不需要为了”以防万一”就上 Redis,多引入一个外部依赖也是多一个故障点。
第二层:重试策略加指数退避和上限
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3), # 最多重试 3 次
wait=wait_exponential(min=1, max=10) # 退避 1s → 2s → 4s
)
def call_api(prompt: str):
...
无上限的重试是造成超额事故的主要元凶之一。wait_exponential(min=1, max=10) 意味着第一次重试等 1 秒,第二次等 2 秒,第三次等 4 秒(如果没到 max 就一直翻倍),这样设计是为了避免”一秒钟内挤爆同一个接口”——如果不做退避、失败了立刻重试,相当于在系统最脆弱的时候(已经触发限流)继续加压,只会让限流窗口一直续期,永远等不到恢复的机会。stop_after_attempt(3) 这个上限同样关键:没有它,遇到那种”确定性会失败”的错误(比如账户已欠费、Key 已失效)时,重试会一直空转到你自己手动叫停,白白消耗时间和日志空间。
这里还有一个容易被忽略的点:tenacity 默认情况下对所有异常都会重试,但有些异常根本不该重试——比如你的 prompt 本身超过了模型的 context 长度限制,这种错误无论重试多少次结果都一样,重试只是在浪费一次计费请求(如果厂商在报错前已经完成了部分 token 计费的话)。建议给 @retry 加上 retry_if_exception_type,只对网络超时、429 这类”重试后可能成功”的异常重试,对参数错误、context 超限这类”重试后必然还是失败”的异常直接放弃并往上抛,让业务代码去处理(比如自动截断历史对话、降级到支持更长 context 的模型)。
并发层:同时在飞的请求数也要设上限
限流控制的是”每分钟能打多少次”,但还有一个维度容易被忽视:同一时刻允许多少个请求同时在等待响应。批量处理任务时,如果直接用 asyncio.gather 把几百个任务一次性丢出去,即便没有触发厂商的 RPM 限制,也可能因为瞬间并发过高导致本地资源(连接数、内存)被打满,或者让重试风暴叠加放大。用信号量(Semaphore)控制并发数是最简单也最实用的办法:
import asyncio
sem = asyncio.Semaphore(5) # 同时最多 5 个请求在飞
async def call_with_limit(prompt: str):
async with sem:
return await call_api_async(prompt)
async def run_batch(prompts: list[str]):
tasks = [call_with_limit(p) for p in prompts]
return await asyncio.gather(*tasks)
Semaphore(5) 的含义是:无论你一次性提交多少个任务,同一时刻真正发出去的请求最多只有 5 个,后面的任务会排队等前面的释放了名额才开始执行。这个并发数设多少合适,没有统一答案,建议从厂商 RPM 上限倒推——如果你的账户 RPM 上限是 60(每分钟 60 次),单次调用平均耗时 3 秒,理论上并发数设到 3 左右就能跑满限额而不触发 429;设太高(比如 20)会导致请求堆积后集中触发限流,设太低又浪费了配额。这个数字建议做成可配置项,上线后根据实际的 429 触发频率动态调整,而不是拍脑袋定死。
第三层:控制台配置硬限额
在各厂商控制台(通常在”用量限制”或”Billing”页):
- 设置每月 Hard Limit(到达后 API 返回错误,不再计费)
- 设置软限额告警(到达 80% 时发邮件/Slack 告警)
- 设置每日预算上限(如果厂商支持)
第四层:环境隔离
| 环境 | 措施 |
|---|---|
| 生产环境 | 使用独立 API Key,设置硬限额 |
| 开发/测试环境 | 使用单独账号或子账号,预算独立管理 |
| CI/CD 测试 | 使用 mock 或最便宜的轻量模型,禁止使用旗舰模型 |
限流、熔断、降级:三个概念不要混为一谈
工程上防止超额,经常会把”限流""熔断""降级”这三个手段混着用,但它们解决的问题不一样,触发条件和恢复方式也不同,分清楚了才能对症下药:
| 手段 | 解决什么问题 | 触发条件 | 恢复方式 |
|---|---|---|---|
| 限流(Rate Limit) | 防止短时间内调用次数/token 量超过设定阈值 | 主动预判,在调用前检查 | 时间窗口滑过去自动恢复 |
| 熔断(Circuit Breaker) | 防止对一个持续失败的下游继续发请求 | 被动响应,连续失败次数达到阈值后触发 | 等待一段冷却时间后放行”探测请求”,成功则恢复 |
| 降级(Fallback) | 保证核心功能可用,即便主模型不可用或成本失控 | 前两者触发后的兜底动作 | 手动或自动切回主链路 |
三者组合起来用,才是完整的防御链:限流负责”防患于未然”,把调用频率摁在安全线以内;熔断负责”止血”,当账户已经连续报错(比如已经欠费、账号被封)时,别再让代码傻乎乎地一次次撞上去,而是在本地直接快速失败,省下无意义的网络往返;降级负责”保业务”,比如熔断触发后自动切换到一个更便宜的备用模型,或者返回一个预设的兜底文案,而不是让整个功能直接挂掉。前面讲的 RateLimiter 和重试退避,本质上分别对应限流和”简易版熔断”,如果你的业务对可用性要求比较高,值得单独引入一个成熟的熔断库(比如 pybreaker),而不是只靠重试次数上限硬撑。
提前估算你的”爆表速度”
与其等出事故才复盘,不如上线前先算一笔账,心里有个数。公式很简单:
每分钟最大可能花费 = 并发数 × (60 / 单次调用平均耗时秒数) × 单次调用平均 token 量 × 单价
举个例子帮你套:如果你的并发上限设的是 5,单次调用平均耗时 3 秒,也就是理论上每分钟这 5 个并发槽位总共能跑完 5 × 60 / 3 = 100 次调用;再假设你的场景里平均每次调用消耗 2000 个 token(输入输出加一起),那么这个服务在满负荷、且完全不受厂商限流拦截的极端情况下,每分钟大概会产生 100 × 2000 = 20 万 token 的消耗——把这个数字乘以你使用的模型的实际单价(去官方定价页查,不同模型价格差异可能有几十倍),就能得到”这个服务在最坏情况下,每分钟最多能烧掉多少钱”。把这个数字和你能接受的月度预算做个对比,反推出你的硬限额、并发数、重试上限应该设成多少,比事后救火要踏实得多。这也是为什么前面反复强调”控制台的软限额告警”不能当唯一防线——告警最快也要几分钟才会触达人,而按上面这个公式,几分钟内就可能已经产生了远超预期的账单。
超额之后怎么办
1. 立即止损:在控制台禁用相关 API Key,或临时设置更低的硬限额。
2. 排查根因:查日志中调用量和 token 量的时间序列,定位异常峰值的触发时间和来源调用。实操上,如果你的日志里记录了每次调用的时间戳、来源函数名和 token 量(建议一开始就把这几项作为标准字段打进日志,事故发生时会救你一命),可以先按分钟粒度聚合调用次数,找到第一次出现异常陡增的时间点,再回头看那个时间点前后几分钟的日志里都是哪个函数在调用、参数是什么。多数情况下你会发现两种典型模式:一种是”同一个 trace_id 或 task_id 反复出现”,说明是重试或循环没有终止;另一种是”短时间内出现大量不同的 trace_id 但调用参数高度相似”,说明是批量任务的并发没控制住。把这两种模式当成排查时的第一假设,能省下很多盲目翻日志的时间。
3. 联系厂商:如果是明显的 Bug 导致的意外超额,部分厂商(尤其国内)会在首次事故时酌情退费,值得尝试联系客服。
4. 加固护栏:事后在代码和控制台同时补充限额配置,两层保障。
常见问题
设置了 Hard Limit,超出后 API 会报什么错误?
通常是 HTTP 429(Too Many Requests)或 HTTP 402(Payment Required),具体错误码各厂商略有差异。建议在测试环境主动触发一次限额,确认错误处理逻辑生效。
用量告警延迟多久?
大多数厂商的用量统计和告警有 5–30 分钟的延迟,极端情况下告警到达时已经超额较多。因此代码层面的限流是必要的第一道防线,不能只依赖控制台告警。
多人团队如何分配 API Key 和预算?
建议按项目或团队创建子账号/子组织,每个子账号有独立的 API Key 和独立的预算上限,方便追踪各项目的成本归因。
限流器要设多大的阈值才合适,是不是越保守越好?
不是。阈值设得太保守(比如远低于账户实际 RPM 上限),日常业务高峰期会被自己的限流器拦下来,用户体验受损,这属于”防御过度”;阈值设得太激进又起不到保护作用。比较实用的做法是:先按厂商公示的账户等级查到真实的 RPM/TPM 上限,把本地限流阈值设成这个上限的 80%~90% 左右,留出安全余量应对账户等级临时被降级、或者同一账户下有其他服务也在共享配额的情况,然后上线后跟踪实际的 429 触发率,如果长期为零可以适当调高,如果偶尔还会触发再适当调低,用真实数据校准,而不是拍一次脑袋就再也不动。
延伸阅读:
- 计费原理总览:大模型 token 计费完全指南
- 各家计费规则:各家计费规则与计费单位
- 预付后付选择:预付费 vs 后付费怎么选
- 成本优化手册:大模型 API 成本优化 10 招
- 在线估算用量:Token 计算器
- 更多成本话题:token 成本专题