Claude API プロダクション耐障害設計パターン — モデルルーティング・サーキットブレーカー・フォールバック戦略の実装ガイド

// model-router.ts — タスク特性に応じた動的モデル選択
interface ModelConfig {
  id: string;
  name: string;
  maxTokens: number;
  costPer1kInput: number;   // USD
  costPer1kOutput: number;  // USD
  avgLatencyMs: number;
  isAvailable: boolean;
  circuitBreaker: ClaudeCircuitBreaker;
}
 
interface RoutingRequest {
  taskType: 'simple' | 'complex' | 'creative' | 'code';
  inputTokenEstimate: number;
  latencyBudgetMs: number;
  qualityPriority: 'cost' | 'balanced' | 'quality';
}
 
class IntelligentModelRouter {
  private models: Map<string, ModelConfig> = new Map();
  private metrics: Map<string, { successRate: number; p95Latency: number }> = new Map();
 
  registerModel(config: ModelConfig): void {
    this.models.set(config.id, config);
    this.metrics.set(config.id, { successRate: 1.0, p95Latency: config.avgLatencyMs });
  }
 
  selectModel(request: RoutingRequest): ModelConfig {
    const candidates = Array.from(this.models.values())
      .filter(m => m.isAvailable)
      .filter(m => m.circuitBreaker.getState() !== 'OPEN');
 
    if (candidates.length === 0) {
      throw new Error('No available models — all circuits are OPEN');
    }
 
    // スコアリングでモデルを選択
    const scored = candidates.map(model => ({
      model,
      score: this.calculateScore(model, request),
    }));
 
    scored.sort((a, b) => b.score - a.score);
    const selected = scored[0].model;
 
    console.log(
      `[Router] Selected ${selected.name} (score: ${scored[0].score.toFixed(2)}) ` +
      `for ${request.taskType} task`
    );
    return selected;
  }
 
  private calculateScore(model: ModelConfig, request: RoutingRequest): number {
    const metric = this.metrics.get(model.id)!;
    let score = 0;
 
    // 1. 信頼性スコア（成功率）
    score += metric.successRate * 30;
 
    // 2. レイテンシスコア（予算内に収まるか）
    const latencyFit = Math.max(
      0,
      1 - metric.p95Latency / request.latencyBudgetMs
    );
    score += latencyFit * 25;
 
    // 3. コストスコア
    const estimatedCost =
      (request.inputTokenEstimate / 1000) * model.costPer1kInput;
    const costScore = 1 / (1 + estimatedCost * 100);
    score += costScore * (request.qualityPriority === 'cost' ? 35 : 15);
 
    // 4. タスク適合スコア
    score += this.taskFitScore(model, request.taskType) * 30;
 
    return score;
  }
 
  private taskFitScore(model: ModelConfig, taskType: string): number {
    // Opus: 複雑なタスク・創造的タスクに最適
    // Sonnet: バランス型、コーディングに強い
    // Haiku: 高速・低コスト、シンプルなタスクに最適
    const fitMatrix: Record<string, Record<string, number>> = {
      'claude-opus-4-6':   { simple: 0.5, complex: 1.0, creative: 1.0, code: 0.9 },
      'claude-sonnet-4-6': { simple: 0.8, complex: 0.8, creative: 0.7, code: 1.0 },
      'claude-haiku-4-5':  { simple: 1.0, complex: 0.4, creative: 0.3, code: 0.6 },
    };
    return fitMatrix[model.id]?.[taskType] ?? 0.5;
  }
 
  // メトリクスの更新（実際の応答結果をフィードバック）
  updateMetrics(
    modelId: string,
    success: boolean,
    latencyMs: number
  ): void {
    const metric = this.metrics.get(modelId);
    if (!metric) return;
 
    // 指数移動平均で更新
    const alpha = 0.1;
    metric.successRate = metric.successRate * (1 - alpha) + (success ? 1 : 0) * alpha;
    metric.p95Latency = metric.p95Latency * (1 - alpha) + latencyMs * alpha;
  }
}
 
// 使用例
const router = new IntelligentModelRouter();
 
router.registerModel({
  id: 'claude-opus-4-6',
  name: 'Claude Opus 4.6',
  maxTokens: 32_000,
  costPer1kInput: 0.015,
  costPer1kOutput: 0.075,
  avgLatencyMs: 3000,
  isAvailable: true,
  circuitBreaker: new ClaudeCircuitBreaker(),
});
 
