AI 应用限流设计：令牌桶与漏桶算法実装完全ガイド

AI API を運用する上で避けて通れない問題が、急激なトラフィック増加によるサービス障害です。私は以前、トークン消費の Spike 時にConnectionError: timeoutが頻発し、夜中の3時に緊急対応を組んだ経験があります。本稿では、HolySheep AI を含む実際の API 呼び出しで可用性を確保するための令牌桶（Token Bucket）と漏桶（Leak Bucket）アルゴリズムについて、Python での実装方法から実践的なTips まで丁寧に解説します。

なぜ AI API には限流設計が重要か

AI API は処理コストが高く、(HolySheep AIの GPT-4.1 で $8/MTok、Claude Sonnet 4.5 で $15/MTok) 意図しない大量リクエストは即座にコスト超過を招きます。また、レイテンシ要件も厳しく、私は過去案件で p99 レイテンシ <50ms を維持するために、サーバーレス環境での限流実装を何度も行ってきました。

令牌桶（Token Bucket）アルゴリズム

アルゴリズム概要

令牌桶は「バケツにトークンが一定速度で補充され、リクエストごとにトークンを消費する」モデルです。バースト（一時的な高負荷）にも対応できるため、DeepSeek V3.2 の超低価格 $0.42/MTok を活用したコスト最適化にも有効です。

import time
import threading
from dataclasses import dataclass, field
from typing import Optional
import requests

