Circuit Breaker を活用した AI サービスの耐障害性設計

AI API を活用したシステムは、外部サービスへの依存度高いため、一時的な障害や高負荷時にシステムが完全に停止するリスクがあります。本稿では、Circuit Breaker パターンを用いた AI サービスの耐障害性設計について、HolySheep AI を具体的な基盤とした実装例を交えて解説します。

なぜ AI サービスに Circuit Breaker が必要か

AI API を呼び出す際、以下のような障害に遭遇する機会は思っている以上に多いです：

• API の一時的な応答遅延（モデル負荷による数百秒のハングアップ）
• レートリミット超過（流量制御による 429 エラー）
• タイムアウト連鎖（1 つの遅い応答がシステム全体を巻き込む）

これらの問題を放置すると、ユーザーのリクエストが永遠に待たされ、リソースが枯渇，最终的にシステム全体が停止します。Circuit Breaker はこの問題を解決する効果的なパターンです。

具体的なユースケース

ケース 1：EC サイトの AI カスタマーサービス急増

私の实战経験では某 EC 사에서、大規模セール時に AI 客服への問い合わせが平时的 50 倍に急増这种事象が発生しました。Circuit Breaker がない状態では、AI モデルの応答遅延時に注文処理システム全体が影響を受けました。阀値设定を適切に调整することで、峰值时也能维持核心业务的可用性。

ケース 2：企業 RAG システムの構築

企业内部のドキュメント検索に RAG を活用する場合、夜間バッチ処理で大量のクエリが発生します。この際、HolySheep AI の <50ms レイテンシを活かしながらも、障害時にフォールバックする設計が重要です。

ケース 3：個人開発者の AI 統合プロジェクト

個人開発者在构建 AI 功能时，预算有限但需要保证服务稳定性。使用 HolySheep AI 的优势在于费用节省（レート¥1=$1、对比官方节省 85%），即使在流量激增时也能通过 Circuit Breaker 有效控制成本。

Circuit Breaker の実装

Python による実装例

import time
import asyncio
from enum import Enum
from typing import Callable, Any
from dataclasses import dataclass, field
import httpx

class CircuitState(Enum):
    CLOSED = "closed"      # 正常動作
    OPEN = "open"          # 遮断中
    HALF_OPEN = "half_open"  # 回復確認中

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5      # OPEN に切り替える連続エラー回数
    success_threshold: int = 2     # CLOSED に復帰する成功回数
    timeout: float = 30.0          # OPEN → HALF_OPEN への移行秒数
    half_open_max_calls: int = 3   # HALF_OPEN 時の最大試行回数

@dataclass
class CircuitBreaker:
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: float = field(default_factory=time.time)
    half_open_calls: int = 0
    config: CircuitBreakerConfig = field(default_factory=CircuitBreakerConfig)

    async def call(
        self,
        func: Callable,
        *args,
        **kwargs
    ) -> Any:
        """Circuit Breaker 経由で関数を実行"""
        
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.config.timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                print("🔄 Circuit Breaker: HALF_OPEN に移行")
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit Breaker is OPEN. Retry after {self.config.timeout}s"
                )
        
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.config.half_open_max_calls:
                raise CircuitBreakerOpenError(
                    "Circuit Breaker is in HALF_OPEN: max calls reached"
                )
            self.half_open_calls += 1
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0
                print("✅ Circuit Breaker: CLOSED に復帰")
        elif self.state == CircuitState.CLOSED:
            self.success_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            print("❌ Circuit Breaker: OPEN に切り替え（HALF_OPEN 失敗）")
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN
            print("❌ Circuit Breaker: OPEN に切り替え（閾値到達）")

class CircuitBreakerOpenError(Exception):
    """Circuit Breaker が OPEN 状态のときに呼び出しを試みた場合のエラー"""
    pass

HolySheep AI API 呼び出しの例
async def call_holysheep_ai(
    circuit_breaker: CircuitBreaker,
    user_message: str
) -> str:
    """HolySheep AI API を Circuit Breaker 経由で呼び出す"""
    
    async def _make_request():
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": user_message}],
                    "max_tokens": 1000
                }
            )
            response.raise_for_status()
            return response.json()["choices"][0]["message"]["content"]
    
    return await circuit_breaker.call(_make_request)

使用例
async def main():
    cb = CircuitBreaker(
        config=CircuitBreakerConfig(
            failure_threshold=3,
            success_threshold=2,
            timeout=30.0
        )
    )
    
    for i in range(10):
        try:
            response = await call_holysheep_ai(cb, f"質問 {i}")
            print(f"Response {i}: {response[:50]}...")
        except CircuitBreakerOpenError as e:
            print(f"⚠️ Circuit Breaker OPEN: {e}")
            await asyncio.sleep(5)
        except Exception as e:
            print(f"❌ Error: {e}")

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

