Karti Tripathi
359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""Run orchestration: target + tasks + perf sweep -> one committed score card."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import uuid
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from . import tasks as task_catalog
|
|
from .perf import run_sweep
|
|
from .registry import Registry, Target
|
|
from .results import (
|
|
PerfResult,
|
|
QualityResult,
|
|
Run,
|
|
SampleOutcome,
|
|
compute_verdict,
|
|
runner_info,
|
|
truncate,
|
|
utc_now,
|
|
)
|
|
|
|
LOG_DIR = Path(__file__).resolve().parent.parent / "logs"
|
|
|
|
# Inspect's canonical scalar score values.
|
|
_CORRECTNESS = {"C": 1.0, "I": 0.0, "P": 0.5, "N": 0.0}
|
|
|
|
|
|
def _normalize_sample_score(value: Any, primary_key: str | None) -> tuple[float, bool, dict]:
|
|
"""Reduce a score of any shape to (score, passed, extra_metadata).
|
|
|
|
Inspect scorers return scalars, correctness letters, or dicts depending on
|
|
the eval. IFEval returns a dict; a simple match scorer returns "C"/"I".
|
|
Normalizing here keeps the results schema uniform, which is what makes
|
|
cross-task and cross-checkpoint comparison possible at all.
|
|
"""
|
|
extra: dict[str, Any] = {}
|
|
|
|
if isinstance(value, dict):
|
|
extra = dict(value)
|
|
if primary_key and primary_key in value:
|
|
inner = value[primary_key]
|
|
else:
|
|
# Fall back to the first bool/numeric entry so an unconfigured task
|
|
# still produces something rather than silently scoring zero.
|
|
inner = next(
|
|
(v for v in value.values() if isinstance(v, (bool, int, float))),
|
|
0.0,
|
|
)
|
|
return _normalize_sample_score(inner, None)[0], bool(inner), extra
|
|
|
|
if isinstance(value, bool):
|
|
return (1.0 if value else 0.0), value, extra
|
|
|
|
if isinstance(value, (int, float)):
|
|
# A non-finite score means the sample was never graded — a rubric sample with no
|
|
# judge configured arrives as NaN. `NaN > 0.0` is False, so it used to be recorded
|
|
# as an ordinary failure, which is a different claim: "the model got this wrong"
|
|
# rather than "nobody scored this". The mean already skips it; this makes the
|
|
# per-sample record say so too.
|
|
if isinstance(value, float) and value != value:
|
|
return value, False, {**extra, "ungraded": True}
|
|
return float(value), float(value) > 0.0, extra
|
|
|
|
if isinstance(value, str):
|
|
score = _CORRECTNESS.get(value.strip().upper()[:1], 0.0)
|
|
return score, score >= 1.0, {"raw": value}
|
|
|
|
return 0.0, False, {"raw": repr(value)}
|
|
|
|
|
|
def _quality_from_log(log: Any, spec: task_catalog.TaskSpec) -> QualityResult:
|
|
"""Extract a QualityResult (including per-sample outcomes) from an EvalLog."""
|
|
if log.status != "success":
|
|
err = getattr(log, "error", None)
|
|
return QualityResult(
|
|
task=spec.name,
|
|
tier=spec.tier,
|
|
error=str(err) if err else f"eval status: {log.status}",
|
|
)
|
|
|
|
metrics: dict[str, float] = {}
|
|
for scorer in (log.results.scores if log.results else []):
|
|
for key, metric in scorer.metrics.items():
|
|
metrics[key] = round(float(metric.value), 6)
|
|
|
|
# Promote the task's headline number to a uniform "accuracy" key so the
|
|
# leaderboard and verdict logic never need per-task special cases.
|
|
if spec.primary_metric in metrics:
|
|
metrics["accuracy"] = metrics[spec.primary_metric]
|
|
|
|
samples: list[SampleOutcome] = []
|
|
for s in (log.samples or []):
|
|
for scorer_name, score in (s.scores or {}).items():
|
|
value, passed, extra = _normalize_sample_score(
|
|
score.value, spec.primary_sample_key
|
|
)
|
|
samples.append(
|
|
SampleOutcome(
|
|
sample_id=str(s.id),
|
|
score=value,
|
|
passed=passed,
|
|
excerpt=truncate(getattr(s.output, "completion", None)),
|
|
metadata={
|
|
"scorer": scorer_name,
|
|
"epoch": s.epoch,
|
|
**extra,
|
|
},
|
|
)
|
|
)
|
|
|
|
stats = getattr(log, "stats", None)
|
|
duration = None
|
|
if stats and getattr(stats, "started_at", None) and getattr(stats, "completed_at", None):
|
|
try:
|
|
from datetime import datetime
|
|
|
|
duration = (
|
|
datetime.fromisoformat(stats.completed_at)
|
|
- datetime.fromisoformat(stats.started_at)
|
|
).total_seconds()
|
|
except Exception: # noqa: BLE001 - duration is nice-to-have, never fatal
|
|
duration = None
|
|
|
|
return QualityResult(
|
|
task=spec.name,
|
|
tier=spec.tier,
|
|
dataset_version=getattr(log.eval, "task_version", None) and str(log.eval.task_version),
|
|
dataset_fingerprint=_dataset_fingerprint(log.samples or []),
|
|
n_samples=len(log.samples or []),
|
|
metrics=metrics,
|
|
samples=samples,
|
|
duration_s=duration,
|
|
)
|
|
|
|
|
|
def _dataset_fingerprint(samples: list[Any]) -> str | None:
|
|
"""Content hash of the samples actually run.
|
|
|
|
`dataset_version` is inspect's static task_version -- it stays "2" whether the family
|
|
has twelve samples or a rewritten scorer. Editing a prompt, tightening a regex, or
|
|
dropping a sample therefore produced two cards that claimed the same dataset version
|
|
and were not comparable, with nothing to reveal it. Comparing those is the exact error
|
|
this bench exists to prevent, so the identity has to come from the content.
|
|
|
|
Hashed over (id, input, target) because those are what determine whether a score means
|
|
the same thing. Sorted, so sample ordering does not change the fingerprint.
|
|
"""
|
|
if not samples:
|
|
return None
|
|
h = hashlib.sha256()
|
|
for key in sorted(
|
|
f"{s.id}\x1f{s.input}\x1f{s.target}" for s in samples
|
|
):
|
|
h.update(key.encode())
|
|
h.update(b"\x1e")
|
|
return f"sha256:{h.hexdigest()[:16]}"
|
|
|
|
|
|
def preflight(specs: list[task_catalog.TaskSpec]) -> None:
|
|
"""Fetch scorer prerequisites before burning GPU time on a doomed run.
|
|
|
|
Scorers often touch their corpora only at scoring time, so a missing
|
|
dependency surfaces after the whole dataset has been generated. Checking
|
|
up front converts a 20-minute failure into a 2-second one.
|
|
"""
|
|
needed = {res for spec in specs for res in spec.nltk_resources}
|
|
if not needed:
|
|
return
|
|
|
|
import nltk
|
|
|
|
for resource in sorted(needed):
|
|
try:
|
|
nltk.data.find(resource)
|
|
except LookupError:
|
|
name = resource.rsplit("/", 1)[-1]
|
|
print(f"[preflight] fetching nltk resource {name}")
|
|
nltk.download(name, quiet=True)
|
|
nltk.data.find(resource) # raise loudly if it still is not there
|
|
|
|
|
|
# Fallback sampling for a target that declares none. Deliberately NOT greedy: temperature 0
|
|
# looks like the reproducible choice and breaks reasoning models, which repeat forever
|
|
# without it. Reproducibility comes from the fixed seed. A target should override this in
|
|
# the registry with the values its model card documents.
|
|
DEFAULT_SAMPLING = {
|
|
"temperature": 0.6,
|
|
"top_p": 0.95,
|
|
"seed": 20260804,
|
|
"max_tokens": 4096,
|
|
}
|
|
|
|
|
|
def resolve_sampling(target: Target) -> dict[str, Any]:
|
|
"""The sampling settings for a target, registry first, defaults second.
|
|
|
|
Returned rather than read inline so the score card records exactly what the run used:
|
|
the two cannot drift, because they are the same call.
|
|
"""
|
|
return {**DEFAULT_SAMPLING, **(target.serving.sampling or {})}
|
|
|
|
|
|
def run_quality(
|
|
target: Target,
|
|
specs: list[task_catalog.TaskSpec],
|
|
limit: int | None = None,
|
|
max_connections: int = 16,
|
|
log_dir: Path | None = None,
|
|
epochs: int = 1,
|
|
) -> list[QualityResult]:
|
|
"""Run quality tasks against a target via Inspect."""
|
|
from inspect_ai import eval as inspect_eval
|
|
|
|
preflight(specs)
|
|
target.apply_env()
|
|
results: list[QualityResult] = []
|
|
sampling = resolve_sampling(target)
|
|
print(f" sampling: {sampling} | epochs: {epochs}")
|
|
if epochs == 1:
|
|
print(
|
|
" NOTE: one epoch. Sampling is not reproducible on this server even with a\n"
|
|
" fixed seed -- continuous batching changes the logits -- so a single\n"
|
|
" epoch is one draw, not a rate. Use --epochs for a signal-tier number."
|
|
)
|
|
|
|
for spec in specs:
|
|
print(f"\n[quality] {spec.name} ({spec.tier}) -> {target.id}")
|
|
logs = inspect_eval(
|
|
tasks=spec.inspect_task,
|
|
model=target.inspect_model,
|
|
limit=limit,
|
|
epochs=epochs,
|
|
log_dir=str(log_dir or LOG_DIR),
|
|
display="plain",
|
|
# Greedy decoding. The perf harness has always pinned temperature 0; the
|
|
# quality half pinned nothing and inherited the server's default, so it
|
|
# sampled. Three runs over the same twelve samples scored 4, 5 and 7 of 11
|
|
# -- 27% of samples flipped -- and a score card that cannot reproduce its
|
|
# own number is not evidence, it is a draw from a distribution nobody
|
|
# recorded. Sampling also makes `compare` meaningless: a per-sample flip
|
|
# caused by temperature is indistinguishable from a real regression.
|
|
#
|
|
# These MUST go through config=GenerateConfig. inspect_eval takes **kwargs,
|
|
# so passing temperature= directly is accepted silently and does nothing --
|
|
# the run would have looked pinned and still sampled.
|
|
# Loose kwargs ARE the interface: inspect_eval collects everything it does
|
|
# not consume into a GenerateConfig itself, so `config=` is rejected as an
|
|
# unknown field and `max_connections` must ride along here rather than as a
|
|
# sibling argument. Verified by running it, not by reading the signature --
|
|
# these names do not appear in it.
|
|
max_connections=max_connections,
|
|
**sampling,
|
|
)
|
|
for log in logs:
|
|
qr = _quality_from_log(log, spec)
|
|
# Record the cap explicitly. A limited run that looks like a full
|
|
# run is the single easiest way to publish a misleading number.
|
|
if limit is not None:
|
|
qr.metrics["_limit"] = float(limit)
|
|
results.append(qr)
|
|
|
|
return results
|
|
|
|
|
|
def run_perf(
|
|
target: Target,
|
|
concurrencies: tuple[int, ...] = (1, 8, 32),
|
|
input_tokens: int = 1024,
|
|
output_tokens: int = 256,
|
|
prompts: list[str] | None = None,
|
|
pin_output: bool = True,
|
|
) -> PerfResult:
|
|
"""Run a serving performance sweep against a target."""
|
|
base_url = target.serving.base_url
|
|
if not base_url:
|
|
raise ValueError(f"target {target.id} has no base_url; cannot run perf sweep")
|
|
|
|
print(f"\n[perf] {target.id} @ {base_url}")
|
|
|
|
# A sweep past the server's own concurrency ceiling measures the queue, not the server.
|
|
# vLLM admits max_num_seqs sequences per step and queues the rest, so those points show
|
|
# flat total throughput and inflating TTFT — which reads as saturation and is really just
|
|
# waiting. your-node runs max_num_seqs=4 against a default sweep of 1,8,32, so two of three
|
|
# points were destined to be misread. Warn rather than refuse: measuring the queue is a
|
|
# legitimate thing to want, as long as nobody mistakes it for the engine's limit.
|
|
max_seqs = (target.serving.flags or {}).get("max_num_seqs")
|
|
if isinstance(max_seqs, int):
|
|
beyond = [c for c in concurrencies if c > max_seqs]
|
|
if beyond:
|
|
print(
|
|
f" WARNING: max_num_seqs={max_seqs}, so concurrency {beyond} exceeds what "
|
|
f"the server admits per step.\n"
|
|
f" Those points measure queueing, not serving capacity. "
|
|
f"Consider --concurrency 1,{max(2, max_seqs // 2)},{max_seqs}."
|
|
)
|
|
|
|
mode = []
|
|
if prompts:
|
|
mode.append(f"{len(prompts)} representative prompts")
|
|
if not pin_output:
|
|
mode.append("natural stop (output length not pinned)")
|
|
print(f" input={input_tokens} output={output_tokens} tokens"
|
|
+ (f" [{'; '.join(mode)}]" if mode else ""))
|
|
return asyncio.run(
|
|
run_sweep(
|
|
base_url=base_url,
|
|
model=target.serving.model_name,
|
|
concurrencies=concurrencies,
|
|
input_tokens=input_tokens,
|
|
output_tokens=output_tokens,
|
|
engine=target.serving.engine,
|
|
prompts=prompts,
|
|
pin_output=pin_output,
|
|
)
|
|
)
|
|
|
|
|
|
def _epochs_of(quality: list[QualityResult]) -> int:
|
|
"""How many times each sample was actually run, read back from the outcomes.
|
|
|
|
Taken from the results rather than the argument so the card cannot claim an averaging
|
|
it did not do -- the same reason resolve_sampling feeds the run and the card from one
|
|
call.
|
|
"""
|
|
per_sample: dict[str, int] = {}
|
|
for q in quality:
|
|
for s in q.samples:
|
|
per_sample[s.sample_id] = per_sample.get(s.sample_id, 0) + 1
|
|
return max(per_sample.values()) if per_sample else 1
|
|
|
|
|
|
def build_run(
|
|
registry: Registry,
|
|
target: Target,
|
|
quality: list[QualityResult],
|
|
perf: PerfResult | None,
|
|
) -> Run:
|
|
host = registry.host(target.host)
|
|
target_snapshot = asdict(target)
|
|
target_snapshot["slug"] = target.slug
|
|
target_snapshot["inspect_model"] = target.inspect_model
|
|
|
|
return Run(
|
|
run_id=uuid.uuid4().hex,
|
|
timestamp=utc_now(),
|
|
target_id=target.id,
|
|
target=target_snapshot,
|
|
host=asdict(host),
|
|
quality=quality,
|
|
perf=perf,
|
|
verdict=compute_verdict(quality, perf),
|
|
runner=runner_info(),
|
|
sampling={**resolve_sampling(target), "epochs": _epochs_of(quality)},
|
|
)
|