AIエージェントのエラーハンドリング設計パターン完全ガイド

「なぜかエージェントが途中で止まる」「APIのタイムアウトで全体が落ちた」——こんな経験、ありませんか？

実際に10社以上のAIエージェント導入を支援する中で、プロダクション投入直後に最も多く遭遇する問題が、エラーハンドリングの設計不足です。LLM APIは1〜5%の確率で失敗し、ツール呼び出しはタイムアウトし、外部サービスは予告なく落ちます。これらに対処する設計がなければ、どれほど優秀なエージェントも本番環境では動きません。

この記事では、AIエージェントのエラーハンドリングを3つのレイヤー（リトライ・サーキットブレーカー・グレースフルデグレード）に分けて、Pythonのコピペ可能なコード例とともに解説します。OpenAI Agents SDK、LangChain、素のPythonどれでも応用できる設計パターンです。

AIエージェントの基本構造については、AIエージェント構築完全ガイドで体系的にまとめていますので、あわせてご確認ください。

まず試したい「5分即効」エラー対策3選

エラーハンドリングの全体設計に入る前に、今日すぐ実装できる3つの対策を紹介します。これだけで稼働率が大幅に改善します。

即効テクニック1：Tenacityでexponential backoff付きリトライ

LLM APIへの呼び出しに指数バックオフを追加する最もシンプルな方法です。tenacityライブラリを使うと1デコレーターで実装できます。

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 動作環境: Python 3.11+, tenacity>=8.2.0, openai>=1.30.0
# pip install tenacity openai

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
from openai import RateLimitError, APITimeoutError, APIConnectionError
import openai

client = openai.OpenAI()

