Claude API のモデル抽象レイヤー設計 — 世代交代に業務ロジックを巻き込まない内部アーキテクチャ

// model-registry.ts — モデルに関する「外部の都合」を閉じ込める唯一の場所
export type ModelRole = "fast" | "balanced" | "deep";
 
interface ModelBinding {
  primary: string;        // 通常使う物理モデルID
  fallbacks: string[];    // 障害・廃止時に降りていく順
  inputPer1M: number;     // 入力 100万トークンあたりの単価（USD）
  outputPer1M: number;    // 出力 100万トークンあたりの単価（USD）
  maxOutputTokens: number;
}
 
// ここを書き換えるだけで全アプリのモデル世代交代が完結する
const REGISTRY: Record<ModelRole, ModelBinding> = {
  fast: {
    primary: "claude-haiku-4-5-20251001",
    fallbacks: ["claude-3-5-haiku-latest"],
    inputPer1M: 1.0,
    outputPer1M: 5.0,
    maxOutputTokens: 4096,
  },
  balanced: {
    primary: "claude-sonnet-4-6",
    fallbacks: ["claude-haiku-4-5-20251001"],
    inputPer1M: 3.0,
    outputPer1M: 15.0,
    maxOutputTokens: 8192,
  },
  deep: {
    primary: "claude-opus-4-6",
    fallbacks: ["claude-sonnet-4-6"],
    inputPer1M: 15.0,
    outputPer1M: 75.0,
    maxOutputTokens: 16384,
  },
};
 
export function resolve(role: ModelRole): ModelBinding {
  const binding = REGISTRY[role];
  if (!binding) throw new Error(`unknown role: ${role}`);
  return binding;
}
 
export function chain(role: ModelRole): string[] {
  const b = resolve(role);
  return [b.primary, ...b.fallbacks];
}

// model-gateway.ts — 業務コードはこの ask() だけを知っていればよい
import Anthropic from "@anthropic-ai/sdk";
import { resolve, chain, type ModelRole } from "./model-registry";
 
const client = new Anthropic();
 
interface AskResult {
  text: string;
  modelUsed: string;
  usdCost: number;
  degraded: boolean;   // primary 以外に落ちたか
}
 
export async function ask(
  role: ModelRole,
  prompt: string,
  opts: { maxTokens?: number } = {},
): Promise<AskResult> {
  const binding = resolve(role);
  const candidates = chain(role);
  let lastErr: unknown;
 
  for (let i = 0; i < candidates.length; i++) {
    const model = candidates[i];
    try {
      const res = await client.messages.create({
        model,
        max_tokens: opts.maxTokens ?? binding.maxOutputTokens,
        messages: [{ role: "user", content: prompt }],
      });
      const inTok = res.usage.input_tokens;
      const outTok = res.usage.output_tokens;
      // フォールバック先は単価が違う場合があるので、実際に使ったモデルで再計算する
      const usdCost = priceFor(model, inTok, outTok);
      const text = res.content
        .filter((b) => b.type === "text")
        .map((b) => (b as { text: string }).text)
        .join("");
      return { text, modelUsed: model, usdCost, degraded: i > 0 };
    } catch (err) {
      lastErr = err;
      // 廃止（404 model_not_found）も過負荷（529）も等しく「次へ降りる」で扱う
      continue;
    }
  }
  throw new Error(`all candidates failed for role=${role}: ${String(lastErr)}`);
}
 
function priceFor(model: string, inTok: number, outTok: number): number {
  // ロールに紐づかない実モデル単価表（フォールバック先も網羅）
  const table: Record<string, [number, number]> = {
    "claude-haiku-4-5-20251001": [1.0, 5.0],
    "claude-3-5-haiku-latest": [0.8, 4.0],
    "claude-sonnet-4-6": [3.0, 15.0],
    "claude-opus-4-6": [15.0, 75.0],
  };
  const [pin, pout] = table[model] ?? [0, 0];
  return (inTok / 1e6) * pin + (outTok / 1e6) * pout;
}

// review-summary.ts — 呼び出し側はロールしか知らない
import { ask } from "./model-gateway";
 
export async function summarizeReview(text: string) {
  const { text: summary, degraded } = await ask(
    "fast",
    `次のアプリレビューを 1 文で要約してください:\n${text}`,
  );
  if (degraded) {
    // 降格時はログだけ残し、ユーザー体験は止めない
    console.warn("[review-summary] running on fallback model");
  }
  return summary;
}

# model_gateway.py — TypeScript 版と同じ境界を Python でも引く
from dataclasses import dataclass
from anthropic import Anthropic
 
client = Anthropic()
 
@dataclass(frozen=True)
class Binding:
    primary: str
    fallbacks: tuple[str, ...]
    max_output: int
 
REGISTRY: dict[str, Binding] = {
    "fast":     Binding("claude-haiku-4-5-20251001", ("claude-3-5-haiku-latest",), 4096),
    "balanced": Binding("claude-sonnet-4-6", ("claude-haiku-4-5-20251001",), 8192),
    "deep":     Binding("claude-opus-4-6", ("claude-sonnet-4-6",), 16384),
}
 
def ask(role: str, prompt: str, max_tokens: int | None = None) -> dict:
    b = REGISTRY[role]
    candidates = (b.primary, *b.fallbacks)
    last_err: Exception | None = None
    for i, model in enumerate(candidates):
        try:
            res = client.messages.create(
                model=model,
                max_tokens=max_tokens or b.max_output,
                messages=[{"role": "user", "content": prompt}],
            )
            text = "".join(blk.text for blk in res.content if blk.type == "text")
            return {"text": text, "model_used": model, "degraded": i > 0}
        except Exception as err:  # 廃止・過負荷を等しく「次へ」で扱う
            last_err = err
            continue
    raise RuntimeError(f"all candidates failed for role={role}: {last_err}")

# test_model_contract.py — 各ロールの primary が実在し応答するかを毎日叩く
import pytest
from model_gateway import REGISTRY, client
 
@pytest.mark.parametrize("role", list(REGISTRY.keys()))
def test_primary_model_alive(role):
    model = REGISTRY[role].primary
    res = client.messages.create(
        model=model,
        max_tokens=8,
        messages=[{"role": "user", "content": "ping"}],
    )
    assert res.stop_reason in ("end_turn", "max_tokens")