AI Agents · 课时

速率限制与配额管理

设置按用户、组织和端点划分的配额,避免某个租户耗尽您的 OpenAI 预算。

第 3 / 4 课15 个步骤

速率限制与配额管理 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

本课时的部分内容尚未翻译,以英文显示。

为什么需要限制?

没有限制时,单个滥用的客户端就可能:

  • 耗尽您的 OpenAI 预算
  • 挤占其他用户的资源
  • 对您的服务发起 DDoS 攻击

速率限制和配额可以保护成本、延迟和公平性。

速率限制与配额

  • 速率限制——每秒或每分钟的请求数(短期)
  • 配额——每天或每月的总预算(长期)

两者都需要。

令牌桶算法

经典算法是:每位用户都有一个按固定速率补充的“桶”。每个请求消耗一个令牌。没有令牌,就不提供服务。

Redis Token Bucket Example

def allow(user_id, rate=10, capacity=30):
    key = f'rate:{user_id}'
    now = time.time()
    pipe = redis.pipeline()
    pipe.hgetall(key)
    state, _ = pipe.execute()
    tokens = float(state.get('tokens', capacity))
    last = float(state.get('last', now))
    tokens = min(capacity, tokens + (now - last) * rate)
    if tokens < 1:
        return False
    tokens -= 1
    redis.hset(key, mapping={'tokens': tokens, 'last': now})
    redis.expire(key, 60)
    return True

FastAPI Integration with slowapi

# pip install slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post('/agents/qa')
@limiter.limit('10/minute')
def qa(req: AgentRequest):
    ...

Different Limits Per Tier

class User:
    def __init__(self, is_pro):
        self.is_pro = is_pro

class State:
    def __init__(self, is_pro):
        self.user = User(is_pro)

class Req:
    def __init__(self, is_pro):
        self.state = State(is_pro)

def limit_for(req):
    return '100/minute' if req.state.user.is_pro else '10/minute'

def qa(req):
    limit = limit_for(req)
    print(f'user is_pro={req.state.user.is_pro} -> rate limit {limit}')
    return limit

qa(Req(is_pro=True))
qa(Req(is_pro=False))

基于成本的配额

限制金额,而不仅仅是请求数:

def check_budget(user_id, predicted_cost):
    key = f'cost:{user_id}:{date.today()}'
    spent = float(redis.get(key) or 0)
    if spent + predicted_cost > daily_budget(user_id):
        raise HTTPException(429, 'Daily budget exceeded')

# After the call:
redis.incrbyfloat(key, actual_cost)
redis.expireat(key, midnight_tomorrow_ts)

并发上限

还要限制每位用户正在处理的请求数:

key = f'inflight:{user_id}'
redis.sadd(key, request_id)
if redis.scard(key) > 3:
    raise HTTPException(429, 'Too many concurrent requests')
# Remove from set when the request finishes.

全局速率限制

保护下游 LLM 接口,避免受到您自己的服务影响:

global_key = 'rate:global:llm'
if redis.incr(global_key) > 60:   # 60 calls per second
    raise HTTPException(503, 'Service overloaded')
redis.expire(global_key, 1)

Return Useful Errors

headers = {
    'X-RateLimit-Limit': '60',
    'X-RateLimit-Remaining': '0',
    'X-RateLimit-Reset': str(reset_time),
    'Retry-After': '30'
}
return JSONResponse({'error': 'rate-limited'}, status_code=429, headers=headers)

分布式速率限制

如果您运行多个接口服务器,速率限制必须共享。Redis 是标准后端;Cloudflare 和 Kong 提供托管替代方案。

突发流量额度

真实使用情况通常会出现突发流量。请允许小规模突发(例如,速率为每秒 10 个请求时,允许一次发送 30 个请求)。令牌桶实现通过更高的 `capacity` 自然地处理这一点。

告警

在以下情况下发出告警:

  • 全局速率限制触发次数每分钟超过 X 次
  • 任何用户持续达到限制(可能需要升级,或者存在错误)
  • 每日预算接近 80%

双层保护

为什么同时需要 BOTH 速率限制 AND 配额?

回顾

使用令牌桶实现速率限制,使用成本计数器实现配额,设置每位用户的并发上限,使用全局熔断器,并通过包含 Retry-After 的实用 429 响应进行提示。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码,获得全天候 AI 导师的即时帮助,并在网页或应用中继续学习。

课程
60
课程
239

常见问题解答

「速率限制与配额管理」课时是免费的吗?

是的 — 「速率限制与配额管理」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「速率限制与配额管理」这节课中我会学到什么?

设置按用户、组织和端点划分的配额,避免某个租户耗尽您的 OpenAI 预算。 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗?

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 3 节课,共 4 节。

「速率限制与配额管理」课时需要多长时间?

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗?

能。每节 AI Agents 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 通过 API 提供智能体服务
  2. 异步工作流与后台任务
  3. 速率限制与配额管理
  4. 智能体的蓝绿部署与金丝雀部署
← 返回 AI Agents