JavaScript/Node.js による実装例

/**
 * HolySheep AI 向け Circuit Breaker 実装
 * Node.js + TypeScript
 */

enum CircuitState {
  CLOSED = 'CLOSED',
  OPEN = 'OPEN',
  HALF_OPEN = 'HALF_OPEN'
}

interface CircuitBreakerOptions {
  failureThreshold?: number;    // OPEN遷移の連続エラー回数
  successThreshold?: number;    // CLOSED復帰の成功回数
  timeout?: number;            // ms (OPEN→HALF_OPENへの移行時間)
  halfOpenMaxCalls?: number;    // HALF_OPEN時の最大試行回数
}

class HolySheepCircuitBreaker {
  private state: CircuitState = CircuitState.CLOSED;
  private failureCount: number = 0;
  private successCount: number = 0;
  private lastFailureTime: number = Date.now();
  private halfOpenCalls: number = 0;

  private readonly failureThreshold: number;
  private readonly successThreshold: number;
  private readonly timeout: number;
  private readonly halfOpenMaxCalls: number;

  constructor(options: CircuitBreakerOptions = {}) {
    this.failureThreshold = options.failureThreshold ?? 5;
    this.successThreshold = options.successThreshold ?? 2;
    this.timeout = options.timeout ?? 30000;
    this.halfOpenMaxCalls = options.halfOpenMaxCalls ?? 3;
  }

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    // OPEN状態の確認
    if (this.state === CircuitState.OPEN) {
      if (Date.now() - this.lastFailureTime >= this.timeout) {
        this.transitionTo(CircuitState.HALF_OPEN);
        console.log('🔄 Circuit Breaker: HALF_OPEN に移行');
      } else {
        throw new CircuitBreakerOpenError(
          Circuit Breaker is OPEN. Retry after ${this.timeout / 1000}s
        );
      }
    }

    // HALF_OPEN状態での呼び出し制限
    if (this.state === CircuitState.HALF_OPEN) {
      if (this.halfOpenCalls >= this.halfOpenMaxCalls) {
        throw new CircuitBreakerOpenError('HALF_OPEN: max calls reached');
      }
      this.halfOpenCalls++;
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess(): void {
    this.failureCount = 0;

    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.successThreshold) {
        this.transitionTo(CircuitState.CLOSED);
        console.log('✅ Circuit Breaker: CLOSED に復帰');
      }
    } else {
      this.successCount = 0;
    }
  }

  private onFailure(): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.state === CircuitState.HALF_OPEN) {
      this.transitionTo(CircuitState.OPEN);
      console.log('❌ Circuit Breaker: OPEN に切り替え（HALF_OPEN 失敗）');
    } else if (this.failureCount >= this.failureThreshold) {
      this.transitionTo(CircuitState.OPEN);
      console.log('❌ Circuit Breaker: OPEN に切り替え（閾値到達）');
    }
  }

  private transitionTo(newState: CircuitState): void {
    this.state = newState;
    this.halfOpenCalls = 0;
    this.successCount = 0;
  }

  getState(): CircuitState {
    return this.state;
  }
}

class CircuitBreakerOpenError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'CircuitBreakerOpenError';
  }
}

// HolySheep AI API 呼び出し関数
async function callHolySheepAI(
  breaker: HolySheepCircuitBreaker,
  message: string
): Promise<string> {
  return breaker.execute(async () => {
    const response = await fetch('https://api.holysheep.ai/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': 'Bearer YOUR_HOLYSHEEP_API_KEY',
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: 'deepseek-v3.2',
        messages: [{ role: 'user', content: message }],
        max_tokens: 500
      })
    });

    if (!response.ok) {
      const error = await response.json();
      throw new Error(API Error: ${response.status} - ${JSON.stringify(error)});
    }

    const data = await response.json();
    return data.choices[0].message.content;
  });
}

// 使用例
async function main() {
  const breaker = new HolySheepCircuitBreaker({
    failureThreshold: 3,
    successThreshold: 2,
    timeout: 30000
  });

  const messages = [
    'こんにちは',
    'おはようございます',
    '今日の天気を教えてください'
  ];

  for (let i = 0; i < messages.length; i++) {
    try {
      const response = await callHolySheepAI(breaker, messages[i]);
      console.log(✅ Response ${i + 1}:, response.substring(0, 100));
    } catch (error) {
      if (error instanceof CircuitBreakerOpenError) {
        console.log(⚠️ Circuit Breaker OPEN: ${error.message});
        await new Promise(resolve => setTimeout(resolve, 5000));
      } else {
        console.error(❌ Error:, error);
      }
    }
  }
}