@retry(
    retry=retry_if_exception_type((RateLimitError, APITimeoutError, APIConnectionError)),
    wait=wait_exponential(multiplier=1, min=1, max=60),  # 1s → 2s → 4s ... 最大60s
    stop=stop_after_attempt(5),  # 最大5回リトライ
)
def call_llm(prompt: str) -> str:
    """LLM呼び出しにリトライ付きラッパー"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        timeout=30,
    )
    return response.choices[0].message.content

ポイント：

• retry_if_exception_typeで「リトライすべきエラー」を明示（401/403はリトライしない）
• wait_exponentialで待機時間を指数的に増やしリトライストームを防ぐ
• stop_after_attemptで無限ループを防ぐ

即効テクニック2：jitter付きリトライで分散実行

複数エージェントが同時に失敗すると、全員が同じタイミングでリトライして再びサーバーを圧迫します。jitter（ランダムなゆらぎ）を加えることで解決できます。

# 動作環境: Python 3.11+, tenacity>=8.2.0
from tenacity import wait_random_exponential

@retry(
    retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
    wait=wait_random_exponential(multiplier=1, max=60),  # jitter付きexponential backoff
    stop=stop_after_attempt(5),
)
async def call_llm_async(prompt: str) -> str:
    """非同期版のリトライ付きLLM呼び出し"""
    response = await client.chat.completions.acreate(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

即効テクニック3：タイムアウトと例外の型安全なラッパー

エラーを握りつぶさず、型安全に伝播させるラッパーパターンです。エージェントの各ステップで使い回せます。

# 動作環境: Python 3.11+
import asyncio
from dataclasses import dataclass
from typing import Optional, TypeVar, Generic

T = TypeVar("T")

@dataclass
class Result(Generic[T]):
    """成功/失敗を型安全に表現するResult型"""
    value: Optional[T] = None
    error: Optional[str] = None

    @property
    def is_ok(self) -> bool:
        return self.error is None

async def safe_tool_call(tool_func, *args, timeout: float = 10.0, **kwargs) -> Result:
    """ツール呼び出しにタイムアウトとエラーキャプチャを付与"""
    try:
        value = await asyncio.wait_for(tool_func(*args, **kwargs), timeout=timeout)
        return Result(value=value)
    except asyncio.TimeoutError:
        return Result(error=f"Timeout after {timeout}s")
    except Exception as e:
        return Result(error=f"{type(e).__name__}: {str(e)}")

エラーハンドリングの3レイヤー設計

プロダクション品質のAIエージェントは、以下の3レイヤーでエラーに対処します。

L2: サーキットブレーカーの実装

リトライだけでは「障害中のサービスへのアクセス」を止められません。サーキットブレーカーは失敗率が閾値を超えたら自動的に遮断し、システム全体への波及を防ぎます。

pybreaker を使ったサーキットブレーカー実装

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 動作環境: Python 3.11+, pybreaker>=1.0.1
# pip install pybreaker

import pybreaker
import logging

logger = logging.getLogger(__name__)

# サーキットブレーカーの設定
# fail_max: 連続失敗回数の閾値
# reset_timeout: OPEN → HALF-OPEN に移行するまでの秒数
llm_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    listeners=[pybreaker.CircuitBreakerListener()]
)

@llm_breaker
def call_external_api(endpoint: str, payload: dict) -> dict:
    """サーキットブレーカー付きの外部API呼び出し"""
    import httpx
    response = httpx.post(endpoint, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()

# 呼び出し例
try:
    result = call_external_api("https://api.example.com/agent", {"query": "hello"})
except pybreaker.CircuitBreakerError:
    # サーキットがOPEN状態のとき: キャッシュ or フォールバック
    logger.warning("Circuit breaker is OPEN. Using fallback.")
    result = get_cached_response()
except Exception as e:
    logger.error(f"API call failed: {e}")
    result = None

状態遷移の説明：

• CLOSED（正常）：全リクエストを通す。失敗カウントを積算
• OPEN（遮断中）：fail_maxを超えたらOPENに。リクエストを即座に拒否
• HALF-OPEN（試行中）：reset_timeout後に1リクエスト試行。成功でCLOSEDへ

L3: グレースフルデグレードとフォールバックチェーン

L1・L2が尽きたとき、エージェントは「落ちる」のではなく「縮退して動き続ける」設計が重要です。フォールバックチェーンは「第一候補→第二候補→最終手段」の順で試みます。

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 動作環境: Python 3.11+, openai>=1.30.0, anthropic>=0.20.0
# pip install openai anthropic

import openai
import anthropic
from typing import Optional

openai_client = openai.AsyncOpenAI()
anthropic_client = anthropic.AsyncAnthropic()

async def llm_with_fallback(prompt: str) -> Optional[str]:
    """
    フォールバックチェーン:
    GPT-4o → Claude 3.5 Sonnet → キャッシュ → None（人間にエスカレーション）
    """
    # 第一候補: GPT-4o
    try:
        resp = await openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            timeout=20,
        )
        return resp.choices[0].message.content
    except Exception as e:
        print(f"GPT-4o failed: {e}. Trying Claude...")

    # 第二候補: Claude 3.5 Sonnet
    try:
        resp = await anthropic_client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.content[0].text
    except Exception as e:
        print(f"Claude failed: {e}. Trying cache...")

    # 第三候補: セマンティックキャッシュ
    cached = await semantic_cache_lookup(prompt)
    if cached:
        return f"[キャッシュ応答] {cached}"

    # 回復不能: Noneを返してヒューマンエスカレーションを起動
    print("All LLMs failed. Escalating to human.")
    return None

エラー種別ごとのハンドリング判断表

【要注意】よくある失敗パターンと回避策

失敗1：全エラーを同じようにリトライする

❌ すべての例外をキャッチしてとにかくリトライ

⭕ エラー種別を判定し、リトライすべきエラーだけリトライ

なぜ重要か：認証エラーや不正リクエストをリトライしても意味がなく、APIコストが無駄に消費されます。

失敗2：リトライ回数を設定しない

❌ while True: try: call_llm() except: continue

⭕ stop_after_attempt(5)で上限を設定

なぜ重要か：無限リトライはAPIコストが青天井になります。サービス障害時に特に危険。

失敗3：エラーをサイレントに握りつぶす

❌ except Exception: pass

⭕ 必ずログを出力し、Result型で上流に伝播させる

なぜ重要か：握りつぶされたエラーはデバッグが極めて困難になります。本番障害で最も時間を取られる失敗パターンです。

失敗4：ジッターなしの固定バックオフ

❌ リトライを1秒→2秒→3秒の固定間隔で実施

⭕ wait_random_exponentialでランダムなゆらぎを追加

なぜ重要か：マルチエージェント環境で全エージェントが同時にリトライするとサーバーを再び圧迫します（thundering herd問題）。

モニタリングとアラートの設定

正直なところ、エラーハンドリングだけでは不十分です。どこで何回失敗しているかを可視化するモニタリングが、長期的な安定稼働には欠かせません。

# 動作環境: Python 3.11+
import functools
import time
from collections import defaultdict

# シンプルなエラー集計（本番ではPrometheus/DatadogのSDKを使用推奨）
error_counts = defaultdict(int)
retry_counts = defaultdict(int)

def monitored_retry(func):
    """リトライ回数とエラーを計測するデコレーター"""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.monotonic()
        attempt = 0
        last_error = None

        while attempt  0:
                    retry_counts[func.__name__] += attempt
                return result
            except Exception as e:
                attempt += 1
                last_error = e
                error_counts[f"{func.__name__}.{type(e).__name__}"] += 1
                await asyncio.sleep(2 ** attempt)

        raise last_error
    return wrapper

参考・出典

• tenacity — Retrying library for Python（GitHub） — Tenacity公式リポジトリ（参照日: 2026-04-09）
• Resilient AI Agents With MCP: Timeout And Retry Strategies — Octopus blog（参照日: 2026-04-09）
• Resilient APIs: Retry Logic, Circuit Breakers, and Fallback Mechanisms — Medium（参照日: 2026-04-09）
• AI Agent Error Handling: 4 Resilience Patterns in Python — DEV Community（参照日: 2026-04-09）

まとめ：今日から始める3つのアクション

1. 今日やること：既存のLLM呼び出しに@retry(wait=wait_random_exponential(...))を1つ追加する
2. 今週中：外部APIへの接続にサーキットブレーカー（pybreaker）を導入し、OPEN/CLOSED状態をログに出力する
3. 今月中：フォールバックチェーンとモニタリングを整備し、エラー率をダッシュボードで可視化する

あわせて読みたい:

• AIエージェント構築完全ガイド — 基本設計パターンから始めたい方はこちら
• AIエージェントのコスト最適化5つの戦略 — コスト管理と安定稼働を両立する

この記事はAIgent Lab編集部がお届けしました。