AI APIを本番運用していると、必ず遭遇するのが一時的なエラーです。私のプロジェクトでは、夜間のバッチ処理で突如 ConnectionError: timeout が頻発し、サービスの可用性が大きく低下した経験があります。本稿では、HolySheep AI を含むAI API呼び出しにおいて可用性を最大化するリトライ戦略を、実体験に基づき詳細に解説します。
リトライ戦略の基礎：なぜbackoffが必要か
AI API呼び出しでエラーが発生した際、即座に再リクエストを送ると以下のような悪影響が生じます：
- レートリミットの超過：短時間的大量リクエストにより429エラーが連鎖
- サーバー負荷の雪だるま式増加：障害中のサーバーに更なる負荷をかける
- コストの無駄遣い：成功確率の低いリクエストでAPI creditsを消費
Backoff（待機）戦略は、指数関数的または線形的に再試行の間隔を広げることで、サーバーへの負荷を最小限に抑えつつ、成功概率を高めます。
Linear Backoff：均等間隔リトライ
Linear Backoffは、每次リトライ時に固定の時間間隔を加算する方式です。
import time
import requests
def linear_backoff_request(
url: str,
headers: dict,
data: dict,
max_retries: int = 5,
base_delay: float = 1.0,
delay_increment: float = 2.0
) -> dict:
"""
Linear Backoff 実装
リトライ間隔: base_delay, base_delay+increment, base_delay+2*increment...
"""
for attempt in range(max_retries):
try:
response = requests.post(
f"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=data,
timeout=30
)
if response.status_code == 200:
return response.json()
# 4xxエラーはリトライしない（客户端エラー）
if 400 <= response.status_code < 500:
print(f"クライアントエラー {response.status_code}: リトライ中止")
return None
except requests.exceptions.Timeout:
print(f"試行 {attempt + 1}/{max_retries}: タイムアウト")
except requests.exceptions.ConnectionError as e:
print(f"試行 {attempt + 1}/{max_retries}: 接続エラー - {e}")
if attempt < max_retries - 1:
# Linear: base_delay + (attempt * increment)
wait_time = base_delay + (attempt * delay_increment)
print(f"{wait_time}秒後にリトライ...")
time.sleep(wait_time)
print("最大リトライ回数に達しました")
return None
使用例
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hello!"}],
"max_tokens": 100
}
result = linear_backoff_request(
url="https://api.holysheep.ai/v1/chat/completions",
headers=headers,
data=payload
)
print(result)Linear Backoffの特徴：
- 間隔パターン：1秒、3秒、5秒、7秒、9秒（base=1, increment=2の場合）
- 予測可能性：次のリトライ時間が明確で、ロギングや監視が容易
- 欠点：サーバー負荷の回復に時間がかからない場合、冗長な待機が発生
Exponential Backoff：指数関数的リトライ
Exponential Backoffは、失敗の度に待機時間を指数関数的に増加させる方式です。
import time
import random
import requests
from typing import Optional
def exponential_backoff_request(
prompt: str,
model: str = "gpt-4.1",
max_retries: int = 6,
base_delay: float = 1.0,
max_delay: float = 64.0,
jitter: bool = True
) -> Optional[dict]:
"""
Exponential Backoff with Jitter 実装
待機時間計算: min(max_delay, base_delay * (2 ** attempt)) + random_jitter
間隔パターン例: 1-2秒, 2-4秒, 4-8秒, 8-16秒, 16-32秒, 32-64秒
"""
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500
}
for attempt in range(max_retries):
try:
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()
# 429 Too Many Requests
if response.status_code == 429:
retry_after = response.headers.get('Retry-After')
if retry_after:
wait_time = float(retry_after)
else:
wait_time = min(max_delay, base_delay * (2 ** attempt))
if jitter:
wait_time += random.uniform(0, 1) # フルジッター
print(f"レートリミット到達。{wait_time:.2f}秒後にリトライ...")
time.sleep(wait_time)
continue
# 401 Unauthorized
if response.status_code == 401:
print("認証エラー: APIキーを確認してください")
return None
# 500番台サーバーエラーはリトライ対象
if 500 <= response.status_code < 600:
wait_time = min(max_delay, base_delay * (2 ** attempt))
if jitter:
wait_time += random.uniform(0, wait_time * 0.1) # частичнаяジッター
print(f"サーバーエラー {response.status_code}。{wait_time:.2f}秒後にリトライ...")
time.sleep(wait_time)
else:
print(f"予期しないエラー {response.status_code}")
return None
except requests.exceptions.Timeout:
wait_time = min(max_delay, base_delay * (2 ** attempt))
if jitter:
wait_time += random.uniform(0, 1)
print(f"試行 {attempt + 1}/{max_retries}: タイムアウト。{wait_time:.2f}秒後にリトライ...")
time.sleep(wait_time)
except requests.exceptions.ConnectionError as e:
print(f"試行 {attempt + 1}/{max_retries}: 接続エラー - {str(e)[:50]}")
if attempt < max_retries - 1:
wait_time = min(max_delay, base_delay * (2 ** attempt))
time.sleep(wait_time)
print("最大