Gemini で組む LLM-as-Judge 評価パイプライン — 本番運用に耐える設計と実装

# eval_pointwise.py
# Gemini を判定者にして「事実性」「丁寧さ」「具体性」を 0-5 で評価する Pointwise ジャッジ
# 必要な環境変数: GEMINI_API_KEY
import os
import json
from google import genai
from google.genai import types
from pydantic import BaseModel, Field
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class JudgeScore(BaseModel):
    factuality: int = Field(ge=0, le=5, description="事実性 0-5")
    politeness: int = Field(ge=0, le=5, description="丁寧さ 0-5")
    specificity: int = Field(ge=0, le=5, description="具体性 0-5")
    rationale: str = Field(description="評価理由を80字以内")
 
JUDGE_PROMPT = """あなたは厳密な評価者です。以下のユーザー質問とアシスタント応答について、3つの観点で 0-5 のスコアを付けてください。
 
評価観点:
- factuality: 応答に書かれている事実が正確か。検証できない場合は中央値の3とする
- politeness: 敬体で書かれているか、相手を尊重しているか
- specificity: 具体例・数値・コードなど踏み込んだ情報があるか
 
ユーザー質問:
{question}
 
アシスタント応答:
{answer}
 
JSON のみを返してください。説明文は禁止です。"""
 
def judge_pointwise(question: str, answer: str) -> JudgeScore:
    prompt = JUDGE_PROMPT.format(question=question, answer=answer)
    try:
        resp = client.models.generate_content(
            model="gemini-2.5-pro",
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.0,  # 評価は決定論的に
                response_mime_type="application/json",
                response_schema=JudgeScore,
                max_output_tokens=512,
            ),
        )
        # Pydantic で型検証してから返す
        return JudgeScore.model_validate_json(resp.text)
    except Exception as e:
        # ジャッジ失敗は黙って欠損にせず、ログ + 中央値で代替する
        print(f"[judge error] {type(e).__name__}: {e}")
        return JudgeScore(factuality=3, politeness=3, specificity=3, rationale="judge_failed")
 
if __name__ == "__main__":
    q = "Cloudflare Workers で Gemini API を呼ぶ最小コードを教えてください"
    a = "fetch を使えばOKです"
    score = judge_pointwise(q, a)
    print(json.dumps(score.model_dump(), ensure_ascii=False, indent=2))

{
  "factuality": 2,
  "politeness": 2,
  "specificity": 1,
  "rationale": "fetch だけでは不十分。エンドポイント・認証・モデル名の説明が欠落"
}

# eval_pairwise.py
# 位置バイアスを排除した Pairwise ジャッジ
import os
import random
from typing import Literal
from google import genai
from google.genai import types
from pydantic import BaseModel, Field
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class PairwiseVerdict(BaseModel):
    winner: Literal["A", "B", "TIE"]
    rationale: str = Field(max_length=200)
 
PAIRWISE_PROMPT = """あなたは応答品質の比較評価者です。同じ質問に対する応答 A と応答 B を比較し、どちらが優れているかを判定してください。
 
判定基準（同じ重みで総合判定）:
1. 事実の正しさ
2. 質問への直接的な回答度合い
3. 必要十分な具体性（過剰な前置きや繰り返しは減点）
 
質問:
{question}
 
応答 A:
{a}
 
応答 B:
{b}
 
A / B / TIE のいずれかを winner に入れてください。"""
 
def judge_pairwise(question: str, ans_a: str, ans_b: str) -> PairwiseVerdict:
    # 位置バイアス対策: 50% の確率で A と B を入れ替える
    swap = random.random() < 0.5
    a_text, b_text = (ans_b, ans_a) if swap else (ans_a, ans_b)
 
    prompt = PAIRWISE_PROMPT.format(question=question, a=a_text, b=b_text)
 
    try:
        resp = client.models.generate_content(
            model="gemini-2.5-pro",
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.0,
                response_mime_type="application/json",
                response_schema=PairwiseVerdict,
                max_output_tokens=400,
            ),
        )
        verdict = PairwiseVerdict.model_validate_json(resp.text)
 
        # 入れ替えていた場合は判定も逆にする
        if swap and verdict.winner == "A":
            return PairwiseVerdict(winner="B", rationale=verdict.rationale)
        if swap and verdict.winner == "B":
            return PairwiseVerdict(winner="A", rationale=verdict.rationale)
        return verdict
    except Exception as e:
        print(f"[pairwise judge error] {type(e).__name__}: {e}")
        return PairwiseVerdict(winner="TIE", rationale="judge_failed")

