Set up for Unit 8: evaluation tools

{"id": "t2", "order": {"SalesOrder": "9000002", "HeaderBillingBlockReason": "02"}, "draft": {"likely_cause": "...", "next_check": "...", "confidence": "low"}, "verdict": "right", "reason": "Admits what the data can't tell."}

Name: inspect_ai
Version: 0.3.276
...
Name: ranx
Version: 0.3.21

inspect --version

"""Unit 8: turn judged triage notes into a golden set, and check that the file is well formed.

A golden set is a fixed list of cases with a known right answer. Here each case is one blocked
sales order, the draft note an AI wrote for it in Unit 1, and your verdict on that note.
The file is unit08/data/triage_golden.jsonl: one JSON object per line.

How to run (from your course folder, with .venv turned on):
    python unit08/golden_set.py --sample     # write six made-up judged notes to start from
    python unit08/golden_set.py              # check the file and print a summary
    python unit08/golden_set.py --sample --force   # overwrite the file with the samples again
"""
import argparse
import json
import sys
from collections import Counter
from pathlib import Path

GOLDEN = Path(__file__).resolve().parent / "data" / "triage_golden.jsonl"
VERDICTS = ("right", "partly", "wrong")
REQUIRED = ("id", "order", "draft", "verdict", "reason")


def order(number, customer, amount, delivery_block, billing_block, credit_status):
    """Build the header fields Unit 1's triage.py read. The codes are placeholders, not SAP meanings."""
    return {"SalesOrder": number, "SoldToParty": customer, "TotalNetAmount": amount,
            "TransactionCurrency": "USD", "TotalCreditCheckStatus": credit_status,
            "DeliveryBlockReason": delivery_block, "HeaderBillingBlockReason": billing_block}


def draft(cause, check, confidence):
    return {"likely_cause": cause, "next_check": check, "confidence": confidence}


# Six made-up judged notes, shaped like Unit 1's output. Replace or extend them with your own.
SAMPLES = [
    {"id": "t1", "order": order("9000001", "CUST-A", "18250.00", "01", "", "B"),
     "draft": draft("Both a delivery block and a credit check status are set, so credit is worth "
                    "checking first.", "Check the customer's credit exposure and open items.", "medium"),
     "verdict": "right", "reason": "Uses only the fields given and points to the right next check."},
    {"id": "t2", "order": order("9000002", "CUST-B", "940.00", "", "02", ""),
     "draft": draft("A billing block is set; the data doesn't say why.",
                    "Ask billing which block code 02 stands for in this system.", "low"),
     "verdict": "right", "reason": "Admits what the data can't tell and asks for the code meaning."},
    {"id": "t3", "order": order("9000004", "CUST-D", "72400.00", "01", "", "B"),
     "draft": draft("Delivery block 01 means the customer is over the credit limit.",
                    "Release the order in credit management.", "high"),
     "verdict": "wrong", "reason": "Guesses what code 01 means and tells the analyst to release, not to check."},
    {"id": "t4", "order": order("9000005", "CUST-E", "3150.00", "03", "02", ""),
     "draft": draft("A delivery block is set.", "Check the delivery block with the sales team.", "medium"),
     "verdict": "partly", "reason": "Misses the billing block on the same order."},
    {"id": "t5", "order": order("9000006", "CUST-A", "12900.00", "01", "", "B"),
     "draft": draft("A delivery block with a non-blank credit status; possibly a credit hold.",
                    "Check credit exposure; CUST-A also has order 9000001 blocked.", "medium"),
     "verdict": "partly", "reason": "Good check, but it cites another order it was never given."},
    {"id": "t6", "order": order("9000007", "CUST-F", "560.00", "", "", ""),
     "draft": draft("No block fields are set, so the data doesn't show why this order is held.",
                    "Confirm with the analyst why it was flagged.", "low"),
     "verdict": "right", "reason": "Says the data is insufficient instead of inventing a cause."},
]