router.registerModel({
  id: 'claude-sonnet-4-6',
  name: 'Claude Sonnet 4.6',
  maxTokens: 16_000,
  costPer1kInput: 0.003,
  costPer1kOutput: 0.015,
  avgLatencyMs: 1200,
  isAvailable: true,
  circuitBreaker: new ClaudeCircuitBreaker(),
});
 
const model = router.selectModel({
  taskType: 'code',
  inputTokenEstimate: 2000,
  latencyBudgetMs: 5000,
  qualityPriority: 'balanced',
});
// 期待出力: [Router] Selected Claude Sonnet 4.6 (score: 82.50) for code task

// fallback-chain.ts — 段階的フォールバック実装
interface FallbackStep {
  name: string;
  execute: (input: string) => Promise<string>;
  timeout: number;
}
 
class FallbackChain {
  private steps: FallbackStep[] = [];
 
  addStep(step: FallbackStep): this {
    this.steps.push(step);
    return this;
  }
 
  async execute(input: string): Promise<{ result: string; usedStep: string }> {
    const errors: Array<{ step: string; error: string }> = [];
 
    for (const step of this.steps) {
      try {
        const result = await Promise.race([
          step.execute(input),
          this.createTimeout(step.timeout, step.name),
        ]);
        console.log(`[Fallback] Success with step: ${step.name}`);
        return { result, usedStep: step.name };
      } catch (error) {
        const message = error instanceof Error ? error.message : String(error);
        console.warn(`[Fallback] ${step.name} failed: ${message}`);
        errors.push({ step: step.name, error: message });
        continue;  // 次のステップへフォールバック
      }
    }
 
    throw new AllFallbacksExhaustedError(errors);
  }
 
  private createTimeout(ms: number, stepName: string): Promise<never> {
    return new Promise((_, reject) =>
      setTimeout(() => reject(new Error(`${stepName} timed out after ${ms}ms`)), ms)
    );
  }
}
 
class AllFallbacksExhaustedError extends Error {
  constructor(public errors: Array<{ step: string; error: string }>) {
    super(`All ${errors.length} fallback steps failed`);
    this.name = 'AllFallbacksExhaustedError';
  }
}
 
// プロダクション構成例
const chain = new FallbackChain()
  .addStep({
    name: 'Claude Opus 4.6 (Primary)',
    timeout: 30_000,
    execute: async (input) => {
      const res = await opusBreaker.execute(() =>
        anthropic.messages.create({
          model: 'claude-opus-4-6',
          max_tokens: 4096,
          messages: [{ role: 'user', content: input }],
        })
      );
      return res.content[0].type === 'text' ? res.content[0].text : '';
    },
  })
  .addStep({
    name: 'Claude Sonnet 4.6 (Secondary)',
    timeout: 15_000,
    execute: async (input) => {
      const res = await sonnetBreaker.execute(() =>
        anthropic.messages.create({
          model: 'claude-sonnet-4-6',
          max_tokens: 4096,
          messages: [{ role: 'user', content: input }],
        })
      );
      return res.content[0].type === 'text' ? res.content[0].text : '';
    },
  })
  .addStep({
    name: 'Cached Response (Emergency)',
    timeout: 1_000,
    execute: async (input) => {
      // 緊急時: キャッシュから類似回答を返却
      const cached = await responseCache.findSimilar(input);
      if (!cached) throw new Error('No cached response available');
      return `[Cached] ${cached.response}`;
    },
  });
 
const { result, usedStep } = await chain.execute('Explain quantum computing');
// 期待出力: [Fallback] Success with step: Claude Opus 4.6 (Primary)

// retry-strategy.ts — プロダクション対応リトライ実装
interface RetryConfig {
  maxRetries: number;
  baseDelayMs: number;
  maxDelayMs: number;
  retryableErrors: Set<number>;  // HTTP ステータスコード
  retryableErrorTypes: Set<string>;
}
 
class ExponentialBackoffRetry {
  private config: RetryConfig;
 
  constructor(config: Partial<RetryConfig> = {}) {
    this.config = {
      maxRetries: 3,
      baseDelayMs: 1_000,
      maxDelayMs: 60_000,
      retryableErrors: new Set([429, 500, 502, 503, 529]),
      retryableErrorTypes: new Set([
        'overloaded_error',
        'api_error',
        'rate_limit_error',
      ]),
      ...config,
    };
  }
 
