最近、オンラインサービスの問題を調査中に、AI推論サービスがトラフィック増加により応答時間が急上昇し、全体の呼び出しチェーンが停止する事態に直面しました。ログを見ると、下流のベクトルデータベースがたまにタイムアウトしており、エージェントはエラー処理を行わず例外をスローしていました。この経験から、エージェントアーキテクチャには強力な「免疫システム」が必要であることを学びました。

なぜこれらの戦略が必要なのか？

フォールトトレランス、リトライ、デグレードはよく聞く話ですが、特にAIエージェントでは重要です。従来のソフトウェアのエラーは多くの場合明確ですが、AIコンポーネントの失敗はより隠れた形で起こります。大規模モデルは誤った答えを返す可能性があり、外部APIはタイムアウトするかもしれませんが再試行で成功するかもしれません。これらの不確定要素を設計に組み込むことで、システムは「玩具」から「エンジニアリング製品」へと進化します。

フォールトトレランスはtry-catchだけではない

フォールトトレランスは単純にtry-catchで包むものではありません。以下は典型的な問題のあるコード例です：

  
  def query_agent(question):
      # こう書くのは良くない：1つのエラーで全てが崩れる
      answer = llm.generate(question)  # 可能なエラー
      embeddings = vector_db.search(answer)  # タイムアウトの可能性
      return format_response(embeddings)
  
  

改善案は故障領域を分離することです。各コンポーネントには独立したエラーハンドリングが必要です。

  
  class FaultTolerantLLM:
      def generate(self, prompt, fallback_strategy="cached"):
          try:
              return self._call_llm_api(prompt)
          except LLMTimeout:
              if fallback_strategy == "cached":
                  return self._get_cached_response(prompt)  # キャッシュへの降格
              elif fallback_strategy == "simple":
                  return "システムが考え中です。後でもう一度お試しください"  # ユーザーへの友好的なメッセージ
              else:
                  raise  # ストラテジーに基づいて例外を投げるかどうか決定
  
  

重要なのは、**1つのコンポーネントの障害が全体に影響を与えないようにすること**です。ベクトルデータベースがダウンしても、大規模モデルの元の回答を返すことができます。大規模モデルが利用できない場合は、ルールエンジンに降格できます。

リトライの科学：適切に試す

リトライは簡単そうに見えますが、多くの落とし穴があります。以下のようなコードは避けてください。

  
  for i in range(10):
      try:
          return call_api()
      except:
          time.sleep(1)
  
  

問題点は以下の通りです： - エラータイプを区別していない（404はリトライすべきでない） - 同時にリトライすると「驚群効果」が発生する - 下流の負荷考慮不足 実際の環境では、戦略的なリトライを使用します。

  
  class RetryPolicy:
      def __init__(self):
          self.max_retries = 3
          self.backoff_factor = 1.5  # 指数退避
          self.retryable_errors = [TimeoutError, ConnectionError]
      
      def should_retry(self, error, attempt):
          if attempt >= self.max_retries:
              return False
          if type(error) not in self.retryable_errors:
              return False
          if hasattr(error, 'response'):
              if error.response.status == 429:  # レート制限
                  self._add_jitter()  # 同期を避けるためのランダムな遅延
          return True
  
  

経験則： - **1）リトライ可能なエラーは主にネットワーク問題や一時的な障害** - **2）必ずバックオフとジャイターを使用する** - **3）リトライ前にヘルスチェックを行う** AIの場合、上下文の長さが有限なので、リトライ時に履歴メッセージを繰り返し連結すると長さ制限に達する可能性があります。適切な上下文管理が必要です。

デグレード戦略：可用性優先