def validate(path: Path) -> list:
    """Read the golden set and return its records, or stop with the line number of the first problem."""
    if not path.exists():
        sys.exit(f"No golden set yet at {path}. Run: python unit08/golden_set.py --sample")
    records, seen = [], set()
    for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not line.strip():
            continue  # blank lines are allowed
        try:
            record = json.loads(line)
        except json.JSONDecodeError as error:
            sys.exit(f"Line {number} is not valid JSON ({error.msg}). Each line must be one complete object.")
        missing = [field for field in REQUIRED if field not in record]
        if missing:
            sys.exit(f"Line {number} is missing: {', '.join(missing)}")
        if record["verdict"] not in VERDICTS:
            sys.exit(f"Line {number}: verdict must be one of {', '.join(VERDICTS)}, not {record['verdict']!r}")
        if record["id"] in seen:
            sys.exit(f"Line {number}: the id {record['id']!r} is used twice")
        seen.add(record["id"])
        records.append(record)
    if not records:
        sys.exit("The golden set is empty. Run with --sample, or add your own judged notes.")
    return records


def main() -> None:
    parser = argparse.ArgumentParser(description="Create or check the Unit 8 golden set of judged triage notes.")
    parser.add_argument("--sample", action="store_true", help="write six made-up judged notes")
    parser.add_argument("--force", action="store_true", help="with --sample, overwrite an existing file")
    args = parser.parse_args()

    if args.sample:
        if GOLDEN.exists() and not args.force:
            sys.exit(f"{GOLDEN.name} already exists. Add --force to overwrite it with the samples.")
        GOLDEN.parent.mkdir(parents=True, exist_ok=True)
        GOLDEN.write_text("".join(json.dumps(r) + "\n" for r in SAMPLES), encoding="utf-8")
        print(f"Wrote {len(SAMPLES)} made-up judged notes to unit08/data/{GOLDEN.name}")

    records = validate(GOLDEN)
    counts = Counter(r["verdict"] for r in records)
    print(f"\nGolden set: {len(records)} judged notes")
    for verdict in VERDICTS:
        print(f"  {verdict:7} {counts[verdict]}")
    share = counts["right"] / len(records)
    print(f"\nShare of drafts you judged right: {share:.0%}")
    print("That share is your first evaluation number: the quality of the Unit 1 drafts, judged by a person.")


if __name__ == "__main__":
    main()

Wrote 6 made-up judged notes to unit08/data/triage_golden.jsonl

Golden set: 6 judged notes
  right   3
  partly  2
  wrong   1

Share of drafts you judged right: 50%
That share is your first evaluation number: the quality of the Unit 1 drafts, judged by a person.

"""Unit 8 smoke test: score a tiny search result with three retrieval metrics, by hand and with ranx.

Three questions were asked of the Unit 7 help notes (IDs n1 to n8). For each, we know which notes
are relevant (the "qrels") and which notes the search returned, best first (the "run").
The script computes Recall@3, MRR and nDCG@3 in plain Python, then asks the ranx library for the
same numbers. If the two columns agree, the library is installed and you read it correctly.

How to run (from your course folder, with .venv turned on):
    python unit08/retrieval_metrics_hello.py              # by hand and with ranx
    python unit08/retrieval_metrics_hello.py --no-ranx    # by hand only (no library needed)
"""
import argparse
import math
import sys
import warnings

K = 3

# Which notes are relevant for each question. 2 = answers it fully, 1 = helps a bit.
QRELS = {
    "q1 why can't this order be delivered": {"n1": 2, "n3": 1},
    "q2 invoice quantity higher than goods receipt": {"n4": 2, "n6": 1},
    "q3 planned order arrives too late": {"n7": 2},
}

# What a search returned for each question, best first, with its similarity score.
RUN = {
    "q1 why can't this order be delivered": {"n3": 0.81, "n2": 0.77, "n1": 0.74, "n6": 0.40},
    "q2 invoice quantity higher than goods receipt": {"n4": 0.88, "n5": 0.71, "n6": 0.69, "n1": 0.30},
    "q3 planned order arrives too late": {"n8": 0.79, "n2": 0.55, "n5": 0.41, "n7": 0.39},
}


def ranked(results: dict) -> list:
    """Note IDs sorted by score, best first."""
    return sorted(results, key=results.get, reverse=True)


def recall_at_k(relevant: dict, results: dict, k: int) -> float:
    """Share of the relevant notes that appear in the top k."""
    top = ranked(results)[:k]
    return sum(1 for note in relevant if note in top) / len(relevant)


def reciprocal_rank(relevant: dict, results: dict) -> float:
    """1 divided by the position of the first relevant note (0 if none is returned)."""
    for position, note in enumerate(ranked(results), 1):
        if note in relevant:
            return 1 / position
    return 0.0


