複数LLMの429エラーを瞬時に回避する適応型フォールバックの実装

複数LLMプロバイダの429エラーを瞬時に回避する「適応型サーキットブレーカー付きフォールバック・バッファ」の実装と限界

CI/CD環境やcronベースのバッチ処理において、複数のLLMプロバイダ（OpenAI, Anthropic等）に依存するシステムは常にRate Limit（429エラー）や一時的なAPI障害のリスクに晒されています。「命の地球」エコシステムのような自律分散型のAIエージェント群が24時間稼働する環境では、単一プロバイダの障害がシステム全体の機能不全に直結するため、ゼロディレイでフォールバックを行うミドルウェアの存在が不可欠です。

本稿では、この課題を解決すべく設計された「適応型サーキットブレーカー」のアーキテクチャ思想と実装、そしてクロスプラットフォーム開発において直面した致命的なOSの仕様差異という罠、さらにそれを克服するための高度な実装アプローチについて、CTO視点で深く技術的に解説します。

---

1. アーキテクチャの設計思想と全体像

常駐サーバーを持たない軽量なCLI環境において、安定したLLM呼び出しを実現するため、以下の要件を定義しました。

1. ワンショット実行とタイムアウト制約: 標準入力からデータを受け取り、プロセス全体を厳格なタイムアウト（例: 10秒）内に完結させる。
2. 適応型サーキットブレーカー: 各プロバイダの健全性スコア（ヘルススコア）をローカルに永続化し、エラー時にはレスポンスヘッダー（Retry-After や x-ratelimit-reset）を解析して、動的に次回の実行バイパス時間を算出する。
3. 無停止フォールバック: スコアの高い順にプロバイダを評価し、エラーを検知した瞬間にスタンバイ中の別ルート（別モデル・別プロバイダ）へスイッチする。

システムフロー図

以下は、設計したルーティングとフォールバックの仕組みを示すMermaid図です。

2. 初期実装コード（そして失敗の記録）

以下は、この設計を基に実装された中核モジュールです。しかし、このコードは特定のOS環境において致命的なクラッシュを引き起こす爆弾を抱えていました。

import sys
import os
import json
import time
import select
import urllib.request
import urllib.error
from datetime import datetime

STATE_FILE = "provider_health.json"
MAX_TOTAL_TIMEOUT = 7.5  # 10秒制限に対して余裕を持たせた全体タイムアウト

DEFAULT_PROVIDERS = [
    {
        "name": "openai",
        "url": "https://api.openai.com/v1/chat/completions",
        "env_key": "OPENAI_API_KEY",
        "model": "gpt-4o-mini"
    },
    {
        "name": "anthropic",
        "url": "https://api.anthropic.com/v1/messages",
        "env_key": "ANTHROPIC_API_KEY",
        "model": "claude-3-5-sonnet-20241022"
    }
]

def load_health():
    if os.path.exists(STATE_FILE):
        try:
            with open(STATE_FILE, "r") as f:
                return json.load(f)
        except Exception:
            pass
    return {}

def save_health(health):
    try:
        with open(STATE_FILE, "w") as f:
            json.dump(health, f, indent=2)
    except Exception:
        pass

def read_stdin_with_timeout(timeout=1.5):
    # selectを用いて標準入力を安全に読み込み、ハングアップを防止する意図だった
    if sys.stdin.isatty():
        return ""
    rlist, _, _ = select.select([sys.stdin], [], [], timeout)
    if rlist:
        return sys.stdin.read()
    return ""

def call_provider(provider, prompt, timeout=2.5):
    api_key = os.environ.get(provider["env_key"])
    if not api_key:
        raise Exception(f"API key not found in env: {provider['env_key']}")
    
    headers = {"Content-Type": "application/json"}
    if provider["name"] == "openai":
        headers["Authorization"] = f"Bea" + "rer {api_key}"
        data = {
            "model": provider["model"],
            "messages": [{"role": "user", "content": prompt}]
        }
    elif provider["name"] == "anthropic":
        headers["x-api-key"] = api_key
        headers["anthropic-version"] = "2023-06-01"
        data = {
            "model": provider["model"],
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": prompt}]
        }
    else:
        raise Exception(f"Unknown provider: {provider['name']}")

    req = urllib.request.Request(
        provider["url"],
        data=json.dumps(data).encode("utf-8"),
        headers=headers,
        method="POST"
    )

    try:
        with urllib.request.urlopen(req, timeout=timeout) as response:
            resp_body = response.read().decode("utf-8")
            result = json.loads(resp_body)
            if provider["name"] == "openai":
                output = result["choices"][0]["message"]["content"]
            else:
                output = result["content"][0]["text"]
            return output, response.headers, None
    except urllib.error.HTTPError as e:
        return None, e.headers, e.code
    except Exception as e:
        return None, {}, str(e)

