Claude API のグレースフル・デグラデーション設計 — AI 機能が静かに動き続ける 4 階層フォールバック

# tier_decider.py
from dataclasses import dataclass
from enum import IntEnum
from typing import Literal
 
class Tier(IntEnum):
    T0_FULL = 0
    T1_REDUCED = 1
    T2_CACHED = 2
    T3_STATIC = 3
 
@dataclass(frozen=True)
class SLISnapshot:
    """直近 1 分の SLI 集計。外部の Prometheus/CW から渡される前提。"""
    p99_latency_ms: float
    err_5xx_rate: float
    err_429_rate: float
    burn_rate_1h: float  # SLO 月次目標に対する直近 1h の消費倍率
 
# Tier 降格判定のルール群（優先順位順）
def decide_tier(
    current: Tier,
    sli: SLISnapshot,
    feature_priority: Literal["critical", "standard", "optional"] = "standard",
) -> tuple[Tier, str]:
    """次に取るべき Tier と、その判断理由のテキストを返す。"""
 
    # 1. SLO バーンレート 14.4 倍超え -> 即時 T2 or T3
    if sli.burn_rate_1h >= 14.4:
        if feature_priority == "optional":
            return Tier.T3_STATIC, f"burn_rate={sli.burn_rate_1h:.1f} optional -> T3"
        return Tier.T2_CACHED, f"burn_rate={sli.burn_rate_1h:.1f} -> T2"
 
    # 2. 5xx 率 2% 超 -> 1 段降格
    if sli.err_5xx_rate >= 0.02:
        return _step_down(current, reason=f"5xx={sli.err_5xx_rate:.1%}")
 
    # 3. 429 率 5% 超 -> 1 段降格
    if sli.err_429_rate >= 0.05:
        return _step_down(current, reason=f"429={sli.err_429_rate:.1%}")
 
    # 4. p99 レイテンシ 800ms 超 -> 1 段降格
    if sli.p99_latency_ms >= 800.0:
        return _step_down(current, reason=f"p99={sli.p99_latency_ms:.0f}ms")
 
    # 5. すべての指標が healthy -> 1 段昇格（30 秒のヒステリシスは呼び出し側で）
    if (
        sli.p99_latency_ms < 500.0
        and sli.err_5xx_rate < 0.005
        and sli.err_429_rate < 0.01
        and sli.burn_rate_1h < 2.0
    ):
        return _step_up(current, reason="all SLIs healthy")
 
    return current, "no change"
 
def _step_down(current: Tier, reason: str) -> tuple[Tier, str]:
    next_tier = Tier(min(current.value + 1, Tier.T3_STATIC.value))
    return next_tier, f"step_down to {next_tier.name}: {reason}"
 
def _step_up(current: Tier, reason: str) -> tuple[Tier, str]:
    next_tier = Tier(max(current.value - 1, Tier.T0_FULL.value))
    return next_tier, f"step_up to {next_tier.name}: {reason}"

// claude-degraded-client.ts
import Anthropic from "@anthropic-ai/sdk";
 
type Tier = 0 | 1 | 2 | 3;
 
interface DegradedClientOptions {
  apiKey: string;
  cacheGet: (key: string) => Promise<string | null>;
  cachePut: (key: string, value: string, ttlSec: number) => Promise<void>;
  currentTier: () => Promise<Tier>; // tier_decider の出力を返す
  featurePriority: "critical" | "standard" | "optional";
}
 
const MODEL_BY_TIER: Record<0 | 1, string> = {
  0: "claude-sonnet-4-6",
  1: "claude-haiku-4-5-20251001",
};
 
export class ClaudeDegradedClient {
  private readonly anthropic: Anthropic;
 
  constructor(private readonly opts: DegradedClientOptions) {
    this.anthropic = new Anthropic({ apiKey: opts.apiKey });
  }
 
  async respond(
    prompt: string,
    cacheKey: string,
    staticFallback: string,
  ): Promise<{ text: string; tierUsed: Tier; cached: boolean }> {
    const tier = await this.opts.currentTier();
 
    // Tier 2: セマンティックキャッシュ優先
    if (tier >= 2) {
      const hit = await this.opts.cacheGet(cacheKey);
      if (hit) return { text: hit, tierUsed: 2, cached: true };
      if (tier === 3) {
        // T3: そもそも API を叩かない
        return { text: staticFallback, tierUsed: 3, cached: false };
      }
      // T2 でキャッシュミスのときは、コスト最小の T1 で再生成
      const text = await this.callModel(prompt, 1, /*thinking=*/ false);
      await this.opts.cachePut(cacheKey, text, 3600);
      return { text, tierUsed: 2, cached: false };
    }
 
    // Tier 0 / 1: 通常の Claude API 呼び出し
    const useThinking = tier === 0 && this.opts.featurePriority === "critical";
    const text = await this.callModel(prompt, tier, useThinking);
 
    // 成功応答は次回のキャッシュ候補として保存
    await this.opts.cachePut(cacheKey, text, 600);
    return { text, tierUsed: tier, cached: false };
  }
 
  private async callModel(
    prompt: string,
    tier: 0 | 1,
    thinking: boolean,
  ): Promise<string> {
    const model = MODEL_BY_TIER[tier];
    const msg = await this.anthropic.messages.create({
      model,
      max_tokens: tier === 0 ? 2048 : 1024,
      messages: [{ role: "user", content: prompt }],
      ...(thinking
        ? { thinking: { type: "enabled", budget_tokens: 4096 } }
        : {}),
    });
    return msg.content
      .filter((b) => b.type === "text")
      .map((b) => (b as { text: string }).text)
      .join("");
  }
}

# .github/workflows/chaos-degradation-drill.yml
name: chaos-degradation-drill
on:
  schedule:
    - cron: "0 18 * * 3"   # 毎週水曜 03:00 JST
  workflow_dispatch:
 
jobs:
  drill:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with:
          node-version: "22"
      - name: Inject synthetic 429 storm
        run: |
          node scripts/inject-fault.mjs \
            --target=preview \
            --pattern=429-storm \
            --duration=180
 
      - name: Verify Tier descended to T2 or below
        run: |
          node scripts/assert-tier.mjs --expected-max=2 --within=60s
 
      - name: Verify cached responses still served
        run: |
          node scripts/assert-cache-hit-rate.mjs --min=0.8
 
      - name: Recovery within 5 minutes
        run: |
          node scripts/wait-tier.mjs --expected=0 --within=300s
 
      - name: Post drill report to ops channel
        if: always()
        run: |
          node scripts/post-drill-report.mjs --to=ops --result="${{ job.status }}"