外部APIやモデルプロバイダの不調で業務が止まってしまう不安。どのタイミングで切り替えるべきか、どのように安全に段階的に移すかは現場でよく迷うポイントです。本稿では、実務で使える視点に絞り、Pythonで実装できるアダプタ層、健全性チェック、サーキットブレーカー、カナリア配信までのワークフローを具体的に示します。読み進めながら自分のサービスに当てはめてください。
導入の背景と適用範囲
いつフェイルオーバー設計が必要かをまず整理します。全てのシステムで完全な多重化が必要なわけではありませんが、以下の条件に該当する場合は設計を検討してください。
| 導入トリガー | 理由 |
|---|---|
| SLA/SLO違反リスクが高い | 外部遅延が業務影響を直ちに与えるため |
| コスト急騰時の保険が必要 | 突発的な利用料上昇時に代替で運用継続するため |
| リージョン障害や法令リスク | 特定のプロバイダを使えなくなるケースに備えるため |
| 既存の監視・コスト管理と連携したい | 第95回（SLO）や第88回（APIコスト）との連携がある場合 |
アーキテクチャ方針（実務で使える抽象化）
設計方針は「統一インターフェース」「ルーティングで切替可能」「最小限のローカル保護」の3点です。
主要コンポーネント
| コンポーネント | 役割 | 注意点 |
|---|---|---|
| Provider Adapter | 各プロバイダを統一インターフェースで扱う | レスポンス正規化とエラーコードのラップ |
| Router | ルーティング／カナリア割合を決定する | 割合/セグメント指定を外部設定可能に |
| Health Checker | 定期的に健全性を評価しRouterへ反映 | API応答時間とエラー率を重視 |
| Circuit Breaker | 障害の連鎖を防ぎ、一時遮断する | 自動回復の閾値と手動操作を両方用意 |
| Local Cache / Backpressure | 短時間の遅延吸収／負荷制御 | 重要なリクエストは優先順位付け |
構成図（テーブルで表す例）
| 外部 | 内部 | 備考 |
|---|---|---|
| Provider A Provider B |
Client → Router → Adapter → Provider ↑ Health Checker → Router ↑ Circuit Breaker 層 |
Router は割合ベース／ユーザセグメントで振り分け |
Python 実装パターン（実務利用向け）
以下はそのまま貼れるシンプルなコード例です。実運用ではログ、メトリクス、認証管理を追加してください。
1) 共通 Adapter の例
# adapter.py
import time
class ProviderResponse:
def __init__(self, ok, result=None, error=None, latency_ms=None):
self.ok = ok
self.result = result
self.error = error
self.latency_ms = latency_ms
class ProviderAdapter:
def __init__(self, name):
self.name = name
def call(self, payload, timeout=5):
"""同期呼び出しの例。各プロバイダ実装はここをoverrideする"""
raise NotImplementedError
2) シンプルなプロバイダ実装の例
# providers/openai_adapter.py
import requests
from adapter import ProviderAdapter, ProviderResponse
import time
class OpenAIAdapter(ProviderAdapter):
def __init__(self, api_key):
super().__init__('openai')
self.api_key = api_key
def call(self, payload, timeout=5):
start = time.time()
try:
r = requests.post('https://api.openai.com/v1/...,', json=payload,
headers={'Authorization': f'Bearer {self.api_key}'}, timeout=timeout)
latency = int((time.time() - start) * 1000)
if r.status_code == 200:
return ProviderResponse(True, result=r.json(), latency_ms=latency)
else:
return ProviderResponse(False, error=f'status:{r.status_code}', latency_ms=latency)
except requests.RequestException as e:
latency = int((time.time() - start) * 1000)
return ProviderResponse(False, error=str(e), latency_ms=latency)
3) 健全性チェッカー（ヘルスチェック）
# health.py
import time
class HealthChecker:
def __init__(self, adapter, window=60):
self.adapter = adapter
self.window = window
self.history = [] # (timestamp, ok, latency_ms)
def probe(self, sample_payload):
resp = self.adapter.call(sample_payload, timeout=3)
self.history.append((time.time(), resp.ok, resp.latency_ms or 9999))
# 保持する履歴は window 秒分に制限
cutoff = time.time() - self.window
self.history = [h for h in self.history if h[0] >= cutoff]
return resp
def metrics(self):
total = len(self.history)
if total == 0:
return {'error_rate': 0.0, 'p99': None}
errs = sum(1 for _, ok, _ in self.history if not ok)
latencies = [lat for _, ok, lat in self.history if ok]
p99 = max(latencies) if latencies else None
return {'error_rate': errs / total, 'p99': p99}
4) 軽量サーキットブレーカー
# circuit.py
import time
class CircuitBreaker:
def __init__(self, fail_threshold=5, recovery_time=30):
self.fail_threshold = fail_threshold
self.recovery_time = recovery_time
self.failure_count = 0
self.opened_at = None
def record_success(self):
self.failure_count = 0
self.opened_at = None
def record_failure(self):
self.failure_count += 1
if self.failure_count >= self.fail_threshold:
self.opened_at = time.time()
def allow(self):
if self.opened_at is None:
return True
if time.time() - self.opened_at > self.recovery_time:
# half-open を簡易に扱う: 許可して試行
return True
return False
5) 非同期呼び出しとタイムアウトのポイント
async実装ではaiohttp等を用います。タイムアウトとキャンセル処理を必ず設け、ルーティング層で遅延しすぎるプロバイダを切り替える設計にしてください。
# async_example.py (抜粋)
import aiohttp
import asyncio
async def async_call(url, data, timeout=3):
try:
async with aiohttp.ClientSession() as s:
async with s.post(url, json=data, timeout=timeout) as r:
return await r.json()
except asyncio.TimeoutError:
raise
運用で使える切替手順（チェックリスト）
以下は現場で使える簡易ルールと手順表です。状況に合わせて閾値はチューニングしてください。
| 項目 | 推奨/例 | 理由 |
|---|---|---|
| フェイルオーバー発動ルール | エラー率 > 5% かつ連続エラー数 >= 10、またはP99 > 2s（5分間） | 短期のスパイクで誤発動しないために複合条件にする |
| カナリア開始 | 初期 5% のトラフィックを代替へ、24時間で観察 | 互換性と品質を段階的に確認するため |
| 割合拡大 | 5%→20%→50%→100%（各段階最低1時間＋品質観察） | 拡大は短時間で行わない |
| ロールバック条件 | 代替でエラー率急増 / 認証失敗 / データ損失疑い | 即時ロールバックを可能にする |
| 関係者通知 | 自動アラート→SRE/プロダクト/法務へ同時通知 | 契約・法令リスクがある場合に迅速に判断を仰ぐ |
運用手順（簡易ランブック）
- 1) モニタリングが閾値を超えた場合、HealthCheckerの最新レポートを確認。自動でCircuitを開くかを判定する。
- 2) Routerでカナリア割合を5%に設定し、代替プロバイダへ流す（ログは必ずトレースする）。
- 3) 1時間ごとに品質メトリクス（エラー率/P99）を確認。問題が無ければ次フェーズへ移行。
- 4) 重大な問題が出たら即時ロールバックし、原因調査と関係者ミーティングを開始。
テストと検証プラン
切替機構は定期的にテストしておくことが重要です。以下のテストを自動化してください。
| テスト | 目的 | 実施方法（例） |
|---|---|---|
| 障害注入（ローカル/ステージング） | サーキット動作とルートの切替確認 | プロバイダをモックしてエラー応答・遅延を注入 |
| 負荷ベンチ | P99やスループットの変化確認 | wrkやlocustで通常時/切替時の比較 |
| 回帰テスト | 互換性チェック | 実リクエストの代表ケースを自動化テストで実行 |
| 監査ログ確認 | 追跡性の担保 | リクエストID、選択プロバイダ、フォールバック理由を検証 |
自動化スクリプト例（障害注入）
# fault_injector.py (ステージング用、単純例)
import random
from adapter import ProviderAdapter, ProviderResponse
class FlakyAdapter(ProviderAdapter):
def call(self, payload, timeout=5):
if random.random() < 0.3: # 30% failure
return ProviderResponse(False, error='injected')
return ProviderResponse(True, result={'ok': True})
# テストスクリプトはCIに組み込む
監視・アラート設計（具体例）
| 指標 | サンプル閾値（例） | 通知内容 |
|---|---|---|
| エラー率 | 5%（警告） / 10%（致命） | 直近5分のエラー率・該当プロバイダID・リクエストIDサンプル |
| P99 レイテンシ | 2s（警告） / 5s（致命） | 遅延が大きいAPIパス・影響割合 |
| 呼び出し数 | baseline の ±50% | 急増はコスト問題のサイン |
| コスト/分 | 予算しきい値超過 | 自動でカナリア割合を引き下げる仕組みを検討 |
ログには必ず以下を残してください：リクエストID、選択プロバイダ名、フォールバック理由、レイテンシ。
セキュリティと契約面のチェックリスト
| 項目 | 確認内容 |
|---|---|
| PII/データフロー | 代替プロバイダへ送るデータにPIIが含まれないか、必要ならマスク/トークン化 |
| 契約制約 | 契約で代替利用が禁止されていないか（地域・用途制限）を法務と確認 |
| 鍵管理 | 各プロバイダ鍵はKMS/シークレット管理で分離し、アクセス権限を制御 |
| データ保持/削除 | 切替時に送信済みデータの扱い（保持が必要か削除か）を定義 |
よくある失敗例と回避策
- 失敗例：瞬時に全面切替して互換性で障害拡大 → 回避策：カナリア段階と回帰テストを必須化
- 失敗例：監視が不十分で切替後にコスト急増 → 回避策：コスト指標の自動連携と割合制御
- 失敗例：ログが残らず責任追跡ができない → 回避策：必須ログ項目を運用ルール化
導入後レビュー（30日/90日チェック項目）
| 期間 | 確認項目 |
|---|---|
| 30日 | 切替テスト結果、カナリアでのエラー率、運用手順の実効性 |
| 90日 | コスト推移、法務リスクの再評価、運用改善の反映 |
まとめ
外部プロバイダに依存する領域は、業務継続とリスク管理の両面で設計が必要です。本稿では実務で使える抽象化（Adapter/Router/Health/Circuit）、Pythonでの実装パターン、段階的切替と運用チェックリスト、テスト計画、監視設計、セキュリティ・契約面の注意点を示しました。まずは小さなカナリアから始め、監視・ログを充実させつつ段階的に拡大するのが安全な導入の近道です。
Manage AI シリーズ「AIとPythonの実務」の一環として、次回は「プロンプト互換性を保つための自動検査パイプライン」を予定しています。この記事のコード例やチェックリストはテンプレートとして調整してお使いください。