Gemini API の出力品質を継続監視する評価基盤の設計

# judge.py — 観点別ジャッジを多数決で集約する最小実装
from dataclasses import dataclass
from typing import Literal
import google.generativeai as genai
 
JUDGE_MODEL = "gemini-2.5-pro"
 
@dataclass
class JudgeVerdict:
    aspect: Literal["constraint", "intent", "naturalness"]
    score: int            # 1-5
    rationale: str
 
CONSTRAINT_PROMPT = """あなたは厳密な制約チェッカーです。
以下の制約を満たしているかを 1〜5 で評価し、根拠を 80 字以内で書いてください。
評価する候補出力: <<<{output}>>>
制約: must_include_intent={intent}, forbidden={forbidden}, max_chars={max_chars}
JSONで返してください: {{"score": int, "rationale": str}}"""
 
INTENT_PROMPT = """以下の入力と出力で、入力から期待される行動目的が出力に表れているかを 1〜5 で評価してください。
入力: <<<{input}>>>
出力: <<<{output}>>>
JSONで返してください: {{"score": int, "rationale": str}}"""
 
NATURALNESS_PROMPT = """以下の日本語出力の自然さを 1〜5 で評価してください。
評価軸: 機械的な定型句がないか、読み手に強要する語気がないか、字数バランス。
出力: <<<{output}>>>
JSONで返してください: {{"score": int, "rationale": str}}"""
 
 
def call_judge(prompt: str) -> JudgeVerdict:
    model = genai.GenerativeModel(
        JUDGE_MODEL,
        generation_config={"temperature": 0.1, "response_mime_type": "application/json"},
    )
    resp = model.generate_content(prompt)
    data = resp.parsed if hasattr(resp, "parsed") else parse_json_fallback(resp.text)
    return JudgeVerdict(
        aspect=prompt_aspect(prompt),
        score=int(data["score"]),
        rationale=data["rationale"],
    )
 
 
def majority_vote(verdicts: list[JudgeVerdict]) -> int:
    """同じ観点で 3 回投票させ、中央値を採用する。極端な外れ値を吸収する設計。"""
    by_aspect = {}
    for v in verdicts:
        by_aspect.setdefault(v.aspect, []).append(v.score)
    return {a: sorted(s)[len(s) // 2] for a, s in by_aspect.items()}

# regression.py — 4週中央値を基準にした劣化検出
import statistics
 
WARN_DROP = 1.0
ALERT_DROP = 1.5
 
def detect_regression(history: list[float], current: float) -> str:
    """history は古い順の直近4週スコア。当週分は含まない。"""
    if len(history) < 4:
        return "insufficient_history"
    baseline = statistics.median(history)
    delta = baseline - current
    if delta >= ALERT_DROP:
        return f"alert (baseline={baseline:.2f}, current={current:.2f})"
    if delta >= WARN_DROP:
        return f"warn (baseline={baseline:.2f}, current={current:.2f})"
    return "ok"

# pipeline.py — 週次評価パイプラインのエントリポイント
import yaml
from pathlib import Path
from typing import Any
 
from judge import call_judge, majority_vote, CONSTRAINT_PROMPT, INTENT_PROMPT, NATURALNESS_PROMPT
from regression import detect_regression
from store import load_history, append_score, send_alert
 
DATASET_DIR = Path("golden_dataset")
 
 
def select_samples_for_week(week: int) -> list[Path]:
    """階層サンプリング。揺れているサンプルは毎週、安定は隔週。"""
    all_paths = list(DATASET_DIR.glob("*.yaml"))
    selected = []
    for p in all_paths:
        history = load_history(p.stem)
        if not history or stdev_recent(history) > 0.4:
            selected.append(p)
            continue
        if week % 2 == 0:
            selected.append(p)
    return selected
 
 
def evaluate_one(sample_path: Path) -> dict[str, Any]:
    sample = yaml.safe_load(sample_path.read_text())
    output = run_production_call(sample["input"])
    verdicts = []
    for prompt_tpl, aspect, model in [
        (CONSTRAINT_PROMPT, "constraint", "gemini-2.5-flash"),
        (INTENT_PROMPT,     "intent",     "gemini-2.5-pro"),
        (NATURALNESS_PROMPT, "naturalness", "gemini-2.5-pro"),
    ]:
        for _ in range(3):  # 観点ごとに3回呼んで中央値
            verdicts.append(call_judge(prompt_tpl.format(**sample, output=output)))
    scores = majority_vote(verdicts)
    return {"id": sample["id"], "scores": scores, "output": output}
 
 
def main(week: int) -> None:
    targets = select_samples_for_week(week)
    for path in targets:
        result = evaluate_one(path)
        for aspect, score in result["scores"].items():
            history = load_history(result["id"], aspect)
            status = detect_regression(history, score)
            append_score(result["id"], aspect, score)
            if status.startswith(("warn", "alert")):
                send_alert(result["id"], aspect, status, result["output"])