在国内调用 AI 大模型 API 时，限流是每个工程师必须面对的挑战。无论是 立即注册 的 HolySheep 中转服务，还是直接对接 OpenAI/Anthropic 官方接口，429 Too Many Requests 错误都像家常便饭。本文从实战角度出发，深入对比令牌桶（Token Bucket）和滑动窗口（Sliding Window）两种主流限流算法的实现、性能和适用场景，附带完整的生产级 Python 代码和 benchmark 数据。
为什么 AI API 限流如此重要
我曾在一次为金融客户构建量化分析系统的项目中，遭遇了严重的限流问题。当时团队日均调用量超过 50 万次，峰值 QPS 达到 200+，但上游 API 的限制只有 100 QPS。由于缺乏有效的限流机制，系统频繁触发 429 错误，导致下游量化模型训练中断，最终造成了数小时的业务损失。
那次教训让我深刻认识到：限流不仅是保护上游 API 的手段，更是保障系统稳定性的核心防线。一个设计良好的限流方案，可以让你在充分利用 API 配额的同时，避免因超限导致的请求失败、重试风暴和用户体验下降。
令牌桶算法：允许适度突发
算法原理
令牌桶的核心思想是：系统以固定速率向桶中添加令牌，桶的容量有上限。每个请求需要消耗一个（或多个）令牌，当桶中有足够令牌时请求通过，否则被拒绝或等待。这种设计允许在令牌充足时处理突发流量，同时长期来看请求速率被平滑限制在令牌添加速率。
生产级 Python 实现
import time
import threading
from typing import Optional, Dict
from dataclasses import dataclass, field
@dataclass
class TokenBucketConfig:
"""令牌桶配置"""
rate: float # 每秒添加的令牌数
capacity: int # 桶的最大容量
name: str = "default"
class TokenBucket:
"""线程安全的令牌桶限流器
适用场景：
- 允许适度突发流量的业务
- 需要平滑限制长期请求速率
- 批量任务和异步处理场景
"""
def __init__(self, config: TokenBucketConfig):
self.rate = config.rate
self.capacity = config.capacity
self.name = config.name
self._tokens: float = float(config.capacity)
self._last_refill_time: float = time.monotonic()
self._lock = threading.RLock()
self._total_requests: int = 0
self._total_allows: int = 0
self._total_rejects: int = 0
def _refill(self) -> None:
"""补充令牌"""
now = time.monotonic()
elapsed = now - self._last_refill_time
self._tokens = min(
self.capacity,
self._tokens + elapsed * self.rate
)
self._last_refill_time = now
def allow_request(self, tokens_needed: int = 1) -> bool:
"""检查是否允许请求（不阻塞）"""
with self._lock:
self._refill()
self._total_requests += 1
if self._tokens >= tokens_needed:
self._tokens -= tokens_needed
self._total_allows += 1