LLM-as-Judge, Done Honestly

# Colab cell — run once. Set your key in the 🔑 panel (name it
# ANTHROPIC_API_KEY) or just paste it when prompted.
!pip install -q anthropic

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
except Exception:
    from getpass import getpass
    os.environ.setdefault("ANTHROPIC_API_KEY", getpass("Anthropic API key: "))

import anthropic

client = anthropic.Anthropic()

# In production these are your 30+ hand-labeled examples in labels.json and
# an anchored rubric file; inlined here as tiny fixtures so the cell runs.
labels = [
    {"id": "grounded",   "output": "Your plan renews on the 1st, per your account page.", "human": "pass"},
    {"id": "invented",   "output": "Your plan renews on the 15th and includes free flights.", "human": "fail"},
    {"id": "abstained",  "output": "I don't see that detail in the provided context.", "human": "pass"},
]
RUBRIC = (
    "You grade whether a support reply stays faithful to known account facts "
    "(renewal date is the 1st; no travel perks). PASS if every claim is "
    "supported or the reply abstains; FAIL if it invents any fact."
)

def run_judge(output: str, rubric: str) -> str:
    # Force a structured verdict via a tool schema (Module 1 pattern).
    # Judges run in volume, so default to the cheap tier — validation
    # against human labels, not judge size, is what earns trust.
    resp = client.messages.create(
        model="claude-haiku-4-5", max_tokens=512,
        tools=[{
            "name": "record_verdict",
            "description": "Record the grading verdict for one answer.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "verdict": {"type": "string", "enum": ["pass", "fail"]},
                    "reason": {"type": "string"},
                },
                "required": ["verdict", "reason"],
            },
        }],
        tool_choice={"type": "tool", "name": "record_verdict"},
        system=rubric,
        messages=[{"role": "user", "content": f"Answer to grade:\n{output}"}],
    )
    block = next(b for b in resp.content if b.type == "tool_use")
    return block.input["verdict"]

def agreement(labels: list[dict], rubric: str) -> float:
    hits = 0
    for ex in labels:
        if run_judge(ex["output"], rubric) == ex["human"]:
            hits += 1
    return hits / len(labels)

rate = agreement(labels, RUBRIC)
print(f"judge-human agreement: {rate:.0%}")
# In your real suite make this a gate: assert rate >= 0.85 so a judge below
# your bar can never be promoted to production.
print("trustworthy" if rate >= 0.85 else "tune the rubric and re-measure")

# Colab cell — pure Python, no key needed. Fake judges make the
# position-bias control visible without spending a token.
import random
from collections import namedtuple

def pairwise(judge_call, prompt: str, answer_a: str, answer_b: str) -> str:
    """Return 'A', 'B', or 'tie', controlling for position bias."""
    # Run once in each order; the labels A/B track the ORIGINAL answers.
    order1 = judge_call(prompt, first=answer_a, second=answer_b)   # -> 'first'/'second'
    order2 = judge_call(prompt, first=answer_b, second=answer_a)

    # Translate each verdict back to the original answer it points at.
    pick1 = "A" if order1 == "first" else "B"
    pick2 = "A" if order2 == "second" else "B"

    if pick1 == pick2:
        return pick1                 # consistent across orders — trustworthy
    return "tie"                     # flipped with position — treat as no signal

def win_rate(cases, judge_call, candidate, baseline) -> float:
    wins = ties = 0
    for c in cases:
        # Randomize which is presented first at the suite level too.
        a, b = candidate[c], baseline[c]
        verdict = pairwise(judge_call, c.prompt, a, b)
        if verdict == "A":
            wins += 1
        elif verdict == "tie":
            ties += 1
    # Ties count as half; a fair coin lands near 0.5.
    return (wins + 0.5 * ties) / len(cases)


# A position-biased judge (always picks whatever is shown first) flips with
# order, so pairwise correctly scores it a tie -- "no signal":
biased = lambda prompt, first, second: "first"
print("position-biased judge ->", pairwise(biased, "q", "answer A", "answer B"))

# A quality judge (prefers the more detailed answer) survives both orders:
quality = lambda prompt, first, second: "first" if len(first) >= len(second) else "second"
print("quality judge        ->", pairwise(quality, "q", "a detailed answer", "short"))

# win_rate over a few cases, candidate more detailed than baseline:
Case = namedtuple("Case", "prompt")
cases = [Case("q1"), Case("q2"), Case("q3")]
candidate = {c: "a thorough, detailed answer" for c in cases}
baseline = {c: "brief" for c in cases}
print("candidate win rate   ->", win_rate(cases, quality, candidate, baseline))

# Colab cell — run block 1 first (it sets up client with your key).

# Holistic — mushy, drifts, disagreement is unexplainable:
BAD_RUBRIC = "You are an expert evaluator. Rate helpfulness from 1 to 10."

# Decomposed — each field is independently checkable and debuggable:
GOOD_RUBRIC = """You are grading a customer-support reply against three
criteria. Answer each independently as true/false, citing the exact
sentence that supports your answer.

1. answers_question: Does the reply directly address what the customer
   asked, without dodging into unrelated information?
2. names_next_step: Does the reply tell the customer what happens next
   or what they should do?
3. no_unsupported_claims: Does the reply avoid stating any policy, price,
   or fact not present in the provided account/policy context?

A reply "passes" only if all three are true."""

def run_decomposed_judge(reply: str, context: str) -> dict:
    resp = client.messages.create(
        model="claude-haiku-4-5", max_tokens=512,
        tools=[{
            "name": "record_criteria",
            "description": "Record the three-criteria verdict.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "answers_question": {"type": "boolean"},
                    "names_next_step": {"type": "boolean"},
                    "no_unsupported_claims": {"type": "boolean"},
                    "evidence": {"type": "string"},
                },
                "required": ["answers_question", "names_next_step",
                            "no_unsupported_claims", "evidence"],
            },
        }],
        tool_choice={"type": "tool", "name": "record_criteria"},
        system=GOOD_RUBRIC,
        messages=[{"role": "user",
                   "content": f"Context:\n{context}\n\nReply:\n{reply}"}],
    )
    block = next(b for b in resp.content if b.type == "tool_use")
    verdict = block.input
    verdict["pass"] = all(verdict[k] for k in
                          ("answers_question", "names_next_step",
                           "no_unsupported_claims"))
    return verdict


print(run_decomposed_judge(
    reply="Your plan renews on the 1st. I've logged your request; support "
          "will follow up within one business day.",
    context="Account: plan renews on the 1st. No travel perks."))