API Gatewayにおける熔断・降級・再試行机制の設計と実装

API Gatewayの安定運用において、熔断（Circuit Breaker）・降級（Fallback）・再試行（Retry）は不可或缺的三大机制です。本稿では、HolySheep AIを例に、production-readyな実装方法を解説します。

結論：どれを選ぶべきか

私自身の実装経験では、HolySheep AIの<50msレイテンシと¥1=$1の為替レート является最適な選択です。公式OpenAI API比85%のコスト削減は、大量リクエストを処理するシステムで年間数百万円の差になります。

価格・性能比較表

熔断机制（Circuit Breaker）の実装

熔断机制は、失敗率が閾値を超えた際にリクエストを即座に遮断し、システム全体の、雪崩れ的な障害を防止します。

import time
import threading
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Optional
import requests

class CircuitState(Enum):
    CLOSED = "closed"      # 正常稼働
    OPEN = "open"          # 熔断中
    HALF_OPEN = "half_open" # 試行状態

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5        # 熔断발동の失敗回数
    success_threshold: int = 3        # 半開→閉店の成功回数
    timeout: float = 30.0            # 熔断持続時間（秒）
    half_open_max_calls: int = 3      # 半開時の最大試行回数

class CircuitBreaker:
    def __init__(self, config: CircuitBreakerConfig):
        self.config = config
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self.half_open_calls = 0
        self._lock = threading.Lock()

    def call(self, func: Callable, *args, **kwargs):
        with self._lock:
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                else:
                    raise CircuitOpenError("Circuit is OPEN - request rejected")

            if self.state == CircuitState.HALF_OPEN:
                if self.half_open_calls >= self.config.half_open_max_calls:
                    raise CircuitOpenError("Circuit is HALF_OPEN - max attempts reached")
                self.half_open_calls += 1

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e

    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return True
        return (time.time() - self.last_failure_time) >= self.config.timeout

    def _on_success(self):
        with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.config.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    self.success_count = 0
            else:
                self.failure_count = 0

    def _on_failure(self):
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
            elif self.failure_count >= self.config.failure_threshold:
                self.state = CircuitState.OPEN

class CircuitOpenError(Exception):
    pass

HolySheep AI API呼び出しの例
def call_holysheep_api(prompt: str, api_key: str):
    url = "https://api.holysheep.ai/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 1000
    }
    response = requests.post(url, json=payload, headers=headers, timeout=30)
    return response.json()

使用例
circuit_breaker = CircuitBreaker(CircuitBreakerConfig(
    failure_threshold=5,
    timeout=30.0
))

try:
    result = circuit_breaker.call(call_holysheep_api, "Hello", "YOUR_HOLYSHEEP_API_KEY")
    print(result)
except CircuitOpenError as e:
    print(f"Fallback triggered: {e}")

降級機制（Fallback）の実装

降級は、主要なAPIが利用不能時に代替応答を返す机制です。キャッシュや軽量モデルへの切り替えを実装します。

import json
import hashlib
from functools import wraps
from typing import Any, Dict, Optional, List

class FallbackManager:
    def __init__(self, cache_size: int = 1000):
        self.cache: Dict[str, Any] = {}
        self.cache_size = cache_size
        self.fallback_chain: List[str] = ["gpt-4.1", "gpt-3.5-turbo", "cached_response"]

    def _get_cache_key(self, prompt: str, model: str) -> str:
        content = f"{model}:{prompt}"
        return hashlib.sha256(content.encode()).hexdigest()

    def get_cached_response(self, prompt: str) -> Optional[
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
DeepSeek API 批量リクエスト処理最適化入門
多模态RAG画像検索とテキストマッチング実践ガイド
2026年4月 AI API 請求書の異常処理と返金手続き完全ガイド