レート制限を処理するためのベストプラクティス

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": "What is exponential backoff retry?"}],
        max_tokens=1024
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"Request failed: {e}")

import time

class TokenBucket:
    """
    1 分あたりのリクエスト数 (RPM) を制御するためのトークンバケット実装。
    高い同時実行性の下で先入れ先出し (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:
    """継続的なウォームアップメカニズムをサポートする高度なトークンバケット。"""
    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. デュアルバジェットチェック：同時実行スロットとトークンバジェットの両方をチェック。
        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  # 例：0.7
            return

        # [通常応答] TPT (トークンあたりの時間) を計算
        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()