Antigravity Self-Healing エージェントの設計実装 — 障害検知・自己診断・自動復旧のプロダクションパターン

# requirements: antigravity-agentkit>=2.0.0, redis>=5.0.0
# 環境変数: REDIS_URL（例: redis://localhost:6379/0）
 
import time
from datetime import datetime, timedelta
from collections import deque
from typing import Literal
import os
import redis
 
HealthStatus = Literal["healthy", "warning", "critical"]
 
class HealthSignals:
    """エージェントのヘルスシグナルを Redis に集約し、状態判定を行うクラス。
 
    Redis を使う理由は2つ:
    - エージェントが複数プロセスで動く場合、メモリ上の deque では情報が分散する
    - プロセスが落ちても直前の状態を失わない
    インメモリで十分な小規模運用なら deque のままでも動作する。
    """
 
    def __init__(self, agent_id: str, redis_url: str | None = None):
        self.agent_id = agent_id
        self.r = redis.from_url(redis_url or os.environ["REDIS_URL"])
        self.window_seconds = 600  # 10分ウィンドウ
 
    def record_tool_call(self, tool: str, success: bool, latency_ms: float) -> None:
        """ツール呼び出しの結果を記録する。"""
        ts = time.time()
        key = f"agent:{self.agent_id}:tool_calls"
        # ZSET に (success, latency, tool) を JSON 文字列で保存し、score を timestamp に
        member = f"{int(ts*1000)}:{success}:{latency_ms}:{tool}"
        self.r.zadd(key, {member: ts})
        # 古いエントリは自動削除（時間ウィンドウ外）
        self.r.zremrangebyscore(key, 0, ts - self.window_seconds)
 
    def record_empty_response(self, was_empty: bool) -> None:
        key = f"agent:{self.agent_id}:empty_responses"
        # LIST で直近100件を保持
        self.r.lpush(key, "1" if was_empty else "0")
        self.r.ltrim(key, 0, 99)
 
    def status(self) -> HealthStatus:
        """4つのシグナルを統合して現在の状態を返す。"""
        success_rate = self._success_rate()
        p95 = self._p95_latency()
        empty_rate = self._empty_rate()
 
        if success_rate < 0.5 or empty_rate > 0.20:
            return "critical"
        if success_rate < 0.8 or p95 > self._baseline_p95() * 3 or empty_rate > 0.05:
            return "warning"
        return "healthy"
 
    def _success_rate(self) -> float:
        # 実装簡略化のため省略。本番ではツール別の成功率を別途計算する
        ...

# 診断レイヤー: エラーパターンと外部サービス状態から復旧戦略を決定する
import re
import httpx
from dataclasses import dataclass
from enum import Enum
 
class FailurePattern(Enum):
    RATE_LIMIT = "rate_limit"
    SERVICE_DOWN = "service_down"
    TIMEOUT = "timeout"
    EMPTY_LLM_RESPONSE = "empty_llm"
    JSON_PARSE_ERROR = "json_parse"
    UNKNOWN = "unknown"
 
@dataclass
class Diagnosis:
    pattern: FailurePattern
    confidence: float  # 0.0〜1.0
    recovery_strategy: str
    cooldown_seconds: int
 
