AI API 429 Too Many Requests の処理と退避戦略完全ガイド

ジッター付き指数バックオフ

基本的な指数バックオフにジッター（乱数）を追加することで、複数のクライアントが同時に再試行する「、雷鳴の問題（thundering herd problem）」を防ぎます。

import random
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Optional
import logging

@dataclass
class RateLimitConfig:
    """レートリミット設定"""
    requests_per_minute: int = 60
    tokens_per_minute: int = 60000
    burst_size: int = 10

class HolySheepAsyncClient:
    """非同期用のHolySheep AIクライアント（ジッター付きバックオフ）"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1"
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.logger = logging.getLogger(__name__)
        self._rate_limiter = None
    
    def _jittered_backoff(
        self,
        attempt: int,
        base: float = 1.0,
        cap: float = 60.0,
        jitter_factor: float = 0.5
    ) -> float:
        """
        ジッター付き指数バックオフ
        
        Full Jitter方式：ランダムな値を加算して山谷を分散
        
        待機時間 = min(cap, base * 2^attempt * random(1-jitter_factor, 1+jitter_factor))
        """
        exponential_delay = base * (2 ** attempt)
        # ジッター適用（±50%の範囲でランダム）
        jitter = random.uniform(1 - jitter_factor, 1 + jitter_factor)
        full_jitter_delay = exponential_delay * jitter
        
        return min(full_jitter_delay, cap)
    
    async def chat_completion_async(
        self,
        messages: list,
        model: str = "gpt-4-turbo",
        max_retries: int = 5,
        **kwargs
    ) -> dict:
        """
        非同期でChat Completion APIを呼び出す
        
        特徴：
        - asyncioによる並行処理対応
        - ジッター付きバックオフ
        - 自動リトライ
        """
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": model,
            "messages": messages,
            **kwargs
        }
        
        async with aiohttp.ClientSession() as session:
            for attempt in range(max_retries):
                try:
                    async with session.post(
                        url,
                        headers=headers,
                        json=payload,
                        timeout=aiohttp.ClientTimeout(total=30)
                    ) as response:
                        
                        if response.status == 200:
                            return await response.json()
                        
                        elif response.status == 429:
                            # Retry-After ヘッダーの確認
                            retry_after = response.headers.get('Retry-After')
                            
                            if retry_after:
                                wait_time = int(retry_after)
                            else:
                                wait_time = self._jittered_backoff(attempt)
                            
                            self.logger.warning(
                                f"429発生: {wait_time:.2f}秒後にリトライ"
                            )
                            
                            if attempt < max_retries - 1:
                                await asyncio.sleep(wait_time)
                                continue
                            else:
                                raise Exception("最大リトライ回数超過")
                        
                        elif response.status >= 500:
                            wait_time = self._jittered_backoff(attempt)
                            if attempt < max_retries - 1:
                                await asyncio.sleep(wait_time)
                                continue
                            else:
                                raise Exception(f"サーバーエラー: {response.status}")
                        
                        else:
                            # クライアントエラーはリトライ不要
                            error_text = await response.text()
                            raise Exception(
                                f"APIエラー {response.status}: {error_text}"
                            )
                
                except aiohttp.ClientError as e:
                    if attempt < max_retries - 1:
                        wait_time = self._jittered_backoff(attempt)
                        await asyncio.sleep(wait_time)
                        continue
                    raise
        
        raise Exception("予期しないエラー")


async def batch_process_example():
    """批量処理の例：多个文档を並行処理"""
    client = HolySheepAsyncClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    documents = [
        {"id": 1, "content": "文档1内容..."},
        {"id": 2, "content": "文档2内容..."},
        {"id": 3, "content": "文档3内容..."},
        {"id": 4, "content": "文档4内容..."},
    ]
    
    tasks = []
    for doc in documents:
        messages = [
            {"role": "system", "content": "あなたはドキュメント分析助手です"},
            {"role": "user", "content": f"次の文档を要約してください：{doc['content']}"}
        ]
        task = client.chat_completion_async(messages)
        tasks.append((doc['id'], task))
    
    results = await asyncio.gather(*[t[1] for t in tasks], return_exceptions=True)
    
    for (doc_id, _), result in zip(tasks, results):
        if isinstance(result, Exception):
            print(f"ドキュメント {doc_id} エラー: {result}")
        else:
            print(f"ドキュメント {doc_id} 成功: {result['choices'][0]['message']['content'][:50]}...")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(batch_process_example())

RAGシステムでのレート制限対策

企业RAG系统在处理大量文档时，推荐采用以下架构：

• セマフォフォア（Semaphore）：同时请求数の上限を制御
• トークンバジェット管理：TPM（每分钟令牌数）を监控
• リクエストキュー：超过限制的请求进行排队

import asyncio
from collections import deque
import time
from typing import List, Dict, Any
import logging

class TokenBucketRateLimiter:
    """
    トークンバケット方式のレートリミッター
    
    指定された速率（1秒あたりのリクエスト数）でリクエストを許可し、
    バースト的なアクセスも一定の範囲内で許容します。
    """
    
    def __init__(
        self,
        rpm: int = 60,      # 1分あたりのリクエスト数
        tpm: int = 60000,   # 1分あたりのトークン数
        burst: int = 10     # バースト許容数
    ):
        self.rpm = rpm
        self.tpm = tpm
        self.burst = burst
        
        # トークンバケットの状態
        self.tokens = burst
        self.last_update = time.time()
        self.request_timestamps = deque()
        
        # ロック（并发控制用）
        self._lock = asyncio.Lock()
    
    async def acquire(self, estimated_tokens: int = 1000):
        """
        レート制限内でリクエストを実行する
        
        Args:
            estimated_tokens: 推定トークン数（入力+出力）
        """
        async with self._lock:
            now = time.time()
            
            # 时间経過でトークンを補充
            elapsed = now - self.last_update
            refill_rate = self.rpm / 60.0  # 每秒补充速率
            self.tokens = min(self.burst, self.tokens + elapsed * refill_rate)
            self.last_update = now
            
            # トークン不足の場合は待機
            if self.tokens < estimated_tokens / 1000:
                wait_time = (estimated_tokens / 1000 - self.tokens) / refill_rate
                logging.warning(f"トークン不足: {wait_time:.2f}秒待機")
                await asyncio.sleep(wait_time)
                self.tokens = 0
            else:
                self.tokens -= estimated_tokens / 1000
            
            # タイムスタンプを記録
            self.request_timestamps.append(now)
            
            # 1分以上古いタイムスタンプを削除
            while self.request_timestamps and 
                  now - self.request_timestamps[0] > 60:
                self.request_timestamps.popleft()
            
            # RPM超えていないか確認
            if len(self.request_timestamps) > self.rpm:
                sleep_time = 60 - (now - self.request_timestamps[0])
                if sleep_time > 0:
                    logging.warning(f"RPM超え: {sleep_time:.2f}秒待機")
                    await asyncio.sleep(sleep_time)


class RAGProcessor:
    """
    RAGシステム用プロセッサー
    
    문서 검색 및 생성 요구를 순차적으로 처리하고
    APIへのリクエストを適切にスロットリングします。
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1"
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.rate_limiter = TokenBucketRateLimiter(rpm=60, tpm=60000)
        self.logger = logging.getLogger(__name__)
    
    async def process_query(
        self,
        query: str,
        retrieved_contexts: List[str]
    ) -> Dict[str, Any]:
        """
        RAGクエリを処理
        
        1. 检索的文書をプロンプトに組込み
        2. レートリミッター経由でAPI呼び出し
        """
        # プロンプト構築
        context_text = "\n\n".join([
            f"[文脈{i+1}] {ctx}" for i, ctx in enumerate(retrieved_contexts)
        ])
        
        messages = [
            {
                "role": "system",
                "content": "提供された文脈のみに基づいて、准确な回答を生成してください。"
            },
            {
                "role": "user",
                "content": f"文脈：\n{context_text}\n\n質問：{query}"
            }
        ]
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
ja o3 api jieruwanzhengjiaochengopenai zuiqiangtuilim 2026 0
ja claude vs gpt 4o api chengbenzhutiaoduibifenxi 2026 04 04