私は普段の業務で每天都大量のAPIリクエストを処理していますが、タイムアウト処理の不出来で本番障害が発生することがしばしばありました。特にAI APIは予測できない処理時間を抱えるため、適切なタイムアウト設計がシステムの安定性を左右します。本稿では、HolySheep AIを例に、本番レベルのリトライ戦略、サーキットブレーカー、パフォーマンス最適化を実装面から詳しく解説します。
なぜAI APIのタイムアウト処理が重要か
AI APIは従来のREST APIと異なり、以下の特性を持っています：
- リクエストボディとレスポンスボディの両方に大容量テキスト
- モデルによって処理時間が大きく変動（DeepSeek V3.2 $0.42/MTok vs GPT-4.1 $8/MTok）
- レート制限による429エラーの頻度
- サーバー側のキュー詰まりによる意図しない遅延
HolySheep AIの場合、<50msの低レイテンシを実現していますが、それでもネットワーク分包이나 서버负载によりタイムアウトは発生します。特に注意すべきは、タイムアウト時にリクエストが「どこで失敗したか」を正確に把握しないと、無限リトライによるコスト爆発を招くことです。
基本的なタイムアウト設定のアーキテクチャ
効果的なタイムアウト処理には、3层次的アプローチが必要です：
- 接続タイムアウト（connect_timeout）：TCP接続確立までの最大待機時間
- 読み取りタイムアウト（read_timeout）：サーバーからのレスポンス受信最大待機時間
- 全体リクエストタイムアウト（total_timeout）：リクエスト全体の最大許容時間
以下のコードは、PythonでHolySheep AI APIに対して適切なタイムアウトを設定する基本パターンです：
import httpx
import asyncio
from typing import Optional
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class TimeoutConfig:
"""HolySheep AI API 用タイムアウト設定"""
connect_timeout: float = 10.0 # TCP接続確立: 10秒
read_timeout: float = 60.0 # レスポンス読み取り: 60秒
pool_timeout: float = 65.0 # プール接続待機: 65秒
total_timeout: float = 120.0 # 全体リミット: 120秒
class HolySheepAIClient:
"""HolySheep AI API クライアント - 適切なタイムアウト処理付き"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, timeout_config: Optional[TimeoutConfig] = None):
self.api_key = api_key
self.timeout = timeout_config or TimeoutConfig()
self._client: Optional[httpx.AsyncClient] = None
async def _get_client(self) -> httpx.AsyncClient:
"""HTTPクライアントの遅延初期化"""
if self._client is None:
self._client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=httpx.Timeout(
connect=self.timeout.connect_timeout,
read=self.timeout.read_timeout,
pool=self.timeout.pool_timeout
),
limits=httpx.Limits(
max_keepalive_connections=20,
max_connections=100
)
)
return self._client
async def chat_completion(
self,
model: str,
messages: list,
max_tokens: int = 1000,
temperature: float = 0.7
) -> dict:
"""
HolySheep AI Chat Completion API呼び出し
Args:
model: モデル名（gpt-4o, claude-3-opus, deepseek-v3等）
messages: メッセージ履歴
max_tokens: 最大出力トークン数
temperature: 生成多様性パラメータ
Returns:
APIレスポンス辞書
Raises:
httpx.TimeoutException: タイムアウト発生時
httpx.HTTPStatusError: HTTPエラー時
"""
client = await self._get_client()
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
try:
response = await client.post("/chat/completions", json=payload)
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
logger.error(f"タイムアウト発生 - model:{model}, type:{type(e).__name__}")
raise
except httpx.HTTPStatusError as e:
logger.error(f"HTTPエラー - status:{e.response.status_code}")
raise
finally:
await self._cleanup()
async def _cleanup(self):
"""リソースクリーンアップ"""
if self._client:
await self._client.aclose()
self._client = None指数バックオフ付きリトライ戦略の実装
タイムアウト後のリトライ処理では、指数バックオフ（Exponential Backoff）とジッター（Jitter）を組み合わせた戦略が重要です。無制御なリトライは以下を招きます：
- HolySheep AIのレート制限（Rate Limit）への抵触
- コストの爆発的増加（特にGPT-4.1 $8/MTokの高コストモデル）
- サーバー負荷の更なる増大
以下の実装では、最大3回のリトライを指数バックオフ＋完全ジッターで行うています：
import asyncio
import random
from typing import TypeVar, Callable, Optional
from functools import wraps
import httpx
import logging
from dataclasses import dataclass
from enum import Enum
logger = logging.getLogger(__name__)
T = TypeVar('T')
class RetryStrategy(Enum):
"""リトライ戦略の選択肢"""
EXPONENTIAL_BACKOFF = "exponential"
LINEAR_BACKOFF = "linear"
FIBONACCI_BACKOFF = "fibonacci"
@dataclass
class RetryConfig:
"""リトライ設定パラメータ"""
max_retries: int = 3
base_delay: float = 1.0 # 基本待機秒数
max_delay: float = 30.0 # 最大待機秒数
exponential_base: float = 2.0 # 指数バックオフの底
jitter_factor: float = 1.0 # ジッター係数（1.0=完全ジッター）
retryable_status_codes: tuple = (429, 500, 502, 503, 504)
class HolySheepRetryClient:
"""
HolySheep AI API 用リトライ機能付きクライアント
指数バックオフ＋ジッターでサーバー負荷を最小化
"""
def __init__(
self,
api_key: str,
retry_config: Optional[RetryConfig] = None,
timeout_config: Optional[TimeoutConfig] = None
):
self.base_client =