dongarandClaude Opus 5.5
50b46821d7
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m11s
Choosing a local executor model, and tuning its knobs, needs measured success rates across repeated runs, comparable with every model measured before. - Four canned milestones under nisse_dev/bench/tasks: write a function, fix a bug a failing test exposes, thread an option through several files, and find one failure in output far past max_output_chars. Each is an approved plan over a seed project, judged by a hidden check the model never sees; a reference solution proves each is solvable and that its check tells the difference - Runs use the agent's own client wiring, now shared as make_llm_clients, with delegation and ask_expert off: an escalation is a failure of the model under test - Each run is stored as one JSON line in bench/results/, committed, with the model's backing name and declared facts, a note on how it was served, the task's content hash, the agent's commit and tuning, the outcome, turns, misbehaviour counts, context trims, tokens, time and LiteLLM's cost; a fallback answer can never count as a success - The report rebuilds the comparison from every stored run, with Wilson 95% intervals, marking runs against older task versions - scripts/benchmark.py runs and reports; scripts/evaluate_model.sh takes a new model through the context probe, the capability matrix and the benchmark in one command - Clients keep LiteLLM's per-call cost, and BudgetedClient counts the requests it had to trim - The tuning guide gains "Benchmarking models" Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
"""Benchmark tasks: a seed project, an approved plan, and a hidden check.
|
|
|
|
Each task is a directory under ``tasks/``:
|
|
|
|
- ``task.yaml``: ``title``, ``plan`` (the approved milestone plan, written the
|
|
way the planner is asked to write them), and ``check`` (a shell command run
|
|
in the workspace afterwards).
|
|
- ``seed/``: the project the executor starts from.
|
|
- ``hidden/``: files the check needs, copied in only after the run, into
|
|
``.bench_check/``. The executor never sees them.
|
|
- ``solution/``: a reference solution, overlaid on the seed by the tests to
|
|
prove each task can be solved and that its check tells the difference.
|
|
"""
|
|
|
|
import hashlib
|
|
import shutil
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
TASKS_DIR = Path(__file__).parent / "tasks"
|
|
CHECK_DIR = ".bench_check"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Task:
|
|
name: str
|
|
title: str
|
|
plan: str
|
|
check: str
|
|
path: Path
|
|
|
|
@property
|
|
def content_hash(self) -> str:
|
|
"""What the task is, so a changed task is never compared with runs of
|
|
its earlier version. Covers everything the model or the check sees;
|
|
not the reference solution."""
|
|
digest = hashlib.sha256()
|
|
for part in ("task.yaml", "seed", "hidden"):
|
|
root = self.path / part
|
|
files = [root] if root.is_file() else sorted(root.rglob("*"))
|
|
for file in files:
|
|
if file.is_file() and "__pycache__" not in file.parts:
|
|
digest.update(str(file.relative_to(self.path)).encode())
|
|
digest.update(file.read_bytes())
|
|
return digest.hexdigest()[:16]
|
|
|
|
def seed_into(self, workspace: Path) -> None:
|
|
shutil.copytree(self.path / "seed", workspace, dirs_exist_ok=True)
|
|
|
|
def apply_solution(self, workspace: Path) -> None:
|
|
shutil.copytree(self.path / "solution", workspace, dirs_exist_ok=True)
|
|
|
|
def run_check(self, workspace: Path, timeout_s: int = 300) -> tuple[bool, str]:
|
|
"""Copy the hidden files in and run the check. Returns whether it
|
|
passed, and its output for the record."""
|
|
target = workspace / CHECK_DIR
|
|
if target.exists():
|
|
shutil.rmtree(target) # never trust a directory the model made
|
|
shutil.copytree(self.path / "hidden", target)
|
|
try:
|
|
result = subprocess.run(
|
|
self.check,
|
|
shell=True,
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout_s,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return False, f"check timed out after {timeout_s}s"
|
|
output = (result.stdout + result.stderr)[-2000:]
|
|
return result.returncode == 0, output
|
|
|
|
|
|
def load_task(path: Path) -> Task:
|
|
spec = yaml.safe_load((path / "task.yaml").read_text())
|
|
return Task(
|
|
name=path.name,
|
|
title=str(spec["title"]),
|
|
plan=str(spec["plan"]).strip(),
|
|
check=str(spec["check"]),
|
|
path=path,
|
|
)
|
|
|
|
|
|
def load_tasks(names: list[str] | None = None, root: Path = TASKS_DIR) -> list[Task]:
|
|
available = {
|
|
p.name: p for p in sorted(root.iterdir()) if (p / "task.yaml").exists()
|
|
}
|
|
if names:
|
|
unknown = [n for n in names if n not in available]
|
|
if unknown:
|
|
raise ValueError(
|
|
f"unknown task(s): {', '.join(unknown)}; "
|
|
f"available: {', '.join(available)}"
|
|
)
|
|
return [load_task(available[n]) for n in names]
|
|
return [load_task(p) for p in available.values()]
|