When the Default Model Silently Upgrades: Catching Prompt Regressions in Numbers

# prompt_registry.py
# Manage prompt + model + sampling as one variant in Firestore
from google.cloud import firestore
from dataclasses import dataclass
import time
 
@dataclass
class PromptVariant:
    variant_id: str          # e.g. v4 / v4-terse
    model: str               # e.g. gemini-3.5-flash / gemini-3-pro
    system_instruction: str
    temperature: float
    thinking_level: str      # "low" / "medium" / "high" (Gen-3 reasoning depth)
    weight: int              # share; sum only the active ones for the denominator
    status: str              # active / shadow / archived
 
class PromptRegistry:
    def __init__(self, collection: str = "prompt_variants"):
        self._db = firestore.Client()
        self._collection = collection
        self._cache: dict[str, list[PromptVariant]] = {}
        self._refreshed_at = 0.0
        self._ttl = 60  # seconds
 
    def _refresh(self, prompt_key: str) -> None:
        docs = (self._db.collection(self._collection)
                .where("prompt_key", "==", prompt_key)
                .where("status", "in", ["active", "shadow"])
                .stream())
        rows = []
        for doc in docs:
            d = doc.to_dict()
            d.pop("prompt_key", None)  # not a dataclass field
            rows.append(PromptVariant(**d))
        self._cache[prompt_key] = rows
        self._refreshed_at = time.time()
 
    def get_variants(self, prompt_key: str) -> list[PromptVariant]:
        if (time.time() - self._refreshed_at) > self._ttl:
            self._refresh(prompt_key)
        rows = self._cache.get(prompt_key)
        if not rows:
            raise RuntimeError(f"no variants for prompt_key: {prompt_key}")
        return rows

# traffic_split.py
# Hash user_id + prompt_key with SHA-256, cut intervals by weight
import hashlib
from typing import Iterable
from prompt_registry import PromptVariant
 
def pick_variant(user_id: str, prompt_key: str,
                 variants: Iterable[PromptVariant]) -> PromptVariant:
    actives = [v for v in variants if v.status == "active"]
    if not actives:
        raise RuntimeError("no active variant")
    total = sum(v.weight for v in actives)
    if total <= 0:
        raise ValueError("weight sum must be positive")
 
    seed = f"{prompt_key}:{user_id}".encode()
    bucket = int(hashlib.sha256(seed).hexdigest(), 16) % total
 
    acc = 0
    for v in actives:
        acc += v.weight
        if bucket < acc:
            return v
    return actives[-1]  # rounding fallback (normally unreachable)

# gemini_client.py
# Wrap the call with variant selection and log writing
import time, uuid
from google import genai
from google.cloud import firestore
from prompt_registry import PromptRegistry
from traffic_split import pick_variant
 
_client = genai.Client()
_registry = PromptRegistry()
_logs = firestore.Client().collection("prompt_logs")
 
def generate(user_id: str, prompt_key: str, user_input: str) -> tuple[str, str]:
    variant = pick_variant(user_id, prompt_key, _registry.get_variants(prompt_key))
    log_id = uuid.uuid4().hex
    started = time.monotonic()
    error, usage, text = None, {}, ""
    try:
        resp = _client.models.generate_content(
            model=variant.model,
            contents=user_input,
            config={
                "system_instruction": variant.system_instruction,
                "temperature": variant.temperature,
                "thinking_config": {"thinking_level": variant.thinking_level},
            },
        )
        text = resp.text or ""
        um = resp.usage_metadata
        if um:
            usage = {
                "prompt_tokens": um.prompt_token_count,
                "output_tokens": um.candidates_token_count,
                "thoughts_tokens": getattr(um, "thoughts_token_count", 0) or 0,
                "cached_tokens": getattr(um, "cached_content_token_count", 0) or 0,
            }
    except Exception as e:
        error = f"{type(e).__name__}: {str(e)[:200]}"  # PII-bearing stack goes elsewhere
        raise
    finally:
        # Always write, even on exception. A spike in error rate is the first regression signal.
        _logs.document(log_id).set({
            "log_id": log_id, "user_id": user_id, "prompt_key": prompt_key,
            "variant_id": variant.variant_id, "model": variant.model,
            "user_input": user_input, "output_text": text,
            "latency_ms": int((time.monotonic() - started) * 1000),
            "error": error, "usage": usage,
            "created_at": firestore.SERVER_TIMESTAMP,
        })
    return text, log_id

# eval_regression.py
# Sample logs, score per variant with an LLM judge, emit mean and a quick z
import json, math, statistics
from google import genai
from google.cloud import firestore
 
_client = genai.Client()
_db = firestore.Client()
 
JUDGE = """You are a strict evaluator of response quality.
Score each of the three axes as an integer from 1 to 5 (5 is best).
- accuracy: is it factually correct
- helpfulness: does it meet the asker's goal
- conciseness: is it free of bloat
Output JSON only, keys as above. Do not write any prose."""
 
def judge(user_input: str, output_text: str) -> dict:
    resp = _client.models.generate_content(
        model="gemini-3-pro",
        contents=f"[Question]\n{user_input}\n\n[Response]\n{output_text}",
        config={"system_instruction": JUDGE,
                "response_mime_type": "application/json",
                "temperature": 0.0},
    )
    try:
        return json.loads(resp.text)
    except (json.JSONDecodeError, TypeError):
        return {"accuracy": 0, "helpfulness": 0, "conciseness": 0, "parse_error": True}
 
def score_overall(d: dict) -> float:
    return (d["accuracy"] + d["helpfulness"] + d["conciseness"]) / 3
 
def evaluate(prompt_key: str, sample: int = 400) -> dict:
    logs = (_db.collection("prompt_logs")
            .where("prompt_key", "==", prompt_key)
            .order_by("created_at", direction=firestore.Query.DESCENDING)
            .limit(sample).stream())
    buckets: dict[str, list[float]] = {}
    for log in logs:
        d = log.to_dict()
        if d.get("error") or not d.get("output_text"):
            continue
        s = judge(d["user_input"], d["output_text"])
        if s.get("parse_error"):
            continue
        buckets.setdefault(d["variant_id"], []).append(score_overall(s))
    return {vid: {"n": len(v), "mean": round(statistics.mean(v), 3),
                  "stdev": round(statistics.stdev(v), 3) if len(v) > 1 else 0.0}
            for vid, v in buckets.items() if v}
 
def compare(base: dict, cand: dict) -> dict:
    diff = cand["mean"] - base["mean"]
    se = math.sqrt(base["stdev"]**2 / max(base["n"], 1) +
                   cand["stdev"]**2 / max(cand["n"], 1))
    z = diff / se if se > 0 else 0.0
    return {"diff": round(diff, 3), "z": round(z, 2),
            "verdict": "regression" if z < -2 else "win" if z > 2 else "inconclusive"}