限流应对最佳实践

import openai
from openai import OpenAI
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type
)

RETRYABLE_ERRORS = (
    openai.RateLimitError,
    openai.InternalServerError,
    openai.APIConnectionError,
)

@retry(
    wait=wait_random_exponential(min=1, max=60),
    stop=stop_after_attempt(6),
    retry=retry_if_exception_type(RETRYABLE_ERRORS)
)
def chat_with_retry(client, model, messages, max_tokens):
    return client.chat.completions.create(
        model=model,
        max_tokens=max_tokens,
        messages=messages
    )

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="YOUR_DASHSCOPE_API_KEY"
)

try:
    response = chat_with_retry(
        client=client,
        model="qwen-plus",
        messages=[{"role": "user", "content": "什么是指数退避重试？"}],
        max_tokens=1024
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"请求失败: {e}")

import time

class TokenBucket:
    """
    令牌桶实现，用于控制每分钟请求数 (RPM)。
    支持预支 (Debt) 机制，以保证高并发下的先进先出 (FIFO) 顺序。
    """
    def __init__(self, quota_per_minute: float, initial_tokens: float = 0.0):
        self.capacity = quota_per_minute
        self.tokens = initial_tokens
        self.refill_rate = quota_per_minute / 60.0
        self.last_refill = time.monotonic()

    def reserve(self, cost: float = 1.0) -> float:
        """
        申请令牌。
        如果令牌不足，返回需要等待的秒数（支持预支）。
        """
        self._refill()

        # 1. 令牌充足：直接扣除
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0

        # 2. 令牌不足：计算等待时间并预支
        # 为当前请求"预定"了未来的令牌，确保 FIFO 顺序
        deficit = cost - self.tokens
        wait_seconds = deficit / self.refill_rate
        self.tokens -= cost
        return wait_seconds

    def _refill(self):
        """根据流逝时间补充令牌"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        if elapsed > 0:
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            self.last_refill = now

import time

class TokenBucket:
    """进阶令牌桶，支持匀速预热 (Continuous Warm-up) 机制。"""
    def __init__(self, quota_per_minute: float, warmup_seconds: float = 0.0):
        self.capacity = quota_per_minute
        self.tokens = 0.0
        self.target_refill_rate = quota_per_minute / 60.0
        self.warmup_seconds = warmup_seconds
        self.start_time = time.monotonic()
        self.last_update_time = self.start_time
        self.cumulative_generated = 0.0

    def _get_cumulative_tokens(self, t: float) -> float:
        if t <= 0:
            return 0.0
        R = self.target_refill_rate
        T = self.warmup_seconds
        if T <= 0:
            return R * t
        if t <= T:
            return (R / (2 * T)) * (t ** 2)
        else:
            warmup_total = (R * T) / 2.0
            return warmup_total + R * (t - T)

    def _get_time_for_cumulative_tokens(self, target_cumulative: float) -> float:
        if target_cumulative <= 0:
            return 0.0
        R = self.target_refill_rate
        T = self.warmup_seconds
        if T <= 0:
            return target_cumulative / R
        warmup_total = (R * T) / 2.0
        if target_cumulative <= warmup_total:
            return ((2 * T * target_cumulative) / R) ** 0.5
        else:
            return (target_cumulative - warmup_total) / R + T

    def reserve(self, cost: float = 1.0) -> float:
        now = time.monotonic()
        relative_now = now - self.start_time
        current_cumulative = self._get_cumulative_tokens(relative_now)
        new_tokens = current_cumulative - self.cumulative_generated
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.cumulative_generated = current_cumulative
        self.last_update_time = now
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0
        deficit = cost - self.tokens
        self.tokens -= cost
        target_cumulative = self.cumulative_generated + deficit
        target_time = self._get_time_for_cumulative_tokens(target_cumulative)
        wait_seconds = target_time - relative_now
        return max(0.0, wait_seconds)

    def adjust(self, amount: float):
        self.tokens = min(self.capacity, self.tokens + amount)

class ElasticCongestionController:
    async def acquire(self):
        """[准入阶段] 请求发起前的检查"""
        # 1. SSR 慢启动重启：若空闲太久，主动衰减上限
        #    防止过时水位导致的突发流量
        if self.is_idle_too_long():
            self.perform_slow_start_restart()

        # 2. 熔断检查：若处于 DRAIN (冷却) 状态，强制等待
        if self.state == CongestionState.DRAIN:
            await self.wait_for_cooldown()

        # 3. 双重预算检查：同时检查并发槽位和 Token 预算
        await self.wait_for_budget(request_tokens)

    async def release(self, latency, actual_tokens, error):
        """[反馈阶段] 请求结束后的决策"""
        if error:
            # [故障响应] 遇限流错误 (429/503)：立即排水 + 乘性回退
            self.state = CongestionState.DRAIN
            self.concurrency_limit *= self.backoff_factor  # e.g. 0.7
            return

        # [正常响应] 计算 TPT (Time-Per-Token)
        current_tpt = latency / actual_tokens

        # [拥塞感知] TPT 突增 (生成变慢)：进入 HOLD 观察
        # 维持并发水平，不退避也不增长
        if current_tpt > self.metrics.ema_tpt * 2.0:
            self.state = CongestionState.HOLD
        else:
            # [稳态探测] 网络健康：执行 EBP 弹性探测
            self.state = CongestionState.PROBING
            self.update_limit_via_ebp()