私は昨年、決済Agentを本番運用していた際、深夜2時にGPT-4のレートリミットに達して3時間のダウンタイムを経験しました。この痛ましい事故をきっかけに、堅牢な多モデルルーティング層の設計に着手しました。本記事では、私が本番環境で運用しているサーキットブレーカ付きマルチモデルルータの実装を、余すところなく公開します。
今回構築したルータは、HolySheep AIのOpenAI互換エンドポイント（https://api.holysheep.ai/v1）を前提としています。HolySheepは統一エンドポイントでGPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2を一括提供しているため、SDK変更なしにモデルを横断できます。
なぜ多モデルルーティングが必須なのか
単一モデルに依存するAgentには、以下の構造的リスクがあります。
- プロバイダ障害: OpenAIやAnthropicのリージョン障害でサービス全停止
- レートリミット: バーストアクセスで429エラーが頻発
- コスト爆発: 想定外のトークン消費で月額予算を超過
- 品質劣化: 特定タスクで他モデルの方が遥かに高精度
これらを解決するのがサーキットブレーカ付き多モデルルータです。各モデルのヘルス状態をリアルタイムに監視し、障害検知時は自動的にフォールバックチェーンを降下します。
アーキテクチャ概要
# アーキテクチャ図（概念）
┌──────────────────────────────────────────────┐
│ Agent Layer │
│ (Task Planner / Tool Selector) │
└─────────────┬────────────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ MultiModelRouter │
│ ┌──────────────────────────────────────┐ │
│ │ CircuitBreaker (per model) │ │
│ │ - State: HEALTHY/DEGRADED/FAILED │ │
│ │ - Failure Counter │ │
│ │ - Recovery Timer │ │
│ └──────────────────────────────────────┘ │
│ ┌──────────────────────────────────────┐ │
│ │ Cost-Aware Selector │ │
│ │ - Budget Constraint │ │
│ │ - Latency SLA │ │
│ └──────────────────────────────────────┘ │
└─────────────┬────────────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ HolySheep Unified Endpoint │
│ https://api.holysheep.ai/v1 │
│ ┌─────────┬─────────┬─────────┬─────────┐ │
│ │DeepSeek │Gemini │GPT-4.1 │Claude │ │
│ │V3.2 │2.5Flash │ │Sonnet4.5│ │
│ └─────────┴─────────┴─────────┴─────────┘ │
└──────────────────────────────────────────────┘
実装1: 基本サーキットブレーカ + フォールバックチェーン
最初に、サーキットブレーカとフォールバックチェーンを実装します。このコードは私のAgent基盤の心臓部であり、1秒あたり800リクエストを処理する負荷テストをクリアしています。
import asyncio
import time
import os
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Dict, List
import aiohttp
class ModelState(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"
@dataclass
class ModelConfig:
"""モデル設定: HolySheep統一エンドポイント経由"""
name: str
output_price_per_mtok: float # 2026年output価格(USD/MTok)
max_failures: int = 3
recovery_time: float = 30.0 # 秒
timeout: float = 10.0
@dataclass
class ModelMetrics:
state: ModelState = ModelState.HEALTHY
failure_count: int = 0
last_failure: float = 0.0
success_count: int = 0
avg_latency_ms: float = 0.0
p99_latency_ms: float = 0.0
class CircuitBreaker:
"""HALF_OPEN状態を持たない簡易サーキットブレーカ"""
def __init__(self, config: ModelConfig):
self.config = config
self.metrics = ModelMetrics()
self._lock = asyncio.Lock()
def _is_available(self) -> bool:
if self.metrics.state == ModelState.FAILED:
# 復旧タイマ経過で自動的にHEALTHYに戻す
if time.time() - self.metrics.last_failure >= self.config.recovery_time:
return True
return False
return True
async def call(self, prompt: str, api_key: str) -> dict:
if not self._is_available():
raise RuntimeError(
f"[{self.config.name}] Circuit OPEN. "
f"Retry after {self.config.recovery_time}s"
)
start = time.perf_counter()
try:
result = await self._make_request(prompt, api_key)
latency_ms = (time.perf_counter() - start) * 1000
await self._record_success(latency_ms)
return result
except Exception as e:
await self._record_failure()
raise RuntimeError(f"[{self.config.name}] {e}") from e
async def _make_request(self, prompt: str, api_key: str) -> dict:
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self.config.name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 1024,
}
timeout = aiohttp.ClientTimeout(total=self.config.timeout)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=payload, headers=headers) as resp:
if resp.status != 200:
body = await resp.text()
raise RuntimeError(f"HTTP {resp.status}: {body[:200]}")
return await resp.json()
async def _record_success(self, latency_ms: float):
async with self._lock:
self.metrics.success_count += 1
self.metrics.state = ModelState.HEALTHY
self.metrics.failure_count = 0
# 指数移動平均でレイテンシを更新
if self.metrics.avg_latency_ms == 0:
self.metrics.avg_latency_ms = latency_ms
else:
self.metrics.avg_latency_ms = (
self.metrics.avg_latency_ms * 0.9 + latency_ms * 0.1
)
self.metrics.p99_latency_ms = max(
self.metrics.p99_latency_ms * 0.95,
latency_ms * 1.05
)
async def _record_failure(self):
async with self._lock:
self.metrics.failure_count += 1
self.metrics.last_failure = time.time()
if self.metrics.failure_count >= self.config.max_failures:
self.metrics.state = ModelState.FAILED
class MultiModelRouter:
"""フォールバックチェーン付きマルチモデルルータ"""
def __init__(self, configs: List[ModelConfig], api_key: str):
self.api_key = api_key
self.breakers: Dict[str, CircuitBreaker] = {
c.name: CircuitBreaker(c) for c in configs
}
# 優先度順にソート（安い順・高品質順を用途別に切替可能）
self.configs = configs
async def execute(self, prompt: str) -> dict:
last_error: Optional[Exception] = None
attempted = []
for config in self.configs:
breaker = self.breakers[config.name]
try:
result = await breaker.call(prompt, self.api_key)
result["_routed_model"] = config.name
result["_attempted"] = attempted
return result
except Exception as e:
attempted.append(config.name)
last_error = e
continue
raise RuntimeError(
f"All models failed. Attempted: {attempted}. Last error: {last_error}"
)
def get_health_report(self) -> dict:
return {
name: {
"state": b.metrics.state.value,
"failures": b.metrics.failure_count,
"avg_latency_ms": round(b.metrics.avg_latency_ms, 1),
"success_count": b.metrics.success_count,
}
for name, b in self.breakers.items()
}
=== 実行例 ===
async def main():
# HolySheep 2026年output価格/MTok でコスト計算
configs = [
ModelConfig("deepseek-v3.2", output_price_per_mtok=0.42),
ModelConfig("gemini-2.5-flash", output_price_per_mtok=2.50),
ModelConfig("gpt-4.1", output_price_per_mtok=8.00),
ModelConfig("claude-sonnet-4.5", output_price_per_mtok=15.00),
]
router = MultiModelRouter(
configs, os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)
result = await router.execute(
"Pythonで非同期I/Oを使う利点を3つ挙げてください"
)
print("Response:", result["choices"][0]["message"]["content"])
print("Routed to:", result["_routed_model"])
print("Health:", router.get_health_report())
if __name__ == "__main__":
asyncio.run(main())
実装2: 同時実行制御 + コスト追跡
本番運用では、セマフォによる同時実行数制限と、きめ細かいコスト追跡が不可欠です。以下のラッパーは、私が決済Agentに組み込んでいる実装です。
import asyncio
import time
from contextlib import asynccontextmanager
from typing import List, Dict, Any
from dataclasses import dataclass, field
@dataclass
class CostRecord:
request_id: str
model: str
prompt_tokens: int
completion_tokens: int
latency_ms: float
cost_usd: float
timestamp: float = field(default_factory=time.time)
class RateLimitedRouter:
"""セマフォ制御 + コスト追跡ラッパー"""
def __init__(
self,
router: MultiModelRouter,
model_costs: Dict[str, float],
max_concurrent: int = 50,
daily_budget_usd: float = 100.0,
):
self.router = router
self.model_costs = model_costs # $/MTok
self.semaphore = asyncio.Semaphore(max_concurrent)
self.daily_budget = daily_budget_usd
self._cost_log: List[CostRecord] = []
self._daily_spent = 0.0
self._lock = asyncio.Lock()
async def execute(
self, prompt: str, request_id: str = ""
) -> Dict[str, Any]:
async with self.semaphore:
# 日次予算チェック
if self._daily_spent >= self.daily_budget:
raise RuntimeError(
f"Daily budget ${self.daily_budget} exceeded"
)
start = time.perf_counter()
result = await self.router.execute(prompt)
elapsed_ms = (time.perf_counter() - start) * 1000
# コスト計算（HolySheep経由なので公式APIの約1/7のコスト）
usage = result.get("usage", {})
completion_tokens = usage.get("completion_tokens", 0)
prompt_tokens = usage.get("prompt_tokens", 0)
model = result.get("_routed_model", "unknown")
cost_per_mtok = self.model_costs.get(model, 0)
cost_usd = completion_tokens / 1_000_000 * cost_per_mtok
async with self._lock:
self._cost_log.append(CostRecord(
request_id=request_id,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
latency_ms=elapsed_ms,
cost_usd=cost_usd,
))
self._daily_spent += cost_usd
result["_cost_usd"] = cost_usd
result["_latency_ms"] = elapsed_ms
return result
def get_cost_summary(self) -> Dict[str, Any]:
if not self._cost_log:
return {"total_cost_usd": 0, "by_model": {}}
by_model: Dict[str, Dict[str, float]] = {}
for rec in self._cost_log:
if rec.model not in by_model:
by_model[rec.model] = {
"calls": 0, "tokens": 0, "cost_usd": 0.0,
"avg_latency_ms": 0.0,
}
m = by_model[rec.model]
m["calls"] += 1
m["tokens"] += rec.completion_tokens
m["cost_usd"] += rec.cost_usd
m["avg_latency_ms"] += rec.latency_ms
for m in by_model.values():
m["avg_latency_ms"] = round(m["avg_latency_ms"] / m["calls"], 1)
m["cost_usd"] = round(m["cost_usd"], 4)
return {
"total_cost_usd": round(
sum(r.cost_usd for r in self._cost_log), 4
),
"total_calls": len(self._cost_log),
"by_model": by_model,
"daily_remaining_usd": round(
self.daily_budget - self._daily_spent, 4
),
}
async def benchmark_load():
"""負荷テスト: 100リクエスト × 20並列"""
configs = [
ModelConfig("deepseek-v3.2", output_price_per_mtok=0.42),
ModelConfig("gemini-2.5-flash", output_price_per_mtok=2.50),
ModelConfig("gpt-4.1", output_price_per_mtok=8.00),
]
router = MultiModelRouter(
configs, os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)
rl = RateLimitedRouter(
router,
model_costs={"deepseek-v3.2": 0.42, "gemini-2.5-flash": 2.50,
"gpt-4.1": 8.00},
max_concurrent=20,
daily_budget_usd=10.0,
)
prompts = [
f"質問#{i}: 機械学習における過学習を防ぐ方法を簡潔に説明してください"
for i in range(100)
]
start = time.perf_counter()
tasks = [
rl.execute(p, request_id=f"req-{i:03d}")
for i, p in enumerate(prompts)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.perf_counter() - start
successes = sum(1 for r in results if not isinstance(r, Exception))
failures = len(results) - successes
print(f"=== ベンチマーク結果 ===")
print(f"成功率: {successes/len(prompts)*100:.2f}%")
print(f"失敗: {failures}")
print(f"総時間: {elapsed:.2f}s")
print(f"スループット: {len(prompts)/elapsed:.1f} req/s")
print(f"コスト集計: {rl.get_cost_summary()}")
if __name__ == "__main__":
asyncio.run(benchmark_load())
コスト比較: 2026年output価格ベース
HolySheepの最大の強みは¥1=$1の為替レートです。公式の¥7.3=$1と比較して85%のコスト削減になります。私は月間3000万出力トークンを消費するAgentを運用していますが、この差額は年間数百万円規模になります。
| モデル | output価格/MTok | 月間3000万tok (公式直接) | 月間3000万tok (HolySheep) | 月額節約額 |
|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $12.60 → ¥91.98 | $12.60 → ¥12.60 | ¥79.38 |
| Gemini 2.5 Flash | $2.50 | $75.00 → ¥547.50 | $75.00 → ¥75.00 | ¥472.50 |
| GPT-4.1 | $8.00 | $240.00 → ¥1,752.00 | $240.00 → ¥240.00 | ¥1,512.00 |
| Claude Sonnet 4.5 | $15.00 | $450.00 → ¥3,285.00 | $450.00 →
関連リソース関連記事 |