_{🧠 记忆锚点：限流控制谁能进，背压防止队列失控，熔断隔离故障上游；Token 配额要先预留、再按实际用量对账。}

💡 答案要点

为什么 LLM API 需要限流？

LLM API 限流有三个特殊性：按 Token 计费（超限直接烧钱）、调用延迟高（3-30秒阻塞会级联放大）、上游 API 有 RPM/TPM 限制（超限直接 429）。例如 1000 QPS × 平均 1000 tokens/请求 = 1M TPM，而 GPT-4o TPM 限制仅 450K，不限流直接爆。

限流算法对比：

算法	原理	优点	缺点	适用场景
固定窗口	每分钟固定配额	简单	边界双倍配额	粗粒度控制
滑动窗口	时间窗口平滑	比固定窗口公平	稍复杂	通用限流
令牌桶	桶内令牌决定能否通过	允许突发	实现稍复杂	突发流量
漏桶	恒定速率消费	平滑输出	突发受限	保护下游
自适应限流	根据 429/成功率动态调整	智能	最复杂	保护多租户

令牌桶实现（Python）：

展开 Python 代码示例（32 行）

import time
import threading

class TokenBucketRateLimiter:
    def __init__(self, rate: int, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    def acquire(self, tokens: int = 1, blocking: bool = True, timeout: float = None) -> bool:
        start = time.time()
        with self.lock:
            while True:
                self._refill()
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
                if not blocking:
                    return False
                wait_time = (tokens - self.tokens) / self.rate
                if timeout and (time.time() - start) >= wait_time:
                    return False
                time.sleep(min(wait_time, 0.1))

Token 速率控制器（LLM API 专用）：

import asyncio
import time
from collections import deque

class TokenRateLimiter:
    def __init__(self, tpm_limit: int, window_seconds: int = 60):
        self.tpm_limit = tpm_limit
        self.window = window_seconds
        self.tokens_used = deque()
        self.lock = asyncio.Lock()

    async def acquire(self, tokens: int, timeout: float = 60) -> bool:
        start = time.time()
        while True:
            async with self.lock:
                now = time.time()
                while self.tokens_used and now - self.tokens_used[0][0] > self.window:
                    self.tokens_used.popleft()
                current_usage = sum(t for _, t in self.tokens_used)
                if current_usage + tokens <= self.tpm_limit:
                    self.tokens_used.append((now, tokens))
                    return True
            if timeout and time.time() - start >= timeout:
                return False
            await asyncio.sleep(0.1)

背压机制（Backpressure）：

展开 Python 代码示例（30 行）

class LLMOverloadedException(Exception):
    pass

class LLMCallWithBackpressure:
    def __init__(self, rate_limiter: TokenRateLimiter, max_queue_size: int = 100):
        self.rate_limiter = rate_limiter
        self.queue = asyncio.Queue(maxsize=max_queue_size)
        self.results = {}

    async def submit(self, request_id: str, prompt: str) -> str:
        try:
            self.queue.put_nowait((time.time(), request_id, prompt))
        except asyncio.QueueFull:
            raise LLMOverloadedException(
                f"Queue full, current load={self.queue.qsize()}, max={self.queue.maxsize}"
            )
        return request_id

    async def process(self):
        while True:
            ts, request_id, prompt = await self.queue.get()
            wait_time = time.time() - ts
            if wait_time > 30:
                self.results[request_id] = {"status": "timeout", "result": None}
                continue
            tokens = estimate_tokens(prompt)
            if await self.rate_limiter.acquire(tokens, timeout=60):
                self.results[request_id] = {"status": "done", "result": llm.generate(prompt)}
            else:
                self.results[request_id] = {"status": "rate_limited", "result": None}

生产级限流配置：

rate_limits:
  gpt-4o:
    tpm: 450000
    rpm: 5000
    effective_limit: 400000  # 安全水位 90%
  claude-3-5-sonnet:
    tpm: 1000000
    effective_limit: 800000

backpressure:
  queue_size: 200
  timeout_seconds: 30
  degrade_to_model: "deepseek-v4-flash"

circuit_breaker:
  error_threshold: 0.5
  recovery_timeout: 60

面试话术：

"LLM 限流的核心是'Token 速率控制'而非'请求速率控制'——因为按 Token 计费，请求大小的差异会导致实际消耗差 10 倍。我的实现用令牌桶控制 TPM，配额用 OpenAI 的 90% 作为安全水位。背压机制是当请求堆积超过阈值时直接拒绝，而不是让用户等待——等待会导致超时窗口更长，用户体验更差。生产环境的教训是：429 一定要立即触发限流，不要重试 10 次才认输，那会瞬间打爆上游。"

📚 参考：OpenAI：Rate Limits（限流与背压）