# パターンマッチ用の正規表現と外部サービス検知をまとめた診断器
class Diagnoser:
    RATE_LIMIT_PATTERNS = [
        re.compile(r"rate.?limit", re.I),
        re.compile(r"too many requests", re.I),
        re.compile(r"429", re.I),
    ]
    TIMEOUT_PATTERNS = [
        re.compile(r"timeout", re.I),
        re.compile(r"deadline exceeded", re.I),
    ]
 
    def diagnose(self, error_message: str, tool_name: str) -> Diagnosis:
        # まずパターンマッチで素早く分類
        if any(p.search(error_message) for p in self.RATE_LIMIT_PATTERNS):
            return Diagnosis(
                pattern=FailurePattern.RATE_LIMIT,
                confidence=0.95,
                recovery_strategy="exponential_backoff_with_jitter",
                cooldown_seconds=60,
            )
        if any(p.search(error_message) for p in self.TIMEOUT_PATTERNS):
            # タイムアウトは外部サービスのステータスを確認する
            if self._service_is_degraded(tool_name):
                return Diagnosis(
                    pattern=FailurePattern.SERVICE_DOWN,
                    confidence=0.85,
                    recovery_strategy="failover_to_secondary",
                    cooldown_seconds=300,
                )
            return Diagnosis(
                pattern=FailurePattern.TIMEOUT,
                confidence=0.70,
                recovery_strategy="retry_with_longer_timeout",
                cooldown_seconds=10,
            )
        # 該当パターンなしは UNKNOWN として扱い、LLM 診断にエスカレーション
        return Diagnosis(
            pattern=FailurePattern.UNKNOWN,
            confidence=0.0,
            recovery_strategy="escalate_to_human",
            cooldown_seconds=0,
        )
 
    def _service_is_degraded(self, tool_name: str) -> bool:
        """外部サービスのステータスページを参照する。
        実装例: Stripe なら status.stripe.com の RSS、
        OpenAI なら status.openai.com の API を叩く。"""
        try:
            # Stripe の例
            r = httpx.get("https://status.stripe.com/api/v2/status.json", timeout=2.0)
            return r.json().get("status", {}).get("indicator") != "none"
        except Exception:
            return False  # 確認できないときは保守的に "正常" 扱い

# 復旧戦略を実行する Recovery Executor
import asyncio
import random
from typing import Any, Callable, Awaitable
 
class RecoveryExecutor:
    def __init__(self, primary_provider: str = "anthropic"):
        self.primary = primary_provider
        self.secondary = "gemini" if primary_provider == "anthropic" else "anthropic"
 
    async def execute(
        self,
        strategy: str,
        operation: Callable[..., Awaitable[Any]],
        *args, **kwargs
    ) -> Any:
        """復旧戦略に従って operation を再実行する。
        operation は冪等性を保証していること（idempotency_key 必須）。"""
 
        if strategy == "exponential_backoff_with_jitter":
            return await self._exp_backoff(operation, *args, **kwargs)
        if strategy == "failover_to_secondary":
            kwargs["provider"] = self.secondary
            return await operation(*args, **kwargs)
        if strategy == "retry_with_longer_timeout":
            kwargs["timeout"] = kwargs.get("timeout", 30) * 2
            return await operation(*args, **kwargs)
        if strategy == "degrade_to_readonly":
            kwargs["mode"] = "readonly"
            return await operation(*args, **kwargs)
        if strategy == "circuit_break":
            raise CircuitBreakerOpen("サーキットブレーカー作動中。15分後に再試行します")
        if strategy == "escalate_to_human":
            await self._notify_slack(operation.__name__, args, kwargs)
            raise HumanEscalationRequired("人間の判断が必要です")
        raise ValueError(f"未知の復旧戦略: {strategy}")
 
    async def _exp_backoff(
        self,
        operation: Callable[..., Awaitable[Any]],
        *args, **kwargs,
        max_attempts: int = 5,
    ) -> Any:
        for attempt in range(max_attempts):
            try:
                return await operation(*args, **kwargs)
            except Exception as e:
                if attempt == max_attempts - 1:
                    raise
                base = 2 ** attempt  # 1, 2, 4, 8, 16秒
                jitter = base * random.uniform(0, 0.3)
                await asyncio.sleep(base + jitter)
 
class CircuitBreakerOpen(Exception):
    pass
 
class HumanEscalationRequired(Exception):
    pass

# AgentKit 2.0 のミドルウェアとして組み込む Self-Healing ラッパー
from antigravity.agentkit import Agent, ToolMiddleware, ToolCallContext
 
