LLM APIのレート制限と再試行への対応

重要な二つの制限：RPMとTPM

ほとんどのLLM APIでは、次の制限のいくつかを組み合わせて適用しています：

• RPM：毎分のリクエスト数
• TPM：毎分のトークン数
• RPD / TPD：一日あたりのリクエスト数またはトークン数
• 同時リクエスト数の制限：処理中の呼び出し数の上限
• モデル固有の制限：小規模モデルでは上限が高く、最先端モデルでは上限が低い

重要な点： RPMとTPMでは、制限に達する状況が異なります.

小さなプロンプトを大量に送信すると、先にRPMの制限に達します。大きなプロンプトを少数送信すると、先にTPMの制限に達します。

例：

Limit:
- 600 requests/minute
- 300,000 tokens/minute

Workload A:
- 1,000 short classification calls/minute
- 200 tokens each
- 200,000 tokens/minute total

Result:
- TPM is fine
- RPM fails

Workload B:
- 100 long document analysis calls/minute
- 5,000 tokens each
- 500,000 tokens/minute total

Result:
- RPM is fine
- TPM fails

よくある落とし穴は、開発者がリクエスト数を制限しても、トークン数は制限しないことです。プロンプトが短いステージング環境では、それで問題ありません。しかし本番環境でユーザーがPDF、チャット履歴、ログ、契約書などを貼り付けると、トークン予算は瞬く間に尽きます。

実際には、私はLLM呼び出しのたびに、次の二つのリソースを予約するものとして扱います：

cost = {
  requests: 1,
  tokens: estimated_prompt_tokens + max_output_tokens
}

モデルが返す出力トークン数が max_output_tokensより少なくても、上限分を予約しておけば、容量を超えた割り当てを防げます。

TPMを無視すると実際に何が起こるのか

次の条件を想定します：

TPM limit: 120,000
Average request: 4,000 input tokens + 1,000 max output tokens
Reserved cost: 5,000 tokens

実際の最大スループットは「できるだけ多くのワーカーを動かすこと」では決まりません。次のようになります：

120,000 / 5,000 = 24 requests per minute

これは平均すると2.5秒ごとに一件のリクエストです。

100のワーカーを動かした場合、プロバイダーがバーストを許容していれば、最初のバッチはすべて正常に開始できるかもしれません。しかしその後、アカウントは厳格な制限に突き当たります。以降のすべてのリクエストで 429が返され、リトライが積み重なります。

ワーカー数を減らすだけでは解決できません。トークンの予約を把握できるスケジューラーが必要です。

最小構成のトークン対応レートリミッター

単一プロセスであれば、通常はトークンバケットで十分です。時間の経過に応じて容量を補充し、各リクエストがリクエスト枠とトークン枠の両方を確保するようにします。

以下は簡略化したPythonの例です：

import asyncio
import time
from dataclasses import dataclass

@dataclass
class Bucket:
    capacity: float
    refill_per_second: float
    tokens: float
    updated_at: float

    def refill(self):
        now = time.monotonic()
        elapsed = now - self.updated_at
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_second)
        self.updated_at = now

    async def acquire(self, amount: float):
        while True:
            self.refill()
            if self.tokens >= amount:
                self.tokens -= amount
                return

            missing = amount - self.tokens
            sleep_for = missing / self.refill_per_second
            await asyncio.sleep(min(sleep_for, 1.0))


class LLMRateLimiter:
    def __init__(self, rpm: int, tpm: int):
        now = time.monotonic()
        self.request_bucket = Bucket(
            capacity=rpm,
            refill_per_second=rpm / 60,
            tokens=rpm,
            updated_at=now,
        )
        self.token_bucket = Bucket(
            capacity=tpm,
            refill_per_second=tpm / 60,
            tokens=tpm,
            updated_at=now,
        )
        self.lock = asyncio.Lock()

    async def acquire(self, estimated_tokens: int):
        async with self.lock:
            await self.request_bucket.acquire(1)
            await self.token_bucket.acquire(estimated_tokens)

使用例：

estimated_tokens = prompt_tokens + max_output_tokens
await limiter.acquire(estimated_tokens)

response = await client.chat.completions.create(
    model="gpt-5.5",
    messages=messages,
    max_tokens=max_output_tokens,
)

これは意図的に単純化しています。分散システムでは、Redis、Postgresのアドバイザリロック、または中央でディスパッチを行うキューを使用してください。重要な原則は同じです。APIを呼び出す前に、リクエスト数とトークン予算の両方に基づいて制限をかけます。

