"""Score a fine-tuned model and the base model it came from on the same tasks, and report the gap. Purpose: the only question this part asks. Runs Part 10's evaluation harness twice over the same task files at the same settings, once against the base model and once against the fine-tune, optionally grades both with the same judge, and prints the difference per category so that a gain on the task you trained for and a loss on everything else are both visible. Appends one record to the lab notebook. Platform: all (pure Python over HTTP; both models are reached through an OpenAI-compatible API, so they may be served by any engine on any track, or by the Part 9 gateway) Minimum memory: 8 GB on the machine serving the models; this script needs very little Assumes: Python 3.10 or newer; Part 10's run-eval.py and judge.py in --harness-dir; both models reachable at --base-url under the names given, one at a time or together; sftlog.py next to this file. Pass every task file you care about: the one that tests the behaviour you trained, and your own Part 10 set, which is where a regression shows up. Usage: python3 evaluate-against-base.py --harness-dir ~/eval \ --base-url http://127.0.0.1:8080/v1 \ --base-model qwen3-1.7b --tuned-model qwen3-1.7b-format \ --quant Q4_K_M --engine llama.cpp --engine-version v0.4.0 \ --tasks format-tasks.json --tasks my-tasks.json \ --out-dir eval-out --labbook labbook.md Add --judge-model qwen3-8b to grade both runs with the same judge. Leave it out and only the deterministic checks are compared, which is a floor rather than a score but is exactly repeatable and needs no third model. """ from __future__ import annotations import argparse import json import os import subprocess import sys import shlex from collections import defaultdict from pathlib import Path from typing import Any import sftlog def run_harness(script: Path, arguments: list[str]) -> None: """Run one harness script, showing the command first so the log says what happened.""" command = [sys.executable, str(script), *arguments] safe_command = list(command) for i, value in enumerate(safe_command[:-1]): if value == "--api-key": safe_command[i + 1] = "[redacted]" print("+ " + shlex.join(safe_command)) subprocess.run(command, check=True) def category_pass_rates(results: list[dict[str, Any]]) -> dict[str, tuple[int, int]]: """Deterministic passes and totals per category. No model involved, exactly repeatable.""" counts: dict[str, list[int]] = defaultdict(lambda: [0, 0]) for item in results: counts[item["category"]][1] += 1 counts[item["category"]][0] += int(item["checks"]["passed"]) return {k: (v[0], v[1]) for k, v in sorted(counts.items())} def evaluate(args, script_dir: Path, out_dir: Path, tasks_path: Path, model: str, label: str) -> dict[str, Any]: """One model, one task file: run it, optionally grade it, and return the numbers.""" stem = f"{label}-{tasks_path.stem}" results_path = out_dir / f"results-{stem}.json" arguments = [ "--base-url", (args.tuned_url or args.base_url) if label == "tuned" else args.base_url, "--model", model, "--quant", args.quant, "--engine", args.engine, "--engine-version", args.engine_version, "--tasks", str(tasks_path), "--out", str(results_path), "--notes", f"part-13 {label} on {tasks_path.name}", ] if args.api_key: arguments += ["--api-key", args.api_key] if args.system: arguments += ["--system", args.system] if args.labbook: arguments += ["--labbook", args.labbook] run_harness(script_dir / "run-eval.py", arguments) payload = json.loads(results_path.read_text(encoding="utf-8")) out: dict[str, Any] = { "model": model, "results_file": str(results_path), "tasks": payload["run"]["tasks"], "checks_passed": payload["run"]["checks_passed"], "by_category_checks": category_pass_rates(payload["results"]), "judge_mean": None, "by_category_judge": {}, } if args.judge_model: judged_path = out_dir / f"judged-{stem}.json" judge_arguments = [ "grade", "--base-url", args.judge_url or args.base_url, "--judge-model", args.judge_model, "--results", str(results_path), "--out", str(judged_path), ] if args.api_key: judge_arguments += ["--api-key", args.api_key] if args.labbook: judge_arguments += ["--labbook", args.labbook] run_harness(script_dir / "judge.py", judge_arguments) judged = json.loads(judged_path.read_text(encoding="utf-8")) out["judge_mean"] = judged["judge"]["judge_mean"] out["by_category_judge"] = judged["judge"]["by_category"] out["judged_file"] = str(judged_path) return out def delta(new: float | None, old: float | None) -> str: if new is None or old is None: return "—" difference = new - old return f"{difference:+.2f}" def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--harness-dir", default=".", help="directory holding Part 10's run-eval.py and judge.py") parser.add_argument("--base-url", default="http://127.0.0.1:8080/v1") parser.add_argument("--tuned-url", default=None, help="fine-tune endpoint; defaults to --base-url for a shared gateway") parser.add_argument("--pause-before-tuned", action="store_true", help="wait for you to stop the base server and start the fine-tune") parser.add_argument("--api-key", default=os.environ.get("OPENAI_API_KEY")) parser.add_argument("--base-model", required=True, help="the name the endpoint answers to for the base model") parser.add_argument("--tuned-model", required=True, help="the name the endpoint answers to for the fine-tune") parser.add_argument("--quant", required=True, help="the quantisation both models are serving at; a comparison " "across two quantisations measures the quantisation as well") parser.add_argument("--engine", default="llama.cpp") parser.add_argument("--engine-version", default="unknown") parser.add_argument("--system", default=None, help="system prompt sent with every task; use the one you trained with") parser.add_argument("--tasks", action="append", required=True, help="a task file; repeat the flag for more than one") parser.add_argument("--judge-model", default=None, help="grade both runs with this model; leave out for checks only") parser.add_argument("--judge-url", default=None, help="endpoint for the judge, if it is not at --base-url") parser.add_argument("--out-dir", default="eval-out") parser.add_argument("--labbook", default=None) parser.add_argument("--notes", default=None) args = parser.parse_args() script_dir = Path(args.harness_dir).expanduser() for name in ("run-eval.py", "judge.py") if args.judge_model else ("run-eval.py",): if not (script_dir / name).is_file(): raise SystemExit(f"{script_dir / name} not found; --harness-dir must point at the " "directory holding Part 10's harness scripts") if args.judge_model in (args.base_model, args.tuned_model): print("WARNING: the judge is one of the models under test. Part 10's lab covers " "self-preference bias; use a third, larger model if you can.") out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) comparison: dict[str, Any] = {} for tasks in args.tasks: tasks_path = Path(tasks) if not tasks_path.is_file(): raise SystemExit(f"{tasks_path} does not exist") print(f"\n=== {tasks_path.name}: base model {args.base_model}") base = evaluate(args, script_dir, out_dir, tasks_path, args.base_model, "base") if args.pause_before_tuned: input("Base results saved. Stop the base server, start the fine-tune, " "verify its readiness, then press Enter: ") print(f"\n=== {tasks_path.name}: fine-tune {args.tuned_model}") tuned = evaluate(args, script_dir, out_dir, tasks_path, args.tuned_model, "tuned") comparison[tasks_path.name] = {"base": base, "tuned": tuned} if args.pause_before_tuned and tasks != args.tasks[-1]: input("Fine-tune results saved. Restore the base server for the next task file, " "verify readiness, then press Enter: ") print("\n" + "=" * 78) print("BASE VERSUS FINE-TUNE, same tasks, same settings, same quantisation") print("=" * 78) for name, pair in comparison.items(): base, tuned = pair["base"], pair["tuned"] print(f"\n{name} ({base['tasks']} task(s))") print(f" deterministic checks base {base['checks_passed']:>3}/{base['tasks']}" f" fine-tune {tuned['checks_passed']:>3}/{tuned['tasks']}" f" change {tuned['checks_passed'] - base['checks_passed']:+d}") if base["judge_mean"] is not None: print(f" judge mean base {base['judge_mean']:>5}" f" fine-tune {tuned['judge_mean']:>5}" f" change {delta(tuned['judge_mean'], base['judge_mean'])}") categories = sorted(set(base["by_category_checks"]) | set(tuned["by_category_checks"])) for category in categories: b_pass, b_total = base["by_category_checks"].get(category, (0, 0)) t_pass, t_total = tuned["by_category_checks"].get(category, (0, 0)) line = (f" {category:14s} checks {b_pass}/{b_total} -> {t_pass}/{t_total}" f" ({t_pass - b_pass:+d})") if base["by_category_judge"] or tuned["by_category_judge"]: b_judge = base["by_category_judge"].get(category) t_judge = tuned["by_category_judge"].get(category) line += f" judge {b_judge} -> {t_judge} ({delta(t_judge, b_judge)})" print(line) print("\nRead the categories, not the total. A fine-tune that gains on the behaviour you") print("trained and loses on two others can leave the overall mean unchanged, and that") print("trade is the finding. Anything smaller than the difference between two runs of") print("the base model at these settings is not a difference.") if args.labbook: record = sftlog.record( labbook=args.labbook, lab="part-13/evaluate-against-base", model=f"{args.tuned_model} vs {args.base_model}", dataset={"task_files": args.tasks, "sha256": {t: sftlog.file_sha256(t) for t in args.tasks}}, hyperparameters={"base_url": args.base_url, "tuned_url": args.tuned_url or args.base_url, "quant": args.quant, "engine": args.engine, "engine_version": args.engine_version, "judge_model": args.judge_model, "system_prompt": args.system}, seed=0, losses={}, scores={name: { "base_checks_passed": pair["base"]["checks_passed"], "tuned_checks_passed": pair["tuned"]["checks_passed"], "tasks": pair["base"]["tasks"], "base_judge_mean": pair["base"]["judge_mean"], "tuned_judge_mean": pair["tuned"]["judge_mean"], "base_by_category_judge": pair["base"]["by_category_judge"], "tuned_by_category_judge": pair["tuned"]["by_category_judge"], } for name, pair in comparison.items()}, config_path=__file__, notes=args.notes, ) print(f"\nrecorded comparison {record['run_id']} in {args.labbook}") if __name__ == "__main__": main()