self-healing-agents

自己修復エージェント

Agent Skills™ の一部です。googleadsagent.ai™ 提供

説明

自己修復エージェントは、自身の障害モードを検知し、人の介入なしに自己修正する自律システムです。本番環境では、エージェントの障害は例外的なものではなく、起こりうるものです。ネットワーク呼び出しはタイムアウトし、API は予期しないスキーマを返し、モデルは自信たっぷりに幻覚を生成し、ツールの出力は想定に反することがあります。プロトタイプと本番環境向けエージェントの違いは、あらゆる種類の障害から適切に復旧できる能力にあります。

このスキルでは、googleadsagent.ai™ の Buddy™ エージェント向けに開発された自己修復パターンを体系化しています。このエージェントは、自律的な Google Ads 分析を確実に完了する必要があり、上流 API の変更、レート制限への到達、モデル出力の構造的なエラーがあっても安定して動作します。このシステムは、生物の免疫反応にならった検知・診断・修復のサイクルで動作します。病原体を特定し、脅威を分類して、適切な対抗策を実行します。

自己修復は単なるリトライロジックではありません。エラー分類、戦略の変更（同じ方法を繰り返すのではなく、異なるアプローチでリトライすること）、フォールバックモデルの選択、自動修復を伴う出力検証、そして完全な復旧が不可能な場合のグレースフルデグラデーションを含みます。これらのパターンを用いて構築されたエージェントは、本番環境で 99% 以上のタスク完了率を達成します。

使用する場面

• 人の監視なしで自律的に動作する必要があるエージェントを構築するとき
• ツール呼び出しや API 統合が不安定、またはレート制限の対象となるとき
• モデル出力が厳格なスキーマに準拠する必要があるものの、準拠しないことがあるとき
• 長時間実行されるワークフローで、実行途中の失敗を許容できないとき
• エージェントを活用する機能の SLA を維持する必要があるとき
• エージェントがこれまで遭遇したことのない新しいエラーの種類に対処する必要があるとき

仕組み

graph TD
    A[Agent Action] --> B[Output Validation]
    B -->|Valid| C[Continue Execution]
    B -->|Invalid| D[Error Classifier]
    D --> E{Error Type}
    E -->|Transient| F[Retry with Backoff]
    E -->|Structural| G[Mutate Strategy]
    E -->|Model Error| H[Fallback Model]
    E -->|Unrecoverable| I[Graceful Degradation]
    F --> J{Retry Budget Remaining?}
    J -->|Yes| A
    J -->|No| G
    G --> K[Modified Prompt/Approach]
    K --> A
    H --> L[Alternative Model Execution]
    L --> B
    I --> M[Partial Result + Error Report]

出力検証で異常が検出されると、自己修復サイクルが開始されます。エラー分類器は障害を 4 種類に分類します。ネットワークタイムアウトやレート制限などの一時的エラーは、指数バックオフを用いて再試行します。スキーマ違反や必須フィールドの欠落などの構造的エラーでは、エージェントがアプローチを変更する戦略変更を行います。幻覚や拒否などのモデルエラーでは、フォールバックモデルを選択します。復旧不可能なエラーでは、明確なエラーレポートとともに、得られる限り最良の部分的な結果を返すグレースフルデグラデーションを実行します。

実装

エラー分類エンジン：

enum ErrorType {
  Transient = "transient",
  Structural = "structural",
  ModelError = "model_error",
  Unrecoverable = "unrecoverable",
}

interface ClassifiedError {
  type: ErrorType;
  message: string;
  retryable: boolean;
  suggestedStrategy: string;
}

function classifyError(error: unknown, context: ExecutionContext): ClassifiedError {
  if (error instanceof NetworkError || error instanceof RateLimitError) {
    return {
      type: ErrorType.Transient,
      message: String(error),
      retryable: true,
      suggestedStrategy: "exponential_backoff",
    };
  }
  if (error instanceof SchemaValidationError) {
    return {
      type: ErrorType.Structural,
      message: `Schema violation: ${error.path} — ${error.message}`,
      retryable: true,
      suggestedStrategy: "mutate_prompt",
    };
  }
  if (error instanceof ModelRefusalError || isHallucination(error, context)) {
    return {
      type: ErrorType.ModelError,
      message: String(error),
      retryable: true,
      suggestedStrategy: "fallback_model",
    };
  }
  return {
    type: ErrorType.Unrecoverable,
    message: String(error),
    retryable: false,
    suggestedStrategy: "graceful_degradation",
  };
}

自己修復実行ループ：