リトライ：バックオフは必要ですが、それだけでは不十分

リトライは、次のような場合に有効です：

• 一時的な 429 レスポンス
• ネットワークのタイムアウト
• 一時的な 500, 502, 503、または 504 エラー
• プロバイダー側の過負荷
• 接続のリセット

リトライは、次のような場合に危険です：

• 検証エラー
• コンテキスト長のエラー
• 認証の失敗
• 安全ポリシーによる拒否
• 確実に失敗する不適切なプロンプト
• 冪等性のないツール呼び出し

リトライポリシーでは、まずエラーを分類する必要があります。

基本的なバックオフの形：

import random

def retry_delay(attempt: int, base=0.5, cap=30.0):
    exponential = min(cap, base * (2 ** attempt))
    return random.uniform(0, exponential)

この random.uniform(0, exponential) がジッターです。省略しないでください。

ジッターなし：

100 workers fail
100 workers sleep 1s
100 workers retry
100 workers fail
100 workers sleep 2s
100 workers retry

ジッターあり：

100 workers fail
workers retry across a spread of 0-1s
then 0-2s
then 0-4s

後者のパターンなら、プロバイダーと自分のキューに回復する余裕を与えられます。

再試行用のヘッダーがある場合は従う

多くのAPIには、次のようなヒントが含まれています。

Retry-After: 12

または、プロバイダー固有のリセット用ヘッダーが含まれています。APIが具体的な待機時間を示している場合は、それを待機時間の下限として使います。

delay = max(header_retry_after_seconds, retry_delay(attempt))
await asyncio.sleep(delay)

実際には、再試行にかける合計時間にも上限を設けています。ユーザー向けのチャットリクエストで、ユーザーがスピナーを見つめている間に、何も知らせずに二分間も再試行を続けるべきではありません。バックグラウンドのバッチジョブなら、もっと長く待てます。

ポリシーの例：

{
  "interactive_chat": {
    "max_attempts": 3,
    "max_total_retry_seconds": 8
  },
  "batch_summarization": {
    "max_attempts": 8,
    "max_total_retry_seconds": 300
  }
}

ワーカーの一斉実行よりキューイング

クォータを超える量の処理がある場合、すべてのワーカーがLLM APIを直接呼び出すべきではありません。手前にキューを配置してください。

実用的なアーキテクチャ：

App/API
  -> Job queue
      -> Rate-aware dispatcher
          -> Provider client
              -> Result store

ディスパッチャーは次の項目を管理します。

• RPMの使用量管理
• TPMの使用量管理
• モデルの選択
• 再試行ポリシー
• プロバイダーのフォールバック
• 同時実行数の制限
• キューの優先度

これにより、過負荷を可視化できます。何千ものタスクが次のエラーで失敗する代わりに、 429キュー内の待機件数が増え、推定待機時間が長くなる様子を確認できます。

Redisを基盤とするシステムでは、「実行可能」なジョブと「再試行待ち」のジョブを分けるようにしています。

llm:ready
llm:delayed
llm:dead_letter

失敗したジョブは、すぐには次の場所へ戻りません。 ready代わりに、再試行予定のタイムスタンプが設定されます。小さなスケジューラーが、予定時刻に達したジョブを実行可能なキューへ戻します。

{
  "job_id": "sum_7f13",
  "model": "claude-sonnet-4.6",
  "estimated_tokens": 6200,
  "attempt": 2,
  "not_before": "2026-02-03T09:14:22Z"
}

この not_before フィールドが、再試行の集中を防ぎます。

同時実行制御とレート制限は異なる

同時実行数の制限は、処理中の作業量に上限を設けます。レート制限は、一定の時間枠内の作業量に上限を設けます。

両方が必要です。

各呼び出しに20秒かかり、RPMが60だとします。RPMだけを制限するリミッターは、毎秒一件のリクエストを許可します。20秒後には、約20件の呼び出しが同時に実行されています。レイテンシが90秒に急増すると、同じRPMでも同時実行される呼び出しは90件になります。これは、プロバイダーやローカルリソースの上限を超える可能性があります。

同時実行の制御にはセマフォを使います。

provider_concurrency = asyncio.Semaphore(20)

async def call_model(job):
    async with provider_concurrency:
        await limiter.acquire(job.estimated_tokens)
        return await send_request(job)

よくある落とし穴として、セマフォを早く取得しすぎると、ワーカーが貴重な同時実行枠を占有したまま、レート制限のトークンを待って何もせずに過ごすことになります。通常は、まずレート制限上の実行枠を確保し、ネットワーク呼び出しの直前に同時実行枠を取得します。リミッターによる待機が長くなる可能性がある場合は、その間にワーカーのリソースを占有しないようにしてください。