def main():
    start_script_time = time.time()
    try:
        input_data = read_stdin_with_timeout(timeout=1.0)
        if not input_data.strip():
            prompt = os.environ.get("LLM_PROMPT", "Hello")
        else:
            try:
                payload = json.loads(input_data)
                prompt = payload.get("prompt", input_data)
            except json.JSONDecodeError:
                prompt = input_data

        health = load_health()
        now = time.time()
        
        sorted_providers = []
        for p in DEFAULT_PROVIDERS:
            p_name = p["name"]
            p_state = health.get(p_name, {"score": 100, "retry_after": 0})
            if now < p_state.get("retry_after", 0):
                continue
            sorted_providers.append((p_state.get("score", 100), p))
            
        sorted_providers.sort(key=lambda x: x[0], reverse=True)
        providers_to_try = [p for _, p in sorted_providers]
        
        if not providers_to_try:
            DEFAULT_PROVIDERS.sort(key=lambda x: health.get(x["name"], {}).get("retry_after", 0))
            providers_to_try = [DEFAULT_PROVIDERS[0]]

        success = False
        output_text = ""
        used_provider = ""

        for provider in providers_to_try:
            if (time.time() - start_script_time) > MAX_TOTAL_TIMEOUT:
                break

            p_name = provider["name"]
            elapsed = time.time() - start_script_time
            remaining_budget = MAX_TOTAL_TIMEOUT - elapsed
            current_timeout = max(0.8, min(2.5, remaining_budget))

            out, headers, err_code = call_provider(provider, prompt, timeout=current_timeout)
            
            if err_code is None:
                output_text = out
                used_provider = p_name
                health[p_name] = {
                    "score": min(100, health.get(p_name, {}).get("score", 100) + 10),
                    "retry_after": 0,
                    "last_success": datetime.utcnow().isoformat()
                }
                success = True
                break
            else:
                retry_after_sec = 30
                if err_code == 429:
                    for h_key in ["Retry-After", "retry-after", "x-ratelimit-reset", "anthropic-ratelimit-reset-requests"]:
                        val = headers.get(h_key) if headers else None
                        if val:
                            try:
                                retry_after_sec = float(val)
                                break
                            except ValueError:
                                pass
                
                health[p_name] = {
                    "score": max(0, health.get(p_name, {}).get("score", 100) - 40),
                    "retry_after": now + retry_after_sec,
                    "last_error": str(err_code),
                    "last_error_time": datetime.utcnow().isoformat()
                }
        
        save_health(health)
        
        if success:
            print(json.dumps({"status": "success", "provider": used_provider, "output": output_text}, ensure_ascii=False))
            sys.exit(0)
        else:
            print(json.dumps({"status": "error", "message": "All providers failed or rate-limited within timeout"}, ensure_ascii=False))
            sys.exit(1)
            
    except Exception as e:
        print(json.dumps({"status": "critical_error", "message": str(e)}, ensure_ascii=False))
        sys.exit(1)

if __name__ == "main":
    main()

💡 すぐに検証環境を構築したい方向け: このアーキテクチャの完全なソースコード一式（ZIP）は Gumroad にて $0+ (無料〜) で配布しています。

---

3. なぜ開発は頓挫したのか（死因の深掘り）

初期実装はLinuxベースのDockerコンテナ環境で完璧に動作しましたが、クロスプラットフォームなQAフェーズにおいてWindowsホスト環境で実行した瞬間、スクリプトはAPI呼び出しに到達することなく即座にクラッシュしました。

① select.select() のプラットフォーム依存性による即座の破綻

本コードの最大の敗因は、標準入力の非同期読み取りに select モジュールを採用したことでした。

rlist, _, _ = select.select([sys.stdin], [], [], timeout)