デグレードは最後の防衛線です。基本的な考え方：システムの負荷が高すぎるときや一部の機能が利用不可になったときに、非核心機能を閉じて主要なパスを確保する。AIエージェントでは、多くの「スマート」機能は段階的に降格することができます。 例えば、カスタマーサポートエージェントの降格パスは以下のようになります。 1. 完全モード：LLM生成 → 知識ベース検索 → 感情分析 → 構造化出力 2. 降格モード1：感情分析をスキップして基本的な回答を返す 3. 降格モード2：キャッシュされた類似質問の回答を使用し、LLMを呼び出さない 4. 降格モード3：事前定義されたFAQリストを返し、ユーザーが選択できるようにする 実装例：

  
  class DegradationManager:
      def __init__(self):
          self.degradation_level = 0  # 0: 正常, 1: 軽度降格...
          self.metrics = SystemMetrics()
      
      def check_and_degrade(self):
          if self.metrics.latency > 1000:  # 遅延閾値
              self.degradation_level = 1
              self._disable_expensive_features()  # 高コスト機能を無効化
          
          if self.metrics.error_rate > 0.1:  # エラーレート閾値
              self.degradation_level = 2
              self._enable_fallback_cache()  # キャッシュ降格を有効化
  
  

**デグレードのキーは事前の計画**です。システムが危機的状況になるまで待つべきではありません。チームでは、各Sprintで「降格演習」を行い、特定のサービスを手動で停止してシステムの対応を確認しています。

組み合わせた使用：実践例

完全な例として、ドキュメント解析エージェントを考えます。

  
  class DocumentAnalysisAgent:
      def analyze(self, doc):
          try:
              summary = self._retry_llm_summary(doc)
              
              if self.degradation_level < 1:
                  insights = self._advanced_analysis(summary)  # 失敗の可能性
              else:
                  insights = self._basic_analysis(summary)  # 降格バージョン
                  
              return {"summary": summary, "insights": insights}
              
          except CriticalError as e:
              return {
                  "summary": "ドキュメントを受け付けました。分析サービスは現在制限されています",
                  "insights": [],
                  "degraded": True  # 降格状態を明示
              }
      
      def _retry_llm_summary(self, doc):
          policy = RetryPolicy()
          attempt = 0
          
          while attempt <= policy.max_retries:
              try:
                  return self.llm.summarize(doc)
              except Exception as e:
                  if not policy.should_retry(e, attempt):
                      raise
                  attempt += 1
                  sleep(policy.backoff(attempt))
  
  

最後の`degraded`フラグは、**呼び出し元に降格状態を明示**する役割を持ちます。これにより、上位システムは適切な対策を取ることができます。

監視とデバッグ

これらの戦略を導入すると、監視がさらに重要になります。以下の指標に注目しましょう。 - リトライ率：急激な上昇は下流サービスの不安定性を示す - 降格比率：長期間の降格は容量不足を示す - エラータイプ分布：ビジネスエラーとシステムエラーを区別 ログにはリクエストID、リトライ回数、降格レベルを含めましょう。問題の原因を追跡しやすくなります。以前、偶発的なタイムアウトがリトライによって隠され、下流サービスが完全にダウンするまで気づかなかったことがあります。その後、リトライマークをログに追加し、どの失敗がリトライ後に成功したかをすぐに確認できるようになりました。

個人的な経験

多年にわたる分散システム開発を通じて、フォールトトレランスはコード内のtry-catchではなく、アーキテクチャ段階での「非機能要件」として考慮すべきだと感じています。AIエージェントプロジェクトでは、以下の提案をします。 1. **早期にカオスエンジニアリングを導入**し、オンライニング後に補うのではなく開発環境でAPIのタイムアウトや誤った回答をシミュレートします。 2. **プロダクトマネージャーと共に降格戦略を設計**し、技術者が「すべてかゼロ」の思考に陥らないようにします。ユーザーは部分的な機能を許容することが多いです。 3. **リトライ戦略は保守的**に設定し、計算密集型のAIサービスでは盲目的なリトライは雪崩現象を引き起こす可能性があります。429ステータスコード（レート制限）を見たら適切に退避します。 4. **セーフティモードのスイッチ**を設け、必要時に最簡バージョンに切り替えられるようにします。昨年の大きなイベントでは、この機能により3倍のトラフィックに対応できました。 最後に、これらの戦略は万能薬ではありません。複雑性を増す可能性があります。私の原則は、**最初はシンプルに始め、問題が発生したら反復改善**することです。まずは基本的なエラーアイソレーションとリトライから始め、必要に応じてデグレードと高度な戦略を追加します。過剰な設計もリスクとなります。