652c13db36
User feedback: the out-dir report-reuse layer was one concept too many.
Now each checkpoint line carries {key, ts, pred, score}:
- --resume restores predictions AND their scores; when every sample's
cached score matches the scoring-setup fingerprint (recipe/extract/
scorers/judge), the report is replayed with NO scorer, extractor or
docker container touching anything
- fingerprint mismatch (recipe or judge changed) -> automatic re-eval
and backfill of the fresh scores
- no --resume -> nothing read, nothing written (full fresh run)
- --rescore = ignore cached scores, re-evaluate, refresh the cache
- aggregation always recomputed from cached per-sample scores (cheap,
survives aggregator changes without invalidating)
- legacy checkpoints without a score field backfill on first evaluation
Removed: the out-dir report-reuse block (superseded; also the source of
the UnboundLocalError path).
Co-Authored-By: Claude <noreply@anthropic.com>
300 lines
13 KiB
Python
300 lines
13 KiB
Python
"""The evaluation runner: Dataset x predictions -> EvalReport.
|
|
|
|
Pure orchestration, no I/O hidden inside: predictions arrive as a list
|
|
(loaded from a jsonl of model outputs, a Session store, or built inline),
|
|
results aggregate into an EvalReport that visualizers consume.
|
|
|
|
Judge wiring: pass judge=<callable(messages)->str> once a ModelAdapter
|
|
exists; llm_judge recipes work immediately after that, no recipe change.
|
|
"""
|
|
|
|
import traceback
|
|
from typing import Callable, Dict, Iterable, List, Optional, Sequence, Union
|
|
|
|
from ..data.dataset import Dataset
|
|
from ..data.sample import Sample
|
|
from .aggregator import mean as _mean_agg
|
|
from .recipe import EvalRecipe
|
|
from .record import EvalReport, SampleResult
|
|
from .scorer import ScoreContext
|
|
|
|
|
|
def evaluate(
|
|
dataset: Union[Dataset, List[Sample]],
|
|
predictions: Sequence[Union[str, Dict]],
|
|
recipe: Optional[EvalRecipe] = None,
|
|
*,
|
|
model: str = '',
|
|
judge: Optional[Callable] = None,
|
|
extra_metadata: Optional[Dict] = None,
|
|
on_scored: Optional[Callable[[int, int], None]] = None,
|
|
) -> EvalReport:
|
|
"""Score a dataset against raw predictions.
|
|
|
|
dataset: a Dataset or a plain list of Samples (views/slices).
|
|
predictions: str per sample (raw model output) or dicts with
|
|
{'raw': str, 'group_key': ..., 'metadata': {...}} overrides.
|
|
"""
|
|
samples: List[Sample] = list(dataset)
|
|
spec = getattr(dataset, 'spec', None)
|
|
ds_name = spec.name if spec is not None else samples[0].metadata.get('dataset', 'adhoc') if samples else 'adhoc'
|
|
ds_subset = spec.subset if spec is not None else ''
|
|
if len(predictions) != len(samples):
|
|
raise ValueError(f'{len(predictions)} predictions for {len(samples)} samples')
|
|
|
|
if recipe is None:
|
|
from .recipe import get_eval
|
|
|
|
recipe = get_eval(ds_name)
|
|
extractor = recipe.resolve_extract()
|
|
scorers = recipe.resolve_scorers()
|
|
aggregators = recipe.resolve_aggregators()
|
|
ctx = ScoreContext(judge=judge, params={})
|
|
|
|
# If any scorer executes in docker with per-sample images, overlap pulls
|
|
# with scoring (run sample N while N+1..N+lookahead images download).
|
|
bp = None
|
|
if _needs_bg_prefetch(recipe, samples):
|
|
from ..sandbox import BackgroundPrefetcher, images_for_samples
|
|
|
|
bp = BackgroundPrefetcher(images_for_samples(samples), workers=4, lookahead=8)
|
|
bp.__enter__()
|
|
|
|
results: List[SampleResult] = []
|
|
|
|
def _shell(sample, pred) -> SampleResult:
|
|
"""SampleResult with everything derivable from (sample, prediction):
|
|
identity, raw text, usage, trajectories. Shared by live scoring and
|
|
the checkpoint-score replay path."""
|
|
raw = pred if isinstance(pred, str) else str(pred.get('raw', ''))
|
|
override = {} if isinstance(pred, str) else pred
|
|
result = SampleResult(
|
|
sample_id=sample.id,
|
|
dataset=ds_name,
|
|
subset=ds_subset,
|
|
task_type=sample.task_type,
|
|
raw_prediction=raw,
|
|
target=sample.target,
|
|
group_key=str(override.get('group_key')
|
|
or sample.metadata.get('group_key')
|
|
or (sample.metadata.get('task_id') or sample.metadata.get('id') or '')),
|
|
metadata={k: v for k, v in (sample.metadata or {}).items()
|
|
if k in ('category', 'subject', 'test_category', 'bin', 'difficulty')},
|
|
)
|
|
if isinstance(pred, dict) and pred.get('metadata'):
|
|
result.metadata.update(pred['metadata'])
|
|
if isinstance(pred, dict) and pred.get('trajectory'):
|
|
result.trajectory = pred['trajectory']
|
|
if isinstance(pred, dict) and pred.get('env_state'):
|
|
result.env_state = pred['env_state']
|
|
if isinstance(pred, dict) and pred.get('usage'):
|
|
result.usage = pred['usage']
|
|
return result
|
|
|
|
def judge_one(sample, pred) -> SampleResult:
|
|
"""Extract + score ONE sample (thread-safe: everything here is local
|
|
except docker/subprocess execution, which parallelizes perfectly --
|
|
each sample gets its own container/workdir)."""
|
|
result = _shell(sample, pred)
|
|
raw = result.raw_prediction
|
|
try:
|
|
if bp is not None and sample.sandbox and sample.sandbox.image:
|
|
bp.ensure(sample.sandbox.image) # wait only if this one still pulling
|
|
value, ok, note = extractor(raw, sample)
|
|
result.extracted_prediction = value
|
|
result.extraction_ok = ok
|
|
result.extraction_note = note
|
|
if not ok:
|
|
result.extraction_note = note or 'extractor returned not-ok'
|
|
for metric, scorer in scorers.items():
|
|
try:
|
|
sctx = ctx
|
|
if result.env_state and 'env_state' not in ctx.params:
|
|
sctx = ScoreContext(judge=ctx.judge, judge_model=ctx.judge_model,
|
|
params={**ctx.params,
|
|
'env_state': result.env_state})
|
|
scores, details = scorer(value if ok else '', sample.target, sample, sctx)
|
|
result.scores.update(scores)
|
|
result.score_details.update(details)
|
|
except Exception as e: # one metric failing must not kill the run
|
|
result.scores[metric] = 0.0
|
|
result.score_details[metric] = {'error': f'{type(e).__name__}: {e}'}
|
|
except Exception as e:
|
|
result.error = f'{type(e).__name__}: {e}\n{traceback.format_exc(limit=2)}'
|
|
return result
|
|
|
|
workers = getattr(recipe, 'exec_workers', 1)
|
|
n_total = len(samples)
|
|
import itertools
|
|
|
|
_scored = itertools.count(1) # next() is atomic: safe from worker threads
|
|
|
|
def _counted(sample, pred):
|
|
# docker/subprocess scoring is minute-scale per sample; surface
|
|
# per-sample progress or the run looks frozen at 'generating 100%'
|
|
r = judge_one(sample, pred)
|
|
if on_scored is not None:
|
|
try:
|
|
on_scored(next(_scored), n_total)
|
|
except Exception:
|
|
pass
|
|
return r
|
|
|
|
if workers > 1 and len(samples) > 1:
|
|
# parallel judging: docker/subprocess execution is embarrassingly
|
|
# parallel (one container per sample); text scorers are cheap and
|
|
# thread-safe enough. Serializes again for judge/dict-dependent runs.
|
|
import concurrent.futures
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
|
|
results = list(pool.map(_counted, samples, predictions))
|
|
else:
|
|
for sample, pred in zip(samples, predictions):
|
|
results.append(_counted(sample, pred))
|
|
|
|
report = EvalReport(
|
|
dataset=ds_name,
|
|
recipe=recipe.name or dataset.spec.name,
|
|
model=model,
|
|
num_samples=len(results),
|
|
num_failed_extractions=sum(1 for r in results if not r.extraction_ok),
|
|
samples=results,
|
|
)
|
|
_aggregate_into(report, results, recipe, aggregators)
|
|
if bp is not None:
|
|
report.metric_groups['run_info'] = {
|
|
**report.metric_groups.get('run_info', {}),
|
|
**{f'img_{k}': v for k, v in bp.stats().items()},
|
|
}
|
|
bp.__exit__(None, None, None)
|
|
if extra_metadata:
|
|
report.metric_groups['run_info'] = {k: v for k, v in extra_metadata.items()
|
|
if isinstance(v, (int, float, str))}
|
|
return report
|
|
|
|
|
|
def score_fingerprint(recipe: EvalRecipe, judge_spec: str = '') -> str:
|
|
"""Identity of the SCORING setup: recipe + extractors + scorers + judge.
|
|
Cached score records carry it; a mismatch means re-evaluate."""
|
|
import hashlib
|
|
import json as _json
|
|
|
|
payload = _json.dumps({
|
|
'recipe': getattr(recipe, 'name', ''),
|
|
'extract': str(getattr(recipe, 'extract', '')),
|
|
'scorers': str(getattr(recipe, 'scorers', '')),
|
|
'judge': judge_spec or '',
|
|
}, sort_keys=True, default=str)
|
|
return hashlib.md5(payload.encode()).hexdigest()[:12]
|
|
|
|
|
|
def score_record_of(result: SampleResult, fp: str) -> Dict:
|
|
"""Extract the cacheable part of an evaluated SampleResult."""
|
|
return {'fp': fp,
|
|
'extracted': result.extracted_prediction,
|
|
'ok': result.extraction_ok,
|
|
'note': result.extraction_note,
|
|
'scores': dict(result.scores),
|
|
'details': {k: (v if isinstance(v, (str, int, float, bool, dict, list, type(None)))
|
|
else str(v))
|
|
for k, v in result.score_details.items()},
|
|
'error': result.error or ''}
|
|
|
|
|
|
def evaluate_cached(dataset, predictions, recipe, score_records, *,
|
|
model: str = '', extra_metadata=None) -> EvalReport:
|
|
"""Rebuild a report from checkpoint-cached scores -- no extractor, no
|
|
scorers, no docker. Shells come from (sample, prediction), scores from
|
|
the cached records; aggregation runs FRESH (cheap, and covers recipe
|
|
aggregation changes without invalidating the cache)."""
|
|
samples = list(dataset)
|
|
if len(predictions) != len(samples) or len(score_records) != len(samples):
|
|
raise ValueError('evaluate_cached: samples/predictions/score_records '
|
|
f'length mismatch ({len(samples)}/'
|
|
f'{len(predictions)}/{len(score_records)})')
|
|
spec = getattr(dataset, 'spec', None)
|
|
ds_name = spec.name if spec is not None else 'adhoc'
|
|
ds_subset = spec.subset if spec is not None else ''
|
|
aggregator_map = recipe.resolve_aggregators() if recipe is not None else {}
|
|
from .aggregator import mean as _mean
|
|
|
|
def _mk(sample, pred, sr):
|
|
r = SampleResult(
|
|
sample_id=sample.id, dataset=ds_name, subset=ds_subset,
|
|
task_type=sample.task_type,
|
|
raw_prediction=pred if isinstance(pred, str) else str(pred.get('raw', '')),
|
|
target=sample.target,
|
|
group_key=str((pred if isinstance(pred, dict) else {}).get('group_key')
|
|
or sample.metadata.get('group_key')
|
|
or (sample.metadata.get('task_id') or sample.metadata.get('id') or '')),
|
|
metadata={k: v for k, v in (sample.metadata or {}).items()
|
|
if k in ('category', 'subject', 'test_category', 'bin', 'difficulty')},
|
|
)
|
|
if isinstance(pred, dict):
|
|
if pred.get('usage'):
|
|
r.usage = pred['usage']
|
|
if pred.get('metadata'):
|
|
r.metadata.update(pred['metadata'])
|
|
r.extracted_prediction = sr.get('extracted', '')
|
|
r.extraction_ok = bool(sr.get('ok', True))
|
|
r.extraction_note = sr.get('note', '')
|
|
r.scores.update(sr.get('scores') or {})
|
|
r.score_details.update(sr.get('details') or {})
|
|
return r
|
|
|
|
results = [_mk(s, p, sr) for s, p, sr in zip(samples, predictions, score_records)]
|
|
report = EvalReport(
|
|
dataset=ds_name,
|
|
recipe=recipe.name if recipe is not None else ds_name,
|
|
model=model,
|
|
num_samples=len(results),
|
|
num_failed_extractions=sum(1 for r in results if not r.extraction_ok),
|
|
samples=results,
|
|
)
|
|
_aggregate_into(report, results, recipe, aggregator_map or {'acc': _mean})
|
|
if extra_metadata:
|
|
report.metric_groups['run_info'] = {k: v for k, v in extra_metadata.items()
|
|
if isinstance(v, (int, float, str))}
|
|
return report
|
|
|
|
|
|
def _needs_bg_prefetch(recipe, samples) -> bool:
|
|
"""True when the recipe executes in docker AND samples declare images."""
|
|
try:
|
|
for spec in recipe.scorers.values():
|
|
params = spec if isinstance(spec, dict) else {}
|
|
if params.get('name') == 'execution' and params.get('sandbox') == 'docker':
|
|
return any(s.sandbox and s.sandbox.image for s in samples[:50])
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
|
|
def _aggregate_into(report: EvalReport, results, recipe: EvalRecipe, aggregators) -> None:
|
|
for metric in recipe.scorers:
|
|
agg = aggregators.get(metric)
|
|
if agg is None:
|
|
agg = _mean_agg
|
|
try:
|
|
out = agg(results, metric)
|
|
except Exception as e:
|
|
report.metric_groups[f'agg_error_{metric}'] = {'error': str(e)[:200]}
|
|
continue
|
|
if isinstance(out, dict):
|
|
report.metric_groups[metric] = out
|
|
# primary metric = the aggregator's same-named entry (e.g.
|
|
# simpleqa_official returns is_correct/is_incorrect/...); the
|
|
# old mean-of-all-values fallback invented nonsense like
|
|
# mean(0.035, 0.945, 0.02, 0.98) for is_correct
|
|
if metric in out and isinstance(out[metric], (int, float)):
|
|
report.metrics[metric] = float(out[metric])
|
|
else:
|
|
vals = [v for v in out.values() if isinstance(v, (int, float))]
|
|
if vals:
|
|
report.metrics[metric] = sum(vals) / len(vals)
|
|
else:
|
|
report.metrics[metric] = float(out)
|
|
report.metrics['extraction_failure_rate'] = (
|
|
report.num_failed_extractions / report.num_samples if report.num_samples else 0.0
|
|
)
|