AI APIを本番環境に組み込むとき、私が最も多くご相談を受けるのが「可用性の確保」と「障害時の continuidad de servicio（サービス継続）」です。ECサイトのAIカスタマーサービスが増患時にScalerしても落ちない理由、企業RAGシステムが99.9% uptimeを実現するための設計、そして個人開発者が最小限の工数で可用性を確保する方法について、今日は具体例と共に解説いたします。
なぜ今、AI APIゲートウェイの可用性が死活問題なのか
私の経験では、AI APIの障害は「突然的黑字転換」→「客服対応急増」→「開発チーム深夜対応」という悪循環を生みます。特にEC系事業者様は、AI客服の停止がそのまま売上損失に直結するため、可用性はビジネス要件そのものと言えます。
具体的なユースケースから見る設計パターン
ケース1：ECサイトのAIカスタマーサービス増患対策
# Python - Circuit Breaker実装例
import httpx
import asyncio
from datetime import datetime, timedelta
from typing import Optional
class HolySheepGateway:
"""HolySheep AI API ゲートウェイ - Circuit Breaker付き"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Circuit Breaker状態
self.failure_count = 0
self.last_failure_time: Optional[datetime] = None
self.circuit_open = False
self.failure_threshold = 5
self.reset_timeout = 60 # 秒
async def chat_completions(self, messages: list, model: str = "gpt-4.1"):
"""Circuit Breaker経由でAPI呼び出し"""
if self.circuit_open:
if self._should_attempt_reset():
self.circuit_open = False
self.failure_count = 0
else:
raise Exception("Circuit Breaker Open: HolySheep APIが一時的に利用不可")
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={"model": model, "messages": messages}
)
if response.status_code == 200:
self.failure_count = 0
return response.json()
else:
self._record_failure()
raise Exception(f"API Error: {response.status_code}")
except httpx.TimeoutException:
self._record_failure()
raise Exception("タイムアウト: HolySheep APIの応答が 없습니다")
except httpx.ConnectError:
self._record_failure()
raise Exception("接続エラー: ネットワークまたはAPIエンドポイントの問題")
def _record_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
def _should_attempt_reset(self) -> bool:
if self.last_failure_time is None:
return True
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
return elapsed >= self.reset_timeout
利用例
gateway = HolySheepGateway("YOUR_HOLYSHEEP_API_KEY")
async def handle_customer_query(user_message: str):
try:
result = await gateway.chat_completions(
messages=[{"role": "user", "content": user_message}],
model="gpt-4.1"
)
return result["choices"][0]["message"]["content"]
except Exception as e:
# フォールバック: 缓存済み回答または有人対応へ
return "ただいま込み合っております。少々お待ちください。"ケース2：企業RAGシステムのマルチリージョン構成
# Python - マルチリージョンフォールバック実装
import asyncio
from dataclasses import dataclass
from typing import List, Optional
import httpx
@dataclass
class RegionEndpoint:
name: str
base_url: str
priority: int # 低いほど優先度高
is_healthy: bool = True
latency_ms: float = 0.0
class MultiRegionGateway:
"""マルチリージョン対応 AI API ゲートウェイ"""
def __init__(self, api_key: str):
self.api_key = api_key
self.regions: List[RegionEndpoint] = [
RegionEndpoint("ap-northeast-1", "https://api.holysheep.ai/v1", 1),
RegionEndpoint("us-east-1", "https://api.holysheep.ai/v1", 2),
RegionEndpoint("eu-west-1", "https://api.holysheep.ai/v1", 3),
]
self._health_check_interval = 30
async def _health_check(self, region: RegionEndpoint):
"""リージョン毎のヘルスチェック"""
try:
async with httpx.AsyncClient(timeout=5.0) as client:
start = asyncio.get_event_loop().time()
response = await client.get(
f"{region.base_url}/models",
headers={"Authorization": f"Bearer {self.api_key}"}
)
region.latency_ms = (asyncio.get_event_loop().time() - start) * 1000
region.is_healthy = (response.status_code == 200)
except:
region.is_healthy = False
async def _get_healthy_region(self) -> Optional[RegionEndpoint]:
"""正常なリージョンを優先度順で取得"""
await asyncio.gather(*[self._health_check(r) for r in self.regions])
healthy = [r for r in self.regions if r.is_healthy]
if not healthy:
return None
# レイテンシ順でソート（<50ms目標）
return min(healthy, key=lambda r: r.latency_ms)
async def embeddings(self, texts: List[str], model: str = "text-embedding-3-large"):
"""Embeddings API - 自動リージョン選択"""
region = await self._get_healthy_region()
if not region:
raise Exception("全リージョン障害: AI API利用不可")
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{region.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={"input": texts, "model": model}
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# レートリミット時は次のリージョンへ
return await self.embeddings(texts, model)
else:
raise Exception(f"Embedding Error: {response.status_code}")
企業RAGシステムでの利用
gateway = MultiRegionGateway("YOUR_HOLYSHEEP_API_KEY")
async def rag_retrieval(query: str, vector_store):
# クエリをEmbedding
query_embedding = await