段階的な機能縮退：重要な処理に品質を割り当てる

レート制限が厳しくなっても、失敗させる以外の選択肢があります。

たとえば、カスタマーサポート用アシスタントなら、次のように段階的に機能を縮退できます。

具体例：

• チケットのタグ付けは、最先端モデルを使う代わりに、Haiku 4.5、Qwen、または小型のLlama派生モデルへ振り分けます。
• 通常の推論にはSonnet 4.6を使い、Opus 4.8またはGPT-5.5はエスカレーション時の処理用に確保します。
• 長いコンテキストの分析をFable 5やGemini 3へ移すのは、プロンプトがそのコンテキストウィンドウを本当に必要とする場合だけにします。
• バッチジョブでは、対話型トラフィックに影響が出る前に、優先度の低いテナントの処理を一時停止します。

その代償は複雑さです。すべてのフォールバック経路に、テスト、可観測性、製品としての受け入れ確認が必要です。分野によっては、回答が遅れるよりも、質の低い回答のほうが大きな害をもたらすことがあります。

複数キーと複数プロバイダーによるフォールバック

複数キーや複数プロバイダーへのルーティングは耐障害性を高められますが、実装を誤りやすい仕組みでもあります。

これには、別々の三つのパターンがあります。

制限をこっそり回避するために複数のキーを使わないでください。テナント、環境、製品、請求単位など、正当な分離のために使ってください。

明確なルーティングポリシーは、たとえば次のようになります。

{
  "task": "support_ticket_summary",
  "primary": {
    "provider": "anthropic",
    "model": "claude-sonnet-4.6"
  },
  "fallbacks": [
    {
      "provider": "openai",
      "model": "gpt-5.5",
      "when": ["provider_5xx", "rate_limit_after_retry_budget"]
    },
    {
      "provider": "google",
      "model": "gemini-3",
      "when": ["context_length_needed", "primary_unavailable"]
    }
  ],
  "degrade_to": {
    "provider": "open_model_cluster",
    "model": "qwen",
    "when": ["batch_low_priority"]
  }
}

複数モデルへのアクセスにアグリゲーターやAI Prime Techのようなプラットフォームを使う場合も、同じエンジニアリング上のルールが適用されます。冪等性、予算、フォールバック条件、最終的な課金の扱いを明示的に追跡してください。

フォールバック時の二重課金を避ける

フォールバックで難しいのは、最初のリクエストが完了したかどうかを把握することです。

次の流れを考えてみてください。

Client sends request to Provider A
Provider A processes it successfully
Network connection drops before client receives response
Client retries with Provider B
Both providers bill
Two completions may exist

プロバイダーが冪等性キーやリクエスト照会をサポートしていない限り、クライアント側だけでこの問題を完全に解決することはできません。ただし、被害を減らすことはできます。

利用可能な場合は冪等性キーを使う

論理的な操作ごとに、変わらないキーを生成します。

import hashlib
import json

def idempotency_key(task_name, tenant_id, input_payload):
    canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(canonical.encode()).hexdigest()
    return f"{tenant_id}:{task_name}:{digest}"

送信前に操作を保存します。

{
  "operation_id": "acme:summarize:9d2a...",
  "status": "in_flight",
  "provider": "anthropic",
  "model": "claude-sonnet-4.6",
  "created_at": "2026-02-03T09:12:01Z"
}

その後、完了時に更新します。

{
  "operation_id": "acme:summarize:9d2a...",
  "status": "completed",
  "provider_request_id": "req_abc123",
  "usage": {
    "input_tokens": 4180,
    "output_tokens": 612
  }
}

別の送信先で再試行する前に、操作テーブルを確認してください。別のワーカーがすでに完了している場合は、保存済みの結果を返します。

タイムアウトの種類を区別する

タイムアウトはすべて同じではありません。

• 接続タイムアウト：リクエストはプロバイダーに届いていない可能性が高い
• ヘッダー受信前の読み取りタイムアウト：不明
• トークンのストリーミング中のタイムアウト：プロバイダーが処理したことはほぼ確実
• クライアントによるキャンセル：キャンセルが伝播したかどうかによる

不確かな場合には、慎重な再試行ポリシーを採用するようにしています。

interactive:
- retry connect timeout
- do not cross-provider retry after partial stream
- show recoverable error if status unknown

batch:
- retry after reconciliation delay
- prefer same provider with idempotency key
- fallback only after retry budget expires