main().catch(console.error);

export { HolySheepCircuitBreaker, CircuitBreakerOpenError, CircuitState };

Circuit Breaker の阀値設計ガイド

AI サービスの特性に応じた阀値設定の指针を示します：

• 高トラフィックECサイト：failure_threshold=5, timeout=30s（短時間で恢复を確認し、用户影響を最小化）
• 企业内部RAGシステム：failure_threshold=10, timeout=60s（夜间バッチ处理で多少の延迟は許容）
• 個人開発プロジェクト：failure_threshold=3, timeout=20s（予算有限、早期遮断でコスト制御）

HolySheep AI 利用時の费用节省効果（レート¥1=$1）を活かすため、Circuit Breaker で API エラー时应时立即遮断し、不要なリクエスト费用的发生防止が重要です。

モデル別の推奨阀値

HolySheep AI で利用可能な主要モデルの特性に応じた阀値設定例：

# モデル別の Circuit Breaker 阀値設定例

CIRCUIT_BREAKER_CONFIGS = {
    # 高コストモデル：より敏感な阀値
    "gpt-4.1": {
        "failure_threshold": 3,      # 3回のエラーでOPEN
        "success_threshold": 3,      # 3回の成功でCLOSED
        "timeout": 45,               # 45秒後にHALF_OPEN
        "half_open_max_calls": 2,    # 小さく試行
    },
    "claude-sonnet-4.5": {
        "failure_threshold": 3,
        "success_threshold": 3,
        "timeout": 45,
        "half_open_max_calls": 2,
    },
    
    # 安価なモデル：より緩やかな阀値
    "deepseek-v3.2": {
        "failure_threshold": 10,     # 10回のエラーでOPEN
        "success_threshold": 2,      # 2回の成功でCLOSED
        "timeout": 30,               # 30秒後にHALF_OPEN
        "half_open_max_calls": 5,    # 積極的に恢复確認
    },
    "gemini-2.5-flash": {
        "failure_threshold": 8,
        "success_threshold": 2,
        "timeout": 30,
        "half_open_max_calls": 4,
    },
}

DeepSeek V3.2 のように $0.42/MTok の低成本モデルはより攻撃的な阀值设定が可能で、GPT-4.1 ($8/MTok) や Claude Sonnet 4.5 ($15/MTok) の高コストモデルは失敗時の损失が大きいため、敏感な阀值设定が推奨されます。

フォールバック戦略の實現

