AIアプリケーションを運用していると、必ずと言っていいほど遭遇するのが 429 Too Many Requests エラーです。本稿では、主要AIプロバイダーのレート制限を比較し、HolySheep AIへの移行によるコスト最適化と安定稼働の実現方法を実践的に解説します。
429エラーの本質的理解
429ステータスコードは、API提供側のリソース保護 mechanism が発動した証です。主な原因は以下の3つに分類されます：
- Request Rate Limit：1秒あたりのリクエスト数超過
- Token Limit：1分/1時間あたりのトークン消費量超過
- Concurrent Connection Limit：同時接続数の上限到達
私は以前、深夜のバッチ処理で429連発により処理が12時間遅延し、ビジネス損失を招いた経験があります。そんな事態を未然に防ぐには、各プロバイダーの制限を正確に把握し、適切なリトライロジックを実装することが不可欠です。
主要AI APIのレート制限比較表
| プロバイダー | 無料枠リミット | 有料枠RPM | 1Mトークン単価 | コスト比率 |
|---|---|---|---|---|
| OpenAI (GPT-4.1) | 100 req/min | 500 RPM | $8.00 | 基準 |
| Anthropic (Claude Sonnet 4) | 50 req/min | 300 RPM | $15.00 | 1.88x |
| Google (Gemini 2.5 Flash) | 15 RPM | 1000 RPM | $2.50 | 0.31x |
| DeepSeek V3.2 | limited | 待望 | $0.42 | 0.05x |
| HolySheep AI | 登録時クレジット | 制限緩和 | DeepSeek同等 | 0.05x |
HolySheep AIはDeepSeek V3.2と同等の価格帯で提供されており、OpenAI公式の約85%安い¥1=$1という為替レートを実現しています。WeChat PayやAlipayにも対応しているため、国内ユーザーにとって格段に調達が容易です。
429エラーの対処パターン別実装コード
パターン1：指数バックオフ＋リトライ
最も一般的に効果的なアプローチがexponential backoffです。HolySheep APIを例に、完全なリトライロジックを実装します。
import time
import httpx
from typing import Optional, Dict, Any
import asyncio
class HolySheepAIClient:
"""
HolySheep AI API クライアント
429エラーに対して指数バックオフで自動リトライ
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url
self.max_retries = 5
self.timeout = 60.0
def _calculate_backoff(self, attempt: int, base_delay: float = 1.0) -> float:
"""
指数バックオフの遅延時間を計算
Jitterを追加して Thundering Herd 問題を回避
"""
import random
exponential_delay = base_delay * (2 ** attempt)
jitter = random.uniform(0, 0.5) # 0〜0.5秒のランダム値
return min(exponential_delay + jitter, 60.0) # 最大60秒
def _is_rate_limit_error(self, status_code: int) -> bool:
"""429エラー判定"""
return status_code == 429
async def chat_completion_with_retry(
self,
messages: list,
model: str = "deepseek-v3.2",
**kwargs
) -> Dict[str, Any]:
"""
429エラー対応のリトライ機能付きチャット完了API
Args:
messages: メッセージリスト
model: モデル名 (deepseek-v3.2, gpt-4.1, claude-sonnet-4等)
**kwargs: 追加パラメータ (temperature, max_tokens等)
Returns:
APIレスポンス辞書
"""
async with httpx.AsyncClient(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=self.timeout
) as client:
for attempt in range(self.max_retries):
try:
response = await client.post(
f"{self.base_url}/chat/completions",
json={
"model": model,
"messages": messages,
**kwargs
}
)
# 成功時
if response.status_code == 200:
return response.json()
# 429 Rate Limit 時の処理
if self._is_rate_limit_error(response.status_code):
retry_after = response.headers.get("Retry-After")
if retry_after and attempt == 0:
# サーバー指定の猶予時間を優先
wait_time = float(retry_after)
print(f"[Rate Limited] Server suggests wait: {wait_time}s")
else:
# 指数バックオフで待機
wait_time = self._calculate_backoff(attempt)
print(
f"[Rate Limited] Attempt {attempt + 1}/{self.max_retries} "
f"- Waiting {wait_time:.2f}s before retry..."
)
await asyncio.sleep(wait_time)
continue
# 429以外のエラー
response.raise_for_status()
except httpx.TimeoutException:
print(f"[Timeout] Attempt {attempt + 1}/{self.max_retries}")
await asyncio.sleep(self._calculate_backoff(attempt))
continue
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
continue
raise
raise RuntimeError(
f"Max retries ({self.max_retries}) exceeded for API call"
)
使用例
async def main():
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = await client.chat_completion_with_retry(
messages=[
{"role": "system", "content": "あなたは помощникです。"},
{"role": "user", "content": "こんにちは"}
],
model="deepseek-v3.2",
temperature=0.7,
max_tokens=500
)
print(f"Success: {result['choices'][0]['message']['content']}")
except Exception as e:
print(f"Failed after retries: {e}")
if __name__ == "__main__":
asyncio.run(main())
パターン2：トークン消費量の監視とバッチ処理
1分あたりのトークン消費量で制限される場合、滑动窗口方式で流量を制御します。
import time
from collections import deque
from dataclasses import dataclass
from typing import Optional
import threading
@dataclass
class TokenBucket:
"""
トークンバケット方式でAPI呼び出しを流量制御
429エラーを事前に防止
"""
capacity: int # バケット容量
refill_rate: float # 毎秒補充されるトークン数
def __post_init__(self):
self._tokens = float(self.capacity)
self._last_refill = time.time()
self._lock = threading.Lock()
def _refill(self):
"""時間経過でトークン補充"""
now = time.time()
elapsed = now - self._last_refill
self._tokens = min(
self.capacity,
self._tokens + elapsed * self.refill_rate
)
self._last_refill = now
def consume(self, tokens: int, timeout: float = 60.0) -> bool:
"""
トークンを消費しようとする
Args:
tokens: 消費したいトークン数
timeout: 最大待機時間（秒）
Returns:
消費成功=True, タイムアウト=False
"""
start_time = time.time()
while True:
with self._lock:
self._refill()
if self._tokens >= tokens:
self._tokens -= tokens
return True
# 満タンになるまでの時間を計算
deficit = tokens - self._tokens
wait_time = deficit / self.refill_rate
if time.time() - start_time + wait_time > timeout:
return False
# 少し待ってから再試行
time.sleep(0.1)
class RateLimitedAPIClient:
"""
トークンバケットによる流量制御付きAPIクライアント
HolySheep AI向け
"""
def __init__(
self,
api_key: str,
rpm_limit: int = 60, # 1分あたりのリクエスト数
tpm_limit: int = 100000, # 1分あたりのトークン数
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url
# 1分あたりに換算した補充レート
self.request_bucket = TokenBucket(
capacity=rpm_limit,
refill_rate=rpm_limit / 60.0
)
self.token_bucket = TokenBucket(
capacity=tpm_limit,
refill_rate=tpm_limit / 60.0
)
def _estimate_tokens(self, messages: list, model: str) -> int:
"""
トークン消費量の概算
簡易計算式（実際はトークナイザー使用を推奨）
"""
total_chars = sum(len(m.get("content", "")) for m in messages)
# 日本語は1文字≈1.5トークン相当
return int(total_chars * 1.5 + 100) # +100はシステムオーバーヘッド
def call_with_limit(
self,
messages: list,
model: str = "deepseek-v3.2",
**kwargs
) -> dict:
"""
流量制御付きでAPI呼び出し
制限超過の場合はブロックして待機
"""
estimated_tokens = self._estimate_tokens(messages, model)
# トークンバケット消費（60秒タイムアウト）
if not self.token_bucket.consume(estimated_tokens, timeout=60.0):
raise TimeoutError(
f"Token limit: Could not acquire {estimated_tokens} tokens "
f"within 60 seconds"
)
# リクエストバケット消費
if not self.request_bucket.consume(1, timeout=60.0):
raise TimeoutError("Request rate limit: too many requests")
# HolySheep API呼び出し
import httpx
with httpx.Client(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=60.0
) as client:
response = client.post(
f"{self.base_url}/chat/completions",
json={
"model": model,
"messages": messages,
**kwargs
}
)
if response.status_code == 429:
# バケットにトークンを返す（リトライ時に再消費）
with self.token_bucket._lock:
self.token_bucket._tokens += estimated_tokens
with self.request_bucket._lock:
self.request_bucket._tokens += 1
raise RuntimeError("API returned 429 after rate limit wait")
response.raise_for_status()
return response.json()
使用例
def batch_process_queries(queries: list[str]):
"""
複数のクエリをレート制限を守りながら処理
"""
client = RateLimitedAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
rpm_limit=30, # 1分30リクエストに制限
tpm_limit=50000 # 1分50000トークン
)
results = []
for i, query in enumerate(queries):
print(f"Processing {i+1}/{len(queries)}...")
try:
result = client.call_with_limit(
messages=[
{"role": "user", "content": query}
],
model="deepseek-v3.2",
max_tokens=1000
)
results.append(result)
except Exception as e:
print(f"Error processing query {i+1}: {e}")
results.append(None)
return results
HolySheep AIへの移行プレイブック
移行メリットの整理
HolySheep AIへの移行を検討する理由は明白です：
- コスト削減：¥1=$1という破格のレートで、公式¥7.3=$1比85%OFF
- 対応モデルの幅広さ：DeepSeek V3.2 ($0.42/MTok)、Gemini 2.5 Flash ($2.50)、GPT-4.1 ($8.00)、Claude Sonnet 4.5 ($15.00)
- 決済の容易さ：WeChat Pay・Alipay対応で国内調達が簡単
- 低レイテンシ：<50msの応答速度でリアルタイム処理も快適
- 制限の緩さ：有料プランではレート制限が大幅に緩和
- 無料クレジット：登録だけで実際に使えるクレジットが付与
移行手順（Step-by-Step）
Step 1：現在の使用量分析
# 現在のAPI使用量を分析するスクリプト
import json
from collections import defaultdict
def analyze_current_usage(log_file: str) -> dict:
"""
既存のAPIログファイルを分析して月次コスト試算
"""
usage_stats = defaultdict(lambda: {
"request_count": 0,
"total_tokens": 0,
"error_count": 0,
"cost_usd": 0
})
with open(log_file, 'r') as f:
for line in f:
entry = json.loads(line)
model = entry.get('model', 'unknown')
usage_stats[model]["request_count"] += 1
usage_stats[model]["total_tokens"] += entry.get(
'tokens_used', 0
)
usage_stats[model]["error_count"] += entry.get(
'errors', 0
)
# コスト計算（公式レート）
official_prices = {
"gpt-4": 30.0, # $30/MTok
"gpt-4-turbo": 10.0,
"gpt-3.5-turbo": 2.0,
"claude-3-sonnet": 15.0,
"claude-3-opus": 75.0,
"gemini-pro": 3.5,
}
# HolySheep相当品のコスト試算
holy_rate = 1.0 # ¥1 = $1 (DeepSeek V3.2水準)
report = {}
for model, stats in usage_stats.items():
mtok = stats["total_tokens"] / 1_000_000
official_cost = mtok * official_prices.get(model, 10.0)
holy_cost = mtok * holy_rate
report[model] = {
**stats,
"official_monthly_cost_usd": official_cost,
"holy_monthly_cost_usd": holy_cost,
"savings_percent": (
(official_cost - holy_cost) / official_cost * 100
if official_cost > 0 else 0
)
}
return report
出力例
{
"gpt-4": {
"request_count": 15000,
"total_tokens": 500_000_000,
"official_monthly_cost_usd": 15000.0,
"holy_monthly_cost_usd": 500.0,
"savings_percent": 96.7
}
}
Step 2：HolySheepクライアントへの切り替え
# adapter_pattern.py
既存コードをHolySheepに=swapするAdapterパターン
class BaseLLMAdapter:
"""LLM API抽象基底クラス"""
def chat(self, messages: list, model: str, **kwargs) -> str:
raise NotImplementedError
class OpenAIAdapter(BaseLLMAdapter):
"""既存のOpenAI向けコード"""
def __init__(self, api_key: str):
import openai
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.openai.com/v1" # 移行後に削除
)
def chat(self, messages: list, model: str, **kwargs) -> str:
response = self.client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
return response.choices[0].message.content
class HolySheepAdapter(BaseLLMAdapter):
"""
HolySheep AI向けAdapter
既存のOpenAIAdapterと同一インターフェース
"""
def __init__(self, api_key: str):
import httpx
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.Client(
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=60.0
)
def chat(self, messages: list, model: str, **kwargs) -> str:
# model名のマッピング
model_map = {
"gpt-4": "gpt-4.1",
"gpt-4-turbo": "gpt-4.1",
"gpt-3.5-turbo": "deepseek-v3.2",
"claude-3-sonnet": "claude-sonnet-4",
"gemini-pro": "gemini-2.5-flash"
}
holy_model = model_map.get(model, model)
response = self.client.post(
f"{self.base_url}/chat/completions",
json={
"model": holy_model,
"messages": messages,
**kwargs
}
)
if response.status_code == 429:
raise RateLimitError("HolySheep rate limit exceeded")
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
def close(self):
self.client.close()
class LLMClientFactory:
"""Provider切替用Factory"""
_providers = {
"openai": OpenAIAdapter,
"holysheep": HolySheepAdapter,
}
@classmethod
def create(
cls,
provider: str = "openai",
**kwargs
) -> BaseLLMAdapter:
"""
利用Providerに応じて適切なAdapterを生成
Args:
provider: "openai" | "holysheep"
"""
adapter_class = cls._providers.get(provider)
if not adapter_class:
raise ValueError(f"Unknown provider: {provider}")
return adapter_class(**kwargs)
использование
def migrate_to_holysheep():
"""
移行の実行例
"""
# Before: 古いコード
# client = OpenAIAdapter(api_key="old-key")
# After: HolySheepへ切り替え（1行変更）
client = LLMClientFactory.create(
provider="holysheep",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
response = client.chat(
messages=[
{"role": "user", "content": "こんにちは"}
],
model="gpt-4", # 自動的にHolySheepのgpt-4.1にマッピング
temperature=0.7
)
print(response)
client.close()
Step 3：並行稼働テスト（Canary Deployment）
全トラフィックを即座に移行するのではなく、A/Bテスト的に少しずつHolySheepへ流すことをおすすめします。
リスク管理与ロールバック計画
| リスク | 発生確率 | 影響度 | 対策 |
|---|---|---|---|
| レスポンスフォーマットの差異 | 中 | 中 | Adapterパターンで吸収、統合テスト実施 |
| モデルの出力品質差 | 中 | 高 | Golden Set比較テスト、段階的移行 |
| 利用不可/障害 | 低 | 高 | マルチプロバイダー構成、Fallback実装 |
| 突発的な制限変更 | 低 | 中 | 月次監視、制限アラート設定 |
# rollback_manager.py
自動ロールバック機構
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import logging
logger = logging.getLogger(__name__)
class HealthStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
CRITICAL = "critical"
@dataclass
class HealthMetrics:
error_rate: float # エラー率 (0.0 - 1.0)
avg_latency_ms: float
timeout_rate: float
class AutomaticRollbackManager:
"""
HolySheep⇔旧API間の自動フェイルオーバー
問題発生時に即座に旧APIへ切り戻し
"""
def __init__(
self,
primary_adapter, # HolySheep
fallback_adapter, # 旧API
health_check_interval: int = 60,
error_threshold: float = 0.05, # 5%エラー率で要注意
latency_threshold_ms: float = 2000, # 2秒以上で要注意
):
self.primary = primary_adapter
self.fallback = fallback_adapter
self.health_check_interval = health_check_interval
self.error_threshold = error_threshold
self.latency_threshold_ms = latency_threshold_ms
self._is_primary_active = True
self._metrics_history: list[HealthMetrics] = []
def _get_current_metrics(self) -> HealthMetrics:
"""直近のメトリクスを取得（実際はモニタリングシステムから取得）"""
# ダミーデータ - 実際はPrometheus等から取得
return HealthMetrics(
error_rate=0.02,
avg_latency_ms=150.0,
timeout_rate=0.005
)
def _evaluate_health(self, metrics: HealthMetrics) -> HealthStatus:
"""健常性評価"""
if metrics.error_rate > 0.1:
return HealthStatus.CRITICAL
elif metrics.error_rate > self.error_threshold:
return HealthStatus.DEGRADED
elif metrics.avg_latency_ms > self.latency_threshold_ms:
return HealthStatus.DEGRADED
return HealthStatus.HEALTHY
def _execute_rollback(self):
"""ロールバック実行"""
logger.warning("🚨 EXECUTING ROLLBACK - Switching to fallback")
self._is_primary_active = False
def _execute_recovery(self):
"""プライマリ回復"""
logger.info("✅ PRIMARY RECOVERED - Switching back to HolySheep")
self._is_primary_active = True
def call(self, messages: list, model: str, **kwargs) -> str:
"""
健常性チェック付きAPI呼び出し
自動ロールバック機能を内包
"""
# 健常性チェック
metrics = self._get_current_metrics()
self._metrics_history.append(metrics)
# 直近10件のみ保持
self._metrics_history = self._metrics_history[-10:]
health = self._evaluate_health(metrics)
# критическое状態で自動ロールバック
if health == HealthStatus.CRITICAL and self._is_primary_active:
self._execute_rollback()
# 回復したら元に戻す
if health == HealthStatus.HEALTHY and not self._is_primary_active:
# 5分連続でhealthyなら回復
recent_healthy = all(
self._evaluate_health(m) == HealthStatus.HEALTHY
for m in self._metrics_history[-5:]
)
if recent_healthy:
self._execute_recovery()
# アクティブなAdapterで実行
active_adapter = (
self.primary if self._is_primary_active
else self.fallback
)
try:
result = active_adapter.chat(messages, model, **kwargs)
# 成功時elemetry
logger.info(
f"Success via {'primary' if self._is_primary_active else 'fallback'}"
)
return result
except Exception as e:
logger.error(f"Error: {e}")
# フォールバック先が可用なら試行
if not self._is_primary_active:
raise
# フォールバックへ切り替え
logger.warning("Primary failed, trying fallback...")
self._execute_rollback()
return self.fallback.chat(messages, model, **kwargs)
ROI試算
実際のケーススタディでROIを見てみましょう。
前提条件：
- 月間リクエスト数：500,000回
- 平均トークン数/リクエスト：2,000
- 使用モデル：GPT-4（50%）、GPT-3.5-turbo（50%）
月次コスト比較：
| プロバイダー | 計算式 | 月額コスト |
|---|---|---|
| OpenAI公式 | (250K×$0.06 + 250K×$0.004)/1M×2K | $8,000 |
| HolySheep AI | 1Bトークン × ¥1 = ¥1,000 | ¥1,000（$1,000相当） |
| 節約額 | $7,000/月（87.5%OFF） |
年間では$84,000のコスト削減が可能です。この節約分で追加機能開発やインフラ強化に投資できます。
よくあるエラーと対処法
エラー1：「429: Rate limit exceeded for model」
# 具体的なエラーメッセージ例
{
"error": {
"message": "Rate limit exceeded for model deepseek-v3.2.
Current limit: 60 requests per minute.",
"type": "rate_limit_exceeded",
"code": "rate_limit_exceeded"
}
}
解決策1: リクエスト間隔の制御
import time
def rate_limited_request(client, messages, delay=1.0):
"""1秒間隔でリクエストを送信"""
time.sleep(delay) # RPM制限を考慮した待機
return client.chat(messages)
解決策2: Rush Hour回避スケジューリング
def smart_schedule():
"""
トラフィックが少ない時間帯（午前6-9時、午後6-10時以外）
にバッチ処理を実行
"""
from datetime import datetime
current_hour = datetime.now().hour
if current_hour < 6 or current_hour > 22:
print("Off-peak hours - proceed with batch")
return True
else:
print("Peak hours - skip or reduce batch size")
return False
エラー2：「401: Invalid authentication」
# エラー例
{
"error": {
"message": "Incorrect API key provided.
You passed: sk-xxxx...
Did you mean to use YOUR_HOLYSHEEP_API_KEY?",
"type": "invalid_request_error",
"code": "invalid_api_key"
}
}
解決策: 環境変数からの安全な読み込み
import os
from dotenv import load_dotenv
load_dotenv() # .envファイルから読み込み
def get_holysheep_client():
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY is not set. "
"Get your key from: https://www.holysheep.ai/register"
)
if api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError(
"Please replace 'YOUR_HOLYSHEEP_API_KEY' with your actual key"
)
return HolySheepAIClient(api_key=api_key)
.envファイル例
HOLYSHEEP_API_KEY=hs_live_xxxxxxxxxxxxxxxxxxxxxx
エラー3：「400: Bad Request - Invalid parameter」
# エラー例
{
"error": {
"message": "temperature must be between 0 and 2,
got 3.5",
"type": "invalid_request_error",
"param": "temperature",
"code": "param_invalid_range"
}
}
解決策: パラメータバリデーション
from typing import Optional
class ParameterValidator:
"""HolySheep API用パラメータバリデーター"""
VALID_RANGES = {
"temperature": (0.0, 2.0),
"max_tokens": (1, 32000),
"top_p": (0.0, 1.0),
"presence_penalty": (-2.0, 2.0),
"frequency_penalty": (-2.0, 2.0),
}
VALID_MODELS = [
"deepseek-v3.2",
"gpt-4.1",
"gpt-4.1-mini",
"claude-sonnet-4",
"gemini-2.5-flash",
]
@classmethod
def validate(cls, **kwargs) -> dict:
"""全てのパラメータをバリデート"""
validated = {}
for key, value in kwargs.items():
if key in cls.VALID_RANGES:
min_val, max_val = cls.VALID_RANGES[key]
if value is not None and not (min_val <= value <= max_val):
raise ValueError(
f"{key} must be between {min_val} and {max_val}, "
f"got {value}"
)
validated[key] = value
# モデル名のバリデーション
if "model" in kwargs and kwargs["model"] not in cls.VALID_MODELS:
raise ValueError(
f"Invalid model: {kwargs['model']}. "
f"Valid models: {', '.join(cls.VALID_MODELS)}"
)
return validated
def safe_chat(client, messages, **kwargs):
"""バリデーション付き безопасный chat呼び出し"""
validated_params = ParameterValidator.validate(**kwargs)
return client.chat(messages=messages, **validated_params)
エラー4：「500: Internal Server Error」
# エラー例
{
"error": {
"message": "An internal error occurred.
Please try again later.",
"type": "internal_error",
"code": "internal_error"
}
}
解決策: リトライ＋サーキットブレーカー
import time
from functools import wraps
class CircuitBreaker:
"""サーキットブレーカーパターン"""
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failures = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def call(self, func, *args, **kwargs):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "HALF_OPEN"
else:
raise Exception("Circuit breaker is OPEN - request blocked")
try:
result = func(*args, **kwargs)
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failures = 0
return result
except Exception as e:
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.failure_threshold:
self.state = "OPEN"
print(f"Circuit breaker OPENED after {self.failures} failures")
raise e
使用例
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
def resilient_call(client, messages):
"""サーキットブレーカー付きのAPI呼び出し"""
return breaker.call(client.chat, messages=messages)
エラー5：モデル不在エラー
# エラー例
{
"error": {
"message": "Model 'gpt-5' not found.
Available models: deepseek-v3.2, gpt-4.1, ...",
"type": "invalid_request_error",
"code": "model_not_found"
}
}
解決策: 利用可能モデル一覧の取得とフォ