AI Agent限流策略：防止API调用超限和费用失控

📘 教程 2026-07-16 · 更新于 2026-10-11 约 8 分钟阅读

Agent 没做限流，10 秒就能把月度额度调完。限流不是限制功能，是保护钱包和下游服务。这篇分并发、速率、预算三层给出具体配置方法和参数推导公式。

💡 你将学到

📜 目录

一个没有限流的 Agent 能多快烧完额度
三层限流一览
第一层：并发控制
第二层：速率限制
第三层：预算熔断
参数怎么定
常见坑
常见问题
相关文章

AI Agent限流策略：并发、速率、预算三层保护的完整配置

一个没有限流的 Agent 能多快烧完额度

限流出问题的场景几乎一样：批量任务跑起来，Agent 开足并发对着 API 猛调，几十秒后开始大面积报 429，再往后是账户被临时封禁，最后账单出来发现超支。整个过程没有一行代码是"错"的，只是没人给出口装阀门。

限流要做三层，缺一层都会漏：并发防的是瞬时压垮，速率防的是持续超配额，预算防的是单次任务失控。三层配合，才既有性能上限也有成本上限。

三层限流一览

层	控制什么	单位	触发后的行为
并发控制	同时在飞的任务数	个	排队等待
速率限制	单位时间调用次数/token	次/分、token/分	等待或拒绝
预算熔断	累计花费	元/天、元/月	停止调用并告警

第一层：并发控制

最简单也最有效的一层。用信号量把同时在跑的任务数钉死：

import asyncio

sem = asyncio.Semaphore(5)   # 最多 5 个任务同时在飞

async def run_one(task):
    async with sem:
        return await agent.run(task)

async def main(tasks):
    return await asyncio.gather(*(run_one(t) for t in tasks))

要点：并发数该按下游 API 的并发上限设，不是按你机器能开多少协程。多数托管 API 对同账号并发有硬限制，设高了就是批量 429。同时给每个任务加独立超时，避免个别任务卡死占着信号量不放。

第二层：速率限制

并发控制不了"每分钟总调用量"，因为每个任务可能连续调多次。两种常见算法：

令牌桶：桶按固定速率补令牌，取不到就等。允许短时突发，实践中最常用
滑动窗口：统计最近 N 秒的调用次数，超了拒绝。更严格，适合配额硬的场景

多进程或多实例部署时，计数器必须放在共享存储里（例如 Redis），否则每个进程各算一份，总量轻松翻几倍。

另一个必做的动作是尊重服务端的退避信号：遇到 429 时读 Retry-After（若有），按指数退避加抖动重试。不带抖动的重试会形成"重试风暴"——所有任务在同一秒一起回来，把刚恢复的配额再次打满。

import random, asyncio

async def call_with_backoff(fn, max_retries=4):
    for i in range(max_retries):
        try:
            return await fn()
        except RateLimited as e:
            wait = getattr(e, "retry_after", None) or (2 ** i) + random.random()
            await asyncio.sleep(wait)
    raise RuntimeError("rate limited after retries")

第三层：预算熔断

前两层管频率，管不了"单次任务又贵又长"。预算层做两件事：

记账：每次调用记录输入/输出 token 与按价目折算的成本，按天累计
熔断：超过日/月上限就停止发起新调用，并告警

关键是记账要覆盖所有出口（流式、工具调用、重试也算），并且要给单个任务设成本上限——一个跑飞的循环能在几分钟里吃掉一天预算，这在 Agent 场景里非常常见。

参数怎么定

不要照抄别人的数字，从你自己的配额反推。三个公式：

目标	算法
每分钟调用上限	每分钟 token 配额 ÷ 单次调用平均 token
并发上限	每分钟调用上限 × 单次平均耗时（秒）÷ 60，再取不超过服务端并发限制
单任务预算上限	日均预算 ×（预期单任务占比），通常取日均预算的 1%-5%