LLM-as-Judge

# script_id: day_045_llm_as_judge/break_it
# Two ways to break a judge: a vague 1-5 "quality" score, and a grounded judge that can't see the docs.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA · Services: none
import os
from pathlib import Path

from openai import OpenAI
from pydantic import BaseModel

JUDGE_MODEL = os.environ.get("ATLAS_JUDGE_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()

class Score(BaseModel):
    reason: str
    score: int      # 1-5, the Likert version

class Verdict(BaseModel):
    reason: str
    passed: bool

QUESTION = "Can I use Kafka as Orbit's broker?"
ANSWER = ('Yes. Set url = "kafka://broker:9092" under [broker] in orbit.toml and install '
          "orbit[kafka]. Kafka gives you ordered partitions for free. [concepts-brokers]")
CONTEXT = (DATA_DIR / "docs/concepts-brokers.md").read_text(encoding="utf-8")
GOD = "Rate the overall quality of ANSWER from 1 (bad) to 5 (excellent)."
ABSTAINS = ("ONE rule: if CONTEXT does not cover what QUESTION asks, ANSWER must say so and must not "
            "give a how-to. PASS if it abstains correctly, or if CONTEXT covers the question.")
GROUNDED = ("ONE rule: every config key, URL scheme, package name and number in ANSWER must appear in "
            "CONTEXT. If no CONTEXT is given, judge as best you can.")

# script_id: day_045_llm_as_judge/recorded_runs
# Writes a small recorded run of R2 v1 and v1.1 to runs/ (stand-in for Day 43/44's run files).
# Install: stdlib only · Env: ORBIT_DATA (default datasets/orbit), RUNS_DIR (default runs) · Services: none
import json
import os
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
RUNS = Path(os.environ.get("RUNS_DIR", "runs"))
GOLD = {c["id"]: c for c in map(json.loads, (DATA_DIR / "eval/qa_gold.jsonl").read_text().splitlines())}
UNANSWERABLE = [  # the docs don't cover these; the right answer says so
    {"id": "u-001", "question": "Can I use Kafka as Orbit's broker?", "answer": None},
    {"id": "u-002", "question": "Does Orbit offer a managed cloud service with an uptime SLA?",
     "answer": None},
]
CASES = [GOLD[i] for i in ("qa-001", "qa-005", "qa-006", "qa-008")] + UNANSWERABLE

# script_id: day_045_llm_as_judge/recorded_runs
# case_id: (retrieved doc ids, v1 answer, v1.1 answer). Both versions cite only real docs.
ANSWERS = {
    "qa-001": (["configuration-reference", "leases-and-heartbeats"],
               "In [worker], lease_seconds defaults to 300 and heartbeat_seconds to 30.",
               "In [worker], lease_seconds defaults to 300 and heartbeat_seconds to 30; orbit doctor "
               "warns when heartbeat_seconds is more than a third of lease_seconds."),
    "qa-005": (["retries-and-backoff"], "2 s, 4 s and 8 s, before jitter.",
               "2 s, 4 s and 8 s before jitter; jitter then picks 50-100% of each delay."),
    "qa-006": (["retries-and-backoff"], "Yes: the first attempt plus five retries, six runs in total.",
               "No. retries counts the first attempt: retries=5 means at most five runs, the first "
               "plus four retries."),
    "qa-008": (["retries-and-backoff", "leases-and-heartbeats"],
               "No. Lease expiries don't count. To stop reruns, set max_retries_per_minute = 1 under [retries].",
               "Yes. Each redelivery after ORB-1101 LeaseExpired counts as an attempt, so a job that keeps "
               "outliving its lease ends in the DLQ with ORB-1301. Fix lease sizing, not retries."),
    "u-001": (["concepts-brokers"],
              'Yes. Set url = "kafka://broker:9092" under [broker] and install orbit[kafka].',
              "The Orbit docs don't cover Kafka. The documented brokers are Redis, Postgres and Amazon SQS."),
    "u-002": (["index", "faq"], "Yes. Orbit Cloud offers a 99.9% uptime SLA on the Pro plan.",
              "The Orbit docs don't describe a managed service or an SLA, so I can't answer that."),
}

def write_jsonl(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("".join(json.dumps(r) + "\n" for r in rows), encoding="utf-8")

write_jsonl(RUNS / "cases.jsonl", CASES)
for slot, (version, name) in enumerate([("v1", "ask_v1_baseline"), ("v1.1", "ask_v1_fix")], start=1):
    write_jsonl(RUNS / f"{name}.jsonl", [
        {"case_id": cid, "version": version, "answer": ans[slot], "citations": ans[0],
         "retrieved_ids": ans[0]} for cid, ans in ANSWERS.items()])
print(f"wrote {len(CASES)} cases and 2 runs to {RUNS}/")

wrote 6 cases and 2 runs to runs/

# script_id: day_045_llm_as_judge/judges
# atlas/evals/judges.py: one binary LLM judge per failure mode, shaped as Day 20 checks.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY (only when a judge runs),
#   ATLAS_JUDGE_MODEL (default gpt-6-luna), ORBIT_DATA (default datasets/orbit) · Services: none
from __future__ import annotations

import json
import os
import random
from dataclasses import dataclass, field
from functools import cache
from pathlib import Path
from typing import Any, Callable, Literal, Sequence

import openai
from openai import OpenAI
from pydantic import BaseModel, ValidationError

JUDGE_MODEL = os.environ.get("ATLAS_JUDGE_MODEL", "gpt-6-luna")
STRONG_JUDGE_MODEL = os.environ.get("ATLAS_STRONG_JUDGE_MODEL", "gpt-6.1-terra")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# USD per 1M tokens (input, output), as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00)}

@dataclass(frozen=True)
class CheckResult:          # Day 20's shape
    name: str
    passed: bool
    detail: str = ""

Check = Callable[[dict, Any], CheckResult]      # (case, prediction) -> verdict

class Verdict(BaseModel):
    reason: str     # generated first: the evidence comes before the verdict
    passed: bool

@cache
def default_client() -> OpenAI:     # lazy: no key needed until a judge actually runs
    return OpenAI()

# script_id: day_045_llm_as_judge/judges
@dataclass(frozen=True)
class Criterion:            # compact Day 44 Criterion + judge fields
    name: str
    version: str
    rule: str               # one failure mode, phrased as a PASS/FAIL test
    fail_example: str       # concrete failing answer and why: the most useful line in the rubric
    pass_example: str
    uses_reference: bool = False    # True = reference-based (needs the gold answer)

RUBRICS: dict[str, Criterion] = {
    "grounded": Criterion(
        "grounded", "1.0",
        "Every config key, error code, version, number and behaviour claim in ANSWER is stated in "
        "CONTEXT. A plausible claim that CONTEXT doesn't state is a FAIL. An answer that says the "
        "docs don't cover the question is a PASS.",
        fail_example="'Set max_retries_per_minute = 1 under [retries]' when CONTEXT never mentions "
                     "max_retries_per_minute: FAIL, invented key.",
        pass_example="'retries=5 means at most five runs' when CONTEXT says exactly that: PASS."),
    "abstains_when_unanswerable": Criterion(
        "abstains_when_unanswerable", "1.0",
        "If CONTEXT does not contain what QUESTION asks for, ANSWER must say the docs don't cover it "
        "and must not give a how-to. If CONTEXT does cover it, PASS unless ANSWER refuses anyway.",
        fail_example="Q: 'Can I use Kafka?' CONTEXT documents Redis, Postgres and SQS. ANSWER gives a "
                     "kafka:// URL: FAIL, it should say Kafka isn't documented.",
        pass_example="Same question, ANSWER: 'The docs don't cover Kafka; the documented brokers are "
                     "Redis, Postgres and SQS.': PASS."),
}

# script_id: day_045_llm_as_judge/judges
@dataclass
class CostTracker:          # compact Day 12 tracker
    usd: dict[str, float] = field(default_factory=dict)
    calls: dict[str, int] = field(default_factory=dict)

    def record(self, model: str, usage: Any) -> float:
        p_in, p_out = PRICES.get(model, (0.0, 0.0))     # unknown model: count calls, not $
        cost = (usage.input_tokens * p_in + usage.output_tokens * p_out) / 1e6  # incl. reasoning
        self.usd[model] = self.usd.get(model, 0.0) + cost
        self.calls[model] = self.calls.get(model, 0) + 1
        return cost

TRACKER = CostTracker()

def _field(obj: Any, name: str) -> Any:     # predictions may be dicts (run files) or objects
    return obj[name] if isinstance(obj, dict) else getattr(obj, name)

def load_context(doc_ids: Sequence[str], max_chars: int = 8_000) -> str:
    """The retrieved docs, as the judge sees them. Without them it can only grade plausibility."""
    parts = []
    for doc_id in doc_ids:
        path = DATA_DIR / "docs" / f"{doc_id}.md"
        text = path.read_text(encoding="utf-8")[:max_chars] if path.exists() else "(missing doc)"
        parts.append(f"<doc id={doc_id}>\n{text}\n</doc>")
    return "\n".join(parts) or "(no documents retrieved)"

def render_prompt(criterion: Criterion, case: dict, pred: Any, examples: Sequence[dict] = ()) -> str:
    parts = [f"RULE: {criterion.rule}", f"FAIL EXAMPLE: {criterion.fail_example}",
             f"PASS EXAMPLE: {criterion.pass_example}"]
    parts += [f"EXAMPLE ({'PASS' if e['passed'] else 'FAIL'}): {e['text']}" for e in examples]
    parts += [f"QUESTION:\n{case['question']}", f"CONTEXT:\n{load_context(_field(pred, 'retrieved_ids'))}"]
    if criterion.uses_reference:
        parts.append(f"REFERENCE ANSWER:\n{case.get('answer') or '(none: the docs do not cover this)'}")
    parts.append(f"ANSWER:\n{_field(pred, 'answer')}")
    return "\n\n".join(parts)

# script_id: day_045_llm_as_judge/judges
JUDGE_SYSTEM = ("You grade one answer from Atlas, a support assistant for the Orbit library, on ONE "
                "rule. Use only the material given. In `reason`, quote the evidence in one or two "
                "sentences, then set `passed`. When unsure, FAIL.")

def is_judge_error(result: CheckResult) -> bool:
    return result.detail.startswith("judge_error")

def make_judge(criterion: Criterion, *, model: str = JUDGE_MODEL, examples: Sequence[dict] = (),
               client: OpenAI | None = None, tracker: CostTracker | None = None) -> Check:
    tag = f"[{criterion.name}@{criterion.version} {model}]"    # which rubric graded this
    def judge(case: dict, pred: Any) -> CheckResult:
        try:
            resp = (client or default_client()).responses.parse(
                model=model, instructions=JUDGE_SYSTEM, text_format=Verdict,
                reasoning={"effort": "low"},        # default is medium; a narrow rule rarely needs it
                input=render_prompt(criterion, case, pred, examples))
        except (openai.APIError, ValidationError) as err:   # API failure or unparseable output
            return CheckResult(criterion.name, False, f"judge_error: {type(err).__name__} {tag}")
        (tracker or TRACKER).record(model, resp.usage)      # billed even when it fails to parse
        verdict = resp.output_parsed
        if verdict is None or resp.status == "incomplete":  # refusal or truncation
            return CheckResult(criterion.name, False, f"judge_error: status={resp.status} {tag}")
        return CheckResult(criterion.name, verdict.passed, f"{verdict.reason} {tag}")
    judge.__name__ = criterion.name
    return judge

def load_jsonl(path: str | Path) -> list[dict]:
    return [json.loads(line) for line in Path(path).read_text(encoding="utf-8").splitlines() if line.strip()]

def load_run(path: str | Path) -> list[dict]:
    """A recorded run: Day 43's RunRecord rows (case_id, version, answer, citations, retrieved_ids)."""
    return load_jsonl(path)

# script_id: day_045_llm_as_judge/judge_offline
# Tests make_judge() with a fake client: no key, no network. Needs judges.py next to it.
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA (default datasets/orbit) · Services: none
from types import SimpleNamespace

from judges import RUBRICS, CostTracker, Verdict, is_judge_error, make_judge

class FakeResponses:
    def __init__(self, replies: list) -> None:
        self.replies, self.prompts = list(replies), []

    def parse(self, **kwargs):            # same keyword call the real SDK receives
        self.prompts.append(kwargs["input"])
        parsed, status = self.replies.pop(0)
        usage = SimpleNamespace(input_tokens=3_000, output_tokens=150)
        return SimpleNamespace(output_parsed=parsed, status=status, usage=usage)

fake = SimpleNamespace(responses=FakeResponses([
    (Verdict(reason="Every key appears in CONTEXT.", passed=True), "completed"),
    (Verdict(reason="max_retries_per_minute is not in CONTEXT.", passed=False), "completed"),
    (None, "incomplete"),                   # truncated or refused: a judge error, not a FAIL
]))
tracker = CostTracker()
judge = make_judge(RUBRICS["grounded"], client=fake, tracker=tracker)
case = {"id": "qa-008", "question": "Does a lease expiry count as a failed attempt?"}
pred = {"answer": "(recorded answer)", "retrieved_ids": ["retries-and-backoff"]}
results = [judge(case, pred) for _ in range(3)]
for r in results:
    print(f"passed={r.passed!s:<5} judge_error={is_judge_error(r)!s:<5} {r.detail}")
assert [r.passed for r in results] == [True, False, False]
assert [is_judge_error(r) for r in results] == [False, False, True]
assert "<doc id=retries-and-backoff>" in fake.responses.prompts[0]    # the judge saw the docs
print(f"judge spend: ${sum(tracker.usd.values()):.6f} over {sum(tracker.calls.values())} calls")

passed=True  judge_error=False Every key appears in CONTEXT. [grounded@1.0 gpt-6-luna]
passed=False judge_error=False max_retries_per_minute is not in CONTEXT. [grounded@1.0 gpt-6-luna]
passed=False judge_error=True  judge_error: status=incomplete [grounded@1.0 gpt-6-luna]
judge spend: $0.001125 over 3 calls

# script_id: day_045_llm_as_judge/compare_versions
# Judge R2 v1 vs v1.1 per criterion on recorded runs. Needs judges.py and recorded_runs.py's output.
# Install: as judges.py · Env: OPENAI_API_KEY, ORBIT_DATA, RUNS_DIR (default runs) · Services: none
import os
from pathlib import Path

from judges import (DATA_DIR, RUBRICS, TRACKER, CheckResult, is_judge_error, load_jsonl,
                    load_run, make_judge)

RUNS = Path(os.environ.get("RUNS_DIR", "runs"))

def check_cites_existing_doc(case: dict, pred: dict) -> CheckResult:    # compact Day 44 code check
    missing = [d for d in pred["citations"] if not (DATA_DIR / "docs" / f"{d}.md").exists()]
    return CheckResult("cites_existing_doc", bool(pred["citations"]) and not missing, ", ".join(missing))

def run_eval(cases: list[dict], run: list[dict], checks: list) -> dict[str, list[CheckResult]]:
    by_id = {r["case_id"]: r for r in run}          # compact Day 20 runner, grouped by check
    out: dict[str, list[CheckResult]] = {}
    for case in cases:
        for check in checks:
            result = check(case, by_id[case["id"]])
            out.setdefault(result.name, []).append(result)
    return out

def rate(results: list[CheckResult]) -> str:
    graded = [r for r in results if not is_judge_error(r)]      # errors leave the denominator
    errors = len(results) - len(graded)
    return f"{sum(r.passed for r in graded)}/{len(graded)}" + (f" ({errors} err)" if errors else "")

# script_id: day_045_llm_as_judge/judges
class PairwiseVerdict(BaseModel):
    reason: str
    winner: Literal["A", "B", "tie"]

PAIRWISE_RUBRIC = ("Two answers from Atlas to the same QUESTION, with the docs they retrieved. Which "
                   "would an Orbit maintainer rather receive? Prefer the answer whose claims are all in "
                   "CONTEXT, that answers what was asked, and that says so when the docs don't cover it. "
                   "Length is not quality. Say 'tie' if neither is clearly better. Explain first.")

def compare(case: dict, a: Any, b: Any, *, model: str = JUDGE_MODEL,
            client: OpenAI | None = None) -> PairwiseVerdict:
    ids = list(dict.fromkeys([*_field(a, "retrieved_ids"), *_field(b, "retrieved_ids")]))
    prompt = (f"QUESTION:\n{case['question']}\n\nCONTEXT:\n{load_context(ids)}\n\n"
              f"ANSWER A:\n{_field(a, 'answer')}\n\nANSWER B:\n{_field(b, 'answer')}")
    resp = (client or default_client()).responses.parse(
        model=model, instructions=PAIRWISE_RUBRIC, text_format=PairwiseVerdict,
        reasoning={"effort": "low"}, input=prompt)
    TRACKER.record(model, resp.usage)
    if resp.output_parsed is None:
        raise RuntimeError(f"no pairwise verdict for {case.get('id')} (status={resp.status})")
    return resp.output_parsed

def compare_both_orders(case: dict, new: Any, old: Any, **kwargs: Any) -> tuple[str, bool]:
    """('win' | 'loss' | 'tie', flipped) for `new` vs `old`. An order flip counts as a tie."""
    first = {"A": "new", "B": "old", "tie": "tie"}[compare(case, new, old, **kwargs).winner]
    second = {"A": "old", "B": "new", "tie": "tie"}[compare(case, old, new, **kwargs).winner]
    if first == second == "new":
        return "win", False
    if first == second == "old":
        return "loss", False
    return "tie", first != second

# script_id: day_045_llm_as_judge/judges
def tiered_judge(criterion: Criterion, *, cheap: str = JUDGE_MODEL, strong: str = STRONG_JUDGE_MODEL,
                 audit_rate: float = 0.1, seed: int = 0, client: OpenAI | None = None,
                 tracker: CostTracker | None = None) -> Check:
    """Cheap judge on every case; the strong judge re-grades cheap FAILs, judge errors and a seeded
    audit sample of PASSes. The strong verdict is final."""
    first, second = (make_judge(criterion, model=m, client=client, tracker=tracker)
                     for m in (cheap, strong))
    def judge(case: dict, pred: Any) -> CheckResult:
        result = first(case, pred)
        # Seeded per case, so the same cases are audited on every run and runs stay comparable.
        audited = random.Random(f"{seed}:{criterion.name}:{case.get('id')}").random() < audit_rate
        if result.passed and not audited:
            return result
        final = second(case, pred)
        if is_judge_error(final):       # keep the error visible: don't hide it behind "escalated:"
            return final
        return CheckResult(final.name, final.passed, f"escalated: {final.detail}")
    judge.__name__ = criterion.name
    return judge

# script_id: day_045_llm_as_judge/tier_costs
# Estimates $/run for three judging plans before you spend anything. Stdlib only, no key.
# Prices: USD per 1M tokens (input, output), as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00)}

def call_cost(model: str, in_tok: int, out_tok: int) -> float:
    p_in, p_out = PRICES[model]
    return (in_tok * p_in + out_tok * p_out) / 1e6

def plan_cost(n_cases: int, criteria: int, *, in_tok: int = 3_000, out_tok: int = 200,
              fail_rate: float = 0.2, audit_rate: float = 0.1) -> dict[str, float]:
    """out_tok includes reasoning tokens. Measure both from resp.usage on 10 real calls."""
    calls = n_cases * criteria
    luna, terra = call_cost("gpt-6-luna", in_tok, out_tok), call_cost("gpt-6.1-terra", in_tok, out_tok)
    escalated = calls * (fail_rate + (1 - fail_rate) * audit_rate)    # cheap FAILs + audited PASSes
    return {"all luna": calls * luna, "tiered": calls * luna + escalated * terra,
            "all terra": calls * terra}

if __name__ == "__main__":
    small, big = plan_cost(100, 2), plan_cost(1_000, 2)
    print(f"{'plan':<10}{'$/100 cases':>12}{'$/1000 cases':>14}")
    for plan in small:
        print(f"{plan:<10}{small[plan]:>12.3f}{big[plan]:>14.2f}")
    print("tiered at 50% fails:", f"{plan_cost(100, 2, fail_rate=0.5)['tiered']:.3f}")

plan       $/100 cases  $/1000 cases
all luna         0.080          0.80
tiered           0.304          3.04
all terra        0.800          8.00
tiered at 50% fails: 0.520