⚡ 背压 & 流控

# 伪代码
counter = 0
window_start = now()

def allow_request():
    if now() - window_start >= WINDOW_SIZE:
        counter = 0
        window_start = now()
    if counter < LIMIT:
        counter += 1
        return True
    return False

# 伪代码
log = []  # 请求时间戳列表

def allow_request():
    now = current_time()
    # 移除窗口外的记录
    log = [t for t in log if now - t < WINDOW_SIZE]
    if len(log) < LIMIT:
        log.append(now)
        return True
    return False

# 伪代码
def allow_request():
    now = current_time()
    current_window = now // WINDOW_SIZE
    elapsed_ratio = (now % WINDOW_SIZE) / WINDOW_SIZE

    current_count = get_count(current_window)
    previous_count = get_count(current_window - 1)

    # 加权估计
    estimated = previous_count * (1 - elapsed_ratio) + current_count
    if estimated < LIMIT:
        increment_count(current_window)
        return True
    return False

# 令牌桶的两个关键参数
RATE = 10       # 令牌填充速率 (tokens/second)
BURST = 20      # 桶容量 (最大突发量)

# 状态
tokens = BURST  # 初始满桶
last_refill = now()

def allow_request():
    # 计算自上次填充以来的新令牌
    now = current_time()
    elapsed = now - last_refill
    tokens = min(BURST, tokens + elapsed * RATE)
    last_refill = now

    if tokens >= 1:
        tokens -= 1
        return True
    return False

# 漏桶参数
RATE = 10        # 固定流出速率 (requests/second)
CAPACITY = 20    # 桶容量 (队列长度)

queue = []
last_leak = now()

def allow_request():
    # 先漏水
    now = current_time()
    elapsed = now - last_leak
    leak_count = int(elapsed * RATE)
    for _ in range(min(leak_count, len(queue))):
        queue.pop(0)  # 处理并移除
    last_leak = now

    if len(queue) < CAPACITY:
        queue.append(request)
        return True
    return False  # 桶满，拒绝

关键：限流在组织级别和项目级别生效，不是用户级别。多个模型可能共享同一限流配额。

随着消费增加，自动升级到更高层级，RPM/TPM 限制也会相应提高。

import time
import random
from openai import OpenAI

client = OpenAI()

def call_llm_with_retry(messages, max_retries=5):
    """带指数退避和抖动的 LLM 调用"""
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                max_tokens=1000
            )
            return response
        except Exception as e:
            if hasattr(e, 'status_code') and e.status_code == 429:
                # 指数退避 + 随机抖动
                base_delay = min(2 ** attempt, 60)  # 最大 60 秒
                jitter = random.uniform(0, base_delay * 0.5)
                delay = base_delay + jitter
                print(f"Rate limited, retrying in {delay:.1f}s (attempt {attempt+1})")
                time.sleep(delay)
            else:
                raise
    raise Exception(f"Max retries ({max_retries}) exceeded")

# 使用 tenacity 库更优雅
from tenacity import retry, wait_exponential_jitter, stop_after_attempt

@retry(
    wait=wait_exponential_jitter(max=60),
    stop=stop_after_attempt(5)
)
def call_llm(messages):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )

# 智能模型路由——高峰期自动降级
import time
from collections import defaultdict

class ModelRouter:
    def __init__(self):
        self.request_counts = defaultdict(int)
        self.window_start = time.time()

    def get_model(self, complexity: str = "normal"):
        """根据系统负载选择模型"""
        now = time.time()
        if now - self.window_start > 60:
            self.request_counts.clear()
            self.window_start = now

        rpm = sum(self.request_counts.values())
        load_factor = rpm / 1000  # 假设 1000 RPM 是警戒线

        if load_factor > 0.8:
            # 高负载：全部使用小模型
            return "gpt-4o-mini"
        elif load_factor > 0.5:
            # 中等负载：复杂任务用大模型，简单任务用小模型
            return "gpt-4o" if complexity == "complex" else "gpt-4o-mini"
        else:
            # 低负载：按需分配
            return "gpt-4o" if complexity != "simple" else "gpt-4o-mini"

