どのプロンプト改訂で品質が動いたかを後から辿る — Gemini 生成パイプラインにプロンプト版管理を入れる

# before — 文面は直書き、出力に出どころが残らない
from google import genai
 
client = genai.Client()
 
def draft_article_before(topic: str) -> str:
    prompt = f"次のトピックで丁寧な日本語の記事下書きを書いてください: {topic}"
    resp = client.models.generate_content(
        model="gemini-flash-latest",
        contents=prompt,
    )
    return resp.text  # この文字列が「どの文面・どのモデル」で出たのか後から不明

# after — 版IDと実モデルIDを出力に同梱する
from google import genai
from prompt_registry import load_prompt
 
client = genai.Client()
 
def draft_article_after(topic: str) -> dict:
    p = load_prompt("article_ja_draft")
    prompt = p["text"].format(topic=topic)
    resp = client.models.generate_content(
        model="gemini-flash-latest",
        contents=prompt,
    )
    # 既定の -latest が実際にどのモデルへ解決されたかも控える
    resolved = getattr(resp, "model_version", None) or "gemini-flash-latest"
    return {
        "text": resp.text,
        "prompt_id": p["id"],
        "prompt_revision": p["revision"],
        "model": resolved,
        "topic": topic,
    }

import json, datetime
 
def append_log(record: dict, log_path="gen_log.jsonl"):
    record["ts"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

# bisect_regression.py
import json
 
def load_records(path="gen_log.jsonl"):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]
 
def find_regression_boundary(records, score_key="score", drop=0.10):
    """時系列に並べたログから、スコアが drop 以上落ちた最初の改訂境界を返す。"""
    rows = sorted(records, key=lambda r: r["ts"])
    # 改訂ごとの平均スコアを、登場順を保ったまま集計する
    seen = []
    agg = {}
    for r in rows:
        rev = r["prompt_revision"]
        if rev not in agg:
            agg[rev] = []
            seen.append(rev)
        agg[rev].append(r.get(score_key, 0.0))
    means = [(rev, sum(v) / len(v)) for rev, v in ((s, agg[s]) for s in seen)]
    # 隣り合う改訂で、落差が drop を超えた最初の境界を返す
    for (prev_rev, prev_m), (cur_rev, cur_m) in zip(means, means[1:]):
        if prev_m - cur_m >= drop:
            return {"from": prev_rev, "to": cur_rev,
                    "before": round(prev_m, 3), "after": round(cur_m, 3)}
    return None
 
if __name__ == "__main__":
    recs = load_records()
    boundary = find_regression_boundary(recs)
    print(boundary or "回帰は検出されませんでした")

# 落ちる前の版（from ハッシュに対応するコミット）に、その1ファイルだけ戻す
git log --oneline -- prompts/article_ja_draft.txt
git checkout <good-commit> -- prompts/article_ja_draft.txt

PINNED_GOOD = {"article_ja_draft": "9f2c1ab33de0"}
 
def assert_not_regressed(p: dict):
    pinned = PINNED_GOOD.get(p["id"])
    if pinned and p["revision"] != pinned:
        # 黙って進めず、まず気づけるようにする（自動運用なら通知へ）
        print(f"⚠️ {p['id']} がピン留め {pinned} から外れています → {p['revision']}")