@dataclass
class TokenBucket:
    """令牌桶算法実装 - スレッドセーフ対応"""
    capacity: float  # 最大トークン数（バケツサイズ）
    refill_rate: float  # 毎秒補充されるトークン数
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    lock: threading.Lock = field(default_factory=threading.Lock)

    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()

    def _refill(self):
        """現在時刻に基づいてトークンを補充"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

    def consume(self, tokens: float = 1.0) -> bool:
        """
        トークンを消費しようとする
        成功: True, 失敗（トークン不足）: False
        """
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def wait_for_token(self, tokens: float = 1.0, timeout: Optional[float] = None):
        """トークンが利用可能になるまでブロック"""
        start = time.monotonic()
        while True:
            if self.consume(tokens):
                return True
            if timeout and (time.monotonic() - start) >= timeout:
                raise TimeoutError("トークン取得がタイムアウトしました")
            time.sleep(0.01)  # CPU 節約


class HolySheepAIClient:
    """HolySheep AI API 用レート制限付きクライアント"""
    
    def __init__(self, api_key: str, max_rps: float = 10.0):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.bucket = TokenBucket(
            capacity=max_rps * 2,  # バースト許容
            refill_rate=max_rps      # 毎秒max_rps補充
        )
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def chat_completion(self, messages: list, model: str = "gpt-4o") -> dict:
        """チャット完了API呼び出し（レート制限付き）"""
        self.bucket.wait_for_token(timeout=30.0)
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={"model": model, "messages": messages},
            timeout=60
        )
        return response.json()


使用例
if __name__ == "__main__":
    client = HolySheepAIClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_rps=5.0  # 毎秒5リクエストまで
    )
    
    messages = [{"role": "user", "content": "こんにちは"}]
    result = client.chat_completion(messages)
    print(result)

令牌桶の適用シーン

• バースト流量対応: 初期補充されたトークンで突発的なリクエストを処理
• コスト制御: DeepSeek V3.2 ($0.42/MTok) など低価格APIの効率的活用
• 段階的制限: 容量を大きくすれば一時的な高負荷も吸収

漏桶（Leak Bucket）アルゴリズム

アルゴリズム概要

漏桶は「一定速度でしかリクエストを処理しない」モデルです。出力速度を強制的に一定に保つため、より厳しい流量制御が必要な場合に適しています。HolySheep AI で複数モデルを同時に使う場合、漏桶を用いて各モデルの総リクエスト数を均一化管理できます。

import time
import threading
from collections import deque
from typing import Optional

class LeakBucket:
    """漏桶算法実装 - 一定速度でのリクエスト処理保証"""
    
    def __init__(self, leak_rate: float, capacity: Optional[int] = None):
        """
        leak_rate: 毎秒処理可能なリクエスト数
        capacity: キューの最大サイズ（Noneは無限）
        """
        self.leak_rate = leak_rate
        self.capacity = capacity
        self.queue: deque = deque()
        self.last_leak_time = time.monotonic()
        self.lock = threading.Lock()

    def add_request(self, request_id: str) -> bool:
        """
        リクエストをキューに追加
        成功: True, 拒否（容量一杯）: False
        """
        with self.lock:
            self._leak()  # 時間経過でキューを排出
            
            if self.capacity and len(self.queue) >= self.capacity:
                return False
            
            self.queue.append(request_id)
            return True

    def _leak(self):
        """経過時間に基づいてキューからリクエストを排出"""
        now = time.monotonic()
        elapsed = now - self.last_leak_time
        requests_to_leak = int(elapsed * self.leak_rate)
        
        if requests_to_leak > 0:
            for _ in range(min(requests_to_leak, len(self.queue))):
                self.queue.popleft()
            self.last_leak_time = now

    def process_next(self) -> Optional[str]:
        """次のリクエストを取り出す（利用可能なければNone）"""
        with self.lock:
            self._leak()
            if self.queue:
                return self.queue.popleft()
            return None

    def get_queue_status(self) -> dict:
        """現在のキュー状態を取得"""
        with self.lock:
            self._leak()
            return {
                "queue_length": len(self.queue),
                "can_accept": not (self.capacity and len(self.queue) >= self.capacity),
                "leak_rate": self.leak_rate
            }


class MultiModelRateLimiter:
    """複数モデル用の漏桶ラッパー"""
    
    def __init__(self, total_rps: float = 20.0):
        self.global_bucket = LeakBucket(leak_rate=total_rps)
        self.model_buckets: dict[str, LeakBucket] = {}
    
    def register_model(self, model: str, rps: float):
        self.model_buckets[model] = LeakBucket(
            leak_rate=rps,
            capacity=int(rps * 2)  # 2秒分のバッファ
        )
    
    def request_allowed(self, model: str) -> bool:
        """リクエストが許可されるかチェック"""
        # グローバルレートのチェック
        if not self.global_bucket.add_request(f"global_{time.time()}"):
            return False
        
        # モデル別レートのチェック
        if model in self.model_buckets:
            return self.model_buckets[model].add_request(f"{model}_{time.time()}")
        
        return True
    
    def get_all_status(self) -> dict:
        return {
            "global": self.global_bucket.get_queue_status(),
            "models": {m: b.get_queue_status() for m, b in self.model_buckets.items()}
        }


使用例
if __name__ == "__main__":
    limiter = MultiModelRateLimiter(total_rps=30.0)
    
    # 各モデルのレート設定
    limiter.register_model("gpt-4o", rps=10.0)  # GPT-4.1 $8/MTok
    limiter.register_model("claude-sonnet", rps=5.0)  # Claude Sonnet 4.5 $15/MTok
    limiter.register_model("gemini-flash", rps=15.0)  # Gemini 2.5 Flash $2.50/MTok
    
    # リクエストチェック
    models = ["gpt-4o", "claude-sonnet", "gemini-flash"]
    for model in models:
        if limiter.request_allowed(model):
            print(f"✅ {model} へのリクエストは許可されました")
        else:
            print(f"❌ {model} へのリクエストは拒否されました（レート制限中）")

令牌桶と漏桶の比較

特徴	令牌桶	漏桶
バースト対応	✅ 可能	❌ 不可
出力平滑化	⚠️ 限定的	✅ 完全
実装複雑さ	中程度	比較的簡単
適切な用途	APIクライアント	APIサーバー

実際の API 呼び出しへの適用例

HolySheep AI での実装では、日本語サポートと<50msの低レイテンシを活かし、バッチ処理とリアルタイム処理を同時に行うアーキテクチャを組めます。以下は実践的な例です：

import asyncio
import aiohttp
from token_bucket import TokenBucket

class AsyncHolySheepClient:
    """非同期対応のHolySheep AIクライアント"""
    
    def __init__(self, api_key: str, requests_per_second: float = 10.0):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.bucket = TokenBucket(
            capacity=requests_per_second * 3,  # 3秒分のバースト
            refill_rate=requests_per_second
        )
        self._session: Optional[aiohttp.ClientSession] = None

    async def _ensure_session(self):
        if self._session is None:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            timeout = aiohttp.ClientTimeout(total=60)
            self._session = aiohttp.ClientSession(headers=headers, timeout=timeout)

    async def chat_completion_async(
        self, 
        messages: list, 
        model: str = "gpt-4o"
    ) -> dict:
        """非同期チャット完了呼び出し"""
        await self._ensure_session()
        
        # トークン確保を待つ
        await asyncio.get_event_loop().run_in_executor(
            None, self.bucket.wait_for_token, 1.0, 30.0
        )
        
        try:
            async with self._session.post(
                f"{self.base_url}/chat/completions",
                json={"model": model, "messages": messages}
            ) as response:
                if response.status == 429:
                    raise RateLimitError("レート制限に達しました")
                if response.status == 401:
                    raise AuthError("APIキーが無効です")
                
                return await response.json()
        
        except aiohttp.ClientError as e:
            raise ConnectionError(f"接続エラー: {e}")

    async def batch_process(
        self, 
        prompts: list[str], 
        model: str = "gpt-4o"
    ) -> list[dict]:
        """一括処理（並行数制限付き）"""
        semaphore = asyncio.Semaphore(5)  # 最大5並行
        
        async def process_single(prompt: str) -> dict:
            async with semaphore:
                return await self.chat_completion_async(
                    [{"role": "user", "content": prompt}],
                    model=model
                )
        
        return await asyncio.gather(*[process_single(p) for p in prompts])

    async def close(self):
        if self._session:
            await self._session.close()


class RateLimitError(Exception):
    """レート制限エラー"""
    pass

class AuthError(Exception):
    """認証エラー"""
    pass


使用例
async def main():
    client = AsyncHolySheepClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        requests_per_second=10.0
    )
    
    try:
        # 単一リクエスト
        result = await client.chat_completion_async([
            {"role": "user", "content": "Hello, HolySheep!"}
        ])
        print(f"単一結果: {result}")
        
        # バッチ処理（最大5並行）
        results = await client.batch_process([
            "質問1",
            "質問2", 
            "質問3",
            "質問4",
            "質問5"
        ])
        print(f"バッチ処理完了: {len(results)}件")
        
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

よくあるエラーと対処法

1. ConnectionError: timeout — タイムアウトエラー

発生原因: レート制限によりリクエストが長時間待機し、タイムアウト発生
解決策: タイムアウト値を適切に設定し、指数バックオフでリトライ

import asyncio
import aiohttp

async def retry_with_backoff(
    session: aiohttp.ClientSession,
    url: str,
    payload: dict,
    max_retries: int = 3,
    base_delay: float = 1.0
) -> dict:
    """指数バックオフ付きリトライ"""
    headers = {
        "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    for attempt in range(max_retries):
        try:
            async with session.post(url, json=payload, headers=headers) as response:
                if response.status == 200:
                    return await response.json()
                elif response.status == 429:
                    # レート制限時はRetry-Afterヘッダを確認
                    retry_after = response.headers.get("Retry-After", base_delay * (2 ** attempt))
                    await asyncio.sleep(float(retry_after))
                else:
                    raise aiohttp.ClientError(f"HTTP {response.status}")
                    
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise ConnectionError(f"最大リトライ回数を超過: {e}")
            delay = base_delay * (2 ** attempt)
            await asyncio.sleep(delay)
    
    raise ConnectionError("予期しないエラーでリトライが失敗しました")

2. 401 Unauthorized — 認証エラー

発生原因: API キーが無効または期限切れ
解決策: API キーの検証と自動ローテーション機構の実装

from typing import Optional

class APIKeyManager:
    """APIキー管理（キーローテーション対応）"""
    
    def __init__(self, api_keys: list[str]):
        self.api_keys = api_keys
        self.current_index = 0
        self.error_counts = {i: 0 for i in range(len(api_keys))}
        self.max_errors = 5  # 5回連続エラーでキーを切り替え
    
    @property
    def current_key(self) -> str:
        return self.api_keys[self.current_index]
    
    def mark_success(self):
        """成功時にエラーカウントをリセット"""
        self.error_counts[self.current_index] = 0
    
    def mark_error(self) -> Optional[str]:
        """エラー時にカウントアップ、問題あるならキーを切り替え"""
        self.error_counts[self.current_index] += 1
        
        if self.error_counts[self.current_index] >= self.max_errors:
            self._rotate_key()
            return self.current_key
        return None
    
    def _rotate_key(self):
        """次の有効なAPIキーに切り替え"""
        original_index = self.current_index
        for _ in range(len(self.api_keys)):
            self.current_index = (self.current_index + 1) % len(self.api_keys)
            if self.error_counts[self.current_index] < self.max_errors:
                print(f"APIキーをローテーション: {original_index} -> {self.current_index}")
                return
        raise RuntimeError("利用可能なAPIキーがありません")

使用例
key_manager = APIKeyManager([
    "YOUR_HOLYSHEEP_API_KEY_1",
    "YOUR_HOLYSHEEP_API_KEY_2",
    "YOUR_HOLYSHEEP_API_KEY_3"
])

3. 429 Too Many Requests — レート制限超過

発生原因: リクエスト頻度が API の制限を超えた
解決策: 分散型レートリミッターとリクエストキュー化

import time
from collections import defaultdict
import threading

class DistributedRateLimiter:
    """スレッドセーフな分散レートリミッター"""
    
    def __init__(self):
        self.request_times: dict[str, list[float]] = defaultdict(list)
        self.lock = threading.Lock()
        self.window_size = 1.0  # 1秒間
        self.max_requests = 50   # 最大リクエスト数/秒

    def is_allowed(self, client_id: str) -> bool:
        """リクエストが許可されるかチェック"""
        now = time.monotonic()
        cutoff = now - self.window_size
        
        with self.lock:
            # 古いリクエスト記録を削除
            self.request_times[client_id] = [
                t for t in self.request_times[client_id] if t > cutoff
            ]
            
            if len(self.request_times[client_id]) >= self.max_requests:
                return False
            
            self.request_times[client_id].append(now)
            return True

    def wait_time(self, client_id: str) -> float:
        """次のリクエストまで待つべき時間（秒）"""
        now = time.monotonic()
        cutoff = now - self.window_size
        
        with self.lock:
            recent = [t for t in self.request_times[client_id] if t > cutoff]
            if len(recent) < self.max_requests:
                return 0.0
            
            oldest = min(recent)
            return (oldest + self.window_size) - now


、実践的な使用例
limiter = DistributedRateLimiter()

def process_request(client_id: str, request_data: dict) -> bool:
    """リクエスト処理のラッパー"""
    if not limiter.is_allowed(client_id):
        wait = limiter.wait_time(client_id)
        raise RateLimitError(
            f"クライアント {client_id} はレート制限中です。"
            f"再試行まで {wait:.2f}秒 お待ちください。"
        )
    # 本来处理を続行
    return True

class RateLimitError(Exception):
    pass

まとめ：HolySheep AI での最佳Practices

本稿で解説した令牌桶と漏桶アルゴリズムは、AI API 運用の信頼性を大きく向上させます。今すぐ登録して、¥1=$1（公式比85%節約）の為替レートと<50msの低レイテンシを体験してみてください。Gemini 2.5 Flash ($2.50/MTok) や DeepSeek V3.2 ($0.42/MTok) などの選択肢も活用し、コスト効率とパフォーマンスのバランスを最適化しましょう。

実際の運用では、複数のアルゴリズムを組み合わせたハイブリッドアプローチも有効です。私の経験では、Webhook 処理には漏桶、定期的なポーリングには令牌桶を採用することで、p99 レイテンシを50ms以下かつリクエスト失敗率0.1%未満で維持できています。

👉 HolySheep AI に登録して無料クレジットを獲得