# eval_pipeline.py
# 評価データセットを並列にジャッジへ流し、結果を BigQuery / SQLite に書き出す
import asyncio
import json
import sqlite3
import time
from typing import List
from google import genai
from google.genai import types
from pydantic import BaseModel
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class EvalRecord(BaseModel):
    case_id: str
    question: str
    answer: str
 
async def judge_async(record: EvalRecord, semaphore: asyncio.Semaphore) -> dict:
    """並列度を制御しながら 1 ケースを評価する"""
    async with semaphore:
        for attempt in range(3):
            try:
                resp = await asyncio.to_thread(
                    client.models.generate_content,
                    model="gemini-2.5-pro",
                    contents=JUDGE_PROMPT.format(question=record.question, answer=record.answer),
                    config=types.GenerateContentConfig(
                        temperature=0.0,
                        response_mime_type="application/json",
                        response_schema=JudgeScore,
                        max_output_tokens=512,
                    ),
                )
                score = JudgeScore.model_validate_json(resp.text)
                return {"case_id": record.case_id, **score.model_dump()}
            except Exception as e:
                # 429 / 503 はバックオフで再試行
                wait = 2 ** attempt
                print(f"[retry {attempt+1}/3] {record.case_id}: {e} — wait {wait}s")
                await asyncio.sleep(wait)
        # 3 回失敗したら欠損として返す
        return {"case_id": record.case_id, "factuality": 3, "politeness": 3, "specificity": 3, "rationale": "max_retry"}
 
async def run_pipeline(dataset: List[EvalRecord], concurrency: int = 8) -> List[dict]:
    sem = asyncio.Semaphore(concurrency)
    return await asyncio.gather(*[judge_async(r, sem) for r in dataset])
 
def save_to_sqlite(results: List[dict], path: str = "eval_results.db") -> None:
    conn = sqlite3.connect(path)
    conn.execute("""CREATE TABLE IF NOT EXISTS scores (
        case_id TEXT, ts INTEGER, factuality INT, politeness INT,
        specificity INT, rationale TEXT
    )""")
    ts = int(time.time())
    conn.executemany(
        "INSERT INTO scores VALUES (?, ?, ?, ?, ?, ?)",
        [(r["case_id"], ts, r["factuality"], r["politeness"], r["specificity"], r["rationale"]) for r in results],
    )
    conn.commit()
    conn.close()

# eval_flakiness.py
# 同一ケースを N 回判定し、スコアのぶれ（フラッキネス）を測る
# 出力: 案件ごとの最頻値・一致率・ぶれ幅と、信頼できない案件のフラグ
import os
import statistics
from collections import Counter
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def judge_once(question: str, answer: str) -> int:
    resp = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=JUDGE_PROMPT.format(question=question, answer=answer),
        config=types.GenerateContentConfig(
            temperature=0.0,
            response_mime_type="application/json",
            response_schema=JudgeScore,
            max_output_tokens=512,
        ),
    )
    return JudgeScore.model_validate_json(resp.text).factuality
 
def measure_flakiness(question: str, answer: str, n: int = 5) -> dict:
    scores = [judge_once(question, answer) for _ in range(n)]
    counter = Counter(scores)
    mode, mode_count = counter.most_common(1)[0]
    agreement = mode_count / n            # 最頻値の占有率 = 一致率
    spread = max(scores) - min(scores)    # ぶれ幅
    return {
        "mode": mode,
        "agreement": round(agreement, 2),
        "spread": spread,
        "stdev": round(statistics.pstdev(scores), 2),
        # 一致率が 0.6 未満、またはぶれ幅が 2 以上なら「信頼できない案件」
        "unreliable": agreement < 0.6 or spread >= 2,
        "raw": scores,
    }
 
if __name__ == "__main__":
    q = "Cloudflare Workers で Gemini API を呼ぶ最小コードを教えてください"
    a = "fetch でエンドポイントに POST し、x-goog-api-key で認証します。モデル名は URL に含めます。"
    print(measure_flakiness(q, a, n=5))