  async execute<T>(fn: () => Promise<T>): Promise<T> {
    let lastError: Error | null = null;
 
    for (let attempt = 0; attempt <= this.config.maxRetries; attempt++) {
      try {
        return await fn();
      } catch (error) {
        lastError = error instanceof Error ? error : new Error(String(error));
 
        if (attempt === this.config.maxRetries) break;
        if (!this.isRetryable(error)) {
          console.warn(`[Retry] Non-retryable error: ${lastError.message}`);
          throw lastError;
        }
 
        // レート制限の場合: Retry-After ヘッダーを優先
        const retryAfter = this.extractRetryAfter(error);
        const delay = retryAfter ?? this.calculateDelay(attempt);
 
        console.log(
          `[Retry] Attempt ${attempt + 1}/${this.config.maxRetries} — ` +
          `waiting ${delay}ms (${retryAfter ? 'Retry-After' : 'exponential backoff'})`
        );
 
        await this.sleep(delay);
      }
    }
 
    throw lastError!;
  }
 
  private calculateDelay(attempt: number): number {
    // 指数バックオフ: baseDelay * 2^attempt
    const exponential = this.config.baseDelayMs * Math.pow(2, attempt);
    // 上限を適用
    const capped = Math.min(exponential, this.config.maxDelayMs);
    // フルジッター: 0〜capped の一様乱数（Thundering Herd防止）
    return Math.random() * capped;
  }
 
  private isRetryable(error: unknown): boolean {
    if (error && typeof error === 'object') {
      const e = error as Record<string, unknown>;
      // Anthropic SDKのエラー型をチェック
      if (typeof e.status === 'number' && this.config.retryableErrors.has(e.status)) {
        return true;
      }
      if (typeof e.type === 'string' && this.config.retryableErrorTypes.has(e.type)) {
        return true;
      }
    }
    return false;
  }
 
  private extractRetryAfter(error: unknown): number | null {
    if (error && typeof error === 'object') {
      const headers = (error as Record<string, unknown>).headers;
      if (headers && typeof headers === 'object') {
        const retryAfter = (headers as Record<string, string>)['retry-after'];
        if (retryAfter) {
          const seconds = parseFloat(retryAfter);
          if (!isNaN(seconds)) return seconds * 1000;
        }
      }
    }
    return null;
  }
 
  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}
 
// 使用例
const retry = new ExponentialBackoffRetry({
  maxRetries: 3,
  baseDelayMs: 1_000,
  maxDelayMs: 30_000,
});
 
const response = await retry.execute(() =>
  anthropic.messages.create({
    model: 'claude-sonnet-4-6',
    max_tokens: 1024,
    messages: [{ role: 'user', content: 'Hello' }],
  })
);
// 期待出力:
// 成功時: APIレスポンス
// リトライ時: [Retry] Attempt 1/3 — waiting 847ms (exponential backoff)

// health-monitor.ts — 統合ヘルスモニタリング
interface HealthStatus {
  overall: 'healthy' | 'degraded' | 'unhealthy';
  models: Record<string, {
    circuitState: CircuitState;
    successRate: number;
    p95LatencyMs: number;
    requestsPerMinute: number;
  }>;
  fallback: {
    cacheHitRate: number;
    cacheSize: number;
  };
  timestamp: string;
}
 
class HealthMonitor {
  private requestLog: Array<{
    modelId: string;
    success: boolean;
    latencyMs: number;
    timestamp: number;
  }> = [];
 
  recordRequest(
    modelId: string,
    success: boolean,
    latencyMs: number
  ): void {
    this.requestLog.push({
      modelId,
      success,
      latencyMs,
      timestamp: Date.now(),
    });
 
    // 直近5分のログのみ保持（メモリ管理）
    const fiveMinAgo = Date.now() - 5 * 60_000;
    this.requestLog = this.requestLog.filter(r => r.timestamp > fiveMinAgo);
  }
 
