"""Score a fine-tuned agent model and the base it came from on the same suites, and report the gap. Purpose: the only question this part asks. Runs Part 26's agent-eval.py over the same task suite twice, once against the base model and once against the fine-tune, runs Part 24's tool-call-reliability.py against both, optionally runs Part 10's run-eval.py over your general task set to find the regressions, and prints the difference per category so that a gain on the work you trained for and a loss on everything else are both visible at once. 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: 16 GB on the machine serving the models; this script needs very little Assumes: Python 3.10 or newer. Part 26's agent-eval.py and agent-tasks.json, and Part 24's tool-call-reliability.py and tool-prompts.json, at the paths you pass. Both models answering under the aliases you give, one at a time or together. agentlog.py sits next to this file. Nothing here starts an engine: a script that starts models is a script that hides which model answered. Usage: python3 compare-agent-models.py \\ --agent-eval ../part-26-building-agent-systems/agent-eval.py \\ --agent ../part-26-building-agent-systems/scaffold-minimal.py \\ --tasks ../part-26-building-agent-systems/agent-tasks.json \\ --reliability ../part-24-tools-mcp-and-the-agent-loop/tool-call-reliability.py \\ --base-url http://127.0.0.1:4000/v1 \\ --base-model local/agent --tuned-model local/agent-tuned \\ --workspace ./agent-workspace --repeats 3 \\ --out-dir compare-out --labbook labbook.md Add --harness-dir ~/eval --general-tasks my-tasks.json to run Part 10's harness over your own general set as well. That is where a fine-tune's regressions show up, and leaving it out is how people ship one without noticing. """ from __future__ import annotations import argparse import json import subprocess import sys from collections import defaultdict from pathlib import Path from typing import Any import agentlog def run(command: list[str], capture: bool = False) -> str: """Run one child process, printing the command first so the log says what happened.""" print("+ " + " ".join(command)) done = subprocess.run(command, capture_output=capture, text=True, check=False) if capture and done.returncode not in (0, 2): sys.exit(f"{command[1]} exited {done.returncode}:\n{(done.stderr or '')[:800]}") return done.stdout if capture else "" def agent_suite(args: argparse.Namespace, out_dir: Path, model: str, label: str) -> dict[str, Any]: """One model through Part 26's harness. Exit code 2 means some tasks failed, which is the normal case here and not an error.""" results_path = out_dir / f"agent-{label}.json" command = [ sys.executable, args.agent_eval, "--agent", args.agent, "--tasks", args.tasks, "--base-url", args.base_url, "--model", model, "--repeats", str(args.repeats), "--out", str(results_path), "--trajectory-dir", str(out_dir / f"trajectories-{label}"), "--quant", args.quant, "--engine", args.engine, "--engine-version", args.engine_version, "--notes", f"part-27 comparison, {label}", ] if args.workspace: command += ["--workspace", args.workspace] if args.index: command += ["--index", args.index] if args.cost_per_million is not None: command += ["--cost-per-million", str(args.cost_per_million)] subprocess.run(command, check=False) if not results_path.is_file(): sys.exit(f"{results_path} was not written; read the harness output above") return json.loads(results_path.read_text(encoding="utf-8")) def reliability(args: argparse.Namespace, out_dir: Path, model: str, label: str) -> dict[str, Any]: """One model through Part 24's tool-calling reliability test, as JSON.""" command = [ sys.executable, args.reliability, "--base-url", args.base_url, "--model", model, "--repeat", str(args.reliability_repeat), "--temperature", str(args.temperature), "--json", ] if args.prompts: command += ["--prompts", args.prompts] output = run(command, capture=True) path = out_dir / f"reliability-{label}.json" path.write_text(output, encoding="utf-8") try: return json.loads(output) except json.JSONDecodeError: sys.exit(f"{args.reliability} did not print JSON; see {path}") def general_suite(args: argparse.Namespace, out_dir: Path, model: str, label: str) -> dict[str, Any] | None: """Part 10's harness over your own general task set: the regression check.""" if not (args.harness_dir and args.general_tasks): return None script = Path(args.harness_dir) / "run-eval.py" if not script.is_file(): sys.exit(f"{script} does not exist; --harness-dir must hold Part 10's run-eval.py") results_path = out_dir / f"general-{label}.json" run([sys.executable, str(script), "--tasks", args.general_tasks, "--base-url", args.base_url, "--model", model, "--quant", args.quant, "--engine", args.engine, "--engine-version", args.engine_version, "--out", str(results_path)]) if not results_path.is_file(): print(f" {script.name} wrote no results for {label}; skipping the general comparison") return None return json.loads(results_path.read_text(encoding="utf-8")) def by_category(payload: dict[str, Any]) -> dict[str, tuple[int, int]]: """Passes and attempts per category, from a Part 26 results file.""" counts: dict[str, list[int]] = defaultdict(lambda: [0, 0]) for row in payload.get("results", []): category = row.get("category", "uncategorised") counts[category][1] += 1 counts[category][0] += int(bool((row.get("checks") or {}).get("passed"))) return {k: (v[0], v[1]) for k, v in sorted(counts.items())} def general_pass_rates(payload: dict[str, Any] | None) -> dict[str, tuple[int, int]]: """Passes and attempts per category from Part 10's harness, whose rows carry the same `category` and `checks.passed` fields.""" if not payload: return {} rows = payload.get("results", payload if isinstance(payload, list) else []) counts: dict[str, list[int]] = defaultdict(lambda: [0, 0]) for row in rows: if not isinstance(row, dict): continue category = row.get("category", "uncategorised") counts[category][1] += 1 counts[category][0] += int(bool((row.get("checks") or {}).get("passed"))) return {k: (v[0], v[1]) for k, v in sorted(counts.items())} def fraction(pair: tuple[int, int] | None) -> float | None: if not pair or not pair[1]: return None return pair[0] / pair[1] def show(value: float | None) -> str: return "n/a" if value is None else f"{value * 100:5.1f}%" def delta(after: float | None, before: float | None) -> str: if after is None or before is None: return " n/a" difference = (after - before) * 100 return f"{difference:+6.1f}" def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--agent-eval", required=True, help="path to Part 26's agent-eval.py") parser.add_argument("--agent", required=True, help="the agent entry point to measure with") parser.add_argument("--tasks", required=True, help="Part 26's agent-tasks.json, or your own") parser.add_argument("--reliability", required=True, help="path to Part 24's tool-call-reliability.py") parser.add_argument("--prompts", default=None, help="Part 24's tool-prompts.json, if not beside it") parser.add_argument("--base-url", default="http://127.0.0.1:4000/v1") parser.add_argument("--base-model", required=True, help="alias the base model answers under") parser.add_argument("--tuned-model", required=True, help="alias the fine-tune answers under") parser.add_argument("--workspace", default=None) parser.add_argument("--index", default=None) parser.add_argument("--repeats", type=int, default=3, help="attempts per task; 3 or more, because one run of a suite this " "size cannot tell a real difference from a sampling one") parser.add_argument("--reliability-repeat", type=int, default=5) parser.add_argument("--temperature", type=float, default=0.0) parser.add_argument("--quant", default="unknown", help="the quantisation actually loaded; not discoverable over the API") parser.add_argument("--engine", default="unknown") parser.add_argument("--engine-version", default="unknown") parser.add_argument("--cost-per-million", type=float, default=None) parser.add_argument("--harness-dir", default=None, help="directory holding Part 10's run-eval.py") parser.add_argument("--general-tasks", default=None, help="your own general task set") parser.add_argument("--out-dir", default="compare-out") parser.add_argument("--labbook", default=None) parser.add_argument("--notes", default=None) args = parser.parse_args() if args.base_model == args.tuned_model: sys.exit("--base-model and --tuned-model are the same alias; serve them under " "different names or the comparison measures nothing") out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) print(f"\n=== agent suite: base ({args.base_model}) ===") base_agent = agent_suite(args, out_dir, args.base_model, "base") print(f"\n=== agent suite: fine-tune ({args.tuned_model}) ===") tuned_agent = agent_suite(args, out_dir, args.tuned_model, "tuned") print("\n=== tool-call reliability: base ===") base_rel = reliability(args, out_dir, args.base_model, "base") print("\n=== tool-call reliability: fine-tune ===") tuned_rel = reliability(args, out_dir, args.tuned_model, "tuned") base_general = general_suite(args, out_dir, args.base_model, "base") tuned_general = general_suite(args, out_dir, args.tuned_model, "tuned") # ---------------------------------------------------------------- the agent suite base_cats, tuned_cats = by_category(base_agent), by_category(tuned_agent) print("\n\nAgent suite, success rate per category") print(f"{'category':<18}{'base':>10}{'fine-tune':>12}{'change':>10}") for category in sorted(set(base_cats) | set(tuned_cats)): before, after = fraction(base_cats.get(category)), fraction(tuned_cats.get(category)) print(f"{category:<18}{show(before):>10}{show(after):>12}{delta(after, before):>10}") base_all = base_agent["run"]["summary"]["success_rate"] tuned_all = tuned_agent["run"]["summary"]["success_rate"] print(f"{'all tasks':<18}{show(base_all):>10}{show(tuned_all):>12}{delta(tuned_all, base_all):>10}") print("\nCost of an attempt") print(f"{'measure':<18}{'base':>10}{'fine-tune':>12}") for key, label in (("mean_steps", "mean steps"), ("mean_tokens", "mean tokens"), ("mean_seconds", "mean seconds")): before = base_agent["run"]["summary"].get(key) after = tuned_agent["run"]["summary"].get(key) print(f"{label:<18}{before!s:>10}{after!s:>12}") # -------------------------------------------------------------------- reliability print("\nTool-call reliability (Part 24's fixed prompt set)") print(f"{'rate':<20}{'base':>10}{'fine-tune':>12}{'change':>10}") for key, label in (("call_rate", "call rate"), ("parse_rate", "parse rate"), ("right_tool_rate", "right tool"), ("schema_valid_rate", "schema valid"), ("args_correct_rate", "arguments correct"), ("false_call_rate", "false calls")): before = base_rel["rates"].get(key) after = tuned_rel["rates"].get(key) print(f"{label:<20}{show(before):>10}{show(after):>12}{delta(after, before):>10}") print("A rise in the false-call rate is a regression even when everything else improves: " "it means the model learned that calling a tool is always the right move.") # ------------------------------------------------------------------- the general set regressions: list[str] = [] base_gen, tuned_gen = general_pass_rates(base_general), general_pass_rates(tuned_general) if base_gen or tuned_gen: print("\nGeneral task set (Part 10's harness), pass rate per category") print(f"{'category':<18}{'base':>10}{'fine-tune':>12}{'change':>10}") for category in sorted(set(base_gen) | set(tuned_gen)): before, after = fraction(base_gen.get(category)), fraction(tuned_gen.get(category)) print(f"{category:<18}{show(before):>10}{show(after):>12}{delta(after, before):>10}") if before is not None and after is not None and after < before: regressions.append(f"{category}: {show(before)} to {show(after)}") else: print("\nGeneral task set: not run. Pass --harness-dir and --general-tasks. Without " "it this comparison can only tell you what got better.") for key in ("false_call_rate",): before, after = base_rel["rates"].get(key), tuned_rel["rates"].get(key) if before is not None and after is not None and after > before: regressions.append(f"tool-call {key}: {show(before)} to {show(after)}") if regressions: print("\nRegressions worth writing down:") for line in regressions: print(f" {line}") summary = { "agent_success_base": base_all, "agent_success_tuned": tuned_all, "agent_success_change": None if None in (base_all, tuned_all) else round(tuned_all - base_all, 4), "agent_by_category_base": {k: list(v) for k, v in base_cats.items()}, "agent_by_category_tuned": {k: list(v) for k, v in tuned_cats.items()}, "reliability_base": base_rel["rates"], "reliability_tuned": tuned_rel["rates"], "general_base": {k: list(v) for k, v in base_gen.items()}, "general_tuned": {k: list(v) for k, v in tuned_gen.items()}, "regressions": regressions, } (out_dir / "comparison.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") print(f"\nwritten to {out_dir / 'comparison.json'}") if args.labbook: record = agentlog.record( labbook=args.labbook, lab="part-27/compare-agent-models", model={"base": args.base_model, "tuned": args.tuned_model}, dataset={"path": args.tasks, "sha256": agentlog.file_sha256(args.tasks), "general_tasks": args.general_tasks}, data_lineage=agentlog.lineage(evaluation_suite=args.tasks, general_tasks=args.general_tasks), hyperparameters={"repeats": args.repeats, "reliability_repeat": args.reliability_repeat, "temperature": args.temperature, "base_url": args.base_url, "quant": args.quant, "engine": args.engine, "engine_version": args.engine_version, "agent_entry_point": args.agent}, seed=None, losses=None, scores=summary, config_path=__file__, notes=args.notes, ) print(f"recorded run {record['run_id']} in {args.labbook}") if __name__ == "__main__": main()