#!/usr/bin/env python3 """Score a run with a judge model, and measure the judge before you believe it. Purpose: the judging half of the personal evaluation harness. Three modes. `grade` rates every answer in a results file against its reference and rubric on a fixed 1-5 scale. `compare` puts two runs head to head and asks the judge twice with the order swapped, so that position bias shows up as a flip rate rather than hiding inside the result. `human` asks you to score a sample yourself and reports how far the judge and you agree, which is the only number that tells you whether the judge is measuring your task or its own preferences. Platform: all (pure Python over HTTP) Minimum memory: 8 GB on the machine running the judge model; this script needs very little Assumes: Python 3.9 or later, results files written by run-eval.py, and an OpenAI-compatible endpoint at --base-url serving the judge model. Use a different and preferably larger model than the one under test: a model grading its own answers scores them generously. Usage: python3 judge.py grade --results results-qwen3-4b.json --judge-model qwen3-8b \ --out judged-qwen3-4b.json --labbook labbook.md python3 judge.py compare --results-a results-qwen3-4b.json \ --results-b results-qwen3-8b.json --judge-model qwen3-8b --labbook labbook.md python3 judge.py human --judged judged-qwen3-4b.json --sample 8 --labbook labbook.md """ from __future__ import annotations import argparse import json import random import sys import time import urllib.error import urllib.request from collections import Counter, defaultdict from pathlib import Path from typing import Optional GRADE_SYSTEM = """You grade one answer against a reference answer and a rubric. Reply with JSON only. Scale: 5 meets the rubric completely 4 meets it with one small omission or one harmless addition 3 partly meets it: one required element missing or wrong 2 mostly fails it, or answers a different question 1 fails it, or contradicts the reference Grade only against the rubric. Length is not quality: a short answer that meets the rubric scores higher than a long one that meets it and adds material the reference does not support. Do not reward confidence, formatting flourishes or politeness.""" COMPARE_SYSTEM = """You are shown one task and two answers, labelled first and second. Decide which better meets the rubric. Reply with JSON only. Judge only against the rubric. Ignore which answer is longer, which sounds more confident, and which is presented first. If they meet the rubric equally well, say tie.""" def post_json(url: str, payload: dict, api_key: Optional[str], timeout: int) -> dict: body = json.dumps(payload).encode("utf-8") headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" request = urllib.request.Request(url, data=body, headers=headers, method="POST") try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:400] raise RuntimeError(f"{url} returned HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"cannot reach {url}: {exc.reason}") from exc def ask_judge(system: str, user: str, schema: dict, args) -> Optional[dict]: payload = { "model": args.judge_model, "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}], "temperature": 0.0, "max_tokens": 300, "seed": args.seed, "response_format": {"type": "json_schema", "json_schema": {"name": "verdict", "schema": schema, "strict": True}}, } try: body = post_json(args.base_url.rstrip("/") + "/chat/completions", payload, args.api_key, args.timeout) return json.loads(body["choices"][0]["message"]["content"]) except (RuntimeError, KeyError, ValueError, TypeError) as exc: print(f" judge call failed: {exc}", file=sys.stderr) return None GRADE_SCHEMA = { "type": "object", "properties": { "score": {"type": "integer", "minimum": 1, "maximum": 5}, "why": {"type": "string"}, }, "required": ["score", "why"], "additionalProperties": False, } COMPARE_SCHEMA = { "type": "object", "properties": { "winner": {"type": "string", "enum": ["first", "second", "tie"]}, "why": {"type": "string"}, }, "required": ["winner", "why"], "additionalProperties": False, } # -------------------------------------------------------------------------------------- # grade # -------------------------------------------------------------------------------------- def mode_grade(args) -> None: payload = json.loads(Path(args.results).read_text(encoding="utf-8")) run, results = payload["run"], payload["results"] started = time.time() by_category = defaultdict(list) for item in results: user = (f"Task: {item['prompt']}\n\nRubric: {item['rubric']}\n\n" f"Reference answer: {item['reference']}\n\n" f"Answer to grade: {item['answer'] or '(empty)'}") verdict = ask_judge(GRADE_SYSTEM, user, GRADE_SCHEMA, args) item["judge"] = verdict score = verdict["score"] if verdict else None if score is not None: by_category[item["category"]].append(score) print(f" {item['id']} [{item['category']:11s}] judge={score if score else '-'} " f"checks={'pass' if item['checks']['passed'] else 'fail'}") scores = [s for values in by_category.values() for s in values] lengths = {s: [] for s in range(1, 6)} for item in results: if item.get("judge"): lengths[item["judge"]["score"]].append(len(item["answer"].split())) summary = { "lab": "part-10/lab-benchmark-models-on-your-own-tasks/judge", "mode": "grade", "model": run["model"], "quant": run["quant"], "engine": run["engine"], "engine_version": run["engine_version"], "settings": run["settings"], "task_set": run["task_set"], "task_set_version": run["task_set_version"], "judge_model": args.judge_model, "graded": len(scores), "judge_mean": round(sum(scores) / len(scores), 2) if scores else None, "checks_passed": run["checks_passed"], "tasks": run["tasks"], "by_category": {k: round(sum(v) / len(v), 2) for k, v in by_category.items() if v}, # The verbosity diagnostic: if answers scored 5 are much longer than answers scored 2, # the judge may be rewarding length rather than quality. Read it alongside the # human-agreement number from `judge.py human`. "mean_words_by_score": {str(s): round(sum(v) / len(v), 1) for s, v in lengths.items() if v}, "self_judged": args.judge_model == run["model"], "seconds": round(time.time() - started, 1), "date": time.strftime("%Y-%m-%d"), } Path(args.out).write_text(json.dumps({"run": run, "judge": summary, "results": results}, indent=2), encoding="utf-8") print("\n" + json.dumps(summary, indent=2)) if summary["self_judged"]: print("\nWARNING: the judge and the model under test are the same. Self-preference " "bias makes this number optimistic. Use a different model.", file=sys.stderr) record(args, summary) # -------------------------------------------------------------------------------------- # compare, with the position swap # -------------------------------------------------------------------------------------- def mode_compare(args) -> None: left = json.loads(Path(args.results_a).read_text(encoding="utf-8")) right = json.loads(Path(args.results_b).read_text(encoding="utf-8")) by_id = {item["id"]: item for item in right["results"]} tally = Counter() started = time.time() for item in left["results"]: other = by_id.get(item["id"]) if other is None: continue def one(first_answer: str, second_answer: str) -> Optional[str]: user = (f"Task: {item['prompt']}\n\nRubric: {item['rubric']}\n\n" f"Reference answer: {item['reference']}\n\n" f"First answer: {first_answer or '(empty)'}\n\n" f"Second answer: {second_answer or '(empty)'}") verdict = ask_judge(COMPARE_SYSTEM, user, COMPARE_SCHEMA, args) return verdict["winner"] if verdict else None # Ask twice with the order swapped, and translate both verdicts back to A or B. A judge # free of position bias gives the same answer both ways; every disagreement is a flip. forward = one(item["answer"], other["answer"]) backward = one(other["answer"], item["answer"]) if forward is None or backward is None: tally["failed"] += 1 continue first_pass = {"first": "A", "second": "B", "tie": "tie"}[forward] second_pass = {"first": "B", "second": "A", "tie": "tie"}[backward] if first_pass == second_pass: tally[f"wins_{first_pass}" if first_pass != "tie" else "ties"] += 1 else: tally["flips"] += 1 tally["compared"] += 1 print(f" {item['id']:5s} forward={first_pass:4s} swapped={second_pass:4s}" f"{' FLIP' if first_pass != second_pass else ''}") compared = tally["compared"] or 1 summary = { "lab": "part-10/lab-benchmark-models-on-your-own-tasks/judge", "mode": "compare", "a": {"model": left["run"]["model"], "quant": left["run"]["quant"], "settings": left["run"]["settings"]}, "b": {"model": right["run"]["model"], "quant": right["run"]["quant"], "settings": right["run"]["settings"]}, "judge_model": args.judge_model, "compared": tally["compared"], "wins_a": tally["wins_A"], "wins_b": tally["wins_B"], "ties": tally["ties"], "order_flips": tally["flips"], "flip_rate": round(tally["flips"] / compared, 3), "judge_failures": tally["failed"], "seconds": round(time.time() - started, 1), "date": time.strftime("%Y-%m-%d"), } print("\n" + json.dumps(summary, indent=2)) print("\nflip_rate is the share of tasks where swapping the order changed the verdict. " "It is a measurement of the judge, not of either model.") record(args, summary) # -------------------------------------------------------------------------------------- # human agreement # -------------------------------------------------------------------------------------- def mode_human(args) -> None: payload = json.loads(Path(args.judged).read_text(encoding="utf-8")) results = [r for r in payload["results"] if r.get("judge")] if not results: sys.exit("no judged results in that file; run `judge.py grade` first") random.seed(args.seed) sample = random.sample(results, min(args.sample, len(results))) print(f"Scoring {len(sample)} answers yourself. The judge's score is hidden until the end.\n" "Use the same 1-5 scale the rubric describes. Enter s to skip.\n") pairs = [] for item in sample: print("=" * 72) print(f"TASK {item['id']} [{item['category']}]") print(item["prompt"][:600]) print(f"\nRUBRIC {item['rubric']}") print(f"\nREFERENCE {item['reference'][:600]}") print(f"\nANSWER {item['answer'][:900] or '(empty)'}\n") while True: raw = input("your score 1-5 (s to skip): ").strip().lower() if raw == "s": break if raw in {"1", "2", "3", "4", "5"}: pairs.append((int(raw), item["judge"]["score"], item["id"])) break print(" 1, 2, 3, 4, 5 or s") if not pairs: sys.exit("nothing scored") exact = sum(1 for mine, theirs, _ in pairs if mine == theirs) within_one = sum(1 for mine, theirs, _ in pairs if abs(mine - theirs) <= 1) disagreements = [(i, mine, theirs) for mine, theirs, i in pairs if abs(mine - theirs) >= 2] summary = { "lab": "part-10/lab-benchmark-models-on-your-own-tasks/judge", "mode": "human-agreement", "judge_model": payload["judge"]["judge_model"], "model": payload["run"]["model"], "scored": len(pairs), "exact_agreement": round(exact / len(pairs), 2), "within_one": round(within_one / len(pairs), 2), "mean_human": round(sum(m for m, _, _ in pairs) / len(pairs), 2), "mean_judge": round(sum(t for _, t, _ in pairs) / len(pairs), 2), "big_disagreements": [{"id": i, "human": m, "judge": t} for i, m, t in disagreements], "date": time.strftime("%Y-%m-%d"), } print("\n" + json.dumps(summary, indent=2)) print("\nRead the big disagreements. Each one is either a rubric that says less than you " "meant, or a judge that cannot grade this category. Both are worth fixing.") record(args, summary) def record(args, summary: dict) -> None: if not args.labbook: return with Path(args.labbook).open("a", encoding="utf-8") as handle: handle.write(json.dumps(summary) + "\n") print(f"\nrecorded in {args.labbook}") def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("mode", choices=["grade", "compare", "human"]) parser.add_argument("--base-url", default="http://127.0.0.1:8080/v1") parser.add_argument("--api-key", default=None) parser.add_argument("--judge-model", default=None, help="the grading model; not the one under test") parser.add_argument("--results", default="results.json", help="grade: the run to score") parser.add_argument("--results-a", default=None, help="compare: the first run") parser.add_argument("--results-b", default=None, help="compare: the second run") parser.add_argument("--judged", default="judged.json", help="human: a file from grade mode") parser.add_argument("--out", default="judged.json", help="grade: where to write the scores") parser.add_argument("--sample", type=int, default=8, help="human: how many to score yourself") parser.add_argument("--seed", type=int, default=7) parser.add_argument("--timeout", type=int, default=300) parser.add_argument("--labbook", default=None) args = parser.parse_args() if args.mode in {"grade", "compare"} and not args.judge_model: sys.exit("--judge-model is required for grade and compare") if args.mode == "compare" and not (args.results_a and args.results_b): sys.exit("compare needs --results-a and --results-b") if args.mode == "grade": mode_grade(args) elif args.mode == "compare": mode_compare(args) else: mode_human(args) if __name__ == "__main__": main()