gemini-flash-latest が 3.5 Flash に切り替わっても深夜に事故らない『昇格ゲート』の設計

# golden/review_triage.jsonl の各行（例）
# {"id": "r001", "input": "起動時に毎回広告が出て使いにくい", "expect_label": "complaint"}
# {"id": "r002", "input": "壁紙の追加ありがとうございます！毎日使ってます", "expect_label": "praise"}
# {"id": "r003", "input": "ダークモードに対応してほしいです", "expect_label": "request"}

# acceptance_harness.py
import json, time, statistics
from pathlib import Path
from google import genai
from google.genai import types
 
client = genai.Client()  # GEMINI_API_KEY は環境変数から
 
# 1,000トークンあたりの単価（USD）。実際の料金は自分の請求条件で埋める。
PRICE_TABLE = {
    "gemini-2.5-flash": {"in": 0.0000, "out": 0.0000},
    "gemini-3.5-flash": {"in": 0.0000, "out": 0.0000},
}
 
RESPONSE_SCHEMA = types.Schema(
    type=types.Type.OBJECT,
    required=["label"],
    properties={
        "label": types.Schema(
            type=types.Type.STRING,
            enum=["complaint", "praise", "request", "spam"],
        )
    },
)
 
def call_once(model_id: str, text: str):
    """1コールを実行し、(ラベル, スキーマ適合, レイテンシ秒, コストUSD) を返す。"""
    started = time.perf_counter()
    try:
        resp = client.models.generate_content(
            model=model_id,
            contents=text,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=RESPONSE_SCHEMA,
                temperature=0.0,  # 評価は決定的に寄せる
            ),
        )
    except Exception as e:
        return None, False, time.perf_counter() - started, 0.0
    latency = time.perf_counter() - started
 
    schema_ok, label = False, None
    try:
        data = json.loads(resp.text)
        label = data["label"]
        schema_ok = label in {"complaint", "praise", "request", "spam"}
    except Exception:
        schema_ok = False
 
    um = resp.usage_metadata
    price = PRICE_TABLE.get(model_id, {"in": 0.0, "out": 0.0})
    cost = (um.prompt_token_count / 1000) * price["in"] \
         + ((um.candidates_token_count or 0) / 1000) * price["out"]
    return label, schema_ok, latency, cost
 
def evaluate(model_id: str, golden: list[dict]) -> dict:
    labels, schema_hits, latencies, costs = [], 0, [], []
    for row in golden:
        label, schema_ok, latency, cost = call_once(model_id, row["input"])
        labels.append(label)
        schema_hits += 1 if schema_ok else 0
        latencies.append(latency)
        costs.append(cost)
    n = len(golden)
    latencies.sort()
    p95 = latencies[min(int(n * 0.95), n - 1)] if n else 0.0
    return {
        "model": model_id,
        "labels": labels,
        "schema_rate": schema_hits / n if n else 0.0,
        "p95_latency": p95,
        "avg_cost": statistics.mean(costs) if costs else 0.0,
    }
 
def agreement(a: list, b: list) -> float:
    """同位置のラベル一致率。None（呼び出し失敗）は不一致扱い。"""
    if not a:
        return 0.0
    hits = sum(1 for x, y in zip(a, b) if x is not None and x == y)
    return hits / len(a)

# promote.py — ゲート判定と registry の書き換え
import json
from pathlib import Path
from acceptance_harness import evaluate, agreement
from model_registry import load_registry, REGISTRY_PATH
 
def run_gate(role: str, golden_path: str) -> bool:
    golden = [json.loads(l) for l in Path(golden_path).read_text(encoding="utf-8").splitlines() if l.strip()]
    reg = load_registry()
    prod_id = reg[role]["prod"]
    cand_id = reg[role]["candidate"]
 
    prod = evaluate(prod_id, golden)
    cand = evaluate(cand_id, golden)
    agree = agreement(prod["labels"], cand["labels"])
 
    checks = {
        "schema_rate":  cand["schema_rate"] >= 0.99,
        "agreement":    agree >= 0.95,
        "p95_latency":  cand["p95_latency"] <= prod["p95_latency"] * 1.3,
        "cost_ok":      cand["avg_cost"] <= prod["avg_cost"] * 1.05,
    }
    print(f"[{role}] {cand_id}  適合率={cand['schema_rate']:.3f} "
          f"一致率={agree:.3f} p95={cand['p95_latency']:.2f}s "
          f"コスト比={cand['avg_cost'] / (prod['avg_cost'] or 1):.2f}")
    for name, ok in checks.items():
        print(f"  {'✅' if ok else '❌'} {name}")
 
    if all(checks.values()):
        # 昇格: candidate を prod スロットへ書き写す
        reg[role]["prod"] = cand_id
        REGISTRY_PATH.write_text(json.dumps(reg, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"[{role}] 昇格しました → prod = {cand_id}")
        return True
 
    print(f"[{role}] 不合格。prod = {prod_id} のまま据え置きます")
    return False