router = ModelRouter()
model = router.get_model(complexity="complex")

# /etc/nginx/conf.d/rate-limit.conf

# 定义限流区域
# $binary_remote_addr — 以 IP 为 key（比 $remote_addr 省内存）
# zone=mylimit:10m — 共享内存区域名和大小（10MB ≈ 16万 IP）
# rate=10r/s — 10 请求/秒 = 每 100ms 允许 1 请求
limit_req_zone $binary_remote_addr zone=mylimit:10m rate=10r/s;

server {
    # 登录接口——严格限流
    location /login/ {
        # burst=20: 允许 20 个请求排队
        # nodelay: 排队的请求立即处理（而非匀速释放）
        limit_req zone=mylimit burst=20 nodelay;
        limit_req_status 429;  # 默认返回 503，改为 429 更语义化

        proxy_pass http://backend;
    }

    # API 接口——两阶段限流（NGINX 1.15.7+）
    location /api/ {
        # delay=8: 前 8 个突发请求立即处理，后续排队等待
        limit_req zone=mylimit burst=12 delay=8;

        proxy_pass http://backend;
    }
}

# 按 API Key 限流（而非 IP）
# 从查询参数或 Header 提取 key
limit_req_zone $api_key zone=api_limit:10m rate=60r/m;

map $http_authorization $api_key {
    ~^Bearer\s+(\S+)$ $1;  # 从 Authorization header 提取
    default $binary_remote_addr;  # fallback 到 IP
}

server {
    location /api/ {
        limit_req zone=api_limit burst=10 nodelay;
        limit_req_status 429;

        # 自定义 429 响应
        error_page 429 /rate_limited.html;

        # 添加限流相关 Header
        add_header X-RateLimit-Limit 60 always;
        add_header X-RateLimit-Remaining $limit_remaining always;

        proxy_pass http://backend;
    }

    location = /rate_limited.html {
        internal;
        default_type application/json;
        return 429 '{"error":"rate_limit_exceeded","message":"Too many requests. Please retry later.","retry_after":60}';
    }
}

# 每个 MCP 工具独立限流
from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class ToolRateLimit:
    """工具级别的限流配置"""
    tool_name: str
    rpm: int                    # 每分钟最大调用次数
    concurrent: int = 5         # 最大并发调用数
    timeout: float = 30.0       # 单次调用超时
    fallback: Optional[str] = None  # 限流时的替代工具

# 工具限流注册表
TOOL_LIMITS = {
    "web_search":      ToolRateLimit("web_search", rpm=30, concurrent=3),
    "code_execute":    ToolRateLimit("code_execute", rpm=60, concurrent=5),
    "llm_generate":    ToolRateLimit("llm_generate", rpm=100, concurrent=10,
                                    fallback="llm_generate_fast"),
    "llm_generate_fast": ToolRateLimit("llm_generate_fast", rpm=200, concurrent=20),
    "database_query":  ToolRateLimit("database_query", rpm=120, concurrent=8),
}

class ToolRateLimiter:
    """带限流的工具调用器"""

    def __init__(self):
        self.semaphores = {
            name: asyncio.Semaphore(limit.concurrent)
            for name, limit in TOOL_LIMITS.items()
        }
        self.counters = defaultdict(int)  # RPM 计数
        self.window_start = time.time()

    async def call(self, tool_name: str, **kwargs):
        limit = TOOL_LIMITS.get(tool_name)
        if not limit:
            return await self._execute(tool_name, **kwargs)

        # 检查 RPM
        if self.counters[tool_name] >= limit.rpm:
            if limit.fallback:
                return await self.call(limit.fallback, **kwargs)
            raise RateLimitError(f"Tool {tool_name} rate limited")

        # 并发控制
        async with self.semaphores[tool_name]:
            self.counters[tool_name] += 1
            try:
                return await asyncio.wait_for(
                    self._execute(tool_name, **kwargs),
                    timeout=limit.timeout
                )
            except asyncio.TimeoutError:
                raise ToolTimeoutError(f"Tool {tool_name} timed out")

