The Day I Stopped Tracking gemini-flash-latest: Batch Design That Survives Silent Model Swaps

# canary_compare.py — replay one golden set against old and new models,
# and compare conformance to the downstream contract.
# Setup: pip install google-genai / put golden_set.jsonl in the same directory
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"    # the currently pinned model
CANDIDATE = "gemini-3.6-flash"   # promotion candidate
 
ALLOWED_CATEGORIES = [
    "nature", "abstract", "minimal", "night", "pastel",
    "vivid", "monochrome", "seasonal", "texture", "gradient",
]
 
# Put constraints in the schema, not in polite prompt wording
# (that distinction is the core lesson of this incident)
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:
    """Whatever the schema cannot express must be re-checked on the receiving side."""
    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:  # length caps are not enforceable in the schema
        v.append("caption_length")
    return v
 
def generate(model: str, item: dict) -> dict:
    """Generate one item. Backs off exponentially on 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,  # we are comparing, so minimize sampling noise
                    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,
                # Log this on every call. It decides the first 30 minutes
                # of your next incident investigation.
                "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, ": every call failed (check key, permissions, model name)")
        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("contract conformance:", len(clean), "/", len(golden))
    print("violation breakdown:", breakdown)
    print("mean output tokens:", 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)