既定モデルが勝手に上がる時代に、プロンプト変更の劣化を数値で捕まえる — バリアント管理とカナリア評価の実装メモ

# prompt_registry.py
# プロンプト + モデル + サンプリングを1つのバリアントとして Firestore で管理する
from google.cloud import firestore
from dataclasses import dataclass
import time
 
@dataclass
class PromptVariant:
    variant_id: str          # 例: v4 / v4-terse
    model: str               # 例: gemini-3.5-flash / gemini-3-pro
    system_instruction: str
    temperature: float
    thinking_level: str      # "low" / "medium" / "high"（3 世代の推論深度指定）
    weight: int              # active のものだけ合計して分母にする配信比率
    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  # 秒
 
    def _refresh(self, prompt_key: str) -> None:
        docs = (self._db.collection(self._collection)
                .where("prompt_key", "==", prompt_key)
                .where("status", "in", ["active", "shadow"])
                .stream())
        # ドキュメントに prompt_key が混ざっていても dataclass 側で弾けるよう間引く
        rows = []
        for doc in docs:
            d = doc.to_dict()
            d.pop("prompt_key", None)
            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
# user_id と prompt_key の SHA-256 を取り、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]  # 丸めへのフォールバック（通常到達しない）

# gemini_client.py
# 呼び出しをバリアント選択とログ書き込みで包む
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 を含みうる stack は別経路へ
        raise
    finally:
        # 例外が出ても必ず残す。エラー率の急増が回帰の一次シグナルになる
        _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
# ログをサンプリングし、LLM-as-judge でバリアント別スコア平均と簡易 z を出す
import json, math, statistics
from google import genai
from google.cloud import firestore
 
_client = genai.Client()
_db = firestore.Client()
 
JUDGE = """あなたは応答品質の厳格な評価者です。
次の3観点を 1〜5 の整数で採点してください（5が最高）。
- accuracy: 事実として正しいか
- helpfulness: 質問者の目的を満たしているか
- conciseness: 冗長でないか
JSON のみ出力。キーは上記3つ。説明文は書かないでください。"""
 
def judge(user_input: str, output_text: str) -> dict:
    resp = _client.models.generate_content(
        model="gemini-3-pro",
        contents=f"【質問】\n{user_input}\n\n【応答】\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:
    # 3観点の単純平均。重み付けしたいときはここを差し替える
    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:
    # base を現行、cand を新版とし、平均差と簡易 z を返す
    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"}