class SelfHealingMiddleware(ToolMiddleware):
    def __init__(
        self,
        health_signals: HealthSignals,
        diagnoser: Diagnoser,
        executor: RecoveryExecutor,
    ):
        self.health = health_signals
        self.diag = diagnoser
        self.exec = executor
 
    async def before_tool_call(self, ctx: ToolCallContext) -> None:
        """ツール呼び出し前にヘルスチェックを実行する。"""
        status = self.health.status()
        if status == "critical":
            # 緊急時は読み取り専用ツールのみ許可
            if not ctx.tool.is_read_only:
                raise HealthCheckFailed(
                    f"エージェントが critical 状態です。書き込みツールは利用できません: {ctx.tool.name}"
                )
 
    async def on_tool_error(self, ctx: ToolCallContext, error: Exception) -> Any:
        """ツール呼び出しが失敗したら診断と復旧を実行する。"""
        # 失敗を記録
        self.health.record_tool_call(
            tool=ctx.tool.name,
            success=False,
            latency_ms=ctx.elapsed_ms,
        )
        # 診断
        diagnosis = self.diag.diagnose(str(error), ctx.tool.name)
        # 信頼度が低い診断は再投げ（人間への通知は escalate_to_human が行う）
        if diagnosis.confidence < 0.5:
            raise error
        # 復旧戦略を実行
        return await self.exec.execute(
            diagnosis.recovery_strategy,
            ctx.tool.invoke,
            **ctx.tool_args,
        )
 
# エージェント生成時にミドルウェアを差し込む
agent = Agent(
    name="payment-support-agent",
    model="claude-sonnet-4-5",
    tools=[stripe_tools, slack_tools],
    middleware=[
        SelfHealingMiddleware(
            health_signals=HealthSignals(agent_id="payment-support"),
            diagnoser=Diagnoser(),
            executor=RecoveryExecutor(primary_provider="anthropic"),
        ),
    ],
)

# 予算上限つきの実行器 — 自己修復が「自己破産」にならないようにする
class BudgetedExecutor(RecoveryExecutor):
    def __init__(self, primary_provider: str, max_recovery_cost_usd: float = 0.10):
        super().__init__(primary_provider)
        self.max_cost = max_recovery_cost_usd
 
    async def execute(self, strategy, operation, *args, **kwargs):
        spent = kwargs.pop("_recovery_spent_usd", 0.0)
        if spent >= self.max_cost:
            await self._notify_slack(
                f"復旧予算超過: ${spent:.4f} / ${self.max_cost}",
                args, kwargs,
            )
            raise HumanEscalationRequired("復旧予算を使い切りました")
        return await super().execute(strategy, operation, *args, **kwargs)

# 復旧パスをテストするための障害注入ハーネス
import pytest
import asyncio
 
class FailureInjector:
    # 任意のツールをラップし、テスト時に特定の失敗モードを注入する
    def __init__(self, tool, failure_mode=None, fail_count=1):
        self.tool = tool
        self.failure_mode = failure_mode
        self.fail_count = fail_count
        self.calls = 0
 
    async def invoke(self, *args, **kwargs):
        self.calls += 1
        if self.failure_mode and self.calls <= self.fail_count:
            if self.failure_mode == "rate_limit":
                raise Exception("429: rate limit exceeded")
            if self.failure_mode == "timeout":
                raise asyncio.TimeoutError("deadline exceeded")
            if self.failure_mode == "empty":
                return ""
        return await self.tool.invoke(*args, **kwargs)
 
@pytest.mark.asyncio
async def test_rate_limit_recovers_via_backoff():
    tool = FailureInjector(stripe_list_invoices, failure_mode="rate_limit", fail_count=2)
    middleware = SelfHealingMiddleware(...)
    result = await middleware.with_recovery(tool.invoke, customer_id="cus_123")
    assert result is not None
    assert tool.calls == 3  # 2回失敗、3回目で成功