class SelfHealingAgent:
    def __init__(self, primary_model, fallback_models, max_retries=3):
        self.primary_model = primary_model
        self.fallback_models = fallback_models
        self.max_retries = max_retries

    async def execute(self, task, validator):
        attempts = []
        current_model = self.primary_model
        current_prompt = task.prompt

        for attempt in range(self.max_retries + len(self.fallback_models)):
            try:
                result = await current_model.generate(current_prompt)
                validation = validator.validate(result)
                if validation.is_valid:
                    return HealingResult(result=result, attempts=attempts)

                error = classify_error(validation.error, task.context)
                attempts.append({"attempt": attempt, "error": error, "model": current_model.name})

                if error.strategy == "mutate_prompt":
                    current_prompt = self.mutate_prompt(current_prompt, error)
                elif error.strategy == "fallback_model":
                    current_model = self.next_fallback(current_model)

            except Exception as e:
                error = classify_error(e, task.context)
                attempts.append({"attempt": attempt, "error": error})
                if not error.retryable:
                    break
                await asyncio.sleep(2 ** attempt)

        return HealingResult(
            result=self.graceful_degradation(task, attempts),
            attempts=attempts,
            degraded=True
        )

    def mutate_prompt(self, prompt, error):
        mutations = {
            "schema_violation": f"{prompt}\n\nPrevious attempt had error: {error.message}. Ensure strict schema compliance.",
            "missing_field": f"{prompt}\n\nYou MUST include all required fields. Missing: {error.message}",
        }
        return mutations.get(error.subtype, f"{prompt}\n\nPrevious error: {error.message}. Adjust approach.")

    def next_fallback(self, current):
        idx = ([self.primary_model] + self.fallback_models).index(current)
        if idx < len(self.fallback_models):
            return self.fallback_models[idx]
        return current

自動修復を伴う出力検証：

def validate_and_repair(output: str, schema: dict) -> tuple[dict, bool]:
    try:
        parsed = json.loads(output)
    except json.JSONDecodeError:
        extracted = extract_json_from_text(output)
        if extracted:
            parsed = extracted
        else:
            raise StructuralError("No valid JSON found in output")

    repaired = False
    for field, rules in schema.get("required_fields", {}).items():
        if field not in parsed:
            if "default" in rules:
                parsed[field] = rules["default"]
                repaired = True
            else:
                raise StructuralError(f"Missing required field: {field}")

    return parsed, repaired

ベストプラクティス

1. 再試行する前に分類する — やみくもな再試行はトークンと時間を浪費します。最適な復旧戦略を選べるよう、必ずエラーの種類を分類してください。
2. 再試行時は方法を変え、繰り返さない — 同じプロンプトで失敗した場合、同じものを再送しても効果はほとんどありません。エラーのコンテキストを追加するか、リクエストを組み直してください。
3. 再試行の上限を厳格に設定する — 自己修復ループの暴走を防ぐため、再試行の合計回数（通常は 3～5 回）とトークン消費の合計量を制限してください。
4. 修復の試行をすべて記録する — 各再試行、変更、フォールバックの呼び出しについて、事後分析に必要なコンテキストをすべて含めて記録してください。
5. グレースフルデグラデーション時の出力を設計する — 明確なエラー注釈が付いた部分的な結果は、空の失敗結果よりもはるかに有用です。
6. 修復経路を明示的にテストする — テスト時に既知の障害モードを発生させ、各修復経路が正しく動作することを確認してください。
7. 修復率を監視する — 修復率が 10% を超える場合は、自己修復で覆い隠すのではなく、発生源で修正すべき上流の問題があることを示しています。

プラットフォームの互換性

機能	Claude Code	Cursor	Codex	Gemini CLI
ミューテーションを伴う再試行	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
フォールバックモデルの選択	✅ API 経由	✅ 拡張機能経由	✅ API 経由	✅ API 経由
出力の検証	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
エラーの分類	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
グレースフルデグラデーション	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応

関連スキル

• 検証ループ - 自己修復エージェントが回復する障害を検出する、多段階の検証パイプライン
• エージェント本能システム - 自己修復が必要になる前にエラーを防ぐ、アクション前の安全ゲート
• 敵対的レジリエンス - 一時的なエラーや構造的なエラーに加え、敵対的な要因による障害にも対処する防御層

キーワード

自己修復、エラー回復、再試行戦略、フォールバックモデル、出力検証、グレースフルデグラデーション、戦略の変更、エラー分類、レジリエンス、エージェントスキル

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License