  getStatus(
    models: Map<string, ModelConfig>,
    cache: { hitRate: number; size: number }
  ): HealthStatus {
    const modelStatuses: HealthStatus['models'] = {};
 
    for (const [id, model] of models) {
      const recent = this.requestLog.filter(
        r => r.modelId === id && r.timestamp > Date.now() - 60_000
      );
      const successRate = recent.length > 0
        ? recent.filter(r => r.success).length / recent.length
        : 1.0;
      const latencies = recent.map(r => r.latencyMs).sort((a, b) => a - b);
      const p95Index = Math.floor(latencies.length * 0.95);
 
      modelStatuses[id] = {
        circuitState: model.circuitBreaker.getState(),
        successRate,
        p95LatencyMs: latencies[p95Index] ?? 0,
        requestsPerMinute: recent.length,
      };
    }
 
    // 全体のヘルス判定
    const states = Object.values(modelStatuses);
    const allOpen = states.every(s => s.circuitState === 'OPEN');
    const anyOpen = states.some(s => s.circuitState === 'OPEN');
 
    return {
      overall: allOpen ? 'unhealthy' : anyOpen ? 'degraded' : 'healthy',
      models: modelStatuses,
      fallback: {
        cacheHitRate: cache.hitRate,
        cacheSize: cache.size,
      },
      timestamp: new Date().toISOString(),
    };
  }
}
 
// Express/Hono エンドポイント例
// GET /health
app.get('/health', (req, res) => {
  const status = monitor.getStatus(router.models, responseCache.stats());
  const httpCode = status.overall === 'healthy' ? 200
    : status.overall === 'degraded' ? 200
    : 503;
  res.status(httpCode).json(status);
});
// 期待出力:
// {
//   "overall": "healthy",
//   "models": {
//     "claude-opus-4-6": { "circuitState": "CLOSED", "successRate": 0.98, ... },
//     "claude-sonnet-4-6": { "circuitState": "CLOSED", "successRate": 1.0, ... }
//   },
//   ...
// }

// disaster-recovery.ts — DR戦略の統合実装
interface DRConfig {
  primaryProvider: 'anthropic';
  queueMaxSize: number;         // キュー最大サイズ
  queueFlushInterval: number;   // キュー処理間隔（ms）
  graceDegradation: boolean;    // グレースフルデグラデーション有効化
}
 
class DisasterRecoveryManager {
  private requestQueue: Array<{
    id: string;
    input: string;
    priority: number;
    enqueuedAt: number;
    callback: (result: string) => void;
  }> = [];
 
  constructor(
    private config: DRConfig,
    private fallbackChain: FallbackChain,
    private healthMonitor: HealthMonitor
  ) {
    // キューの定期フラッシュ
    setInterval(() => this.flushQueue(), config.queueFlushInterval);
  }
 
  async handleRequest(input: string, priority = 5): Promise<string> {
    const health = this.healthMonitor.getStatus(/* ... */);
 
    switch (health.overall) {
      case 'healthy':
        // 通常運用: フォールバックチェーン経由で処理
        return (await this.fallbackChain.execute(input)).result;
 
      case 'degraded':
        // 縮退運用: 高優先度のみ即時処理、低優先度はキュー
        if (priority >= 7) {
          return (await this.fallbackChain.execute(input)).result;
        }
        return this.enqueue(input, priority);
 
      case 'unhealthy':
        // 全停止: すべてキューに入れ、キャッシュ応答を返却
        if (this.config.graceDegradation) {
          this.enqueue(input, priority);
          return this.getGracefulResponse(input);
        }
        throw new ServiceUnavailableError('All API providers are down');
    }
  }
 
  private enqueue(input: string, priority: number): string {
    if (this.requestQueue.length >= this.config.queueMaxSize) {
      // キュー溢れ: 最低優先度を除去
      this.requestQueue.sort((a, b) => a.priority - b.priority);
      this.requestQueue.shift();
    }
 
    const id = crypto.randomUUID();
    this.requestQueue.push({
      id,
      input,
      priority,
      enqueuedAt: Date.now(),
      callback: () => {},
    });
 
    console.log(
      `[DR] Request ${id} queued (priority: ${priority}, ` +
      `queue size: ${this.requestQueue.length})`
    );
    return `リクエストを受け付けました（ID: ${id}）。サービス復旧後に処理されます。`;
  }
 
  private async flushQueue(): Promise<void> {
    const health = this.healthMonitor.getStatus(/* ... */);
    if (health.overall === 'unhealthy' || this.requestQueue.length === 0) return;
 
    // 優先度順に処理（高い方から）
    this.requestQueue.sort((a, b) => b.priority - a.priority);
    const batch = this.requestQueue.splice(0, 5);  // 5件ずつ処理
 
    for (const req of batch) {
      try {
        const { result } = await this.fallbackChain.execute(req.input);
        req.callback(result);
        console.log(`[DR] Flushed queued request ${req.id}`);
      } catch {
        // 失敗したらキューに戻す
        this.requestQueue.push(req);
      }
    }
  }
 
