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

# requirements: antigravity-agentkit>=2.0.0, redis>=5.0.0
# 環境変数: REDIS_URL（例: redis://localhost:6379/0）
 
import os
import time
import uuid
from typing import Literal
 
import redis
 
HealthStatus = Literal["healthy", "warning", "critical"]
 
class HealthSignals:
    """エージェントのヘルスシグナルを Redis に集約し、状態判定を行うクラス。
 
    Redis を使う理由は2つ:
    - エージェントが複数プロセスで動く場合、メモリ上の deque では情報が分散する
    - プロセスが落ちても直前の状態を失わない
    インメモリで十分な小規模運用なら、同じインターフェースで list に置き換えられる。
    """
 
    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"], decode_responses=True)
        self.window_seconds = 600  # 10分ウィンドウ
        self.k_calls = f"agent:{agent_id}:tool_calls"
        self.k_empty = f"agent:{agent_id}:empty_responses"
        self.k_base = f"agent:{agent_id}:baseline_p95_ms"
 
    def record_tool_call(self, tool: str, success: bool, latency_ms: float) -> None:
        ts = time.time()
        member = f"{uuid.uuid4().hex}:{int(success)}:{latency_ms:.1f}:{tool}"  # uuid を付けて、同一ミリ秒の呼び出しが ZSET 上で潰れないようにする
        self.r.zadd(self.k_calls, {member: ts})
        self.r.zremrangebyscore(self.k_calls, 0, ts - self.window_seconds)
 
    def record_empty_response(self, was_empty: bool) -> None:
        self.r.lpush(self.k_empty, "1" if was_empty else "0")
        self.r.ltrim(self.k_empty, 0, 99)
 
    def set_baseline_p95(self, ms: float) -> None:
        self.r.set(self.k_base, ms)
 
    def status(self) -> HealthStatus:
        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 _window(self) -> list[tuple[bool, float]]:
        now = time.time()
        rows = self.r.zrangebyscore(self.k_calls, now - self.window_seconds, now)
        out = []
        for m in rows:
            _, ok, lat, _tool = m.split(":", 3)  # ツール名に ':' が入っていても壊れないよう、先頭3区切りだけ分割する
            out.append((ok == "1", float(lat)))
        return out
 
    def _success_rate(self) -> float:
        rows = self._window()
        if not rows:
            return 1.0  # データがないうちは異常扱いにしない
        return sum(1 for ok, _ in rows if ok) / len(rows)
 
    def _p95_latency(self) -> float:
        lats = sorted(lat for _, lat in self._window())
        if not lats:
            return 0.0
        return lats[min(len(lats) - 1, int(len(lats) * 0.95))]
 
    def _baseline_p95(self) -> float:
        v = self.r.get(self.k_base)
        return float(v) if v else 2000.0  # 基準が未登録の間の仮値（ms）。実測の P95 を set_baseline_p95() で登録する
 
    def _empty_rate(self) -> float:
        vals = self.r.lrange(self.k_empty, 0, 99)
        if not vals:
            return 0.0
        return sum(1 for v in vals if v == "1") / len(vals)

# 診断レイヤー: エラーパターンと外部サービス状態から復旧戦略を決定する
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:
    # ツール名 → 依存先ステータスページの JSON API の URL（Atlassian Statuspage 形式 {"status": {"indicator": "none"}} を想定）
    STATUS_API: dict[str, str] = {}
 
    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:
        """依存先のステータス API を参照する。URL は STATUS_API に登録したものだけを見る。
        ステータスページの形式は依存先ごとに違うため、登録前に実際の JSON を確認すること。"""
        url = self.STATUS_API.get(tool_name)
        if not url:
            return False  # 参照先が未登録なら判断材料なし
        try:
            r = httpx.get(url, timeout=2.0)
            return r.json().get("status", {}).get("indicator", "none") != "none"
        except Exception:
            return False  # 確認できないときは保守的に「正常」扱い（ステータスページ自体も落ちうる）

# 復旧戦略を実行する Recovery Executor
import asyncio
import os
import random
from typing import Any, Callable, Awaitable
 
import httpx
 
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 _notify_slack(self, name: str, args: tuple, kwargs: dict) -> None:
        """Slack の Incoming Webhook に人間への引き継ぎを通知する。URL は環境変数 SLACK_WEBHOOK_URL。"""
        url = os.environ.get("SLACK_WEBHOOK_URL")
        if not url:
            return
        safe = {k: v for k, v in kwargs.items() if k not in ("api_key", "token")}  # 秘匿値は載せない
        async with httpx.AsyncClient(timeout=5.0) as client:
            await client.post(url, json={"text": f"[self-healing] {name} を人間に引き継ぎます: {safe}"})
 
    async def _exp_backoff(
        self,
        operation: Callable[..., Awaitable[Any]],
        *args,
        max_attempts: int = 5,
        **kwargs,
    ) -> 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秒（最後の試行のあとは待たずに送出する）
                jitter = base * random.uniform(0, 0.3)
                await asyncio.sleep(base + jitter)
 
class CircuitBreakerOpen(Exception):
    pass
 
class HumanEscalationRequired(Exception):
    pass
 
class HealthCheckFailed(Exception):
    pass

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
 
def retry_after_seconds(headers: dict[str, str], default: float) -> float:
    """Retry-After（秒数 or HTTP 日付）を秒に直す。無い・読めないときは default。"""
    raw = headers.get("Retry-After")
    if not raw:
        return default
    if raw.isdigit():
        return float(raw)
    try:
        when = parsedate_to_datetime(raw)
        return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return default
 
# _exp_backoff の中での使い方（例外から headers を取り出せる HTTP クライアントを前提）
# wait = max(base + jitter, retry_after_seconds(headers, 0.0))
# if wait > 120:           # 上限を超える待ちは、待たずに人間へ引き継ぐ
#     raise HumanEscalationRequired("Retry-After が長すぎます")

# AgentKit 2.0 のミドルウェアとして組み込む Self-Healing ラッパー
from typing import Any
 
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 after_tool_call(self, ctx: ToolCallContext, result: Any) -> None:
        """成功も記録する。失敗しか記録しないと、成功率の分母が偏って常に低く見える。"""
        self.health.record_tool_call(tool=ctx.tool.name, success=True, latency_ms=ctx.elapsed_ms)
        self.health.record_empty_response(result in ("", None))
 
    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)
 
class FakeInvoiceTool:
    async def invoke(self, *args, **kwargs):
        return {"ok": True}
 
@pytest.mark.asyncio
async def test_rate_limit_recovers_via_backoff(monkeypatch):
    async def no_sleep(_seconds):  # 待ち時間をテストから消す
        return None
    monkeypatch.setattr(asyncio, "sleep", no_sleep)
 
    tool = FailureInjector(FakeInvoiceTool(), failure_mode="rate_limit", fail_count=2)
    executor = RecoveryExecutor()
    result = await executor.execute("exponential_backoff_with_jitter", tool.invoke, customer_id="cus_123")
    assert result is not None
    assert tool.calls == 3  # 2回失敗、3回目で成功