Gemini API 構造化出力スキーマを本番アプリで安全に進化させる設計記録

// workers/recommend.ts
import { Recommendation, RecommendationV2 } from "../schemas/recommendation";
 
export async function handleRecommend(req: Request, env: Env) {
  const raw = await callGeminiAPI(env, prompt);
  const ir = await normalizeToIR(raw); // 内部 IR への正規化
 
  // 旧版（v1）と新版（v2）を同時に整形して返す
  const v1Payload = toV1Shape(ir);
  const v2Payload = RecommendationV2.parse({
    schemaVersion: "v2",
    theme: ir.theme,
    subtitle: ir.subtitle ?? "",
    confidence: ir.confidence,
  });
 
  return Response.json({
    // 旧クライアント向け
    ...v1Payload,
    // 新クライアント向け（同じレスポンスに同居）
    v2: v2Payload,
    _meta: {
      generated_at: new Date().toISOString(),
      ir_source: ir.source,
    },
  });
}

async function logSample(env: Env, payload: SamplePayload) {
  // 1% サンプリング。本番運用で多すぎず少なすぎないライン
  if (Math.random() > 0.01) return;
 
  const date = new Date().toISOString().slice(0, 10);
  const key = `gemini-samples/${date}/${crypto.randomUUID()}.jsonl`;
 
  const line = JSON.stringify({
    ts: Date.now(),
    schema_version: "v2",
    gemini_raw: payload.raw,
    ir: payload.ir,
    v1_output: payload.v1,
    v2_output: payload.v2,
    // app バージョンや locale も必ず一緒に
    app_version: payload.appVersion,
    locale: payload.locale,
  });
 
  await env.SAMPLES.put(key, line, {
    httpMetadata: { contentType: "application/x-ndjson" },
  });
}

# scripts/schema_compat.py
import json
import re
from pathlib import Path
from collections import Counter
 
# 1. クライアントコードからフィールド参照を抽出
FIELD_RE = re.compile(r"\.recommendation\.(\w+)|response\[['\"](\w+)['\"]\]")
 
def extract_field_usage(app_dirs: list[Path]) -> Counter:
    usage = Counter()
    for d in app_dirs:
        for ext in ("*.swift", "*.kt"):
            for f in d.rglob(ext):
                text = f.read_text(encoding="utf-8", errors="ignore")
                for m in FIELD_RE.finditer(text):
                    field = m.group(1) or m.group(2)
                    if field:
                        usage[field] += 1
    return usage
 
# 2. R2 のサンプリングログから「実際に返している値の分布」を取る
def sample_field_distribution(log_dir: Path) -> dict[str, dict]:
    dist = {}
    for f in log_dir.rglob("*.jsonl"):
        for line in f.read_text().splitlines():
            row = json.loads(line)
            for k, v in (row.get("v1_output") or {}).items():
                d = dist.setdefault(k, {"null": 0, "empty": 0, "value": 0})
                if v is None:
                    d["null"] += 1
                elif v == "":
                    d["empty"] += 1
                else:
                    d["value"] += 1
    return dist
 
# 3. 突き合わせて「破壊的変更か非破壊的変更か」を判定
def classify_change(field: str, removed: bool, became_required: bool,
                    usage: Counter, dist: dict) -> str:
    used = usage.get(field, 0)
    nulls = dist.get(field, {}).get("null", 0)
    total = sum(dist.get(field, {}).values()) or 1
    null_ratio = nulls / total
 
    if removed and used > 0:
        return f"BREAKING: field '{field}' removed but referenced {used} times"
    if became_required and null_ratio > 0.05:
        return f"BREAKING: '{field}' required but {null_ratio:.1%} of past responses were null"
    return "SAFE"
 
if __name__ == "__main__":
    apps = [Path("./apps/wallpaper-a"), Path("./apps/healing-b")]
    logs = Path("./samples/2026-05")
    usage = extract_field_usage(apps)
    dist = sample_field_distribution(logs)
    for field in ["subtitle", "confidence", "theme"]:
        print(field, classify_change(field, False, True, usage, dist))

// schemas/v1/article-classification.ts
export const GeminiSchemaV1 = {
  type: "object",
  properties: {
    category: { type: "string", enum: ["tech", "lifestyle", "business"] },
    confidence: { type: "number" },
  },
  required: ["category", "confidence"],
} as const;
 
export interface StoredV1 {
  schemaVersion: 1;
  category: "tech" | "lifestyle" | "business";
  confidence: number;
  createdAt: string;
}
 
// schemas/v2/article-classification.ts
export const GeminiSchemaV2 = {
  type: "object",
  properties: {
    category: { type: "string", enum: ["tech", "lifestyle", "business"] },
    subCategory: { type: "string" },
    confidence: { type: "number" },
  },
  required: ["category", "confidence"],
} as const;
 
export interface StoredV2 {
  schemaVersion: 2;
  category: "tech" | "lifestyle" | "business";
  subCategory: string | null;
  confidence: number;
  createdAt: string;
}
 
// domain/article-classification.ts
export interface Classification {
  primary: string;
  secondary: string | null;  // V1 では常に null
  confidence: number;
}
 
export function toDomain(stored: StoredV1 | StoredV2): Classification {
  return {
    primary: stored.category,
    secondary: stored.schemaVersion === 2 ? stored.subCategory : null,
    confidence: stored.confidence,
  };
}

function migrateV1toV2(v1: { tags: string }): { tags: string[] } {
  return { tags: v1.tags ? v1.tags.split(",").map(s => s.trim()) : [] };
}

// Phase 1: 両方書く
function write(record: { headline?: string; title?: string }) {
  const value = record.headline ?? record.title;
  return { headline: value, title: value };  // 旧読み出し向けに残す
}
 
// Phase 2: 新フィールドだけを読む（旧読み出しコードは消す）
// Phase 3: 旧フィールドを書かなくなる
// Phase 4: 旧フィールドを削除

async function classify(text: string, version: 1 | 2 = 2): Promise<Classification> {
  const schema = version === 2 ? GeminiSchemaV2 : GeminiSchemaV1;
  const response = await ai.models.generateContent({
    model: "gemini-2.5-flash",
    contents: text,
    config: {
      responseMimeType: "application/json",
      responseSchema: schema,
    },
  });
 
  const parsed = JSON.parse(response.text!);
  const stored = saveWithVersion(parsed, version);
  await logUsage({
    schemaVersion: version,
    inputTokens: response.usageMetadata?.promptTokenCount,
    outputTokens: response.usageMetadata?.candidatesTokenCount,
  });
  return toDomain(stored);
}

async function migrateBatch(size: number) {
  const records = await db.query<StoredV1>({
    where: { schemaVersion: 1 },
    limit: size,
  });
 
  const failures: Array<{ id: string; error: string }> = [];
  for (const record of records) {
    try {
      const upgraded = migrateV1toV2(record);
      await db.update(record.id, upgraded);
    } catch (err) {
      failures.push({ id: record.id, error: String(err) });
    }
  }
 
  console.log(`Migrated ${records.length - failures.length}/${records.length}`);
  if (failures.length > 0) {
    await db.insertFailureQueue(failures);
  }
  return { processed: records.length, failures: failures.length };
}

// 最低限これだけ追加しておく
interface Stored {
  schemaVersion: number;
  // ... 他のフィールド
}