Claude の品質回帰を eval ハーネスで自動検出する設計

# golden.jsonl — 1行1ケース。expected は文字列一致ではなくルーブリックで持つ
# {"id": "cat-012", "category": "返金", "input": "...", "rubric": ["返金カテゴリに分類している", "理由を1文で述べている", "個人情報を出力に含めない"]}
 
import json
from pathlib import Path
 
def load_golden(path: str) -> list[dict]:
    cases = []
    for line in Path(path).read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line:
            cases.append(json.loads(line))
    return cases

import anthropic
 
client = anthropic.Anthropic()
CANDIDATE_MODEL = "claude-sonnet-4-6"
 
def run_candidate(system_prompt: str, user_input: str) -> str:
    resp = client.messages.create(
        model=CANDIDATE_MODEL,
        max_tokens=1024,
        temperature=0,            # 評価の再現性のため固定
        system=system_prompt,
        messages=[{"role": "user", "content": user_input}],
    )
    return resp.content[0].text

JUDGE_MODEL = "claude-opus-4-6"
 
def judge(user_input: str, output: str, rubric: list[str]) -> dict:
    rubric_lines = "\n".join(f"{i+1}. {r}" for i, r in enumerate(rubric))
    judge_system = (
        "あなたは厳密な採点者です。各ルーブリック項目について、"
        "まず判定理由を1文書き、その後に true/false を決めてください。"
        "迷ったら false にしてください。出力は指定 JSON のみ。"
    )
    judge_user = (
        f"# 入力\n{user_input}\n\n# 被験モデルの出力\n{output}\n\n"
        f"# ルーブリック\n{rubric_lines}\n\n"
        '# 出力形式\n{"items":[{"reason":"...","pass":true}], "notes":"..."}'
    )
    resp = client.messages.create(
        model=JUDGE_MODEL,
        max_tokens=1024,
        temperature=0,
        system=judge_system,
        messages=[{"role": "user", "content": judge_user}],
    )
    data = json.loads(resp.content[0].text)
    items = data["items"]
    passed = sum(1 for it in items if it["pass"])
    return {"score": passed / len(items), "items": items, "notes": data.get("notes", "")}

import statistics
 
def evaluate(system_prompt: str, cases: list[dict]) -> dict:
    per_case, per_category = [], {}
    for c in cases:
        out = run_candidate(system_prompt, c["input"])
        r = judge(c["input"], out, c["rubric"])
        per_case.append({"id": c["id"], "category": c.get("category", "-"), "score": r["score"]})
        per_category.setdefault(c.get("category", "-"), []).append(r["score"])
    scores = [p["score"] for p in per_case]
    return {
        "mean": round(statistics.mean(scores), 4),
        "p10": round(sorted(scores)[max(0, len(scores) // 10 - 1)], 4),
        "fails": [p for p in per_case if p["score"] < 0.6],
        "by_category": {k: round(statistics.mean(v), 4) for k, v in per_category.items()},
    }
 
def detect_regression(baseline: dict, candidate: dict) -> list[str]:
    alerts = []
    if candidate["mean"] < baseline["mean"] - 0.02:          # 平均が2pt以上低下
        alerts.append(f"平均スコア低下: {baseline['mean']} -> {candidate['mean']}")
    if candidate["p10"] < baseline["p10"] - 0.05:            # 下位10%が悪化
        alerts.append(f"下位帯の悪化: p10 {baseline['p10']} -> {candidate['p10']}")
    for cat, sc in candidate["by_category"].items():
        base = baseline["by_category"].get(cat)
        if base is not None and sc < base - 0.05:            # 特定カテゴリの劣化
            alerts.append(f"カテゴリ「{cat}」劣化: {base} -> {sc}")
    return alerts

import sys
 
def main():
    cases = load_golden("eval/golden.jsonl")
    system_prompt = Path("prompts/classify.txt").read_text(encoding="utf-8")
    candidate = evaluate(system_prompt, cases)
    baseline = json.loads(Path("eval/baseline.json").read_text(encoding="utf-8"))
    alerts = detect_regression(baseline, candidate)
    print(json.dumps(candidate, ensure_ascii=False, indent=2))
    if alerts:
        print("回帰を検出しました:", file=sys.stderr)
        for a in alerts:
            print("  - " + a, file=sys.stderr)
        sys.exit(1)
    Path("eval/candidate.json").write_text(json.dumps(candidate, ensure_ascii=False))
    print("回帰なし。baseline 更新候補を candidate.json に出力しました。")
 
if __name__ == "__main__":
    main()