ほとんどのAIエージェントのチュートリアルはハッピーパスを示しています。あなたのエージェントはLLMを呼び出し、応答を受け取り、やるべきことを実行します。出荷しましょう。
そして本番がやってきます。レート制限。タイムアウト。壊れた(不正な形式の)応答。コンテキストウィンドウのオーバーフロー。あなたのエージェントは、約48時間で「デモ用に準備できた」から「インシデント発生中」に変わります。
私は小さな運用をしています――最大5つのエージェント、創業者は私ひとりです。午前3時に起こされる失敗は、すべてコードで事前に扱うべきだったものです。実際に機能するパターンを紹介します。
エラーをまず分類する
すべてのエラーが同じ扱いを必要とするわけではありません。私があらゆるエージェントシステムで最初にやることは、失敗を2つのバケットに分類することです:
一時的なエラー:レート制限(429)、タイムアウト、一時的なネットワークのちょっとした不調、モデルの過負荷。もう一度試せばたぶん動きます。
恒久的なエラー:不正なAPIキー、壊れた(不正な形式の)プロンプト、コンテキストウィンドウの超過、モデルが存在しない。リトライしても役に立ちません。
class ErrorClassifier:
TRANSIENT_CODES = {429, 500, 502, 503, 504}
@staticmethod
def classify(error):
if hasattr(error, 'status_code'):
if error.status_code in ErrorClassifier.TRANSIENT_CODES:
return "transient"
if "timeout" in str(error).lower():
return "transient"
return "permanent"
この分類が、下流のあらゆる処理を決めます。一時的なエラーにはリトライを行い、恒久的なエラーにはログ出力し、レポートし、適切にフォールバック(段階的に機能を落とす)します。エージェントのセキュリティのパターンを考えるときも、エラー分類は重要です――恒久的な認証エラーは、一時的なネットワークのちょっとした不調とは別のアラートが必要だからです。
悪化させないリトライ戦略
素朴なアプローチ――即リトライして、永遠にリトライする――は、レート制限を「BAN」へ変えるやり方です。ジッター付きの指数バックオフがベースラインです:
import random
import time
def retry_with_backoff(fn, max_retries=3, base_delay=1.0):
for attempt in range(max_retries):
try:
return fn()
except Exception as e:
if ErrorClassifier.classify(e) == "permanent":
raise # 恒久的なエラーはリトライしない
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
jitter = random.uniform(0, delay * 0.5)
time.sleep(delay + jitter)
重要なポイント:複数のエージェントが同じ制限に同時にぶつかったときの「大群効果(thundering herd)」を、ジッターが防いでくれます。そしてリトライ回数には必ず上限を設けてください――通常は3回で十分です。3回でダメなら30回でもダメです。
LLM呼び出しにサーキットブレーカーを
リトライは個別の失敗を処理します。サーキットブレーカーはシステム的な問題を処理します。もしLLMプロバイダの調子が悪いなら、すべてのリクエストをキューに積んでタイムアウトさせたいわけではありません。
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_time=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_time = recovery_time
self.last_failure_time = None
self.state = "closed" # closed = normal, open = blocking
def call(self, fn):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_time:
self.state = "half-open"
else:
raise CircuitOpenError("Circuit breaker is open")
try:
result = fn()
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
raise
私は外部のLLM呼び出しをすべてサーキットブレーカーでラップしています。サーキットがオープンになったら、エージェントは失敗を積み重ねる代わりに、キャッシュされたレスポンスやよりシンプルなロジックへフォールバックします。observability-first approach を採用しているなら、サーキットの状態遷移を追跡したくなるはずです。これは最良の早期警告サインの1つです。
Fallback Chains: Your Safety Net
プライマリモデルが失敗したとき、フォールバックチェーンがあれば完全な停止(アウトage)を防げます:
FALLBACK_CHAIN = [
{"provider": "anthropic", "model": "claude-sonnet-4-20250514"},
{"provider": "openai", "model": "gpt-4o-mini"},
{"provider": "local", "model": "cached_response"},
]
返却形式: {"translated": "翻訳されたHTML"}def call_with_fallback(prompt, chain=FALLBACK_CHAIN):
errors = []
for option in chain:
try:
return call_model(option["provider"], option["model"], prompt)
except Exception as e:
errors.append(f"{option['provider']}: {e}")
continue
raise AllProvidersFailedError(
f"All {len(chain)} providers failed: {'; '.join(errors)}"
)
チェーンは優雅に劣化します。プレミアムモデル → より安価なモデル → キャッシュ/静的レスポンス。すべてが燃え尽きている状況でも、ユーザーには何かを返せます。
Timeout Handling
LLM呼び出しは遅いです。120秒待っても応答が返ってこないエージェントは、リソースを無駄に消費し、下流の作業をブロックしてしまいます。
import asyncio
async def call_with_timeout(coro, timeout_seconds=30):
try:
return await asyncio.wait_for(coro, timeout=timeout_seconds)
except asyncio.TimeoutError:
raise TimeoutError(f"LLM call exceeded {timeout_seconds}s limit")
タイムアウトは厳しめに設定します。ほとんどのエージェントタスクでは、30秒以内に応答がなければ何かがおかしいはずです。完了(completions)についてはデフォルトで30秒、埋め込み(embeddings)については10秒にしています。
Putting It All Together
これらのパターンが実際のエージェントでどのように組み合わさるかを示します。
async def agent_execute(task):
breaker = get_circuit_breaker("llm_calls")
try:
result = breaker.call(
lambda: retry_with_backoff(
lambda: call_with_fallback(task.prompt),
max_retries=3
)
)
return AgentResult(status="success", data=result)
except CircuitOpenError:
return AgentResult(
status="degraded",
data=get_cached_response(task),
note="Using cached response - LLM circuit open"
)
except AllProvidersFailedError:
return AgentResult(
status="failed",
data=None,
note="All providers unavailable"
)
重要な洞察は、あらゆる層に「定義された失敗モード」があることです。タイムアウトはハングを防ぎます。リトライは一時的な不具合を処理します。サーキットブレーカーは連鎖的な障害を防ぎます。フォールバックは、劣化しているものの機能するレスポンスを提供します。
What I Track
エラーハンドリングは、それが機能していると分かっている場合にのみ役に立ちます。私の小規模なセットアップでは、以下を追跡しています。
- エラー分類の分布 — 一時的なエラーが増えているのか、それとも恒久的なエラーが増えているのか?
- サーキットブレーカーの状態変化 — 回路はどれくらいの頻度で開きますか?
- フォールバックチェーンの深さ — リクエストはチェーンのどこまで降りていきますか?
- リトライ成功率 — リトライは実際にエラーを回復させていますか?
リアルタイムのエラーモニタリング が、エージェントの作り方を変えました。ユーザーから失敗を知るのではなく、障害になる前にパターンを捕まえるのです。
退屈だけど本当の話
これらのパターンはどれも新しいものではありません。サーキットブレーカーは分散システムから来ています。バックオフ付きのリトライは、私たちの多くよりずっと前からあります。フォールバックチェーンは、単に別名でのフェイルオーバーです。
しかし、これらを特にAIエージェントに適用するのがポイントです。失敗は確率的で、レスポンスは非決定的で、リトライをするたびにコストが積み上がります。そこに職人技があります。まずエラー分類から始め、リトライを重ね、サーキットブレーカーを追加し、フォールバックチェーンを構築してください。あなたの午前3時の自分がきっと感謝します。



