AI APIのレート制限：仕組みと対処法

すべてのAI APIにはレート制限があります。開発中に制限に達するとイライラしますし、本番環境で制限に達するとユーザーにエラーが表示されます。各プロバイダーのレート制限の仕組みを理解し、適切なリトライロジックを構築することが、デモと本番アプリケーションの違いです。

各プロバイダーの制限方法

OpenAI

OpenAIはアカウントの利用履歴と支払いレベルに基づく階層型のレート制限を採用しています。

階層	RPM（リクエスト/分）	TPM（トークン/分）	到達条件
無料	3	40,000	新規アカウント
Tier 1	500	200,000	$5支払い
Tier 2	5,000	2,000,000	$50支払い
Tier 3	5,000	10,000,000	$100支払い
Tier 4	10,000	50,000,000	$250支払い
Tier 5	10,000	300,000,000	$1,000支払い

制限はモデルごとに適用されます。GPT-4.1を使用してもGPT-4.1-miniのクォータは消費されません。

Anthropic

AnthropicもRPMとTPMの両方の制限を持つ類似の階層システムを使用しています。低階層では1日のトークン制限もあります。

Google (Gemini)

Google AI Studioはモデルごとの制限があります。無料階層は1日のリクエスト数は寛大ですが、1分あたりのレートは厳しいです（Gemini 2.5 Flash無料階層は15 RPM）。

アグリゲータープラットフォーム

LemonDataやOpenRouterのようなアグリゲーターは、上流の制限に加えて独自のレート制限レイヤーを設けています。LemonDataは役割ベースの制限を使用しています：

役割	RPM
ユーザー	1,000
パートナー	3,000
VIP	10,000

利点：アグリゲーターの制限は通常、個々のプロバイダーの無料階層よりも高く、マルチチャネルルーティングにより、ある上流チャネルがレート制限にかかっても別のチャネルにリクエストがルーティングされます。

レート制限ヘッダーの読み方

主要なプロバイダーはすべてレスポンスヘッダーにレート制限情報を返します：

x-ratelimit-limit-requests: 500
x-ratelimit-remaining-requests: 499
x-ratelimit-reset-requests: 60s
x-ratelimit-limit-tokens: 200000
x-ratelimit-remaining-tokens: 199500

これらのヘッダーを積極的に利用しましょう。429エラーが出るまで速度を落とさないのは避けてください。

リトライロジックの構築

間違った方法

# これはやめましょう
import time

def call_api(messages):
    while True:
        try:
            return client.chat.completions.create(
                model="gpt-4.1",
                messages=messages
            )
        except Exception:
            time.sleep(1)  # 固定遅延、バックオフなし、すべての例外をキャッチ

問題点：指数的バックオフなし、リトライ不可能なエラーもキャッチ、最大リトライ回数なし、ジッターなし。

正しい方法

import time
import random
from openai import RateLimitError, APIError, APIConnectionError

def call_with_retry(messages, model="gpt-4.1", max_retries=3):
    """指数的バックオフとジッターを用いたリトライ"""
    for attempt in range(max_retries + 1):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError as e:
            if attempt == max_retries:
                raise
            # レスポンスのretry_afterがあれば使用
            wait = getattr(e, 'retry_after', None)
            if wait is None:
                wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"レート制限に達しました。{wait:.1f}秒待機します（試行 {attempt + 1}）")
            time.sleep(wait)
        except APIConnectionError:
            if attempt == max_retries:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
        except APIError as e:
            # クライアントエラー（400, 401, 403）はリトライしない
            if e.status_code and 400 <= e.status_code < 500:
                raise
            if attempt == max_retries:
                raise
            time.sleep((2 ** attempt) + random.uniform(0, 1))

重要なポイント：

指数的バックオフ：1秒、2秒、4秒、8秒
ジッター：0〜1秒のランダムな遅延を追加し、一斉アクセスを防止
retry_afterヘッダーがあれば尊重する
クライアントエラー（不正リクエスト、認証失敗）はリトライしない
最大リトライ回数を設定する

非同期版

import asyncio
import random
from openai import AsyncOpenAI, RateLimitError

async_client = AsyncOpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

async def call_with_retry_async(messages, model="gpt-4.1", max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            return await async_client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError:
            if attempt == max_retries:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(wait)

高度な技術：トークンバケットレートリミッター

高スループットのアプリケーションでは、クライアント側でレート制限を実装してサーバーの制限に達するのを防ぎます：

import time
import asyncio

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # トークン/秒
        self.capacity = capacity  # 最大バーストサイズ
        self.tokens = capacity
        self.last_refill = time.monotonic()

    async def acquire(self, tokens: int = 1):
        while True:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return
            # 十分なトークンが貯まるまで待機
            wait = (tokens - self.tokens) / self.rate
            await asyncio.sleep(wait)

# 1分あたり500リクエスト = 約8.3リクエスト/秒
limiter = TokenBucket(rate=8.0, capacity=20)

async def rate_limited_call(messages, model="gpt-4.1"):
    await limiter.acquire()
    return await async_client.chat.completions.create(
        model=model,
        messages=messages
    )

レート制限時のモデルフォールバック

メインモデルがレート制限にかかった場合は代替モデルにフォールバックします：

FALLBACK_CHAIN = [
    "claude-sonnet-4-6",
    "gpt-4.1",
    "gpt-4.1-mini",
]

async def call_with_fallback(messages):
    for model in FALLBACK_CHAIN:
        try:
            return await async_client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError:
            continue
    raise Exception("すべてのモデルがレート制限にかかっています")

ここでAPIアグリゲーターの強みが発揮されます。300以上のモデルが1つのエンドポイントの背後にあるため、常にフォールバックが利用可能です。

レート制限使用状況の監視

レート制限の消費状況を追跡し、ユーザーに影響が出る前に問題を検知しましょう：

import logging

def log_rate_limits(response):
    headers = response.headers
    remaining = headers.get("x-ratelimit-remaining-requests")
    limit = headers.get("x-ratelimit-limit-requests")
    if remaining and int(remaining) < int(limit) * 0.1:
        logging.warning(
            f"レート制限警告: 残り {remaining}/{limit} リクエスト"
        )

残り容量が10%を下回ったらアラートを設定しましょう。これによりユーザーが429エラーを見る前にスロットリングを実装する時間が確保できます。

まとめ

戦略	使用タイミング
指数的バックオフ	常に（基本）
クライアント側レートリミッター	高スループットアプリ（100 RPM超）
モデルフォールバック	SLA要件のある本番アプリ
積極的な監視	すべての本番環境
バッチAPI	リアルタイムでないワークロード

目標はレート制限を完全に避けることではなく、ユーザーが気づかないように優雅に対処することです。

堅牢なAIアプリケーションを構築しましょう：lemondata.ccはマルチチャネルルーティングを提供し、上流のレート制限を自動で処理します。1つのAPIキーで300以上のモデルを利用可能です。

AI APIのレート制限：仕組みと対処方法