async def call_with_fallback(
    circuit_breaker: CircuitBreaker,
    user_message: str
) -> str:
    """フォールバック機能付きの HolySheep AI 呼び出し"""
    
    # 優先順位：DeepSeek → Gemini Flash → キャッシュ
    models_priority = [
        "deepseek-v3.2",      # 最安値 $0.42/MTok
        "gemini-2.5-flash",   # 高速 $2.50/MTok
        "gpt-4.1",            # 高品質 $8/MTok
    ]
    
    errors = []
    
    for model in models_priority:
        try:
            async def _request():
                async with httpx.AsyncClient(timeout=30.0) as client:
                    response = await client.post(
                        "https://api.holysheep.ai/v1/chat/completions",
                        headers={
                            "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                        },
                        json={
                            "model": model,
                            "messages": [{"role": "user", "content": user_message}],
                        }
                    )
                    response.raise_for_status()
                    return response.json()["choices"][0]["message"]["content"]
            
            return await circuit_breaker.call(_request)
            
        except CircuitBreakerOpenError:
            raise  # Circuit Breaker遮断時は上位に丸投げ
        except Exception as e:
            errors.append(f"{model}: {str(e)}")
            continue
    
    # 全モデル失敗時のフォールバック
    return {
        "status": "fallback",
        "message": "ただいま混み合っています。後ほど再度お試しください。",
        "original_errors": errors
    }

よくあるエラーと対処法

エラー 1：Circuit Breaker が OPEN のまま恢复しない

# 問題：timeout を设定しても OPEN から HALF_OPEN に移行しない

原因：time.time() と Date.now() の单位違い（秒 vs ミリ秒）

修正例（Python）
@dataclass
class CircuitBreaker:
    # ...
    def _check_timeout(self) -> bool:
        # time.time() は秒単位なので、ミリ秒比较の場合は * 1000
        elapsed_ms = (time.time() - self.last_failure_time) * 1000
        return elapsed_ms >= self.config.timeout * 1000  # ミリ秒に変換

修正例（JavaScript/TypeScript）
class HolySheepCircuitBreaker {
  // ...
  private checkTimeout(): boolean {
    // Date.now() はミリ秒単位
    return Date.now() - this.lastFailureTime >= this.timeout;
  }
}

エラー 2：HALF_OPEN 状態で無限ループが発生する

# 問題：HALF_OPEN 状态下で常に OPEN に返回され，恢复できない

原因：success_threshold と half_open_max_calls のバランス不良

修正：HALF_OPEN 時は success_threshold=1 に一时的に変更
async def call_with_adaptive_threshold(
    breaker: CircuitBreaker,
    func: Callable
):
    original_threshold = breaker.config.success_threshold
    
    if breaker.state == CircuitState.HALF_OPEN:
        # HALF_OPEN 時は1回の成功でCLOSEDに
        breaker.config.success_threshold = 1
    
    try:
        result = await breaker.call(func)
        return result
    finally:
        breaker.config.success_threshold = original_threshold

エラー 3：API レスポンスは正常だが ApplicationError が発生する

# 問題：HTTP 200 でもビジネスロジックエラーがある場合

解決：Circuit Breaker の例外判定をカスタマイズ
class HolySheepCircuitBreaker:
    # ...
    async execute<T>(
        fn: () => Promise<T>,
        isRetryableError: (error: any) => boolean = () => true
    ): Promise<T> {
        try {
            const result = await fn();
            
            // ビジネスロジックエラーの判定
            if (result.error || result.status === 'error') {
                if (isRetryableError(result)) {
                    this.onFailure();
                }
            }
            
            this.onSuccess();
            return result;
        } catch (error) {
            // レートリミットエラーは即座に OPEN
            if (error.status === 429 || error.code === 'rate_limit_exceeded') {
                this.transitionTo(CircuitState.OPEN);
            } else if (isRetryableError(error)) {
                this.onFailure();
            }
            throw error;
        }
    }

// 使用例
const breaker = new HolySheepCircuitBreaker();
await breaker.execute(
    () => callHolySheepAI(message),
    (error) => error.status >= 500  // サーバーエラー时만 Circuit Breaker 作動
);

エラー 4：同時多数リクエストによる阀値突破

# 問題：多个リクエストが同時に失敗し、正常なリクエストまで遮断される

解決：スライディングウィンドウ方式の導入
from collections import deque
from datetime import datetime

class SlidingWindowCircuitBreaker:
    def __init__(self, window_size: int = 60, failure_threshold: int = 10):
        self.window_size = window_size  # 秒
        self.failure_threshold = failure_threshold
        self.failures: deque = deque()  # 失敗时刻のタイムスタンプ
    
    def _clean_old_failures(self):
        """ウィンドウ外の失敗记录を削除"""
        cutoff = time.time() - self.window_size
        while self.failures and self.failures[0] < cutoff:
            self.failures.popleft()
    
    def record_failure(self):
        self.failures.append(time.time())
    
    def should_open(self) -> bool:
        self._clean_old_failures()
        return len(self.failures) >= self.failure_threshold
    
    def record_success(self):
        """成功時は最古の失敗记录を1つ削除（乐观的恢复）"""
        self._clean_old_failures()
        if self.failures:
            self.failures.popleft()

監視とアラート設定

Circuit Breaker の状態变化を監視し、適切なアラートを設定することで、問題の早期発見が可能になります：

# Prometheus/Metrics エクスポート例
from prometheus_client import Counter, Gauge

circuit_breaker_state = Gauge(
    'circuit_breaker_state',
    'Circuit Breaker state (0=CLOSED, 1=HALF_OPEN, 2=OPEN)',
    ['service', 'model']
)

circuit_breaker_transitions = Counter(
    'circuit_breaker_transitions_total',
    'Circuit Breaker state transitions',
    ['service', 'model', 'from_state', 'to_state']
)

circuit_breaker_calls = Counter(
    'circuit_breaker_calls_total',
    'Circuit Breaker calls',
    ['service', 'model', 'result']  # result: success, failure, blocked
)

Circuit Breaker 监视ダッシュボード用クエリ
OPEN状态的持续时间
circuit_breaker_open_duration_seconds{model="deepseek-v3.2"}

遮断されたリクエスト数
rate(circuit_breaker_calls_total{result="blocked"}[5m])

まとめ

Circuit Breaker パターンは、AI API を活用したシステムの安定性を確保するために不可欠な技术です。本稿で示した実装例と阀値設定の指针を活かし、HolySheep AI の高パフォーマンス（<50ms レイテンシ）と成本効果（¥1=$1、レート85%節約）を活かした、より堅牢な AI 服务を構築去吧。

特に、DeepSeek V3.2 ($0.42/MTok) や Gemini 2.5 Flash ($2.50/MTok) などの低成本モデルを活用する際は、Circuit Breaker による適切な流量制御と組み合わせることで、コストと可用性のバランスを最適化できます。

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