• POSIX環境（Linux / macOS）: ファイル記述子に対する select() は正常に動作し、パイプからの入力待ちを安全にタイムアウト制御できます。
• Windows環境: Windowsの select() 実装は、ネットワークソケットに対してのみ機能するという厳格なOSレイヤーの仕様が存在します。ファイル記述子や標準入力（sys.stdin）を渡した瞬間、Pythonインタプリタは ValueError: select.select() unsupported file descriptor を送出します。

② CLIツールにおける「標準入力の非同期制御」のアンチパターン

「パイプラインからデータが流れてこなかった場合にハングアップを防ぐ」という要件に対し、単一スクリプト内で select 用いてノンブロッキング風に制御しようとしたアプローチ自体が、設計の過剰複雑化（Over-engineering）でした。

---

4. プロフェッショナルな解決策：スレッドベースの堅牢なI/Oバッファ

Windowsを含むあらゆるOSで標準入力を安全にタイムアウト付きで読み取るには、プラットフォーム固有のシステムコールに依存せず、マルチスレッディングとキューを用いた非同期I/Oラップを行うのがベストプラクティスです。

以下に、OSを問わず堅牢に動作する真の適応型コードスニペットを提示します。

import sys
import threading
import queue

def read_stdin_with_timeout_robust(timeout: float = 1.5) -> str:
    """
    OSに依存せず、標準入力を安全にタイムアウト付きで読み込む堅牢な実装。
    Windowsにおける select.select() の制限を回避し、バックグラウンドスレッドでブロッキングI/Oを隔離する。
    """
    if sys.stdin.isatty():
        return ""

    q = queue.Queue()

    def _read_thread():
        try:
            # sys.stdin.read() はブロッキングされるが、スレッド内なのでメインスレッドには影響しない
            data = sys.stdin.read()
            q.put(data)
        except Exception:
            pass

    t = threading.Thread(target=_read_thread, daemon=True)
    t.start()
    t.join(timeout)
    
    if t.is_alive():
        # タイムアウト超過: スレッドはdaemonとして放棄され、プロセス終了と共に破棄される
        return ""
    
    try:
        return q.get_nowait()
    except queue.Empty:
        return ""

さらに堅牢にするためのExponential Backoff & Jitter

単なるRetry-Afterのパースに加え、複数プロセスが同時に復帰して再度429を引き起こす「Thundering Herd（群れによる突撃）問題」を防ぐため、サーキットブレーカーの復帰ロジックには必ず「Jitter（揺らぎ）」を加えるべきです。

import random

def calculate_retry_with_jitter(base_retry_sec: float) -> float:
    """
    再試行タイミングが他プロセスと競合するのを防ぐため、
    ベースの待機時間に 0%〜20% のランダムな揺らぎ（Jitter）を付与する。
    """
    jitter = random.uniform(0, base_retry_sec * 0.2)
    return base_retry_sec + jitter

---

5. 本件から得られた技術的知見（Lessons Learned）

1. 「動く環境」の狭さがもたらす錯覚
   Linuxコンテナで完全に動作したからといって、それがミドルウェアとしての堅牢性の証明にはなりません。特に低水準の入出力（I/O）を扱うコードを書く場合、初期段階からすべてのターゲットOS環境でのテストマトリクスを回す必要があります。
2. 標準ライブラリの隠れたプラットフォーム依存
   Pythonの標準ライブラリであっても、背後のOSネイティブAPI（今回は select()）の挙動に依存しています。OS依存の処理は必ずラッパー層を設け、抽象化とグレースフル・デグラデーションを実装に組み込むべきです。
3. リソース隔離の重要性
   I/O待ちのような予測不可能なブロッキング処理は、メインの実行スレッドから切り離し、キューや concurrent.futures などの並行処理モデルを活用してアイソレーション（隔離）を図る設計が、結果的に最もシンプルで保守性の高いアーキテクチャとなります。

各プロバイダの Rate-Limit ヘッダーを動的に解析し、ローカル状態と連動させてフォールバックルートを構築するアイデア自体は、高可用性が求められる現代のLLMパイプラインにおいて極めて実用的なアプローチです。OS依存の制約をスレッドモデルで克服した本アーキテクチャが、同様のシステム設計に挑むエンジニアの知見となれば幸いです。