Karti Tripathi
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
"""The sample schema and its validation. Standard library only, on purpose.
|
|
|
|
This used to live in `kbench/tasks/signal.py`, which imports `inspect_ai` at module scope —
|
|
so checking whether a JSONL record was well-formed required the whole eval framework to be
|
|
installed. That is backwards: the schema is ours and the runner is swappable, so the thing
|
|
that defines what a sample *is* must not depend on the thing that happens to execute it.
|
|
|
|
Practically it means authoring tools, CI checks and editors can validate data without
|
|
resolving a heavy dependency tree, and that swapping the execution backend later touches
|
|
`tasks/`, not this file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
|
|
|
|
VALID_SPLITS = {"train", "dev", "test"}
|
|
VALID_SCORERS = {"exact", "includes", "regex", "rubric"}
|
|
|
|
|
|
def family_path(family: str, tier: str = "private") -> Path:
|
|
return DATA_DIR / tier / f"{family}.jsonl"
|
|
|
|
|
|
def validate_record(rec: dict[str, Any], where: str) -> None:
|
|
"""Fail loudly on a malformed sample.
|
|
|
|
A silently-skipped sample shrinks the eval set without changing the score's appearance,
|
|
which is the worst possible failure mode for a benchmark.
|
|
"""
|
|
for field in ("id", "input", "split", "scorer"):
|
|
if not rec.get(field):
|
|
raise ValueError(f"{where}: missing required field {field!r}")
|
|
if rec["split"] not in VALID_SPLITS:
|
|
raise ValueError(f"{where}: split must be one of {sorted(VALID_SPLITS)}")
|
|
if rec["scorer"] not in VALID_SCORERS:
|
|
raise ValueError(f"{where}: scorer must be one of {sorted(VALID_SCORERS)}")
|
|
if rec["scorer"] != "rubric" and not rec.get("target"):
|
|
raise ValueError(f"{where}: scorer {rec['scorer']!r} requires a target")
|
|
if rec["scorer"] == "rubric" and not (rec.get("rubric") or rec.get("target")):
|
|
raise ValueError(f"{where}: rubric scorer requires a rubric or a target")
|
|
|
|
|
|
def read_records(path: Path) -> list[tuple[int, dict[str, Any]]]:
|
|
"""Parse and validate a family file. Returns (line number, record) pairs.
|
|
|
|
Line numbers are carried so an error names the line a human has to open.
|
|
"""
|
|
import json
|
|
|
|
out: list[tuple[int, dict[str, Any]]] = []
|
|
for lineno, line in enumerate(path.read_text().splitlines(), start=1):
|
|
line = line.strip()
|
|
if not line or line.startswith("//"):
|
|
continue
|
|
rec = json.loads(line)
|
|
validate_record(rec, f"{path.name}:{lineno}")
|
|
out.append((lineno, rec))
|
|
return out
|