# 不同优先级的 Agent 使用不同的队列
import asyncio
from enum import IntEnum

class Priority(IntEnum):
    CRITICAL = 0  # 支付、安全检查
    HIGH = 1      # 用户直接请求
    NORMAL = 2    # 后台分析
    LOW = 3       # 批量任务

class PriorityExecutor:
    """优先级执行器——高负载时丢弃低优先级任务"""

    def __init__(self, max_workers: int = 10):
        self.queues = {p: asyncio.Queue() for p in Priority}
        self.max_workers = max_workers
        self.active_workers = 0
        self.running = True

    async def submit(self, priority: Priority, coro):
        """提交任务到优先级队列"""
        await self.queues[priority].put(coro)
        self._schedule()

    def _schedule(self):
        """调度：优先处理高优先级"""
        if self.active_workers >= self.max_workers:
            return

        # 从高到低检查队列
        for priority in sorted(Priority):
            if not self.queues[priority].empty():
                task = asyncio.create_task(self._worker(priority))
                self.active_workers += 1
                break

    async def _worker(self, priority: Priority):
        """工作线程"""
        try:
            while self.running:
                try:
                    coro = self.queues[priority].get_nowait()
                    await coro
                except asyncio.QueueEmpty:
                    break
        finally:
            self.active_workers -= 1
            self._schedule()

    def shed_load(self, keep_up_to: Priority = Priority.HIGH):
        """负载脱落：丢弃低优先级任务"""
        for priority in Priority:
            if priority > keep_up_to:
                q = self.queues[priority]
                dropped = 0
                while not q.empty():
                    try:
                        q.get_nowait()
                        dropped += 1
                    except asyncio.QueueEmpty:
                        break
                if dropped:
                    print(f"Shed {dropped} tasks from priority {priority.name}")

# 断路器：当下游持续失败时停止发送请求
from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"      # 正常：允许请求
    OPEN = "open"          # 熔断：拒绝请求
    HALF_OPEN = "half_open"  # 探测：允许少量请求测试

class CircuitBreaker:
    """断路器——保护下游服务"""

    def __init__(self, failure_threshold: int = 5,
                 recovery_timeout: float = 30.0,
                 half_open_max: int = 3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max = half_open_max

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.half_open_count = 0

    async def call(self, func, *args, **kwargs):
        """通过断路器调用函数"""
        if self.state == CircuitState.OPEN:
            # 检查是否到恢复时间
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_count = 0
            else:
                raise CircuitOpenError("Circuit breaker is OPEN")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_count >= self.half_open_max:
                raise CircuitOpenError("Circuit breaker HALF_OPEN limit reached")
            self.half_open_count += 1

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        """成功回调"""
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED  # 恢复正常
        self.failure_count = 0

    def _on_failure(self):
        """失败回调"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN  # 熔断
            print(f"⚠️ Circuit OPEN after {self.failure_count} failures")

# 使用：保护 LLM API 调用
llm_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)

async def safe_llm_call(prompt):
    try:
        return await llm_breaker.call(call_openai, prompt)
    except CircuitOpenError:
        # 断路器打开时使用缓存或降级
        return get_cached_response(prompt)

你需要限流吗？
├── 是 → 需要允许突发吗？
│   ├── 是 → 用令牌桶 ⭐
│   │   └── 例：API 限流、LLM 调用控制
│   └── 否 → 需要严格匀速吗？
│       ├── 是 → 用漏桶
│       │   └── 例：数据库写入限流、消息匀速发送
│       └── 否 → 需要精确计数吗？
│           ├── 是 → 用滑动窗口日志
│           │   └── 例：金融合规、审计
│           └── 否 → 用固定窗口或滑动窗口计数
│               └── 例：简单 API 限流
└── 不确定 → 先用固定窗口，够用就行

独立开发者推荐起步方案：
  NGINX 漏桶（层 1） + Redis 令牌桶（层 2） + 指数退避（客户端）