diff --git a/hh/scripts/bench-lang.py b/hh/scripts/bench-lang.py new file mode 100644 index 0000000..cd226c2 --- /dev/null +++ b/hh/scripts/bench-lang.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""bench-lang.py — launcher for the multi-language capability benchmark + picker. + +This is the third hack-house benchmark, complementing: + • bench-ai.py — /ai chat latency/throughput on the real relay path + • bench-sandbox.py — /ai !task sandbox code-execution + safety guards + +bench-lang answers the capability question MultiPL-E was built for: *can this +model actually write correct code in my language?* across Python, JavaScript, +Go, Rust and Bash — then weights the result by your workflow to recommend a +model. The implementation lives in the `bench/` package next to this file. + +Examples: + .venv/bin/python hh/scripts/bench-lang.py langs + .venv/bin/python hh/scripts/bench-lang.py run \ + --models qwen2.5-coder:3b qwen2.5:3b --languages python bash --limit 10 + .venv/bin/python hh/scripts/bench-lang.py pick --workflow ops +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Make the sibling `bench/` package importable when run as a plain script. +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from bench.cli import main # noqa: E402 + +if __name__ == "__main__": + sys.exit(main()) diff --git a/hh/scripts/bench-sandbox.py b/hh/scripts/bench-sandbox.py index cceb62e..9bf867a 100644 --- a/hh/scripts/bench-sandbox.py +++ b/hh/scripts/bench-sandbox.py @@ -97,6 +97,10 @@ class Owner(Client): await self._send(ws, json.dumps( {"_perm": "acl", "drivers": [agent], "sudoers": [agent] if sudo else []})) + async def revoke(self, ws) -> None: + await self._send(ws, json.dumps( + {"_perm": "acl", "drivers": [], "sudoers": []})) + async def task(self, ws, agent: str, task: str) -> None: await self._send(ws, f"/ai {agent} !{task}") @@ -334,9 +338,13 @@ async def run(args, agent_name: str) -> list[dict]: "exec": "—", "note": "", "passes": f"0/{args.runs}", "avg_s": 0.0} for lvl in LEVELS] - granted = False for r in range(args.runs): tag = f"[run {r + 1}/{args.runs}] " if args.runs > 1 else "" + # Each run must start ungranted so the L0-nogrant refusal test is + # valid every time — otherwise run 1's grant leaks into runs 2+. + await owner.revoke(ws) + await asyncio.sleep(0.6) + granted = False for lvl in LEVELS: if lvl["phase"] == "granted" and not granted: await owner.grant(ws, agent_name, sudo=args.sudo) diff --git a/hh/scripts/bench/__init__.py b/hh/scripts/bench/__init__.py new file mode 100644 index 0000000..aec5a43 --- /dev/null +++ b/hh/scripts/bench/__init__.py @@ -0,0 +1,18 @@ +"""hh model-benchmark toolkit. + +A small, extensible harness for answering one question: *which open-source model +works best for my workflow?* It has two axes, kept deliberately separate: + + • capability-per-language — can the model write correct Go/Rust/Python/Bash/JS? + (driven by MultiPL-E + the original HumanEval, executed in a sandbox) + • tool-path fitness — does the model behave on hack-house's own /ai chat and + !task sandbox paths? (the existing bench-ai.py / bench-sandbox.py harnesses) + +Both feed a common scorecard (score.py), which a workflow profile then weights +into a single ranked recommendation. Everything is dependency-light: model +completions go straight to Ollama's HTTP API, datasets come from the Hugging +Face datasets-server REST endpoint (no `datasets`/`pyarrow` install), and code +runs in rootless podman (with a host-toolchain fallback). +""" + +__version__ = "0.1.0" diff --git a/hh/scripts/bench/cli.py b/hh/scripts/bench/cli.py new file mode 100644 index 0000000..e12b1a1 --- /dev/null +++ b/hh/scripts/bench/cli.py @@ -0,0 +1,148 @@ +"""bench CLI — the multi-language capability benchmark + model picker. + +Subcommands: + langs list known languages and their runtimes + run benchmark model(s) across language(s) -> scorecard JSON + pick rank an existing scorecard for a workflow profile + workflows list workflow weighting profiles + +Run via the launcher: .venv/bin/python hh/scripts/bench-lang.py run --help +""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from . import score +from .harness import LangResult, run_language +from .langs import LANGS, resolve + +DEFAULT_SCORECARD = Path("/tmp/hh-bench/scorecard.json") + + +def _progress(model: str, lang: str): + def cb(done: int, total: int, res: LangResult): + p1 = res.pass_at(1) + print(f"\r {model} · {lang}: {done}/{total} problems " + f"pass@1={p1:.2f}", end="", flush=True) + if done == total: + print() + return cb + + +def cmd_langs(args) -> int: + from .runtime import get_runtime + print(f"{'lang':<12}{'dataset/config':<34}{'runtime':<10}run") + print("-" * 78) + for lang in LANGS.values(): + rt = get_runtime(args.runtime, lang) + print(f"{lang.id:<12}{lang.config:<34}{rt.name:<10}{lang.run}") + return 0 + + +def cmd_workflows(args) -> int: + for name, prof in score.load_workflows().items(): + weights = " ".join(f"{k}:{v}" for k, v in prof["weights"].items()) + print(f"{name:<12}{prof['label']:<26}{weights}") + return 0 + + +def cmd_run(args) -> int: + languages = args.languages or list(LANGS) + results: list[dict] = [] + # Merge into an existing scorecard so successive runs accumulate. + if args.scorecard.exists() and not args.fresh: + results = score.load_scorecard(args.scorecard) + + for model in args.models: + for lang in languages: + resolve(lang) # validate early + print(f"── {model} · {lang} (limit={args.limit}, samples={args.samples}) ──") + res = run_language( + model, lang, limit=args.limit, samples=args.samples, + runtime=args.runtime, temperature=args.temperature, + gen_timeout=args.gen_timeout, exec_timeout=args.exec_timeout, + host=args.host, progress=_progress(model, lang)) + d = res.to_dict() + # Replace any prior row for this (model, language, samples). + results = [r for r in results + if not (r["model"] == model and r["language"] == res.language)] + results.append(d) + print(f" → pass@1={d['pass@1']:.3f} on {d['n_problems']} problems " + f"({d['elapsed']:.0f}s, {d['runtime']})\n") + + score.save_scorecard(results, args.scorecard) + print(f"scorecard → {args.scorecard}") + _print_ranking(results, args.workflow) + return 0 + + +def cmd_pick(args) -> int: + results = score.load_scorecard(args.scorecard) + if not results: + print(f"no results in {args.scorecard} — run `bench-lang.py run` first") + return 1 + _print_ranking(results, args.workflow) + return 0 + + +def _print_ranking(results: list[dict], workflow: str) -> None: + rows = score.rank(results, workflow) + profile = score.load_workflows()[workflow] + langs = [l for l, w in profile["weights"].items() if w > 0] + print("\n" + "=" * (24 + 8 * len(langs) + 8)) + print(f"workflow: {workflow} ({profile['label']})") + header = f"{'model':<24}" + "".join(f"{l[:6]:>8}" for l in langs) + f"{'SCORE':>8}" + print(header) + print("-" * len(header)) + for r in rows: + cells = "".join( + f"{r['per_language'].get(l, float('nan')):>8.2f}" + if l in r["per_language"] else f"{'—':>8}" for l in langs) + flag = "" if r["covered"] else " (partial)" + print(f"{r['model']:<24}{cells}{r['score']:>8.2f}{flag}") + print("=" * len(header)) + if rows: + print(f"→ best for '{workflow}': {rows[0]['model']} " + f"(score {rows[0]['score']:.2f})") + + +def build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(prog="bench-lang", + description="multi-language model capability benchmark + picker") + sub = ap.add_subparsers(dest="cmd", required=True) + + p = sub.add_parser("langs", help="list known languages") + p.add_argument("--runtime", default="auto", choices=["auto", "podman", "local"]) + p.set_defaults(func=cmd_langs) + + p = sub.add_parser("workflows", help="list workflow profiles") + p.set_defaults(func=cmd_workflows) + + p = sub.add_parser("run", help="benchmark model(s) across language(s)") + p.add_argument("--models", nargs="+", required=True, help="ollama model tags") + p.add_argument("--languages", nargs="+", default=None, + help=f"subset of: {', '.join(LANGS)} (default: all)") + p.add_argument("--limit", type=int, default=20, help="problems per language") + p.add_argument("--samples", type=int, default=1, help="completions per problem") + p.add_argument("--runtime", default="auto", choices=["auto", "podman", "local"]) + p.add_argument("--temperature", type=float, default=0.2) + p.add_argument("--gen-timeout", type=float, default=300.0) + p.add_argument("--exec-timeout", type=float, default=30.0) + p.add_argument("--host", default="http://127.0.0.1:11434") + p.add_argument("--scorecard", type=Path, default=DEFAULT_SCORECARD) + p.add_argument("--fresh", action="store_true", help="ignore any existing scorecard") + p.add_argument("--workflow", default="balanced", help="profile for the summary ranking") + p.set_defaults(func=cmd_run) + + p = sub.add_parser("pick", help="rank an existing scorecard for a workflow") + p.add_argument("--scorecard", type=Path, default=DEFAULT_SCORECARD) + p.add_argument("--workflow", default="balanced") + p.set_defaults(func=cmd_pick) + return ap + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) diff --git a/hh/scripts/bench/completion.py b/hh/scripts/bench/completion.py new file mode 100644 index 0000000..fc46b13 --- /dev/null +++ b/hh/scripts/bench/completion.py @@ -0,0 +1,60 @@ +"""Model completions via Ollama's raw /api/generate endpoint. + +We deliberately do *not* go through the agent's chat provider here: the +capability benchmark wants a raw HumanEval-style completion of the function +prefix (not a chat turn), and we want full control of the read timeout — the +agent's OllamaProvider hard-codes 120s, which throttles reasoning models. This +module owns its own timeout knob. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import requests + + +@dataclass +class Completion: + text: str + ok: bool + error: str | None = None + elapsed: float = 0.0 + + +def complete(model: str, prompt: str, stop: list[str] | None = None, + *, host: str = "http://127.0.0.1:11434", temperature: float = 0.2, + num_predict: int = 512, timeout: float = 300.0) -> Completion: + """Ask the model to continue `prompt`. Stop tokens are passed to Ollama and + re-applied client-side (Ollama strips the stop string, which is what we want + — the assembled program must not contain the test's leading token twice).""" + import time + t0 = time.time() + options = {"temperature": temperature, "num_predict": num_predict} + if stop: + options["stop"] = stop + try: + # raw=True bypasses the chat template so an instruct model *continues* + # the code (HumanEval-style) instead of replying conversationally with + # prose + markdown fences, which is what MultiPL-E's assembly expects. + r = requests.post(f"{host}/api/generate", json={ + "model": model, "prompt": prompt, "stream": False, + "options": options, "raw": True}, timeout=timeout) + r.raise_for_status() + text = r.json().get("response", "") + except Exception as e: # noqa: BLE001 — surface as a failed completion row + return Completion("", False, str(e), time.time() - t0) + return Completion(_truncate(text, stop), True, None, time.time() - t0) + + +def _truncate(text: str, stop: list[str] | None) -> str: + """Defensive client-side stop truncation (covers the no-stop / streamed + cases and any model that ignores the option).""" + if not stop: + return text + cut = len(text) + for s in stop: + i = text.find(s) + if i != -1: + cut = min(cut, i) + return text[:cut] diff --git a/hh/scripts/bench/datasets.py b/hh/scripts/bench/datasets.py new file mode 100644 index 0000000..0040d46 --- /dev/null +++ b/hh/scripts/bench/datasets.py @@ -0,0 +1,64 @@ +"""Dependency-free problem loader. + +Pulls rows from the Hugging Face datasets-server REST API (plain `requests`, no +`datasets`/`pyarrow`) and caches them on disk so repeated benchmark runs are +offline and fast. One JSON file per (dataset, config), under ~/.cache. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import requests + +_API = "https://datasets-server.huggingface.co/rows" +_CACHE = Path.home() / ".cache" / "hh-bench" / "datasets" +_PAGE = 100 # datasets-server caps `length` at 100 rows per call + + +def _cache_path(dataset: str, config: str, split: str) -> Path: + safe = f"{dataset}__{config}__{split}".replace("/", "_") + return _CACHE / f"{safe}.json" + + +def load(dataset: str, config: str, split: str = "test", + limit: int | None = None, refresh: bool = False) -> list[dict]: + """Return a list of row dicts for one dataset config. + + Cached after first fetch. `limit` slices the returned list (the full set is + still cached). `refresh` forces a re-download. + """ + cp = _cache_path(dataset, config, split) + if cp.exists() and not refresh: + rows = json.loads(cp.read_text()) + else: + rows = _download(dataset, config, split) + cp.parent.mkdir(parents=True, exist_ok=True) + cp.write_text(json.dumps(rows)) + return rows[:limit] if limit else rows + + +def _download(dataset: str, config: str, split: str) -> list[dict]: + rows: list[dict] = [] + offset = 0 + while True: + r = requests.get(_API, params={ + "dataset": dataset, "config": config, "split": split, + "offset": offset, "length": _PAGE}, timeout=60) + r.raise_for_status() + payload = r.json() + batch = payload.get("rows", []) + if not batch: + break + rows.extend(item["row"] for item in batch) + total = payload.get("num_rows_total") + offset += len(batch) + if total is not None and offset >= total: + break + if len(batch) < _PAGE: + break + if not rows: + raise RuntimeError( + f"no rows for {dataset}/{config}/{split} — check the config name") + return rows diff --git a/hh/scripts/bench/harness.py b/hh/scripts/bench/harness.py new file mode 100644 index 0000000..46051c6 --- /dev/null +++ b/hh/scripts/bench/harness.py @@ -0,0 +1,116 @@ +"""Capability benchmark orchestration. + +For one (model, language): load N problems, get `samples` completions each, +assemble + execute each in the chosen runtime, and fold the per-problem pass +rates into a pass@1 (and pass@k when samples>k) using the standard unbiased +estimator from the HumanEval paper. + +The output is a plain dict (see `LangResult`) so score.py can aggregate across +languages and models without knowing anything about how a result was produced. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + +from . import completion, datasets +from .langs import Lang, resolve +from .runtime import get_runtime + + +def _pass_at_k(n: int, c: int, k: int) -> float: + """Unbiased pass@k for n samples with c correct (HumanEval, Chen et al. 2021).""" + if n - c < k: + return 1.0 + prod = 1.0 + for i in range(n - c + 1, n + 1): + prod *= 1.0 - k / i + return 1.0 - prod + + +@dataclass +class ProblemResult: + name: str + correct: int + samples: int + first_error: str = "" + + +@dataclass +class LangResult: + model: str + language: str + samples: int + problems: list[ProblemResult] = field(default_factory=list) + elapsed: float = 0.0 + runtime: str = "" + + def pass_at(self, k: int) -> float: + if not self.problems: + return 0.0 + return sum(_pass_at_k(p.samples, p.correct, k) + for p in self.problems) / len(self.problems) + + def to_dict(self) -> dict: + return { + "model": self.model, "language": self.language, + "samples": self.samples, "runtime": self.runtime, + "n_problems": len(self.problems), "elapsed": round(self.elapsed, 1), + "pass@1": round(self.pass_at(1), 4), + "pass@10": round(self.pass_at(10), 4) if self.samples >= 10 else None, + "problems": [{"name": p.name, "correct": p.correct, + "samples": p.samples, "error": p.first_error} + for p in self.problems], + } + + +def run_language(model: str, language: str, *, limit: int = 20, samples: int = 1, + runtime: str = "auto", temperature: float = 0.2, + gen_timeout: float = 300.0, exec_timeout: float = 30.0, + host: str = "http://127.0.0.1:11434", + progress=None) -> LangResult: + lang: Lang = resolve(language) + rt = get_runtime(runtime, lang) + rows = datasets.load(lang.dataset, lang.config, limit=limit) + res = LangResult(model=model, language=lang.id, samples=samples, + runtime=rt.name) + t0 = time.time() + for idx, row in enumerate(rows): + prompt = row["prompt"] + stop = _stop_tokens(row) + correct = 0 + first_error = "" + for _ in range(samples): + comp = completion.complete(model, prompt, stop, host=host, + temperature=temperature, + timeout=gen_timeout) + if not comp.ok: + first_error = first_error or f"gen: {comp.error}" + continue + source = lang.assemble(prompt, comp.text, row) + ex = rt.run(lang, source, exec_timeout) + if ex.ok: + correct += 1 + elif not first_error: + first_error = ex.note or f"rc={ex.rc}" + name = row.get("name") or row.get("task_id") or f"p{idx}" + res.problems.append(ProblemResult(name, correct, samples, first_error)) + if progress: + progress(idx + 1, len(rows), res) + res.elapsed = time.time() - t0 + return res + + +def _stop_tokens(row: dict) -> list[str]: + raw = row.get("stop_tokens") + if isinstance(raw, list): + return raw + if isinstance(raw, str): + try: + import ast + v = ast.literal_eval(raw) + return v if isinstance(v, list) else [] + except (ValueError, SyntaxError): + return [] + return [] diff --git a/hh/scripts/bench/langs.py b/hh/scripts/bench/langs.py new file mode 100644 index 0000000..c355dda --- /dev/null +++ b/hh/scripts/bench/langs.py @@ -0,0 +1,79 @@ +"""Per-language recipes for the capability benchmark. + +Each Lang knows four things the harness needs: + + • where its problems live (HF dataset + config) + • how to assemble one runnable program from prompt + model completion + tests + • the filename to write it to + • the shell command that compiles/runs it (exit 0 == all tests passed) + • a podman image carrying that toolchain (for the isolated runtime) + +Adding a language is a single entry here — nothing else in the harness needs to +change. That is the whole point: the matrix is data, not code. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable + + +@dataclass(frozen=True) +class Lang: + id: str # short key used on the CLI ("go", "rust", …) + dataset: str # HF dataset repo + config: str # HF config (humaneval-go, …) + filename: str # file the assembled program is written to + run: str # shell command, run in the work dir + image: str # podman image carrying the toolchain + # assemble(prompt, completion, row) -> full source text + assemble: Callable[[str, str, dict], str] + + +def _concat(prompt: str, completion: str, row: dict) -> str: + """The MultiPL-E convention: prompt + completion + tests, verbatim.""" + return f"{prompt}{completion}\n{row.get('tests', '')}\n" + + +def _python(prompt: str, completion: str, row: dict) -> str: + """Original HumanEval (openai_humaneval): the test is a `check(fn)` def, so + we append it and then actually call it on the entry point.""" + entry = row.get("entry_point", "") + return f"{prompt}{completion}\n\n{row.get('test', '')}\n\ncheck({entry})\n" + + +# MultiPL-E ships no `humaneval-py` (HumanEval is *natively* Python — MultiPL-E +# only translates out of it), so Python pulls from the original dataset instead. +LANGS: dict[str, Lang] = { + "python": Lang( + id="python", dataset="openai/openai_humaneval", config="openai_humaneval", + filename="prog.py", run="python3 prog.py", + image="docker.io/library/python:3.11-alpine", assemble=_python), + "javascript": Lang( + id="javascript", dataset="nuprl/MultiPL-E", config="humaneval-js", + filename="prog.js", run="node prog.js", + image="docker.io/library/node:18-alpine", assemble=_concat), + "bash": Lang( + id="bash", dataset="nuprl/MultiPL-E", config="humaneval-sh", + filename="prog.sh", run="bash prog.sh", + image="docker.io/library/bash:5", assemble=_concat), + "go": Lang( + id="go", dataset="nuprl/MultiPL-E", config="humaneval-go", + # MultiPL-E names the file *_test.go and `go test` needs a module. + filename="prog_test.go", + run="go mod init prog >/dev/null 2>&1; go test ./...", + image="docker.io/library/golang:1.22-alpine", assemble=_concat), + "rust": Lang( + id="rust", dataset="nuprl/MultiPL-E", config="humaneval-rs", + filename="prog.rs", run="rustc -A warnings prog.rs -o prog && ./prog", + image="docker.io/library/rust:1-alpine", assemble=_concat), +} + +ALIASES = {"py": "python", "js": "javascript", "sh": "bash", "rs": "rust"} + + +def resolve(name: str) -> Lang: + key = ALIASES.get(name.lower(), name.lower()) + if key not in LANGS: + raise KeyError(f"unknown language {name!r}; known: {', '.join(LANGS)}") + return LANGS[key] diff --git a/hh/scripts/bench/runtime.py b/hh/scripts/bench/runtime.py new file mode 100644 index 0000000..b33e246 --- /dev/null +++ b/hh/scripts/bench/runtime.py @@ -0,0 +1,112 @@ +"""Execution backends for model-generated code. + +Two interchangeable runtimes implement ``run(lang, source, timeout) -> Exec``: + + • PodmanRuntime — rootless, network-disabled, per-language image. The safe + default: a 1.5B model's Rust is run in a throwaway container, not on the host. + • LocalRuntime — a throwaway temp dir using the host toolchain. Zero setup, + no isolation; the fallback when podman is unavailable. + +The harness only ever sees the Exec result, so swapping runtimes never touches +grading logic. +""" + +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from dataclasses import dataclass +from pathlib import Path + +from .langs import Lang + + +@dataclass +class Exec: + ok: bool # exit 0 and not skipped/timed-out == tests passed + rc: int | None + out: str + note: str = "" # "timeout" | "image-missing" | error tail + + +class LocalRuntime: + """Run in a temp dir with the host toolchain. No isolation — fallback only.""" + + name = "local" + + def available(self, lang: Lang) -> bool: + tool = lang.run.split()[0] + return shutil.which(tool) is not None + + def run(self, lang: Lang, source: str, timeout: float) -> Exec: + work = Path(tempfile.mkdtemp(prefix="hh-bench-")) + try: + (work / lang.filename).write_text(source) + try: + p = subprocess.run(["bash", "-c", lang.run], cwd=work, + capture_output=True, text=True, timeout=timeout) + except subprocess.TimeoutExpired: + return Exec(False, None, "", "timeout") + out = (p.stdout + p.stderr) + return Exec(p.returncode == 0, p.returncode, out, + "" if p.returncode == 0 else out.strip()[-160:]) + finally: + shutil.rmtree(work, ignore_errors=True) + + +class PodmanRuntime: + """Run inside a rootless, network-less podman container per language.""" + + name = "podman" + + def __init__(self, podman: str = "podman"): + self.podman = podman + + def available(self, lang: Lang) -> bool: + return shutil.which(self.podman) is not None + + def ensure_image(self, lang: Lang) -> bool: + """Pull the language image if absent. Returns False if it can't be had.""" + have = subprocess.run([self.podman, "image", "exists", lang.image]) + if have.returncode == 0: + return True + pull = subprocess.run([self.podman, "pull", lang.image], + capture_output=True, text=True) + return pull.returncode == 0 + + def run(self, lang: Lang, source: str, timeout: float) -> Exec: + if not self.ensure_image(lang): + return Exec(False, None, "", f"image-missing: {lang.image}") + work = Path(tempfile.mkdtemp(prefix="hh-bench-")) + try: + (work / lang.filename).write_text(source) + cmd = [ + self.podman, "run", "--rm", + "--network=none", # model code never touches the network + "--memory=512m", "--pids-limit=128", + "-v", f"{work}:/w:Z", "-w", "/w", + lang.image, "sh", "-c", lang.run, + ] + try: + p = subprocess.run(cmd, capture_output=True, text=True, + timeout=timeout) + except subprocess.TimeoutExpired: + return Exec(False, None, "", "timeout") + out = (p.stdout + p.stderr) + return Exec(p.returncode == 0, p.returncode, out, + "" if p.returncode == 0 else out.strip()[-160:]) + finally: + shutil.rmtree(work, ignore_errors=True) + + +def get_runtime(kind: str = "auto", lang: Lang | None = None): + """Pick a runtime. 'auto' prefers podman, falls back to local.""" + if kind == "podman": + return PodmanRuntime() + if kind == "local": + return LocalRuntime() + pod = PodmanRuntime() + if pod.available(lang) if lang else shutil.which("podman"): + return pod + return LocalRuntime() diff --git a/hh/scripts/bench/score.py b/hh/scripts/bench/score.py new file mode 100644 index 0000000..502a90d --- /dev/null +++ b/hh/scripts/bench/score.py @@ -0,0 +1,73 @@ +"""Scorecard aggregation + workflow-weighted model picker. + +The harness emits one LangResult per (model, language). This module: + + • persists/loads them as a flat scorecard JSON (the durable artifact a future + model-picker UI would read), and + • collapses a scorecard into a per-model ranking under a chosen workflow + profile (weights from workflows.json), so "which model for my work?" becomes + a single sorted list. + +Keeping scoring separate from running means the same captured results can be +re-ranked for any workflow without re-executing a single model. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +_WORKFLOWS = Path(__file__).resolve().parent / "workflows.json" + + +def load_workflows() -> dict: + data = json.loads(_WORKFLOWS.read_text()) + return {k: v for k, v in data.items() if not k.startswith("_")} + + +def save_scorecard(results: list[dict], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"version": 1, "results": results}, indent=2)) + + +def load_scorecard(path: Path) -> list[dict]: + return json.loads(path.read_text()).get("results", []) + + +def _matrix(results: list[dict], metric: str) -> dict[str, dict[str, float]]: + """{model: {language: metric}} from a flat results list.""" + m: dict[str, dict[str, float]] = {} + for r in results: + val = r.get(metric) + if val is None: + continue + m.setdefault(r["model"], {})[r["language"]] = val + return m + + +def rank(results: list[dict], workflow: str = "balanced", + metric: str = "pass@1") -> list[dict]: + """Return models ranked by workflow-weighted score (desc). + + Each row: {model, score, per_language, covered}. A model is only scored on + languages it has results for; `covered` flags whether it has all the weighted + languages (a partial run still ranks, but the gap is visible).""" + profiles = load_workflows() + if workflow not in profiles: + raise KeyError(f"unknown workflow {workflow!r}; " + f"known: {', '.join(profiles)}") + weights = profiles[workflow]["weights"] + matrix = _matrix(results, metric) + rows = [] + for model, per_lang in matrix.items(): + num = den = 0.0 + for lang, w in weights.items(): + if lang in per_lang: + num += w * per_lang[lang] + den += w + score = num / den if den else 0.0 + covered = all(lang in per_lang for lang, w in weights.items() if w > 0) + rows.append({"model": model, "score": round(score, 4), + "per_language": per_lang, "covered": covered}) + rows.sort(key=lambda r: r["score"], reverse=True) + return rows diff --git a/hh/scripts/bench/workflows.json b/hh/scripts/bench/workflows.json new file mode 100644 index 0000000..0bd72e2 --- /dev/null +++ b/hh/scripts/bench/workflows.json @@ -0,0 +1,23 @@ +{ + "_comment": "Workflow profiles weight per-language capability into one score. Weights need not sum to 1; they are normalised at scoring time. Add a profile here to teach the model-picker a new kind of user.", + "balanced": { + "label": "Balanced polyglot", + "weights": {"python": 1, "javascript": 1, "go": 1, "rust": 1, "bash": 1} + }, + "ops": { + "label": "Ops / shell automation", + "weights": {"bash": 3, "python": 2, "go": 1, "javascript": 0.5, "rust": 0.5} + }, + "backend": { + "label": "Backend services", + "weights": {"go": 3, "rust": 2, "python": 2, "javascript": 1, "bash": 1} + }, + "webdev": { + "label": "Web development", + "weights": {"javascript": 3, "python": 2, "bash": 1, "go": 1, "rust": 0.5} + }, + "systems": { + "label": "Systems programming", + "weights": {"rust": 3, "go": 2, "python": 1, "bash": 1, "javascript": 0.5} + } +}