インテリジェントモデルルーティングの設計
プロダクション環境では、単一モデルに依存するのではなく、タスクの特性やシステム負荷に応じて最適なモデルを動的に選択する「インテリジェントルーティング」が有効です。
// 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)
フォールバックチェーンの最終段に「キャッシュからの応答」を配置することで、APIが完全にダウンした場合でもユーザーに何らかのレスポンスを返せるようになります。これはUX上重要な設計です。
指数バックオフ × ジッター付きリトライ戦略
リトライは耐障害設計の基本ですが、実装を誤るとかえって障害を悪化させます。特に多数のクライアントが同時にリトライする「Thundering Herd（雷鳴の群れ）」問題を防ぐため、指数バックオフ + フルジッターの組み合わせが必須です。
// 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)
Retry-After ヘッダーを最優先で利用する点が重要です。Claude APIはレート制限時に正確なリトライ時間をヘッダーで返却するため、これを無視して固定の待機時間を使うと、不必要に長い待機や早すぎるリトライの原因になります。レート制限の仕組みとベストプラクティスについてはAPI レート制限とベストプラクティスも参考にしてください。
ヘルスチェックとメトリクス収集
耐障害設計を運用するには、各コンポーネントの状態を常時監視するヘルスチェックが不可欠です。
// 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, ... }
// },
// ...
// }
このヘルスチェックエンドポイントをGrafanaやDatadogなどのモニタリングツールと連携させることで、障害の予兆を早期に検出し、サーキットブレーカーが動作する前に対処することも可能になります。
ディザスタリカバリ戦略
最悪のシナリオ——Claude APIのサービス全体が長時間停止した場合——に備えたディザスタリカバリ（DR）戦略も設計しておきましょう。
// 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アシスタントの応答に一時的な遅延が発生しています。' +
'リクエストはキューに登録されました。復旧次第、結果をお届けします。';
}
}
グレースフルデグラデーション（段階的な品質低下）を取り入れることで、完全なサービス停止を避け、ユーザーには「遅延はあるが応答は得られる」という体験を提供できます。
全パターンを統合するResilienceManagerの構築
ここまで紹介した5つのパターンを1つのクラスに統合し、プロダクションで使いやすいインターフェースにまとめましょう。
// 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 キュー登録メッセージ
公式ドキュメントには書かれていない6つの運用知見
SDK や公式ドキュメントの記述だけでは見えにくい挙動がいくつかあります。ここでは、設計判断に直接効く6つの注意点を整理します。
1. 429だけでなく529（overloaded_error）も無視できない
レート制限は429として返るのが一般的ですが、ピーク時間帯（特にUTC 13:00〜16:00、米西海岸の業務開始タイム）になると529 overloaded_error の比率が増えます。529には retry-after ヘッダが付かないため、429と同じハンドリングだけでは適切なバックオフがかかりません。私の運用では status === 529 を別ルートに分岐させ、baseDelay を 2.0 倍に広げてジッターも拡大したところ、平均リトライ回数を 1.7 回から 0.9 回まで下げられました。
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);
}
2. サーキットブレーカー閾値は「ユースケース」で動的に変える
公式ドキュメントの推奨値（5エラー/30秒）は中規模システム向けの一般値です。バッチ的・低頻度の処理とリアルタイムの応答が混在するシステムでは、用途別に閾値を分けるのが無難です。
- バッチ処理: 3エラー/60秒・半開復帰120秒 — バッチなので即停止で問題なし
- 読者AIアシスタント: 7エラー/30秒・半開復帰30秒 — UX重視で早めに復帰
- 管理画面（内部用）: 10エラー/60秒・半開復帰60秒 — 利用者が少なく寛容で良い
これを一律「5エラー/30秒」で運用したときは、バッチが無駄にAPIを叩いたり、UI側で過剰にCLOSEDに戻ったりして、かえって体験が悪くなりました。閾値はSLOから逆算するもので、テンプレ的な一律値ではうまくいきません。
3. Cloudflare WorkersではResilienceManagerをそのまま動かせない
CloudflareのCPU time limitはFreeプラン10ms、Bundledプラン50ms、Unboundでも30秒という上限があります。ResilienceManager を setInterval でヘルスチェックを回したり、フォールバックチェーンで複数モデルを順番に試したりすると、CPU時間を使い切ってハードに落ちます。Cloudflare Workers + OpenNext の構成では、次のように組み替えると落ちにくくなります。
- ヘルスチェックは別の Cron Triggers Worker へ切り出し、Durable Objects に記録
- リクエスト中のフォールバックは最大2モデル（Sonnet → Haiku）まで
- Supervisor ロジックは別ワーカーへ分離し、メイン側はステートレスのまま維持
Workers のステートレス前提はこういう設計でしか守れません。逆にこれを守ると、デプロイのロールアウトもブルー/グリーンで安全になります。
4. Supervisor用LLMにはHaikuを使う方が現実的
「監視ロジックもLLMでやる」構成のとき、Supervisor に Sonnet を使うと意外と高くつきます。1回の障害判定で約3,000トークン消費し、Sonnet 価格で約¥1.5。1日10〜20回呼ばれると月額¥600〜¥1,000になります。同じ判断を Haiku に任せると約1/4のコスト（月額¥150〜¥250）で、精度差はほぼ気になりませんでした。Supervisor タスクは「YES/NO/RETRY」を返すだけのシンプルなプロンプトなので、Haiku で十分機能します。Sonnet を使うなら本体の生成タスクに回した方が、ユーザー体験への投資対効果が高いと感じます。
5. フォールバックの「品質劣化」をユーザーに伝えるかどうか
技術的にはOpus → Sonnet → Haikuと段階的に落とせますが、ユーザーに「今は軽量モデルで応答しています」と伝えるかどうかは別問題です。ユースケースごとに使い分けるのが現実的です。
- 無料の対話機能: 伝えません。「少し遅いですね」で収まる範囲に留めます
- 有料会員向けの生成処理: 劣化中であることを伝えます。「現在Sonnetで生成中（通常時はOpus）」を表示します
- 管理画面: 詳細な状態を表示
この設計は、個人開発で自分のアプリやサービスを長く運営するなかで得た「ユーザーは詳細な技術情報より、何が起きていて何ができるかを知りたい」という体感から来ています。透明性は誠実さの表れですが、過剰な開示は逆にノイズになります。
6. ディザスタリカバリのRTO/RPOは「サイトの収益構造」で決まる
ResilienceManager設計の最終目標はRTO（Recovery Time Objective）とRPO（Recovery Point Objective）の達成ですが、これは技術的な目標ではなく事業的な目標です。個人運営のサービスを例にすると、以下のように整理できます。
- Stripe決済フロー: RTO 5分以内・RPO 0 — 完全冗長 + 即時キュー登録
- 夜間バッチ処理: RTO 1時間以内・RPO 1時間 — キューバックフィル + Cron復旧
- 無料の対話機能: RTO 15分以内・RPO なし — グレースフルデグラデーション
- ブログ閲覧（静的）: RTO 30秒以内・RPO 0 — Cloudflare CDNフォールバック
「全部を5分以内」にしようとすると過剰投資になります。個人開発で収益が立ち始めた頃に学んだのは、「収益にどう効くか」を起点に逆算する習慣でした。Stripe 決済フローは1分のダウンが¥10,000の機会損失になるなら投資が正当化されますが、内部ツールに同じ予算をかけるのは見合いません。RTO/RPO は技術カタログから選ぶものではなく、事業の数字から導かれるものだと考えています。
ストリーミングを本番で落とさない — 切断・重複・途中終了への備え
ここまでは通常のリクエスト・レスポンスを前提にした耐障害設計を扱ってきました。ストリーミングは、その前提が崩れる場所です。
サーバーとクライアントが長時間つながり続けるため、切断・重複・途中終了が「正常動作の範囲」で起こります。同じ ResilienceManager の発想を、ストリーミング特有の壊れ方に合わせて作り直します。
なぜストリーミングは本番で崩れるのか
ストリーミング API は、ふつうのリクエスト・レスポンスモデルよりも壊れやすい構造を持っています。サーバーとクライアントが長時間つながり続けることを前提にしているため、次のような事象がすべて「正常動作の範囲で起こりうる」からです。
- ISP・中間プロキシによる、30 秒以上の無通信でのコネクション切断
- デプロイ時のロードバランサによるコネクション切断
- クライアント側 JavaScript runtime の GC・suspend によるバッファの読み落とし
- HTTP/2 の再試行で起きる、サーバーからの同一イベント再送
一括応答ならタイムアウトと再送で済む問題が、ストリーミングでは「部分的に受け取って、部分的に失敗する」という中途半端な状態を生みます。この前提を受け入れた上で、壊れても取り乱さずに立て直せる設計が必要です。
ストリーミング実装を「失敗しないように作る」のではなく「失敗を前提に壊れた時の立て直し方を設計する」と考え方を切り替えるだけで、コードの構造が大きく変わります。
SSE イベントの構造をもう一度整理する
Claude API のストリーミングは Server-Sent Events (SSE) で配信され、代表的なイベントは次の種類です。
message_start — 応答メッセージのメタデータcontent_block_start — テキストやツール呼び出しのブロック開始content_block_delta — ブロックの差分（token ごと）content_block_stop — ブロック終了message_delta — stop_reason など応答全体の差分message_stop — 応答全体の終了ping — keep-aliveerror — API 側エラー
本番で重要なのは「どのイベントの途中で接続が切れたか」を常に把握できる状態に保つことです。特に content_block_delta の途中で切れた場合は、そのブロックの開始インデックスだけを持っていれば、再接続後に継続できます。この「ブロック単位の状態管理」が、堅牢なストリーミング実装の土台になります。
基本構造 — 状態を持つ StreamReader を作る
SDK の生の stream をそのまま使うと、エラーハンドリングが散らばります。私はいつも「現在のメッセージ状態」を保持するクラスでラップしています。
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
このクラスの要点は3つです。第一に、接続が切れたら自動でリトライします。第二に、リトライ時には既に受け取った部分応答を assistant メッセージとして履歴に積み、Claude に続きを書いてもらいます。第三に、状態をすべて StreamState に集約しているので、外部からは「完全な応答が来た」ように見えます。
なぜ再送要求ではなく履歴を積む方式にするのか。Claude API はイベントレベルの再送機構を持っていないので、同じリクエスト ID で再接続しても重複配信が起きる保証がありません。履歴を積む方式は確定的で、リクエスト回数こそ増えますが冪等性が保てます。
重複イベントを安全に取り込む
HTTP/2 のリトライ、プロキシの挙動、あるいはクライアント再接続の実装ミスで、同じ content_block_delta が二度流れてくることがあります。テキストが二重に連結されるとユーザーは即座に気づきますから、重複排除は必須です。
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
ここで大事なのは「何を持って同一とみなすか」を明示的に決めることです。イベントのタイムスタンプは再送時に変わりうるので、タイムスタンプをキーにすると検出漏れが発生します。message_id + index + 累計位置 の3つで識別するのが、私の経験では最も安全でした。
tool_use が途中で切れたときの判定
ツール呼び出しの引数は input_json_delta で流れてきますが、これは JSON の文字列断片が分割されて届きます。途中で切れた場合、partial_json が不完全な JSON のままになり、そのまま json.loads すれば例外です。
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
「ツール呼び出しの実行条件」を「safe_parse_tool_input が None でないこと」に置き換えるだけで、ストリーム中断でツールを暴発させる事故を防げます。私自身、過去に途中切断時に半分のパラメータでツールを実行してしまい、本番データベースに意図しない書き込みをしかけたことがあります。この関数をアプリケーションの境界に置くだけで、その類の事故はほぼゼロにできます。
レート制限と重なった場合の多段バックオフ
ストリーミングのリトライとレート制限の再試行は、素朴に両方実装すると相互作用して嵐のように再試行が発生することがあります。私はレート制限（429）と接続エラーを分離し、前者はより長いバックオフを設定しています。
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
ジッタを入れているのは、複数のクライアントが同時に 429 から復帰しようとすると、再度のスパイクでまた 429 を食らう現象（thundering herd）を避けるためです。小さなジッタ（0〜5秒）を入れるだけで、ピーク集中が目に見えて緩和されます。
よくある3つの落とし穴
最後に、私がチームコードレビューで何度も指摘してきた、ストリーミング実装の典型的なミスを3つ挙げます。
落とし穴1: async for の外で close を呼び忘れる
# ❌ 悪い例 — 途中で break すると接続がリークする
stream = client.messages.stream(**kwargs)
async for event in stream:
if something:
break # この時点でコネクションが宙に浮く
コンテキストマネージャ（async with）を必ず使ってください。break しても自動で close されます。私は本番サーバーで「30分ごとにコネクションが漏れる」現象を追いかけて、最終的にこの1行のミスに辿り着いた経験があります。
落とし穴2: クライアント側でテキストを都度 concat してしまう
# ❌ 悪い例 — 毎イベントで巨大な文字列連結が走り、遅くなる
accumulated = ""
async for event in stream:
if event.type == "content_block_delta":
accumulated += event.delta.text
render(accumulated) # 毎回フル文字列を描画
大規模応答では O(n^2) の計算量になります。list.append で delta を貯めて、表示用には別スレッド or requestIdleCallback でまとめて結合するのが定石です。
落とし穴3: error イベントを例外と混同する
Claude API の error イベントは、HTTP ステータスが 200 のまま SSE ボディに埋め込まれます。SDK によっては例外として上げてくれますが、自前 HTTP クライアントを使う場合は event.type == "error" を自分でチェックする必要があります。
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}")
overloaded_error だけはリトライで回復する可能性が高く、それ以外の error は失敗として扱います。この区別がないと、プロンプトバグで永久にリトライループに入るアプリケーションができあがります。
本番運用で実際に採用しているモニタリング指標
コードを堅牢にするだけでは足りません。壊れている兆候を検知できないと、結局ユーザーが先に気づくことになります。私たちのチームでは、ストリーミングに対して次の4指標を監視しています。
- 成功率:
message_stop まで到達したストリームの割合（目標 99.5% 以上）
- 平均再接続回数: 1ストリームあたりの再接続回数（0.1 を超えたら警告）
- token 出力遅延 P95: 最初の token が出るまでの時間 P95（6秒を超えたら警告）
- 重複イベント率: dedup buffer がヒットした割合（1% 未満を維持）
これらを Datadog や Grafana に送っておくと、ユーザーから「なんか遅い」と言われる前に自分たちで気づけます。監視は味方です。
リトライ予算 — 障害のときだけ請求書が跳ねる仕組みを止める
ここまでの5つのパターンには、手当てしていない穴がひとつ残っています。冒頭で挙げた「コスト爆発リスク」です。サーキットブレーカーは連鎖を断ちますが、開くまでのあいだは通します。指数バックオフは間隔を空けますが、回数そのものは減らしません。
私がこの穴に落ちたのは、上流が 20 秒ほど詰まった夜でした。翌朝いちばんに目へ飛び込んできたのは、落ちた生成の数ではなく、API コール数のグラフです。届いたリクエストは増えていないのに、コールだけが膨らんでいました。リトライは、上流がいちばん弱っている時間帯を狙って負荷を足していたのです。
回数の上限だけでは、この膨らみは止まりません。必要だったのは予算でした。成功したリクエストが少しずつ予算を貯め、リトライがそれを取り崩します。上流が健康なら予算は余り、上流が倒れていれば成功が止まるので予算も止まる——gRPC が retry budget と呼ぶ考え方を、Claude API 向けに小さく書き直したのが次のクラスです。
// retry-budget.ts — 成功が貯め、リトライが取り崩すトークンバケット
export class BudgetExhaustedError extends Error {
constructor(public readonly reason: Error) {
super(`Retry budget exhausted: ${reason.message}`);
this.name = 'BudgetExhaustedError';
}
}
export class RetryBudget {
private tokens: number;
constructor(
private readonly max = 500, // 予算の上限（＝初期値）
private readonly ratio = 0.25, // 成功 1 回あたりに戻すトークン
) {
this.tokens = max;
}
/** リトライしてよいか。よければ 1 トークン消費する */
tryAcquire(): boolean {
if (this.tokens < 1) return false;
this.tokens -= 1;
return true;
}
/** 成功したぶんだけ予算を戻す */
onSuccess(): void {
this.tokens = Math.min(this.max, this.tokens + this.ratio);
}
/** 監視に出す値。0 に張り付いたら上流が倒れているサイン */
get remaining(): number {
return Math.round(this.tokens);
}
}
前に載せた ExponentialBackoffRetry には、2 行足すだけで入ります。
// retry-strategy.ts の execute() — 追加①と追加②の 2 行だけ
async execute<T>(fn: () => Promise<T>): Promise<T> {
let lastError: Error | null = null;
for (let attempt = 0; attempt <= this.config.maxRetries; attempt++) {
try {
const result = await fn();
this.budget.onSuccess(); // 追加①: 成功が予算を戻す
return result;
} catch (error) {
lastError = error instanceof Error ? error : new Error(String(error));
if (attempt === this.config.maxRetries) break;
if (!this.isRetryable(error)) throw lastError;
if (!this.budget.tryAcquire()) { // 追加②: 予算切れなら粘らない
throw new BudgetExhaustedError(lastError);
}
const retryAfter = this.extractRetryAfter(error);
await this.sleep(retryAfter ?? this.calculateDelay(attempt));
}
}
throw lastError ?? new Error('Retry failed');
}
予算がどこから効き始めるかを、手元で測る
上限と ratio をいくつにするかは、勘で決めると外します。実際の障害を待たずに確かめたかったので、同じ障害シナリオを3つの方針で流す小さなシミュレータを書きました。乱数はシード固定なので、手元でも同じ数字が出ます。
// retry-budget-sim.mjs — 同じ障害を3方針で流し、上流コール数と取りこぼしを数える
// node retry-budget-sim.mjs
const mulberry32 = (a) => () => {
a |= 0; a = (a + 0x6D2B79F5) | 0;
let t = Math.imul(a ^ (a >>> 15), 1 | a);
t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
const ARRIVALS = 5; // 秒あたりの到着数
const HORIZON = 300; // 観測する秒数
const MAX_ATTEMPTS = 5;
const OUTAGE_END = 20; // 0〜20 秒: 全滅（529）
const DEGRADED_END = 120; // 20〜120 秒: 40% が 529
const failRate = (t) => (t < OUTAGE_END ? 1.0 : t < DEGRADED_END ? 0.4 : 0.0);
function run(policy, { init = 500, ratio = 0.25 } = {}, seed = 20260921) {
const rnd = mulberry32(seed);
let tokens = init;
const queue = Array.from(
{ length: ARRIVALS * HORIZON },
(_, i) => ({ t: i / ARRIVALS, n: 0 }),
);
let calls = 0, retries = 0, failed = 0, denied = 0;
while (queue.length) {
queue.sort((a, b) => a.t - b.t);
const e = queue.shift();
calls++;
if (rnd() >= failRate(e.t)) { // 成功
tokens = Math.min(init, tokens + ratio);
continue;
}
if (e.n + 1 >= MAX_ATTEMPTS) { failed++; continue; }
if (policy === 'budget') {
if (tokens < 1) { denied++; failed++; continue; }
tokens -= 1;
}
const delay = policy === 'naive' ? 0.05 : rnd() * Math.min(30, 2 ** e.n);
retries++;
queue.push({ t: e.t + delay, n: e.n + 1 });
}
const total = ARRIVALS * HORIZON;
return {
calls, retries, denied,
amp: (calls / total).toFixed(2),
loss: ((failed / total) * 100).toFixed(1),
};
}
const rows = [
['即時リトライのみ', run('naive')],
['バックオフ＋フルジッター', run('jitter')],
['＋予算 200 / ratio 0.1', run('budget', { init: 200, ratio: 0.1 })],
['＋予算 400 / ratio 0.2', run('budget', { init: 400, ratio: 0.2 })],
['＋予算 500 / ratio 0.25', run('budget', { init: 500, ratio: 0.25 })],
['＋予算 600 / ratio 0.3', run('budget', { init: 600, ratio: 0.3 })],
];
for (const [name, r] of rows) {
console.log(`${name} calls=${r.calls} retries=${r.retries} denied=${r.denied} amp=${r.amp} loss=${r.loss}%`);
}
届いたリクエストは 1,500 件です。amp は「リクエスト 1 件あたり上流を何回叩いたか」、loss は最終的に返せなかった割合を表します。
最初に目を留めたのは、いちばん下の行でした。予算 600 / ratio 0.3 は、バックオフだけの行と上流コールも取りこぼしも完全に一致しています。予算切れが一度も起きていないからです。一度も枯れない予算は、入れていないのと同じです。
逆に予算 200 まで絞ると、上流への負荷は 1.49 倍から 1.15 倍へ落ちる代わりに、取りこぼしは 5.2% から 18.7% へ跳ね上がります。リトライ予算は上流を守る魔法ではなく、上流を守るぶんだけ自分の可用性を差し出す取引でした。効きはじめるのは予算切れが数十件だけ出る 500 前後で、ここが探すべき折り返し点です。
私はこの取引を、エンドポイントごとに別々に結ぶようにしています。Stripe の決済まわりは予算を薄く（早く枯れるように）して、上流が倒れているなら即座に諦めてキューへ逃がします。読者向けの生成は予算を広めに取り、多少コールが増えても最後まで粘ります。守りたいのが上流なのか自分の可用性なのかは、エンドポイントごとに違うからです。
監視へ足す指標はひとつで足ります。RetryBudget.remaining を定期的に吐き出し、0 に張り付いた秒数を数えるだけです。サーキットブレーカーが OPEN になるより先に、予算のほうが枯れます。
全体を振り返って
Claude API を本番で支える耐障害設計は、二つの層に分けて考えると整理できます。
ひとつはリクエスト単位の層です。サーキットブレーカーで障害の連鎖を断ち、モデルルーティングとフォールバックチェーンで応答を絶やさず、指数バックオフで安全に retry し、ヘルスチェックで全体を見張ります。これらを ResilienceManager に束ねると、個々のパターンを意識せずに高い可用性を保てます。
もうひとつはストリーミングの層です。状態を持つラッパーで部分応答を保持し、重複を前提に dedup し、エラー種別でバックオフを分けます。
共通する姿勢は「壊れないように作る」ではなく「壊れたときに静かに立て直す」ことです。まずはサーキットブレーカーと retry 戦略から入れ、次にストリーミング実装を一か所だけ ResilientStream に置き換え、最後に監視指標をひとつ足します。この順で進めるだけで、可用性は段階的に底上げされていきます。コスト面もあわせて設計したい方はClaude API コスト最適化プロダクションガイドもご覧ください。