diff --git a/evalharness/cli.py b/evalharness/cli.py index b8ea9b8..03da0e3 100644 --- a/evalharness/cli.py +++ b/evalharness/cli.py @@ -728,37 +728,8 @@ def _cmd_eval_run(args) -> int: _rep_secs = 0.0 _rep_tin = _rep_tout = 0 - report = None - # score cache: the checkpoint stores PREDICTIONS, not scores -- - # but when every prediction is already checkpointed AND a saved - # report matches (same model, same sample count), re-scoring - # reproduces the same numbers, so reuse the report and skip the - # whole docker/exec scoring pass. --rescore forces evaluation - # (recipe/judge changed, or just paranoia). - if model_spec and out_dir and _repeats == 1 \ - and not getattr(args, 'rescore', False): - from pathlib import Path as _P2 - - _rp = _P2(out_dir) / name / 'report.jsonl' - if _rp.exists(): - try: - from evalharness.eval.record import EvalReport as _ER - - _old = _ER.load(str(_rp)) - _m_old = (_old.model or '').split('?')[-1] - _m_new = model_spec.split('?')[-1] - if _m_old == _m_new \ - and _old.num_samples == (args.limit or sample_count): - report = _old - _emit('Reusing saved report -- predictions all ' - 'checkpointed (--rescore to re-evaluate)') - except Exception: - pass # unreadable/stale report: score normally - - # progress reporter setup runs for EVERY path (report reuse - # included): the success path advances the overall bar even when - # nothing was generated -- previously this block lived inside the - # generation branch and reuse blew up with UnboundLocalError + # progress reporter setup runs for every path: the success path + # advances the overall bar even when nothing was generated progress_reporter = None if args.progress and model_spec: from evalharness.progress import PROGRESS_REGISTRY @@ -785,7 +756,7 @@ def _cmd_eval_run(args) -> int: _shared_reporter.owned_externally = True progress_reporter = _shared_reporter - if report is None and model_spec: # generate + score in one go + if model_spec: # generate + score in one go from evalharness.model import run_eval @@ -850,6 +821,7 @@ def _cmd_eval_run(args) -> int: progress_reporter=progress_reporter, status_callback=status_callback, on_scored=on_scored, + rescore=getattr(args, 'rescore', False), repeat=_rep + 1)) _m = next((v for k, v in report.metrics.items() if k != 'extraction_failure_rate'), None) @@ -892,7 +864,7 @@ def _cmd_eval_run(args) -> int: if _primary: report.metrics[f'{_primary}_last_run'] = report.metrics[_primary] report.metrics[_primary] = _mean - elif report is None: + else: from evalharness.eval import evaluate if not args.predictions: diff --git a/evalharness/eval/checkpoint.py b/evalharness/eval/checkpoint.py index 4e97c72..aa44d03 100644 --- a/evalharness/eval/checkpoint.py +++ b/evalharness/eval/checkpoint.py @@ -28,6 +28,7 @@ class CheckpointStore: self.model = model self.dataset = dataset self._entries: Dict[str, Dict[str, Any]] = {} + self._scores: Dict[str, Dict[str, Any]] = {} self._fh = None @staticmethod @@ -47,6 +48,7 @@ class CheckpointStore: def load(self) -> Dict[str, Dict[str, Any]]: """Read all checkpointed predictions (idempotent).""" self._entries = {} + self._scores = {} if not self.path.exists(): return self._entries with open(self.path, encoding='utf-8') as f: @@ -57,6 +59,8 @@ class CheckpointStore: try: rec = json.loads(line) self._entries[rec['key']] = rec.get('pred', {}) + if rec.get('score'): + self._scores[rec['key']] = rec['score'] except (ValueError, KeyError): continue # torn tail line from a crash -- safe to skip return self._entries @@ -69,6 +73,41 @@ class CheckpointStore: f.write(json.dumps(rec, ensure_ascii=False) + '\n') self._entries[key] = pred + def scores(self) -> Dict[str, Dict[str, Any]]: + """Cached per-sample score records ({key: {'fp', 'scores', ...}}). + + Scores are bound to predictions and live in the SAME file -- one + --resume flag controls both layers.""" + return self._scores + + def put_scores(self, records: Dict[str, Dict[str, Any]]) -> None: + """Attach score records to checkpoint entries (end-of-evaluation + writeback). Rewrites the file atomically; legacy lines without a + score field simply gain one.""" + if not records: + return + lines: Dict[str, str] = {} + if self.path.exists(): + with open(self.path, encoding='utf-8') as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except ValueError: + continue + if rec.get('key') in records: + rec['score'] = records[rec['key']] + lines[rec['key']] = json.dumps(rec, ensure_ascii=False, + default=str) + tmp = self.path.with_suffix('.tmp') + with open(tmp, 'w', encoding='utf-8') as f: + for v in lines.values(): + f.write(v + '\n') + os.replace(tmp, self.path) + self._scores.update(records) + def __len__(self) -> int: return len(self._entries) diff --git a/evalharness/eval/runner.py b/evalharness/eval/runner.py index fe7e871..323925f 100644 --- a/evalharness/eval/runner.py +++ b/evalharness/eval/runner.py @@ -62,10 +62,10 @@ def evaluate( results: List[SampleResult] = [] - 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).""" + 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( @@ -89,6 +89,14 @@ def evaluate( 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 @@ -165,6 +173,91 @@ def evaluate( 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: diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py index 5b65a21..1e9ca83 100644 --- a/evalharness/model/runner.py +++ b/evalharness/model/runner.py @@ -466,7 +466,10 @@ async def generate_predictions( latency_s=float(u.get('latency_s', 0) or 0)) if status_callback and pending: status_callback(f'Generation complete: {len(preds)} responses collected') - return preds, usages, total_usage + # ckpt info (store + per-position keys) so run_eval can read/write + # SCORES bound to these predictions; None when checkpointing is off + ckpt_info = (ckpt_store, keys) if ckpt_store is not None else None + return preds, usages, total_usage, ckpt_info finally: # reporter lifecycle belongs to the CALLER (CLI reuses one reporter # across benchmarks and closes it after the whole run); only close @@ -630,6 +633,7 @@ async def run_eval( gen_profile: str = '', repeat: int = 1, on_scored=None, + rescore: bool = False, ) -> EvalReport: """Generate + score in one call. Model spec examples: 'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'. @@ -742,7 +746,7 @@ async def run_eval( try: from .gen_profiles import merge_gen_kwargs - preds, _usages, usage = await generate_predictions( + preds, _usages, usage, ckpt_info = await generate_predictions( adapter, list(raw_samples), concurrency, progress=progress, progress_reporter=progress_reporter, status_callback=status_callback, @@ -762,7 +766,27 @@ async def run_eval( repeat=repeat) finally: await adapter.close() - if judge is None and judge_spec: + + # SCORES ARE BOUND TO PREDICTIONS in the checkpoint: when every sample's + # cached score matches the current scoring setup (recipe/extract/scorers/ + # judge fingerprint), replay them without touching a single scorer -- + # docker exec benches skip their containers entirely. --resume controls + # the whole stack (no checkpoint -> nothing cached -> evaluate + backfill) + _fp = None + _records = None + if ckpt_info is not None and not rescore: + from ..eval.runner import score_fingerprint + + store, ck_keys = ckpt_info + _fp = score_fingerprint(recipe, judge_spec or '') + cached = store.scores() + if ck_keys and all(cached.get(k, {}).get('fp') == _fp for k in ck_keys): + _records = [cached[k] for k in ck_keys] + if status_callback: + status_callback('Scores cached in checkpoint -- replaying ' + '(no scorers run; --rescore re-evaluates)') + + if judge is None and judge_spec and _records is None: if status_callback: status_callback('loading judge model') judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key) @@ -770,19 +794,35 @@ async def run_eval( if status_callback: status_callback('Scoring predictions against the benchmark recipe') - # scoring off the event loop: math_equal/sympy equivalence can chew a - # single hard problem for minutes (es's checker famously hangs on one) -- - # running it inline froze the progress bar's clock for the whole bench - report = await asyncio.to_thread( - evaluate, - samples, preds, recipe, - model=model_spec, - judge=judge, - extra_metadata={'gen_input_tokens': usage.input_tokens, - 'gen_output_tokens': usage.output_tokens, - 'gen_total_tokens': usage.total_tokens}, - on_scored=on_scored, - ) + _meta = {'gen_input_tokens': usage.input_tokens, + 'gen_output_tokens': usage.output_tokens, + 'gen_total_tokens': usage.total_tokens} + if _records is not None: + from ..eval.runner import evaluate_cached + + report = evaluate_cached(samples, preds, recipe, _records, + model=model_spec, extra_metadata=_meta) + else: + # scoring off the event loop: math_equal/sympy equivalence can chew a + # single hard problem for minutes (es's checker famously hangs on one) -- + # running it inline froze the progress bar's clock for the whole bench + report = await asyncio.to_thread( + evaluate, + samples, preds, recipe, + model=model_spec, + judge=judge, + extra_metadata=_meta, + on_scored=on_scored, + ) + # writeback: bind these scores to the predictions in the checkpoint + if ckpt_info is not None: + from ..eval.runner import score_fingerprint, score_record_of + + store, ck_keys = ckpt_info + if _fp is None: + _fp = score_fingerprint(recipe, judge_spec or '') + store.put_scores({ck_keys[i]: score_record_of(report.samples[i], _fp) + for i in range(min(len(ck_keys), len(report.samples)))}) report.model = model_spec report.dataset = name if status_callback: