Gemini APIの『デフォルトモデルが上がる』を事故にしない — モデルIDの固定と既定変更の検知設計

import logging
 
logger = logging.getLogger("gemini.model_guard")
 
def generate_with_guard(prompt: str):
    resp = client.models.generate_content(
        model=EXPECTED_MODEL,
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.4,
            max_output_tokens=2048,
        ),
    )
    actual = resp.model_version or "unknown"
    um = resp.usage_metadata
    logger.info(
        "model=%s in_tok=%s out_tok=%s",
        actual,
        getattr(um, "prompt_token_count", None),
        getattr(um, "candidates_token_count", None),
    )
    if not actual.startswith(EXPECTED_VERSION_PREFIX):
        # 落とすほどではないが、必ず気づける経路に流す
        logger.error("MODEL DRIFT at runtime: got %s", actual)
        notify_ops(f"Gemini model drift: {actual}")  # Slack 等へ
    return resp

import json, re, sys
 
FORBIDDEN = re.compile(r"(latest|preview|exp)$")
 
def check_registry(path="model_registry.json") -> int:
    reg = json.load(open(path, encoding="utf-8"))
    errors = []
    for env, cfg in reg.items():
        model = cfg.get("model", "")
        if FORBIDDEN.search(model):
            errors.append(f"{env}: エイリアス禁止 -> {model}")
        if "expected_version_prefix" not in cfg:
            errors.append(f"{env}: expected_version_prefix が未定義")
    for e in errors:
        print("NG:", e)
    return 1 if errors else 0
 
if __name__ == "__main__":
    sys.exit(check_registry())

import json, statistics, datetime, pathlib
 
BASELINE_PATH = pathlib.Path("behavior_baseline.jsonl")
PROBES = [...]  # 本番の代表入力を20〜50件。内容は凍結して変更しない
 
def run_probes() -> dict:
    lengths, valid, thinking_ratio = [], 0, []
    for p in PROBES:
        resp = generate_with_guard(p)
        text = resp.text or ""
        lengths.append(len(text))
        try:
            json.loads(text)
            valid += 1
        except json.JSONDecodeError:
            pass
        um = resp.usage_metadata
        think = getattr(um, "thoughts_token_count", 0) or 0
        out = getattr(um, "candidates_token_count", 0) or 1
        thinking_ratio.append(think / out)
    return {
        "date": datetime.date.today().isoformat(),
        "model": EXPECTED_MODEL,
        "len_median": statistics.median(lengths),
        "json_valid_rate": valid / len(PROBES),
        "thinking_ratio_median": statistics.median(thinking_ratio),
    }

def mad(xs):
    m = statistics.median(xs)
    return statistics.median([abs(x - m) for x in xs]) or 1e-9
 
def check_drift(today: dict, history: list[dict], k: float = 4.0) -> list[str]:
    alerts = []
    for key in ("len_median", "json_valid_rate", "thinking_ratio_median"):
        past = [h[key] for h in history[-14:]]
        if len(past) < 7:
            continue          # 履歴が足りないうちは記録のみに徹する
        base, spread = statistics.median(past), mad(past)
        if abs(today[key] - base) > k * spread:
            alerts.append(
                f"{key}: today={today[key]:.4f} base={base:.4f} mad={spread:.4f}"
            )
    return alerts

import re, pathlib, sys
 
MODEL_LITERAL = re.compile(r"""["'](gemini-[a-z0-9.\-]+)["']""")
ALLOWED = {"model_registry.json", "model_registry.py"}
 
def audit(root: str = "src") -> int:
    hits = []
    for path in pathlib.Path(root).rglob("*.py"):
        if path.name in ALLOWED:
            continue
        for i, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            m = MODEL_LITERAL.search(line)
            if m:
                hits.append(f"{path}:{i}: {m.group(1)}")
    for h in hits:
        print("直書き検出:", h)
    return 1 if hits else 0
 
if __name__ == "__main__":
    sys.exit(audit())