  private getGracefulResponse(input: string): string {
    return '現在、AIアシスタントの応答に一時的な遅延が発生しています。' +
      'リクエストはキューに登録されました。復旧次第、結果をお届けします。';
  }
}

// resilience-manager.ts — 統合耐障害マネージャー
class ClaudeResilienceManager {
  private router: IntelligentModelRouter;
  private retry: ExponentialBackoffRetry;
  private fallbackChain: FallbackChain;
  private healthMonitor: HealthMonitor;
  private drManager: DisasterRecoveryManager;
 
  constructor() {
    this.router = new IntelligentModelRouter();
    this.retry = new ExponentialBackoffRetry({ maxRetries: 3 });
    this.healthMonitor = new HealthMonitor();
 
    // モデル登録（各モデルにサーキットブレーカーを設定）
    this.setupModels();
    // フォールバックチェーン構築
    this.fallbackChain = this.buildFallbackChain();
    // DR マネージャー
    this.drManager = new DisasterRecoveryManager(
      { primaryProvider: 'anthropic', queueMaxSize: 1000,
        queueFlushInterval: 10_000, graceDegradation: true },
      this.fallbackChain,
      this.healthMonitor
    );
  }
 
  async chat(
    input: string,
    options: { taskType?: 'simple' | 'complex' | 'creative' | 'code';
              priority?: number } = {}
  ): Promise<string> {
    const { taskType = 'simple', priority = 5 } = options;
    const startTime = Date.now();
 
    try {
      const result = await this.drManager.handleRequest(input, priority);
      this.healthMonitor.recordRequest('primary', true, Date.now() - startTime);
      return result;
    } catch (error) {
      this.healthMonitor.recordRequest('primary', false, Date.now() - startTime);
      throw error;
    }
  }
 
  getHealth(): HealthStatus {
    return this.healthMonitor.getStatus(
      this.router['models'],
      { hitRate: 0, size: 0 }
    );
  }
 
  private setupModels(): void { /* 前述のモデル登録 */ }
  private buildFallbackChain(): FallbackChain { /* 前述のチェーン構築 */ }
}
 
// アプリケーション全体で1つのインスタンスを共有
export const claude = new ClaudeResilienceManager();
 
// 使用例
const answer = await claude.chat(
  'TypeScriptで型安全なAPIクライアントを書いてください',
  { taskType: 'code', priority: 8 }
);
// 期待出力: （通常時）Claude Opus/Sonnet の応答
//         （障害時）自動フォールバック or キュー登録メッセージ

if (error.status === 529) {
  // overloaded_error は retry-after がない。広めのジッターで待つ
  const wait = baseDelay * 2.0 * (1 + Math.random());
  await sleep(wait);
  return this.execute(input, attempt + 1);
}

import asyncio
import logging
from dataclasses import dataclass, field
from typing import AsyncIterator, Optional
import anthropic
 
log = logging.getLogger(__name__)
 
@dataclass
class StreamState:
    """ストリーミング中の部分的な応答を保持する。
    再接続時はこの状態を参照して続きを再構築する。
    """
    message_id: Optional[str] = None
    model: Optional[str] = None
    blocks: list[dict] = field(default_factory=list)
    current_block_index: int = -1
    stop_reason: Optional[str] = None
    last_event_id: Optional[str] = None  # SSE の再接続ヒントに使う
 
    def append_delta(self, index: int, delta: dict) -> None:
        """content_block_delta を現在のブロックに追加する。"""
        while len(self.blocks) <= index:
            self.blocks.append({"type": "text", "text": ""})
        block = self.blocks[index]
        if delta.get("type") == "text_delta":
            block["text"] = block.get("text", "") + delta.get("text", "")
        elif delta.get("type") == "input_json_delta":
            block["partial_json"] = block.get("partial_json", "") + delta.get("partial_json", "")
 
    def text(self) -> str:
        return "".join(b.get("text", "") for b in self.blocks if b.get("type") == "text")
 