def ndcg_at_k(relevant: dict, results: dict, k: int) -> float:
    """Graded gain, discounted by position, divided by the best possible ordering."""
    gains = [relevant.get(note, 0) for note in ranked(results)[:k]]
    dcg = sum(g / math.log2(i + 2) for i, g in enumerate(gains))
    ideal = sorted(relevant.values(), reverse=True)[:k]
    idcg = sum(g / math.log2(i + 2) for i, g in enumerate(ideal))
    return dcg / idcg if idcg else 0.0


def by_hand() -> dict:
    n = len(QRELS)
    return {
        f"recall@{K}": sum(recall_at_k(QRELS[q], RUN[q], K) for q in QRELS) / n,
        "mrr": sum(reciprocal_rank(QRELS[q], RUN[q]) for q in QRELS) / n,
        f"ndcg@{K}": sum(ndcg_at_k(QRELS[q], RUN[q], K) for q in QRELS) / n,
    }


def with_ranx() -> dict:
    try:
        from ranx import Qrels, Run, evaluate
    except ImportError:
        sys.exit("ranx is not installed. Run: pip install -r requirements.txt  (or add --no-ranx)")
    warnings.filterwarnings("ignore", message="unsafe cast")  # a harmless notice from ranx's compiler
    print("Asking ranx (the first call compiles some code, so it can take several seconds)...")
    return evaluate(Qrels(QRELS), Run(RUN), [f"recall@{K}", "mrr", f"ndcg@{K}"])


def main() -> None:
    parser = argparse.ArgumentParser(description="Compute retrieval metrics by hand and with ranx.")
    parser.add_argument("--no-ranx", action="store_true", help="skip the ranx library")
    args = parser.parse_args()

    print("Per question (by hand):")
    for q in QRELS:
        print(f"  {q[:2]}  top {K}: {', '.join(ranked(RUN[q])[:K]):12}  "
              f"recall {recall_at_k(QRELS[q], RUN[q], K):.2f}  RR {reciprocal_rank(QRELS[q], RUN[q]):.2f}  "
              f"nDCG {ndcg_at_k(QRELS[q], RUN[q], K):.2f}")

    mine = by_hand()
    theirs = {} if args.no_ranx else with_ranx()
    print(f"\n{'metric':10} {'by hand':>8} {'ranx':>8}")
    for metric, value in mine.items():
        other = f"{theirs[metric]:8.3f}" if metric in theirs else "       -"
        print(f"{metric:10} {value:8.3f} {other}")
    if theirs:
        same = all(abs(mine[m] - theirs[m]) < 1e-6 for m in mine)
        print("\nThe two columns agree." if same else "\nThe columns differ: check the data you changed.")


if __name__ == "__main__":
    main()

Per question (by hand):
  q1  top 3: n3, n2, n1    recall 1.00  RR 1.00  nDCG 0.76
  q2  top 3: n4, n5, n6    recall 1.00  RR 1.00  nDCG 0.95
  q3  top 3: n8, n2, n5    recall 0.00  RR 0.25  nDCG 0.00
Asking ranx (the first call compiles some code, so it can take several seconds)...

metric      by hand     ranx
recall@3      0.667    0.667
mrr           0.750    0.750
ndcg@3        0.570    0.570

The two columns agree.

"""Unit 8 smoke test: run a first evaluation with Inspect, using your golden set.

The task asks a model to act as a judge: for each judged triage note in
unit08/data/triage_golden.jsonl, it reads the order and the draft note and replies right, partly
or wrong. Inspect compares the model's verdict with yours and reports the share that agree.

How to run (from your course folder, with .venv turned on):
    python unit08/inspect_hello.py                                    # no account: a mock judge
    python unit08/inspect_hello.py --model anthropic/claude-opus-5-5  # a real model (small charge)
Then look at the results in your browser:
    inspect view --log-dir unit08/logs
"""
import argparse
import json
import os
import sys
from pathlib import Path

from dotenv import load_dotenv

HERE = Path(__file__).resolve().parent
GOLDEN = HERE / "data" / "triage_golden.jsonl"
LOGS = HERE / "logs"

JUDGE = (
    "You review draft notes that an assistant wrote for credit and order-management analysts. "
    "A good note uses only the order fields given, does not guess what block or status codes mean, "
    "says when the data is insufficient, and suggests a sensible next check. "
    "Judge the draft as right, partly (useful but incomplete or with a small error) or wrong. "
    "Think briefly, then end with one line exactly like: VERDICT: right"
)


