gemini-flash-latest への追従をやめた日 — モデルの静かな置き換えに耐えるバッチ設計

# canary_compare.py — 同一ゴールデンセットを新旧モデルへ流し、下流契約への適合を比較する
# 事前準備: pip install google-genai / golden_set.jsonl を同じディレクトリに置く
import json
import time
import pathlib
import statistics
from google import genai
from google.genai import types, errors
 
client = genai.Client(api_key="YOUR_API_KEY")
 
BASELINE = "gemini-3.5-flash"    # ピン留め中の現行モデル
CANDIDATE = "gemini-3.6-flash"   # 昇格候補
 
ALLOWED_CATEGORIES = [
    "nature", "abstract", "minimal", "night", "pastel",
    "vivid", "monochrome", "seasonal", "texture", "gradient",
]
 
# 制約は「プロンプトのお願い」ではなくスキーマに置く（今回の事故で学んだ核心）
SCHEMA = types.Schema(
    type=types.Type.OBJECT,
    properties={
        "category": types.Schema(type=types.Type.STRING, enum=ALLOWED_CATEGORIES),
        "tags": types.Schema(
            type=types.Type.ARRAY,
            items=types.Schema(type=types.Type.STRING),
            min_items=5,
            max_items=5,
        ),
        "caption": types.Schema(type=types.Type.STRING),
    },
    required=["category", "tags", "caption"],
)
 
def validate(data: dict) -> list:
    """スキーマで表現しきれない契約は、受信側で必ず二重に検証します。"""
    v = []
    if data.get("category") not in ALLOWED_CATEGORIES:
        v.append("enum_out")
    if len(data.get("tags", [])) != 5:
        v.append("tag_count")
    if len(data.get("caption", "")) > 120:  # 文字数上限はスキーマでは強制しきれない
        v.append("caption_length")
    return v
 
def generate(model: str, item: dict) -> dict:
    """1件を生成。429/503 は指数バックオフで粘ります。"""
    last_err = None
    for attempt in range(5):
        try:
            t0 = time.monotonic()
            resp = client.models.generate_content(
                model=model,
                contents=[
                    types.Part.from_bytes(
                        data=pathlib.Path(item["image"]).read_bytes(),
                        mime_type="image/jpeg",
                    ),
                    item["prompt"],
                ],
                config=types.GenerateContentConfig(
                    temperature=0.0,  # 比較が目的なので出力の揺れを最小化
                    response_mime_type="application/json",
                    response_schema=SCHEMA,
                ),
            )
            latency = time.monotonic() - t0
            data = json.loads(resp.text)
            return {
                "ok": True,
                "violations": validate(data),
                "latency": latency,
                "out_tokens": resp.usage_metadata.candidates_token_count or 0,
                # これを毎回ログへ。障害調査の最初の30分がここで決まります
                "model_version": getattr(resp, "model_version", "unknown"),
            }
        except errors.APIError as e:
            last_err = e
            if e.code in (429, 503) and attempt < 4:
                time.sleep(min(2 ** attempt * 5, 60))
                continue
            break
    return {"ok": False, "violations": ["api_error: " + str(last_err)],
            "latency": None, "out_tokens": 0, "model_version": "n/a"}
 
def run(model: str, golden: list) -> None:
    results = [generate(model, item) for item in golden]
    ok = [r for r in results if r["ok"]]
    if not ok:
        print(model, ": 全件失敗（APIキー・権限・モデル名を確認）")
        return
    clean = [r for r in ok if not r["violations"]]
    breakdown = {k: sum(k in r["violations"] for r in ok)
                 for k in ("tag_count", "enum_out", "caption_length")}
    lat = sorted(r["latency"] for r in ok)
    print("---", model, "( model_version:", ok[0]["model_version"], ")")
    print("契約適合:", len(clean), "/", len(golden))
    print("逸脱内訳:", breakdown)
    print("出力トークン平均:", round(statistics.mean(r["out_tokens"] for r in ok)))
    print("latency p50/p95:",
          round(lat[len(lat) // 2], 2), "s /",
          round(lat[int(len(lat) * 0.95) - 1], 2), "s")
 
if __name__ == "__main__":
    golden = [json.loads(line) for line in open("golden_set.jsonl", encoding="utf-8")]
    run(BASELINE, golden)
    run(CANDIDATE, golden)