class ResilientStream:
    """Claude API のストリーミングを、切断・再送・部分応答に対して堅牢に扱うラッパー。"""
 
    def __init__(self, client: anthropic.AsyncAnthropic, max_retries: int = 3):
        self.client = client
        self.max_retries = max_retries
        self.state = StreamState()
 
    async def iterate(self, **create_kwargs) -> AsyncIterator[dict]:
        """完了まで安全にイベントを返す。再接続は内部で隠蔽する。"""
        attempt = 0
        while True:
            try:
                async with self.client.messages.stream(**create_kwargs) as stream:
                    async for event in stream:
                        self._update_state(event)
                        yield event
                    # 正常に終了
                    return
            except (anthropic.APIConnectionError, anthropic.APITimeoutError) as e:
                attempt += 1
                if attempt > self.max_retries:
                    log.error("stream failed after %d retries: %s", attempt, e)
                    raise
                backoff = min(2 ** attempt, 30)
                log.warning("stream disconnected (attempt %d): %s. retrying in %ds",
                            attempt, e, backoff)
                await asyncio.sleep(backoff)
                # 再接続時には現時点までの差分を引き継いで再送要求する
                create_kwargs = self._rebuild_kwargs(create_kwargs)
 
    def _update_state(self, event) -> None:
        et = event.type
        if et == "message_start":
            self.state.message_id = event.message.id
            self.state.model = event.message.model
        elif et == "content_block_start":
            self.state.current_block_index = event.index
        elif et == "content_block_delta":
            self.state.append_delta(event.index, event.delta.model_dump())
        elif et == "message_delta":
            if event.delta.stop_reason:
                self.state.stop_reason = event.delta.stop_reason
 
    def _rebuild_kwargs(self, original: dict) -> dict:
        """途中切断後の再接続用にメッセージ配列を組み直す。
        最後のユーザーメッセージに「続きを生成してほしい」という指示を付けず、
        これまでに得た部分応答を assistant メッセージとして追加する。
        """
        rebuilt = dict(original)
        partial_text = self.state.text()
        if not partial_text:
            return rebuilt
        messages = list(rebuilt.get("messages", []))
        messages.append({"role": "assistant", "content": partial_text})
        rebuilt["messages"] = messages
        return rebuilt

from hashlib import sha256
 
class DedupBuffer:
    """SSE イベントの重複配信を検出し、2回目以降を捨てる。"""
 
    def __init__(self, window_size: int = 512):
        self.seen: set[str] = set()
        self.order: list[str] = []
        self.window_size = window_size
 
    def _key(self, event) -> str:
        # message_id + index + 位置（累計 delta サイズ）で一意化
        body = f"{event.message_id}:{event.index}:{event.position}"
        return sha256(body.encode()).hexdigest()
 
    def is_duplicate(self, event) -> bool:
        k = self._key(event)
        if k in self.seen:
            return True
        self.seen.add(k)
        self.order.append(k)
        if len(self.order) > self.window_size:
            old = self.order.pop(0)
            self.seen.discard(old)
        return False

import json
 
def safe_parse_tool_input(block: dict) -> Optional[dict]:
    """tool_use ブロックの partial_json を安全にパースする。
    途中で切れていれば None を返す。"""
    raw = block.get("partial_json", "").strip()
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        log.warning("incomplete tool input: %s (raw=%r)", e, raw[:200])
        return None

import random
 
async def smart_sleep(attempt: int, error: Exception) -> float:
    """エラー種別に応じたバックオフを返す（実際に sleep する）。"""
    if isinstance(error, anthropic.RateLimitError):
        # レート制限はサーバーのヘッダ指示を尊重し、不足分をジッタで足す
        retry_after = getattr(error, "retry_after", None) or 30
        delay = retry_after + random.uniform(0, 5)
    elif isinstance(error, (anthropic.APIConnectionError, anthropic.APITimeoutError)):
        # 接続系は短めの指数バックオフ
        delay = min(2 ** attempt, 20) + random.uniform(0, 1)
    else:
        delay = min(2 ** attempt, 60)
    log.info("sleeping %.2fs before retry (attempt=%d, error=%s)", delay, attempt, type(error).__name__)
    await asyncio.sleep(delay)
    return delay

# ❌ 悪い例 — 途中で break すると接続がリークする
stream = client.messages.stream(**kwargs)
async for event in stream:
    if something:
        break  # この時点でコネクションが宙に浮く

# ❌ 悪い例 — 毎イベントで巨大な文字列連結が走り、遅くなる
accumulated = ""
async for event in stream:
    if event.type == "content_block_delta":
        accumulated += event.delta.text
        render(accumulated)  # 毎回フル文字列を描画

async for event in stream:
    if event.type == "error":
        err = event.error
        log.error("api error in stream: %s (%s)", err.message, err.type)
        if err.type == "overloaded_error":
            # 容量不足はリトライ価値あり
            raise anthropic.APIStatusError(err.message)
        raise RuntimeError(f"fatal api error: {err}")