本番環境で大規模言語モデル（LLM）API を運用する際、突然のバージョン変更、服务停止、レスポンス品質の低下に遭遇することは珍しくありません。私は以前，某社のプロダクションシステムで GPT-4 への突然の仕様変更により毎日凌晨にアラームが鳴り響いていた 경험がありますが，適切なロールバック戦略を実装したことで这些问题を完全に解决しました。
本稿では，今すぐ登録して利用できる HolySheep AI を例に，LLM API のロールバック戦略と実践的な実装方法を解説します。HolySheep AI は ¥1=$1 という業界最安水準の料金体系（公式比85%節約）で，WeChat Pay や Alipay にも対応しており，<50ms の低レイテンシを実現しています。
なぜ API ロールバック戦略が必要なのか
LLM API 運用において，以下のような状況が発生します：
- 新しいモデルバージョンで意図しない出力傾向の変化
- API プロバイダーの突然の仕様変更やデプロイ
- ネットワーク障害による断続的なエラー
- コスト最適化のた目のモデル切り替え
特に2026年現在の市場では，各プロバイダーが频繁にモデルを更新しており，Claude Sonnet 4.5（$15/MTok）から DeepSeek V3.2（$0.42/MTok）まで価格幅が大きいため，状況に応じた柔軟な切り替えが求められます。
実践的なロールバック戦略の実装
1. 基本的なフォールバック機構
まずは，最もシンプルなフォールバックパターンを実装します。API呼び出しに失敗した場合，自動的に备用モデルに切り替えくれます。
import requests
import time
from typing import Optional, Dict, Any
from enum import Enum
class ModelTier(Enum):
PRIMARY = "gpt-4.1"
FALLBACK_1 = "claude-sonnet-4.5"
FALLBACK_2 = "gemini-2.5-flash"
FALLBACK_3 = "deepseek-v3.2"
class LLMRollbackClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model_tiers = [
ModelTier.PRIMARY,
ModelTier.FALLBACK_1,
ModelTier.FALLBACK_2,
ModelTier.FALLBACK_3
]
self.fallback_log = []
def chat_completion_with_fallback(
self,
messages: list,
system_prompt: str = "You are a helpful assistant."
) -> Dict[str, Any]:
"""
フォールバック機能付きのchat completion
Raises:
Exception: 全モデルで失敗した場合
"""
full_messages = [{"role": "system", "content": system_prompt}] + messages
last_error = None
for i, tier in enumerate(self.model_tiers):
try:
response = self._call_api(tier.value, full_messages)
if i > 0:
self.fallback_log.append({
"tier": tier.value,
"attempt": i + 1,
"timestamp": time.time()
})
print(f"⚠️ フォールバック発動: {tier.value} (Attempt {i+1})")
return {
"model": tier.value,
"response": response,
"fallback_used": i > 0
}
except requests.exceptions.Timeout as e:
last_error = f"TimeoutError: {tier.value} - {str(e)}"
print(f"⏱️ タイムアウト: {tier.value}")
continue
except requests.exceptions.ConnectionError as e:
last_error = f"ConnectionError: {tier.value} - {str(e)}"
print(f"🔌 接続エラー: {tier.value}")
continue
except Exception as e:
last_error = f"UnexpectedError: {tier.value} - {str(e)}"
print(f"❌ エラー: {tier.value} - {str(e)}")
continue
raise Exception(f"全モデルの呼び出しに失敗: {last_error}")
def _call_api(self, model: str, messages: list) -> str:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 1000,
"temperature": 0.7
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 401:
raise Exception("401 Unauthorized - API キーが無効です")
elif response.status_code == 429:
raise Exception("429 Rate Limited - レート制限に達しました")
elif response.status_code >= 500:
raise Exception(f"{response.status_code} Server Error")
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
使用例
client = LLMRollbackClient("YOUR_HOLYSHEEP_API_KEY")
result = client.chat_completion_with_fallback([
{"role": "user", "content": "こんにちは，自我介绍してください。"}
])
print(f"使用モデル: {result['model']}")
print(f"フォールバック使用: {result['fallback_used']}")
print(f"レスポンス: {result['response']}")2. 高度なサーキットブレーカーパターン
連続的な失敗を検出した場合，該当モデルへのリクエストを一時的に遮断するサーキットブレーカーパターンを実装します。これにより，障害波及を防ぎ，系统の安定性を向上させます。
import threading
import time
from datetime import datetime, timedelta
from collections import defaultdict
class CircuitBreaker:
"""
サーキットブレーカーパターン実装
States:
CLOSED: 正常稼働，新しいリクエストを許可
OPEN: 障害検出，全リクエストを遮断
HALF_OPEN: 試験的にリクエストを許可
"""
CLOSED = "CLOSED"
OPEN = "OPEN"
HALF_OPEN = "HALF_OPEN"
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
success_threshold: int = 2
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self._state = self.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time = None
self._lock = threading.Lock()
self.stats = defaultdict(lambda: {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"circuit_trips": 0
})
@property
def state(self) -> str:
with self._lock:
if self._state == self.OPEN:
if self._should_attempt_reset():
self._state = self.HALF_OPEN
return self._state
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.recovery_timeout
def record_success(self, model: str):
with self._lock:
self.stats[model]["successful_requests"] += 1
if self._state == self.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = self.CLOSED
self._failure_count = 0
self._success_count = 0
print(f"✅ サーキットブレーカー回復: {model}")
def record_failure(self, model: str):
with self._lock:
self.stats[model]["failed_requests"] += 1
self._failure_count += 1
self._last_failure_time = datetime.now()
if self._state == self.CLOSED:
if self._failure_count >= self.failure_threshold:
self._state = self.OPEN
self.stats[model]["circuit_trips"] += 1
print(f"🚨 サーキットブレーカー遮断: {model}")
elif self._state == self.HALF_OPEN:
self._state = self.OPEN
self._success_count = 0
print(f"🔴 HALF_OPEN → OPEN: {model}")
def can_execute(self) -> bool:
return self.state != self.OPEN
def get_stats(self) -> dict:
return dict(self.stats)
class AdvancedLLMClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.circuit_breakers = {}
self._init_circuit_breakers()
def _init_circuit_breakers(self):
models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
for model in models:
self.circuit_breakers[model] = CircuitBreaker(
failure_threshold=3,
recovery_timeout=30
)
def get_available_model(self) -> Optional[str]:
"""利用可能な最初のモデルを取得"""
priority = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
for model in priority:
if model in self.circuit_breakers:
cb = self.circuit_breakers[model]
if cb.can_execute():
return model
# 全モデル遮断の場合，最も古い遮断から試行
for model in priority:
if model in self.circuit_breakers:
return model
return None
def call_with_protection(self, messages: list) -> dict:
"""保護付きのAPI呼び出し"""
max_attempts = 10
for attempt in range(max_attempts):
model = self.get_available_model()
if model is None:
raise Exception("全モデルが利用不可です")
cb = self.circuit_breakers[model]
if not cb.can_execute():
print(f"⏳ 待機中: {model} (サーキットブレーカー OPEN)")
time.sleep(1)
continue
try:
response = self._make_request(model, messages)
cb.record_success(model)
return {"model": model, "response": response, "attempt": attempt + 1}
except requests.exceptions.Timeout:
print(f"⏱️ タイムアウト: {model}")
cb.record_failure(model)
continue
except requests.exceptions.ConnectionError as e:
print(f"🔌 接続エラー: {model}")
cb.record_failure(model)
continue
except Exception as e:
error_msg = str(e)
if "401" in error_msg:
raise Exception(f"認証エラー: API キーを確認してください - {error_msg}")
if "429" in error_msg:
cb.record_failure(model)
time.sleep(2)
continue
cb.record_failure(model)
continue
raise Exception("最大試行回数に達しました")
def _make_request(self, model: str, messages: list) -> str:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 1500,
"temperature": 0.7
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
def get_health_report(self) -> dict:
"""システム健全性レポート"""
report = {}
for model, cb in self.circuit_breakers.items():
stats = cb.get_stats().get(model, {})
report[model] = {
"circuit_state": cb.state,
"stats": stats
}
return report
使用例
client = AdvancedLLMClient("YOUR_HOLYSHEEP_API_KEY")
try:
result = client.call_with_protection([
{"role": "user", "content": "今日の天気を教えてください。"}
])
print(f"成功: {result['model']} (試行回数: {result['attempt']})")
except Exception as e:
print(f"システムエラー: {str(e)}")
health = client.get_health_report()
print(f"健全性レポート: {health}")3. レート制限を考慮したコスト最適化戦略
HolySheep AI の ¥1=$1 という料金体系を活用し，成本を最適化しながら可用性を維持する戦略を実装します。DeepSeek V3.2 は $0.42/MTok と非常にコスト効率が良いため，適切に活用することで月間コストを大幅に削減できます。
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Optional
import json
@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float # USD
max_tokens_per_minute: int
priority: int
reliability_score: float # 0.0 - 1.0
class CostOptimizedRouter:
"""
コストと可用性のバランスを取る intelligent ルーター
2026年 参考価格 (/MTok output):
- GPT-4.1: $8.00 (高品質・高額)
- Claude Sonnet 4.5: $15.00 (最高品質)
- Gemini 2.5 Flash: $2.50 (バランス型)
- DeepSeek V3.2: $0.42 (最安値)
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.models = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
cost_per_1k_tokens=8.00,
max_tokens_per_minute=500,
priority=1,
reliability_score=0.95
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
cost_per_1k_tokens=15.00,
max_tokens_per_minute=300,
priority=2,
reliability_score=0.92
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
cost_per_1k_tokens=2.50,
max_tokens_per_minute=1000,
priority=3,
reliability_score=0.98
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
cost_per_1k_tokens=0.42,
max_tokens_per_minute=2000,
priority=4,
reliability_score=0.90
)
}
self.usage_stats = {model: {"tokens": 0, "requests": 0, "errors": 0}
for model in self.models}
self.request_timestamps = []
def calculate_score(self, model: ModelConfig, context: dict) -> float:
"""モデルのスコアを計算（高いほど選択されやすい）"""
# 基本コストスコア（安いは加点）
cost_score = 10.0 / model.cost_per_1k_tokens
# 信頼性スコア
reliability = model.reliability_score
# 現在の負荷スコア
current_usage = self.usage_stats[model.name]["tokens"]
load_factor = 1.0 - (current_usage / model.max_tokens_per_minute)
load_factor = max(0.1, load_factor)
# コンテキストに応じた重み付け
quality_needed = context.get("quality_needed", 0.5)
speed_needed = context.get("speed_needed", 0.5)
cost_sensitivity = context.get("cost_sensitivity", 0.5)
# 最終スコア計算
final_score = (
cost_score * cost_sensitivity * 2 +
reliability * 3 +
load_factor * 2 +
(1 / model.cost_per_1k_tokens if quality_needed < 0.3 else 0)
)
return final_score
def select_model(self, context: dict = None) -> str:
"""最適なモデルを選択"""
if context is None:
context = {}
scores = {}
for model_name, model_config in self.models.items():
if self.usage_stats[model_name]["errors"] >= 5:
continue
scores[model_name] = self.calculate_score(model_config, context)
if not scores:
raise Exception("利用可能なモデルがありません")
selected = max(scores.items(), key=lambda x: x[1])
return selected[0]
async def async_call(
self,
messages: list,
context: dict = None,
use_fallback: bool = True
) -> dict:
"""非同期API呼び出し"""
if context is None:
context = {}
model = self.select_model(context)
async def attempt_call(model_name: str, attempt: int) -> dict:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model_name,
"messages": messages,
"max_tokens": 1000,
"temperature": 0.7
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 401:
raise Exception("401 Unauthorized - API キーを確認してください")
elif response.status == 429:
raise Exception("429 Rate Limited")
elif response.status >= 500:
raise Exception(f"{response.status} Server Error")
data = await response.json()
return data["choices"][0]["message"]["content"]
# フォールバックリスト
fallback_order = ["gpt-4.1", "gemini-2.5-flash", "deepseek-v3.2", "claude-sonnet-4.5"]
if model in fallback_order:
fallback_order.remove(model)
fallback_order.insert(0, model)
if not use_fallback:
fallback_order = [model]
last_error = None
for i, model_name in enumerate(fallback_order):
try:
result = await attempt_call(model_name, i)
# 成功統計更新
tokens_used = sum(len(m.get("content", "").split()) * 1.3
for m in messages)
self.usage_stats[model_name]["tokens"] += int(tokens_used)
self.usage_stats[model_name]["requests"] += 1
return {
"model": model_name,
"response": result,
"cost_estimate": self.models[model_name].cost_per_1k_tokens,
"fallback_used": i > 0
}
except Exception as e:
last_error = str(e)
self.usage_stats[model_name]["errors"] += 1
print(f"⚠️ {model_name} エラー: {last_error}")
continue
raise Exception(f"全モデル呼び出し失敗: {last_error}")
def get_cost_report(self) -> dict:
"""コストレポート生成"""
total_cost = 0
total_tokens = 0
report = {}
for model_name, stats in self.usage_stats.items():
model_cost = (stats["tokens"] / 1000) * self.models[model_name].cost_per_1k_tokens
total_cost += model_cost
total_tokens += stats["tokens"]
report[model_name] = {
"tokens_used": stats["tokens"],
"requests": stats["requests"],
"errors": stats["errors"],
"cost_usd": round(model_cost, 4),
"cost_yen": round(model_cost * 7.3, 2) # 2026年参考レート
}
report["summary"] = {
"total_tokens": total_tokens,
"total_cost_usd": round(total_cost, 4),
"total_cost_yen": round(total_cost * 7.3, 2),
"avg_cost_per_1k": round((total_cost / total_tokens * 1000) if total_tokens > 0 else 0, 4)
}
return report
使用例
async def main():
router = CostOptimizedRouter("YOUR_HOLYSHEEP_API_KEY")
# 高品質が必要だがコストも意識
result1 = await router.async_call(
messages=[{"role": "user", "content": "複雑なコードのレビューをお願いします。"}],
context={"quality_needed": 0.9, "cost_sensitivity": 0.3}
)
print(f"高品質リクエスト: {result1['model']}")
# コスト最優先
result2 = await router.async_call(
messages=[{"role": "user", "content": "簡単な質問です。"}],
context={"quality_needed": 0.3, "cost_sensitivity": 0.9}
)
print(f"コスト最適化リクエスト: {result2['model']}")
# コストレポート表示
report = router.get_cost_report()
print(f"\n💰 コストレポート:")
print(f"総コスト: ¥{report['summary']['total_cost_yen']}")
print(f"平均コスト/1Kトークン: ${report['summary']['avg_cost_per_1k']}")
asyncio.run(main())よくあるエラーと対処法
エラー1: ConnectionError: timeout
ネットワークタイムアウト是最も一般的なエラーの一つです。HolySheep AI の <50ms レイテンシでも，网络状况や大規模リクエスト時に発生することがあります。
# 対処方法1: タイムアウト設定の最適化
import requests
def call_with_adaptive_timeout(api_key: str, payload: dict) -> dict:
base_url = "https://api.holysheep.ai/v1"
headers = {"Authorization": f"Bearer {api_key}"}
# ペイロードサイズに応じたタイムアウト設定
input_size = len(str(payload))
if input_size < 1000:
timeout = (10, 30) # (connect_timeout, read_timeout)
elif input_size < 10000:
timeout = (15, 60)
else:
timeout = (30, 120)
try:
response = requests.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload,
timeout=timeout
)
return response.json()
except requests.exceptions.Timeout:
print("タイムアウト発生 - リトライを実行")
# リトライロジック
return call_with_retry(api_key, payload, max_retries=3)
except requests.exceptions.ConnectTimeout:
print("接続タイムアウト - ネットワークを確認")
raise
対処方法2: exponential backoff によるリトライ
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_resilient_session():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return sessionエラー2: 401 Unauthorized
認証エラーは主に API キーの問題で発生します。HolySheep AI では，ダッシュボードから API キーを再生成するвойдите。
# 対処方法: API キー検証と再認証
import os
def validate_and_refresh_api_key(current_key: str) -> str:
"""API キーの有効性を検証し，必要に応じて更新"""
def verify_key(api_key: str) -> bool:
base_url = "https://api.holysheep.ai/v1"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
try:
response = requests.get(
f"{base_url}/models",
headers=headers,
timeout=10
)
return response.status_code == 200
except:
return False
# 現在のキーをテスト
if verify_key(current_key):
return current_key
# 環境変数からの再読み込み
env_key = os.environ.get("HOLYSHEEP_API_KEY")
if env_key and verify_key(env_key):
print("環境変数から新しいAPIキーをロードしました")
return env_key
# 最終手段: デフォルトキーでテスト（実際の運用では非推奨）
default_key = os.environ.get("HOLYSHEEP_API_KEY_FALLBACK")
if default_key and verify_key(default_key):
print("フォールバックAPIキーを使用します")
return default_key
raise Exception(
"API キーが無効です。\n"
"1. https://www.holysheep.ai/register にアクセス\n"
"2. ダッシュボードから新しいAPIキーを生成\n"
"3. 環境変数 HOLYSHEEP_API_KEY を設定"
)
運用環境での使用例
API_KEY = validate_and_refresh_api_key("YOUR_HOLYSHEEP_API_KEY")
print(f"✅ API キー検証成功")エラー3: 429 Rate Limit Exceeded
レート制限は短時間に大量のリクエストを送信した際に発生します。HolySheep AI は高周波アクセスに対応していますが，それでも制限を超えると一時的な遮断されます。
# 対処方法: レート制限対応のトークンバケット実装
import time
import threading
from collections import deque
class TokenBucket:
"""
スレッドセーフなトークンバケットによるレート制限
"""
def __init__(self, rate: float, capacity: int):
"""
Args:
rate: 毎秒補充されるトークン数
capacity: バケットの最大容量
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self.lock = threading.Lock()
self.wait_times = deque(maxlen=100) # 待機時間履歴
def acquire(self, tokens: int = 1, timeout: float = 30) -> bool:
"""トークンを取得、成功まで待機可能"""
start_time = time.time()
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
wait_time = time.time() - start_time
self.wait_times.append(wait_time)
return True
# 次のトークン補充までの時間を計算
needed = tokens - self.tokens
sleep_time = needed / self.rate
if time.time() - start_time + sleep_time > timeout:
return False
time.sleep(min(sleep_time, 0.1))
def _refill(self):
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
def get_stats(self) -> dict:
with self.lock:
avg_wait = sum(self.wait_times) / len(self.wait_times) if self.wait_times else 0
return {
"current_tokens": round(self.tokens, 2),
"avg_wait_time": round(avg_wait, 3),
"utilization": round((1 - self.tokens / self.capacity) * 100, 1)
}
class RateLimitedClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# モデル別のレート制限設定
self.buckets = {
"gpt-4.1": TokenBucket(rate=50, capacity=100), # 毎秒50リクエスト
"claude-sonnet-4.5": TokenBucket(rate=30, capacity=60),
"gemini-2.5-flash": TokenBucket(rate=100, capacity=200),
"deepseek-v3.2": TokenBucket(rate=200, capacity=400)
}
def call_with_rate_limit(
self,
model: str,
payload: dict,
timeout: float = 60
) -> dict:
"""レート制限付きでAPI呼び出し"""
if model not in self.buckets:
model = "deepseek-v3.2" # デフォルト
bucket = self.buckets[model]
if not bucket.acquire(tokens=1, timeout=timeout):
raise Exception(f"レート制限によりタイムアウト: {model}")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 429:
# サーバーサイドのレイト制限 - バックオフ
time.sleep(5)
return self.call_with_rate_limit(model, payload, timeout=timeout)
return response.json()
def get_all_stats(self) -> dict:
return {model: bucket.get_stats() for model, bucket in self.buckets()}
使用例
client = RateLimitedClient("YOUR_HOLYSHEEP_API_KEY")
批量リクエストの処理
for i in range(10):
try:
result = client.call_with_rate_limit(
model="deepseek-v3.2",
payload={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": f"質問 {i}"}]
}
)
print(f"✅ リクエスト {i} 成功")
except Exception as e:
print(f"❌ リクエスト {i} 失敗: {str(e)}")
レート制限状況の確認
stats = client.get_all_stats()
for model, stat in stats.items():
print(f"{model}: 利用率 {stat['utilization']}%, 平均待機 {stat['avg_wait_time']}s")モニタリングとアラート設定
ロールバック戦略の効果を確認するため，持续的なモニタリングとアラートが必要です。
import logging
from datetime import datetime
import json
class LLMOpsMonitor:
"""
LLM API 運用のためのモニタリングシステム
"""
def __init__(self, log_file: str = "llm_operations.log"):
self.log_file = log_file
self.logger = logging.getLogger("LLMOps")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_file)
handler.setFormatter(
logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
)
self.logger.addHandler(handler)
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"fallback_count": 0,
"model_usage": {},
"error_types": {}
}
def log_request(
self,
model: str,
success: bool,
error_type: str = None,
fallback_used: bool = False,
latency_ms: float = None,
tokens_used: int = None
):
"""リクエストの詳細を記録"""
self.metrics["total_requests"] += 1
if success:
self.metrics["successful_requests"] += 1
else:
self.metrics["failed_requests"] += 1
if error_type:
self.metrics["error_types"][error_type] = \
self.metrics["error_types"].get(error_type, 0) + 1
if fallback_used:
self.metrics["fallback_count"] += 1
self.metrics["model_usage"][model] = \
self.metrics["model_usage"].get(model, 0) + 1
log_entry = {
"timestamp": datetime.now().isoformat(),
"model": model,
"success": success,
"error_type": error_type,
"fallback_used": fallback_used,
"latency_ms": latency_ms,
"tokens_used": tokens_used
}
self.logger.info(json.dumps(log_entry))
# アラート条件のチェック
self._check_alerts()
def _check_alerts(self):
"""アラート条件をチェック"""
total = self.metrics["total_requests"]
if total == 0:
return
failure_rate = self.metrics["failed_requests"] / total
fallback_rate = self.metrics["fallback_count"] / total
# 失敗率 > 10% で警告
if failure_rate > 0.1:
self.logger.warning(
f"🚨 異常: 失敗率 {failure_rate*100:.1f}% がしきい値10%を超過"
)
# フォールバック率 > 30% で警告
if fallback_rate > 0.3:
self.logger.warning(
f"⚠️ 注意: フォールバック率 {fallback_rate*100:.1f}% がしきい値30%を超過"
)
# 特定エラーの急増検出
for error_type, count in self.metrics["error_types"].items():
if count > 10:
self.logger.error(
f"🔴 重大: {error_type} が {count} 回発生"
)
def get_summary(self) -> dict:
"""サマリーレポートを取得"""
total = self.metrics["total_requests"]
return {
"total_requests": total,
"success_rate": f"{(self.metrics['successful_requests']/total*100):.2f}%" if total > 0 else "N/A",
"failure_rate": f"{(self.metrics['failed_requests']/total*100):.2f}%" if total > 0 else "N/A",
"fallback_rate": f"{(self.metrics['fallback_count