ストリーミングは慎重に行う

ストリーミングは体感レイテンシを改善しますが、再試行を複雑にします。出力の一部をユーザーに送信した後で別のモデルを使って再試行すると、異なる続きが生成される可能性があります。

ストリーミング形式のチャットでは、通常「最初のトークンを受信した時点」を、ユーザーに意識させずに再試行できる限界としています。それ以降の失敗は、ユーザーに見える失敗として扱います。

"The response was interrupted. Retry?"

これはユーザーに気づかれない復旧ほど洗練されてはいませんが、異なるモデルの互換性のない出力をつなぎ合わせずに済みます。

可観測性：実際に役立つ指標

成功率だけでは不十分です。

少なくとも次を追跡しましょう：

llm_requests_total{provider,model,status}
llm_tokens_reserved_total{provider,model}
llm_tokens_used_total{provider,model}
llm_retries_total{provider,model,error_type}
llm_queue_depth{priority,task_type}
llm_queue_wait_seconds{priority,task_type}
llm_rate_limit_hits_total{provider,model,limit_type}
llm_fallbacks_total{from_provider,to_provider,reason}
llm_duplicate_suppressed_total{task_type}

実際のインシデントで最も役立つダッシュボードのパネルは、レイテンシではないことがよくあります。それは次のものです。

queue_wait_seconds p50 / p95
rate_limit_hits by provider and model
tokens_reserved vs tokens_used
fallback rate by reason

もし tokens_reserved が一貫して次の値を大幅に上回るなら tokens_used、予約量を保守的に見積もりすぎています。実際の使用量に近すぎるか、それを下回る場合は、処理を過剰に受け入れており、予期せずTPM上限に達します。

実用的なリトライラッパー

リトライ対象となる失敗だけを再試行する、簡潔なパターンを紹介します。

import asyncio
import random
import time

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

class LLMError(Exception):
    def __init__(self, status, message="", retry_after=None):
        self.status = status
        self.retry_after = retry_after
        super().__init__(message)

def jittered_backoff(attempt, base=0.5, cap=20):
    return random.uniform(0, min(cap, base * (2 ** attempt)))

async def call_with_retries(fn, max_attempts=4, max_elapsed=20):
    started = time.monotonic()

    for attempt in range(max_attempts):
        try:
            return await fn()
        except LLMError as error:
            elapsed = time.monotonic() - started
            if error.status not in RETRYABLE_STATUS:
                raise
            if attempt == max_attempts - 1 or elapsed >= max_elapsed:
                raise

            delay = jittered_backoff(attempt)
            if error.retry_after is not None:
                delay = max(delay, error.retry_after)

            remaining = max_elapsed - elapsed
            await asyncio.sleep(min(delay, remaining))

このラッパーは意図的に単純にしています。高度な処理は、分類、キューイング、冪等性、ルーティングに持たせるべきであり、いつまでも試行を続ける魔法のようなリトライループに持たせるべきではありません。

実践上のポイント

• レート制限を次の両方でモデル化します： RPMとTPM。トークン数を考慮したスロットリングにより、大きなプロンプトによる予想外の問題を防げます。
• 使用するモデルは、 ジッター付き指数バックオフを使用し、プロバイダーがリトライ用ヘッダーを提供している場合は、その指示に従います。
• 次の仕組みを配置します： キューとディスパッチャー 。バッチ処理や大量のワークロードでは、アプリとLLMプロバイダーの間に配置します。
• 制御手段 同時実行数を個別に管理します 。レート制限とは別に管理しないと、レイテンシの急増によって処理中の呼び出しが多くなりすぎる可能性があります。
• デザイン 段階的な機能縮退 を意図的に設計します。小型モデル、短いコンテキスト、キューに入れたバックグラウンド処理、キャッシュ済みの応答などを活用します。
• 次のものを 複数プロバイダーへのフォールバック は信頼性を高める機能として扱い、重複リクエストを無差別に送るための口実にしないでください。
• 使用するモデルは、 冪等性キーと操作記録 を使用して、二重課金を減らし、重複した生成結果を抑制します。
• トークンがすでに出力された後でストリーミング応答が失敗した場合は、その状況をユーザーに正直に伝えてください。
• 次の指標を中心にダッシュボードを構築します： キューの待機時間、トークン予約量、レート制限への到達、リトライ、フォールバックの理由.
• リトライロジックはシンプルに保ちましょう。本番環境で本当に必要な作業は、予算管理、スケジューリング、そしてリトライすべきでないタイミングの判断です。