def load_samples() -> list:
    """Turn each golden-set line into an Inspect Sample: input = order + draft, target = your verdict."""
    from inspect_ai.dataset import Sample

    if not GOLDEN.exists():
        sys.exit("No golden set yet. Run: python unit08/golden_set.py --sample")
    samples = []
    for line in GOLDEN.read_text(encoding="utf-8").splitlines():
        if line.strip():
            record = json.loads(line)
            text = (f"Order fields:\n{json.dumps(record['order'], indent=2)}\n\n"
                    f"Draft note:\n{json.dumps(record['draft'], indent=2)}")
            samples.append(Sample(id=record["id"], input=text, target=record["verdict"]))
    return samples


def lazy_judge():
    """A stand-in model for the no-account path: it answers 'partly' every time, without reading."""
    from inspect_ai.model import ModelOutput, ModelUsage, get_model

    def always_partly(messages, tools, tool_choice, config):
        output = ModelOutput.from_content(model="mockllm", content="I did not read it.\nVERDICT: partly")
        output.usage = ModelUsage(input_tokens=0, output_tokens=0, total_tokens=0)  # nothing to count offline
        return output

    return get_model("mockllm/model", custom_outputs=always_partly)


def main() -> None:
    parser = argparse.ArgumentParser(description="Run a first Inspect evaluation over the golden set.")
    parser.add_argument("--model", help="for example anthropic/claude-opus-5-5; leave out for the mock judge")
    args = parser.parse_args()

    load_dotenv()  # ANTHROPIC_API_KEY comes from .env, as in Unit 1
    if args.model and args.model.startswith("anthropic/") and not os.getenv("ANTHROPIC_API_KEY"):
        sys.exit("ANTHROPIC_API_KEY is not set in .env (see Set up your computer). Or leave out --model.")
    try:
        from inspect_ai import Task, eval
        from inspect_ai.scorer import pattern
        from inspect_ai.solver import generate, system_message
    except ImportError:
        sys.exit("inspect-ai is not installed. Run: pip install -r requirements.txt")

    task = Task(
        dataset=load_samples(),
        solver=[system_message(JUDGE), generate()],
        scorer=pattern(r"VERDICT:\s*(right|partly|wrong)"),  # pull the verdict out of the reply
        name="triage_judge",
    )
    model = args.model or lazy_judge()
    print(f"Judging {len(task.dataset)} notes with {args.model or 'a mock judge that always says partly'}...")
    logs = eval(task, model=model, log_dir=str(LOGS), display="plain")

    log = logs[0]
    if log.status != "success":
        sys.exit(f"The evaluation did not finish ({log.status}). Read the error above, or run: inspect view --log-dir unit08/logs")
    accuracy = log.results.scores[0].metrics["accuracy"].value
    print(f"\nAgreement with your verdicts: {accuracy:.0%} of {len(task.dataset)} notes")
    print("Log saved in unit08/logs. Browse it with: inspect view --log-dir unit08/logs")


if __name__ == "__main__":
    main()

Judging 6 notes with a mock judge that always says partly...
...
triage_judge (6 samples): mockllm/model
...
pattern
accuracy  0.333
stderr    0.211
Log:
unit08/logs/2026-10-04T18-24-25-00-00_triage-judge_....eval

Agreement with your verdicts: 33% of 6 notes
Log saved in unit08/logs. Browse it with: inspect view --log-dir unit08/logs

"""Check that your computer is ready for Unit 8 (evaluation).

Run it from your course folder:  python check_unit08.py
It uses built-in Python only. It looks for the Unit 8 libraries and files, checks that your golden
set is well formed, and reads the names (not the values) of the keys in .env. It changes nothing.
"""
import importlib.metadata
import importlib.util
import json
import os
import sys

problems = 0
SAMPLE_IDS = {"t1", "t2", "t3", "t4", "t5", "t6"}


def report(ok: bool, label: str, fix: str = "", optional: bool = False) -> None:
    """Print one line: OK, MISSING (must fix) or LATER (optional for now)."""
    global problems
    if ok:
        print(f"  OK       {label}")
    elif optional:
        print(f"  LATER    {label}  ->  {fix}")
    else:
        problems += 1
        print(f"  MISSING  {label}  ->  {fix}")


def library(module: str, package: str) -> str:
    """Return the installed version of a library, or '' if it isn't installed."""
    if importlib.util.find_spec(module) is None:
        return ""
    try:
        return importlib.metadata.version(package)
    except importlib.metadata.PackageNotFoundError:
        return "installed"


