LLM APIをアプリケーションから呼び出すとき、単にHTTPリクエストを送るだけでは本番運用で不安が残ります。通信の一時的な失敗、応答の遅延、429 Too Many Requests、プロバイダー側の障害などが発生するためです。
この記事では、Pythonとhttpxを使って、LLM APIクライアントに以下を実装します。
- 接続・読み取りタイムアウト
- 指数バックオフとジッター付きリトライ
-
Retry-Afterへの対応 - クライアント側のレート制限
- 1回の処理に対する全体期限
手元で簡単な失敗ケースを再現しながら確認したところ、特に重要だったのは「何でもリトライしないこと」と「リクエスト単位のタイムアウトとは別に、処理全体の期限を持つこと」でした。
まず失敗を分類する
LLM API呼び出しの全体的な処理と、失敗時の分岐を図にすると次の通りです。
LLM APIのエラーは、すべて同じように扱うべきではありません。
一時的な通信エラーやサーバー側のエラーは、少し待って再試行すると成功する可能性があります。一方、認証エラーや不正なリクエストは、同じ内容を再送しても成功しません。
| 状況 | 代表例 | 基本方針 |
|---|---|---|
| 一時的な通信エラー | 接続失敗、タイムアウト | リトライ候補 |
| サーバー側エラー | 500、502、503、504 | リトライ候補 |
| レート制限 | 429 | 待機後にリトライ |
| 認証エラー | 401、403 | 即時失敗 |
| 入力エラー | 400、422 | 即時失敗 |
ただし、LLM APIの呼び出しが副作用を持つ設計の場合は注意が必要です。同じリクエストを再送すると、課金や外部処理が重複する可能性があります。可能であればリクエストIDや冪等性キーを付与し、再送しても安全な設計にします。
タイムアウトは接続と読み取りを分ける
タイムアウトを設定しないHTTPクライアントは、相手の応答を待ち続ける可能性があります。LLMは通常のAPIより応答に時間がかかるため、短すぎるタイムアウトも問題になります。
私は次のように、接続・書き込み・読み取り・プール待ちを分けて設定しています。
pip install httpx
import httpx
timeout = httpx.Timeout(
connect=5.0, # 接続確立まで
read=60.0, # レスポンス本文を待つ時間
write=10.0, # リクエスト送信
pool=5.0, # コネクションプール待ち
)
client = httpx.Client(timeout=timeout)
LLMでは生成時間が長くなるため、readだけは十分な値にします。ただし、これだけでは「1回の処理に何秒まで許容するか」は制御できません。リトライを追加すると、各リクエストが60秒ずつ待機して処理全体が数分になることもあります。
そこで、アプリケーション側に全体期限を持たせます。
import time
deadline = time.monotonic() + 90.0
if time.monotonic() >= deadline:
raise TimeoutError("処理全体の期限を超えました")
指数バックオフとジッターでリトライする
リトライ間隔を固定値にすると、多数のクライアントが同時に再送する「リトライの集中」が起きます。基本的には指数バックオフを使い、さらにランダムな揺らぎであるジッターを加えます。
例えば、待機時間を次のようにします。
min(最大待機時間, 基本待機時間 × 2^試行回数) + ランダム値
Retry-Afterがレスポンスに含まれている場合は、サービス側の指示を優先します。以下は、429と一部の5xxだけをリトライする実装例です。
import random
import time
from typing import Any
import httpx
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
class LLMClient:
def __init__(
self,
endpoint: str,
api_key: str,
max_retries: int = 3,
) -> None:
self.endpoint = endpoint
self.max_retries = max_retries
self.http = httpx.Client(
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
timeout=httpx.Timeout(
connect=5.0,
read=60.0,
write=10.0,
pool=5.0,
),
)
def request(self, payload: dict[str, Any]) -> dict[str, Any]:
deadline = time.monotonic() + 90.0
last_error: Exception | None = None
for attempt in range(self.max_retries + 1):
if time.monotonic() >= deadline:
raise TimeoutError("LLM API処理の全体期限を超えました")
try:
response = self.http.post(
self.endpoint,
json=payload,
)
if response.status_code < 400:
return response.json()
if response.status_code not in RETRYABLE_STATUS:
response.raise_for_status()
retry_after = response.headers.get("Retry-After")
if retry_after is not None:
try:
wait_seconds = float(retry_after)
except ValueError:
wait_seconds = 1.0
else:
backoff = min(16.0, 0.5 * (2 ** attempt))
wait_seconds = backoff + random.uniform(0.0, 0.5)
remaining = deadline - time.monotonic()
if remaining <= 0:
break
time.sleep(min(wait_seconds, remaining))
except (httpx.TimeoutException, httpx.NetworkError) as exc:
last_error = exc
if attempt >= self.max_retries:
break
backoff = min(16.0, 0.5 * (2 ** attempt))
wait_seconds = backoff + random.uniform(0.0, 0.5)
remaining = deadline - time.monotonic()
if remaining <= 0:
break
time.sleep(min(wait_seconds, remaining))
if last_error is not None:
raise RuntimeError("LLM APIへの接続に失敗しました") from last_error
raise RuntimeError("リトライ回数を超えました")
ここで重要なのは、リトライ回数を無制限にしないことです。私の検証では、最大3回程度に抑え、最終的には呼び出し元へエラーを返すほうが、ユーザー体験と障害調査の両面で扱いやすくなりました。
クライアント側でもレート制限する
サーバーから429が返ってくるまで送信し続けるのではなく、クライアント側でもリクエスト頻度を抑えます。簡単な実装なら、トークンバケット方式が使えます。
import threading
import time
class TokenBucket:
def __init__(self, rate: float, capacity: int) -> None:
self.rate = rate
self.capacity = float(capacity)
self.tokens = float(capacity)
self.updated_at = time.monotonic()
self.lock = threading.Lock()
def acquire(self) -> None:
while True:
with self.lock:
now = time.monotonic()
elapsed = now - self.updated_at
self.updated_at = now
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate,
)
if self.tokens >= 1.0:
self.tokens -= 1.0
return
wait_seconds = (1.0 - self.tokens) / self.rate
time.sleep(wait_seconds)
例えば、毎秒2リクエスト、バーストは最大5件まで許可する場合は次のように作ります。
bucket = TokenBucket(rate=2.0, capacity=5)
bucket.acquire()
result = client.request({
"model": "example-model",
"input": "短い要約を作成してください。",
})
マルチプロセスや複数ホストで動かす場合、このインメモリ実装だけでは全体の送信量を制御できません。その場合は、共有ストレージやキューを使ってレート制限を一元化します。
ログと検証ポイント
リトライを実装したら、成功したかどうかだけでなく、次の情報をログに残します。
- 試行回数
- HTTPステータス
- 待機時間
- 処理全体の経過時間
- リクエストを識別するID
- 最終的な失敗理由
APIキーやプロンプト本文をそのままログに出すのは避けます。特にユーザー入力を含む場合は、必要な長さに切り詰めるか、ハッシュ化して記録します。
簡単な検証では、モックサーバーを使って429、503、タイムアウトを順番に返すケースを作りました。
python -m pytest -q
テストでは、少なくとも以下を確認します。
def test_retry_on_503(mock_server):
mock_server.responses = [503, 503, 200]
result = call_llm()
assert result["status"] == "ok"
assert mock_server.request_count == 3
def test_no_retry_on_401(mock_server):
mock_server.responses = [401]
with pytest.raises(httpx.HTTPStatusError):
call_llm()
assert mock_server.request_count == 1
実運用では、待機時間をそのままテストすると遅くなるため、スリープ関数を差し替えられる設計にしておくとテストしやすくなります。
まとめ
LLM APIの安定性を高めるには、リトライだけを追加するのでは不十分です。
- リトライ対象のエラーを限定する
-
Retry-Afterを尊重する - 指数バックオフとジッターを使う
- 接続・読み取りタイムアウトを設定する
- リトライを含めた全体期限を設ける
- クライアント側でもレート制限する
- 試行回数と待機時間をログに残す
今回の実装をベースに、非同期処理、キュー、共有レートリミッターへ段階的に拡張できます。まずは「無限に待たない」「何でも再送しない」の2点を守るだけでも、LLM APIを呼び出すアプリケーションの挙動はかなり予測しやすくなります。