def env_names(path: str = ".env") -> set:
    """Names of the settings in .env that have a value (the values are never printed)."""
    names = set()
    if os.path.exists(path):
        with open(path, encoding="utf-8") as handle:
            for line in handle:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    name, value = line.split("=", 1)
                    if value.strip().strip('"').strip("'"):
                        names.add(name.strip())
    return names


def golden_set(path: str):
    """Return (number of records, number of your own records, error text)."""
    if not os.path.exists(path):
        return 0, 0, "not found"
    total = own = 0
    with open(path, encoding="utf-8") as handle:
        for number, line in enumerate(handle, 1):
            if not line.strip():
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                return total, own, f"line {number} is not valid JSON"
            if record.get("verdict") not in ("right", "partly", "wrong"):
                return total, own, f"line {number} has no valid verdict"
            total += 1
            own += record.get("id") not in SAMPLE_IDS
    return total, own, ""


print("\n1. Python")
v = sys.version_info
report(v >= (3, 11), f"Python {v.major}.{v.minor}.{v.micro}",
       "the course needs Python 3.11 or newer (see Set up for Unit 2, Step 1)")
report(sys.prefix != sys.base_prefix, "virtual environment is active", "activate .venv (Step 1)")

print("\n2. Python libraries")
for module, package, fix in [
    ("inspect_ai", "inspect-ai", "pip install -r requirements.txt (Step 2)"),
    ("ranx", "ranx", "pip install -r requirements.txt (Step 2)"),
    ("sklearn", "scikit-learn", "added in Set up for Unit 2; pip install -r requirements.txt"),
    ("pytest", "pytest", "added in Testing AI applications (Unit 6); add pytest to requirements.txt"),
    ("dotenv", "python-dotenv", "added in Set up your computer; pip install -r requirements.txt"),
    ("anthropic", "anthropic", "added in Set up your computer; pip install -r requirements.txt"),
]:
    version = library(module, package)
    report(bool(version), f"{package} {version}".strip(), fix)

print("\n3. Course folder")
for name, step in [("golden_set.py", "3"), ("retrieval_metrics_hello.py", "4"), ("inspect_hello.py", "5")]:
    path = os.path.join("unit08", name)
    report(os.path.exists(path), path, f"create it (Step {step})")
total, own, error = golden_set(os.path.join("unit08", "data", "triage_golden.jsonl"))
report(total > 0 and not error, f"unit08/data/triage_golden.jsonl ({total} judged notes)",
       error or "run python unit08/golden_set.py --sample (Step 3)")
report(own >= 5, f"your own judged notes in the golden set ({own})",
       "add at least 5 of your own (Exercise)", optional=True)
report(os.path.isdir(os.path.join("unit08", "logs")), "unit08/logs (your first Inspect run)",
       "run python unit08/inspect_hello.py (Step 5)", optional=True)
ignored = os.path.exists(".gitignore") and "unit08/logs" in open(".gitignore", encoding="utf-8").read()
report(ignored, ".gitignore leaves out unit08/logs", "add the line unit08/logs/ (Step 5)", optional=True)

print("\n4. A model for judging (needed from LLM evaluation fundamentals)")
names = env_names()
report("ANTHROPIC_API_KEY" in names, "ANTHROPIC_API_KEY in .env",
       "add it as in Set up your computer; until then use the mock judge", optional=True)

print()
if problems:
    print(f"{problems} item(s) to fix. Fix them in order, then run this again.")
    sys.exit(1)
print("All set. Your computer is ready for Unit 8.")

1. Python
  OK       Python 3.13.16
  OK       virtual environment is active

2. Python libraries
  OK       inspect-ai 0.3.276
  OK       ranx 0.3.21
  OK       scikit-learn 1.9.1
  OK       pytest 9.1.1
  OK       python-dotenv 1.2.4
  OK       anthropic 1.11.0

3. Course folder
  OK       unit08/golden_set.py
  OK       unit08/retrieval_metrics_hello.py
  OK       unit08/inspect_hello.py
  OK       unit08/data/triage_golden.jsonl (6 judged notes)
  LATER    your own judged notes in the golden set (0)  ->  add at least 5 of your own (Exercise)
  OK       unit08/logs (your first Inspect run)
  OK       .gitignore leaves out unit08/logs

4. A model for judging (needed from LLM evaluation fundamentals)
  LATER    ANTHROPIC_API_KEY in .env  ->  add it as in Set up your computer; until then use the mock judge

All set. Your computer is ready for Unit 8.