Meta-Eval/EvalHarness
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: Meta-Eval:e9b79a2a412a2d8ccc40cad24d0653bb848509e1
Branches Tags
Meta-Eval:main
..
compare: Meta-Eval:9a64896c97aedfb11f950e7e63ac18c9816f188b
Branches Tags
Meta-Eval:main

No commits in common. "e9b79a2a412a2d8ccc40cad24d0653bb848509e1" and "9a64896c97aedfb11f950e7e63ac18c9816f188b" have entirely different histories.
e9b79a2a41 ... 9a64896c97

17 changed files with 192 additions and 1027 deletions
Show Stats Expand all files Collapse all files

2
README.md
View File

@ -161,7 +161,7 @@ rep.save('gsm8k.report.json')
| `--limit N` / `--limit-per-task N` | 全局前 N / 每子集前 N（多科目 bench 用后者；可组合取交集） | | `--limit N` / `--limit-per-task N` | 全局前 N / 每子集前 N（多科目 bench 用后者；可组合取交集） |
| `--subset` `--split` `--source` | 覆盖子集 / split / 数据源（可指本地路径离线跑） | | `--subset` `--split` `--source` | 覆盖子集 / split / 数据源（可指本地路径离线跑） |
| `--concurrency N` | 并发（默认 32；长输出 bench 建议 8-16） | | `--concurrency N` | 并发（默认 32；长输出 bench 建议 8-16） |
| `--concurrency auto` | 自适应并发门：从 2 起步，健康且供不应求时 +1 爬坡，请求失败 ×0.7 退避（服务端 `/metrics` 可用时按排队信号调节）；当前值显示在进度条 `gate N` | | `--auto-concurrency` | 自适应并发门：按端点健康状况自动决定并发（健康且供不应求时 +1 爬坡，请求失败 ×0.7 退避，服务端 `/metrics` 可用时按排队信号调节）；当前值显示在进度条 `gate N`。此时 `--concurrency` 是起点不是上限 |
| `--resume [PATH]` | 断点续跑；默认 `<cache-dir>/ckpt/<bench>.jsonl` | | `--resume [PATH]` | 断点续跑；默认 `<cache-dir>/ckpt/<bench>.jsonl` |
| `--env NAME` | agent 环境（`bfcl_mock` 等） | | `--env NAME` | agent 环境（`bfcl_mock` 等） |
| `--perf` | 采集流式 TTFT / ITL / 重试率入报告 | | `--perf` | 采集流式 TTFT / ITL / 重试率入报告 |

349
evalharness/cli.py
View File

@ -163,30 +163,6 @@ def _rich_console():
return None return None
BENCH_CATEGORIES = {
'bigcodebench': 'Code & Engineering', 'humaneval': 'Code & Engineering',
'live_code_bench': 'Code & Engineering',
'swe_bench_verified': 'Code & Engineering',
'aime24': 'Reasoning & Math', 'aime25': 'Reasoning & Math',
'aime26': 'Reasoning & Math', 'hmmt26': 'Reasoning & Math',
'imo_answerbench': 'Reasoning & Math', 'hle': 'Reasoning & Math',
'gsm8k': 'Reasoning & Math', 'competition_math': 'Reasoning & Math',
'bbh': 'Reasoning & Math', 'drop': 'Reasoning & Math',
'gpqa_diamond': 'Knowledge & Language', 'mmlu_pro': 'Knowledge & Language',
'simple_qa': 'Knowledge & Language', 'mmlu': 'Knowledge & Language',
'cmmlu': 'Knowledge & Language', 'arc': 'Knowledge & Language',
'hellaswag': 'Knowledge & Language', 'trivia_qa': 'Knowledge & Language',
'winogrande': 'Knowledge & Language',
'longbench_v2': 'Long Context', 'openai_mrcr': 'Long Context',
'tau2_bench': 'Agents & Tools', 'general_fc': 'Agents & Tools',
'bfcl_v3': 'Agents & Tools',
}
def bench_category(name: str) -> str:
return BENCH_CATEGORIES.get(name, '')
def _load_bench_cfg(args, name: str) -> dict: def _load_bench_cfg(args, name: str) -> dict:
"""Merged YAML config for one bench: {default 段, bench 段}. """Merged YAML config for one bench: {default 段, bench 段}.
@ -199,14 +175,8 @@ def _load_bench_cfg(args, name: str) -> dict:
cfg_name = getattr(args, 'config', '') cfg_name = getattr(args, 'config', '')
if not cfg_name and cfg_dir.exists(): if not cfg_name and cfg_dir.exists():
yamls = sorted(cfg_dir.glob('*.yaml')) yamls = sorted(cfg_dir.glob('*.yaml'))
# auto-load: a lone config wins; otherwise default.yaml wins.
# (NB: non-config yaml sidecars must NOT land in config/ -- a
# sample-counts manifest here once disabled auto-load entirely and
# silently dropped every bench's repeats/temperature/max_tokens)
if len(yamls) == 1: if len(yamls) == 1:
cfg_name = yamls[0].stem cfg_name = yamls[0].stem # auto: the only config
elif any(y.stem == 'default' for y in yamls):
cfg_name = 'default'
if not cfg_name: if not cfg_name:
return {} return {}
cfg_path = cfg_dir / f'{cfg_name}.yaml' cfg_path = cfg_dir / f'{cfg_name}.yaml'
@ -223,36 +193,25 @@ def _load_bench_cfg(args, name: str) -> dict:
def _plan_sample_counts(args): def _plan_sample_counts(args):
"""(total_samples, total_generations, n_uncached) across the planned """(total_samples, total_generations, n_uncached) across the planned
benches. Cache-first (exact); uncached benches fall back to the shipped benches, counted from LOCAL cache entries only -- never touches the
sample-counts manifest so the plan still shows CONCRETE numbers on a network, so the run plan stays instant on cold machines. Uncached
cold machine instead of 'counts when datasets load'. benches simply don't contribute yet.
""" """
from pathlib import Path
from evalharness.data import get_dataset from evalharness.data import get_dataset
manifest = {}
mf = Path(__file__).parent / 'config' / 'sample_counts.json'
if mf.exists():
try:
manifest = json.load(open(mf)) or {}
except Exception:
manifest = {}
total = gens = uncached = 0 total = gens = uncached = 0
for name in getattr(args, 'datasets', []) or []: for name in getattr(args, 'datasets', []) or []:
n = None
try: try:
ds = get_dataset(name, **_overrides(args)) ds = get_dataset(name, **_overrides(args))
cache_file = ds.cache_dir / 'samples.jsonl' cache_file = ds.cache_dir / 'samples.jsonl'
if cache_file.exists(): if not cache_file.exists():
with open(cache_file, 'rb') as f: uncached += 1
n = sum(1 for _ in f) continue
with open(cache_file, 'rb') as f:
n = sum(1 for _ in f)
except Exception: except Exception:
n = None
if n is None:
uncached += 1 uncached += 1
n = int(manifest.get(name, 0) or 0) # estimate; 0 = unknown continue
if getattr(args, 'limit', None): if getattr(args, 'limit', None):
n = min(n, args.limit) n = min(n, args.limit)
total += n total += n
@ -273,26 +232,21 @@ def _print_run_plan(console, args, model_spec):
cap = f' · ≤{args.limit} per bench (--limit)' cap = f' · ≤{args.limit} per bench (--limit)'
elif getattr(args, 'limit_per_task', None): elif getattr(args, 'limit_per_task', None):
cap = f' · ≤{args.limit_per_task} per subject (--limit-per-task)' cap = f' · ≤{args.limit_per_task} per subject (--limit-per-task)'
src = 'cached' if not n_uncached else ('cache+est.' if n_samples else 'est.')
if n_samples: if n_samples:
samples = f'{n_samples:,} samples ({src}){cap}' samples = f'{n_samples:,} samples (cached){cap}'
if n_gens > n_samples: # repeats multiply the real work if n_gens > n_samples: # repeats multiply the real work
samples = (f'{n_samples:,} samples ({src}){cap} → ' samples = (f'{n_samples:,} samples (cached){cap} → '
f'{n_gens:,} generations (repeats)') f'{n_gens:,} generations (repeats)')
elif getattr(args, 'limit', None): if n_uncached:
samples = f'up to {args.limit} per bench (--limit)' samples += f' · {n_uncached} bench(es) not cached yet'
elif getattr(args, 'limit_per_task', None): elif n_uncached:
samples = f'up to {args.limit_per_task} per subject (--limit-per-task)' if getattr(args, 'limit', None):
else: samples = f'up to {args.limit} per bench (--limit), counts when datasets load'
samples = '? (unknown benchmarks)' elif getattr(args, 'limit_per_task', None):
if n_uncached: samples = (f'up to {args.limit_per_task} per subject '
samples += f' · {n_uncached} bench(es) not cached yet' '(--limit-per-task), counts when datasets load')
_eff = getattr(args, 'reasoning_effort', '') or '' else:
_think_txt = ('disabled' if getattr(args, 'disable_thinking', False) samples = 'full dataset, counts when datasets load (none cached yet)'
else (f'enabled · effort={_eff}' if _eff else 'enabled'))
_auto = getattr(args, 'auto_concurrency', False)
_conc = (f'auto (start {args.concurrency}, gate decides)' if _auto
else str(args.concurrency))
if console is None: if console is None:
print(f'=== {title} ===') print(f'=== {title} ===')
print(f'Provider: {provider}') print(f'Provider: {provider}')
@ -300,7 +254,8 @@ def _print_run_plan(console, args, model_spec):
print(f'Model: {model_name}') print(f'Model: {model_name}')
print(f'Benchmarks: {len(args.datasets)} -> {", ".join(args.datasets)}') print(f'Benchmarks: {len(args.datasets)} -> {", ".join(args.datasets)}')
print(f'Samples: {samples}') print(f'Samples: {samples}')
print(f'Concurrency: {_conc} | Thinking: {_think_txt} | ' print(f'Concurrency: {args.concurrency} | Thinking: '
f'{"enabled" if not args.disable_thinking else "disabled"} | '
f'Performance: {"on" if args.perf else "off"}') f'Performance: {"on" if args.perf else "off"}')
print(f'Resume: {"on" if args.resume else "off"} | Output: {args.out_dir or "(none)"}') print(f'Resume: {"on" if args.resume else "off"} | Output: {args.out_dir or "(none)"}')
return return
@ -316,11 +271,8 @@ def _print_run_plan(console, args, model_spec):
table.add_row('Model', model_name) table.add_row('Model', model_name)
table.add_row('Benchmarks', f'{len(args.datasets)} · {", ".join(args.datasets)}') table.add_row('Benchmarks', f'{len(args.datasets)} · {", ".join(args.datasets)}')
table.add_row('Samples', samples) table.add_row('Samples', samples)
table.add_row('Concurrency', f'[magenta]{_conc}[/magenta]' if _auto else _conc) table.add_row('Concurrency', str(args.concurrency))
table.add_row('Thinking', table.add_row('Thinking', '[red]disabled[/red]' if args.disable_thinking else '[green]enabled[/green]')
'[red]disabled[/red]' if getattr(args, 'disable_thinking', False)
else (f'[green]enabled · effort={_eff}[/green]' if _eff
else '[green]enabled[/green]'))
table.add_row('Performance', '[green]enabled[/green]' if args.perf else '[dim]disabled[/dim]') table.add_row('Performance', '[green]enabled[/green]' if args.perf else '[dim]disabled[/dim]')
table.add_row('Checkpoint', '[green]resume[/green]' if args.resume else '[dim]new run[/dim]') table.add_row('Checkpoint', '[green]resume[/green]' if args.resume else '[dim]new run[/dim]')
table.add_row('Output', args.out_dir or '[dim](not specified)[/dim]') table.add_row('Output', args.out_dir or '[dim](not specified)[/dim]')
@ -669,55 +621,6 @@ def _cmd_eval_run(args) -> int:
rows = [] rows = []
all_reports = [] all_reports = []
def _flush_summaries():
"""Rewrite summary.csv/xlsx from the benches finished SO FAR.
Called after every benchmark instead of only at the very end:
multi-hour runs (mmlu_pro 12k samples) leave the summary stale for
hours otherwise -- and a crashed run would leave the LAST run's
corpses in place instead of partial results."""
if not out_dir:
return
if all_reports:
try:
from evalharness.viz import render as _render
_render(all_reports, style='excel', out=f'{out_dir}/summary.xlsx')
except Exception as e:
print(f'excel export skipped: {type(e).__name__}: {str(e)[:80]}',
file=sys.stderr)
if not rows:
return
import csv as _csv
with open(f'{out_dir}/summary.csv', 'w', newline='', encoding='utf-8') as f:
w = _csv.writer(f)
w.writerow(['benchmark', 'category', 'score', 'metric', 'num_samples',
'time_h', 'time_s', 'extract_fail',
'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s',
'categories'])
for r in rows:
perf = (r.get('groups') or {}).get('perf') or {}
cats = '; '.join(f'{g}={_f3(v)}'
for gname, gv in (r.get('groups') or {}).items()
if gname != 'perf'
for g, v in (gv or {}).items()
if isinstance(v, (int, float)))[:2000]
w.writerow([r['name'], r.get('category', ''), _f3(r.get('value')),
r['metric'], r.get('n', ''),
r.get('hours', ''), r.get('secs', ''),
r.get('extract_fail', 0)] +
[perf.get(k, '') for k in (
'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] +
[cats])
run_started = _time.time() run_started = _time.time()
total_runs = len(args.datasets) total_runs = len(args.datasets)
model_spec = _compose_model_spec(args) model_spec = _compose_model_spec(args)
@ -726,7 +629,7 @@ def _cmd_eval_run(args) -> int:
# adapts on its own signals (see pool.AdaptiveGate) # adapts on its own signals (see pool.AdaptiveGate)
from evalharness.model.pool import AdaptiveGate from evalharness.model.pool import AdaptiveGate
AdaptiveGate.INITIAL = float(max(1, getattr(args, 'concurrency', 1))) AdaptiveGate.INITIAL = float(max(2, getattr(args, 'concurrency', 8)))
if not out_dir and model_spec and not args.out: if not out_dir and model_spec and not args.out:
# always persist results: default dir = evalharness-results/<stamp>-<model>/ # always persist results: default dir = evalharness-results/<stamp>-<model>/
import re as _re import re as _re
@ -777,40 +680,35 @@ def _cmd_eval_run(args) -> int:
_rep_secs = 0.0 _rep_secs = 0.0
_rep_tin = _rep_tout = 0 _rep_tin = _rep_tout = 0
# 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
_pname = getattr(args, 'progress_plugin', 'rich') \
if args.progress is True else args.progress
try:
_P = PROGRESS_REGISTRY.get(_pname)
except KeyError:
raise SystemExit(f'unknown progress plugin {_pname!r}; '
f'available: {", ".join(PROGRESS_REGISTRY.names())}')
if _P is not None:
# share ONE console: phase lines printed by another
# writer during the live bar interleave incorrectly.
# One reporter for the WHOLE run: overall bar (which
# benchmark) + sample bar (which sample), reused per
# benchmark via reset_samples().
if _shared_reporter is None and (
_pname == 'plain' or console.is_terminal):
# live bars only on a real terminal: through pipes
# (| grep, > log) rich's refresh thread misbehaves
# and stalls the run -- plain phases instead
_shared_reporter = _P(console=console)
_shared_reporter.owned_externally = True
progress_reporter = _shared_reporter
if getattr(args, 'reasoning_effort', ''):
args._gen_override = {**(getattr(args, '_gen_override', {}) or {}),
'reasoning_effort': args.reasoning_effort}
if model_spec: # generate + score in one go if model_spec: # generate + score in one go
from evalharness.model import run_eval from evalharness.model import run_eval
progress_reporter = None
if args.progress:
from evalharness.progress import PROGRESS_REGISTRY
_pname = getattr(args, 'progress_plugin', 'rich') \
if args.progress is True else args.progress
try:
_P = PROGRESS_REGISTRY.get(_pname)
except KeyError:
raise SystemExit(f'unknown progress plugin {_pname!r}; '
f'available: {", ".join(PROGRESS_REGISTRY.names())}')
if _P is not None:
# share ONE console: phase lines printed by another
# writer during the live bar interleave incorrectly.
# One reporter for the WHOLE run: overall bar (which
# benchmark) + sample bar (which sample), reused per
# benchmark via reset_samples().
if _shared_reporter is None and (
_pname == 'plain' or console.is_terminal):
# live bars only on a real terminal: through pipes
# (| grep, > log) rich's refresh thread misbehaves
# and stalls the run -- plain phases instead
_shared_reporter = _P(console=console)
_shared_reporter.owned_externally = True
progress_reporter = _shared_reporter
def status_callback(msg, _idx=i + 1, _name=name, def status_callback(msg, _idx=i + 1, _name=name,
_reporter=progress_reporter, _reporter=progress_reporter,
@ -873,21 +771,11 @@ def _cmd_eval_run(args) -> int:
progress_reporter=progress_reporter, progress_reporter=progress_reporter,
status_callback=status_callback, status_callback=status_callback,
on_scored=on_scored, on_scored=on_scored,
rescore=getattr(args, 'rescore', False),
repeat=_rep + 1)) repeat=_rep + 1))
_m = next((v for k, v in report.metrics.items() _m = next((v for k, v in report.metrics.items()
if k != 'extraction_failure_rate'), None) if k != 'extraction_failure_rate'), None)
if _m is not None: if _m is not None:
_scores.append(_m) _scores.append(_m)
if _repeats > 1 and out_dir:
# per-run report: <out-dir>/<bench>/reps/repNN.report.jsonl
# (each carries THAT run's own score + samples; the
# top-level report.jsonl stays the mean-summary view)
import pathlib as _pl
_rdir = _pl.Path(out_dir) / name / 'reps'
_rdir.mkdir(parents=True, exist_ok=True)
report.save(str(_rdir / f'rep{_rep + 1:02d}.report.jsonl'))
_rep_info = report.metric_groups.get('run_info', {}) or {} _rep_info = report.metric_groups.get('run_info', {}) or {}
_rep_secs += sum(float((s.usage or {}).get('latency_s', 0) or 0) _rep_secs += sum(float((s.usage or {}).get('latency_s', 0) or 0)
for s in report.samples) for s in report.samples)
@ -954,13 +842,15 @@ def _cmd_eval_run(args) -> int:
print(render(report, style=args.style)) print(render(report, style=args.style))
all_reports.append(report) all_reports.append(report)
primary = next(iter(report.metrics), '') primary = next(iter(report.metrics), '')
secs_total = sum(float((s.usage or {}).get('latency_s', 0) or 0)
for s in report.samples)
groups = {k: v for k, v in report.metric_groups.items() groups = {k: v for k, v in report.metric_groups.items()
if isinstance(v, dict) and k not in ('run_info',) if isinstance(v, dict) and k not in ('run_info',)
and not k.startswith('agg_error')} and not k.startswith('agg_error')}
info = report.metric_groups.get('run_info', {}) or {} info = report.metric_groups.get('run_info', {}) or {}
if _repeats > 1: if _repeats > 1 and _rep_secs:
# repeats: tokens are the SUM over all runs (real cost of the # repeats: report the SUM over all runs, not the last one
# predictions used) secs_total = _rep_secs
info = {**info, 'gen_input_tokens': _rep_tin, info = {**info, 'gen_input_tokens': _rep_tin,
'gen_output_tokens': _rep_tout, 'gen_output_tokens': _rep_tout,
'gen_total_tokens': _rep_tin + _rep_tout} 'gen_total_tokens': _rep_tin + _rep_tout}
@ -971,24 +861,12 @@ def _cmd_eval_run(args) -> int:
return lats[min(int(len(lats) * q), len(lats) - 1)] if lats else 0.0 return lats[min(int(len(lats) * q), len(lats) - 1)] if lats else 0.0
fins = [(s.usage or {}).get('finish_reason', '') fins = [(s.usage or {}).get('finish_reason', '')
for s in report.samples] for s in report.samples]
# time = THIS run's wall clock everywhere: summing per-prediction rows.append({'name': name, 'metric': primary, 'value': report.metrics.get(primary),
# latency_s counts RESTORED predictions' original generation time 'n': report.num_samples,
# (days old, slower setup) -- that once reported 15.9h for a
# one-hour run
_wall = round(_time.time() - t0, 1)
_cat = bench_category(name)
report.metric_groups.setdefault('run_info', {})['category'] = _cat
rows.append({'name': name, 'metric': primary, 'category': _cat,
'cached': (report.metric_groups.get('run_info', {})
.get('gen_fresh') == 0) if _repeats <= 1 else False,
'value': report.metrics.get(primary),
# repeats evaluate the SAME N problems k times: the
# count people expect is the generations, not N
'n': report.num_samples * _repeats,
'extract_fail': report.num_failed_extractions, 'extract_fail': report.num_failed_extractions,
'secs': _wall, 'secs': round(secs_total, 1),
'wall': _wall, 'wall': round(_time.time() - t0, 1),
'hours': round(_wall / 3600, 2), 'hours': round(secs_total / 3600, 2),
'tok_in': info.get('gen_input_tokens', 0) or 0, 'tok_in': info.get('gen_input_tokens', 0) or 0,
'tok_out': info.get('gen_output_tokens', 0) or 0, 'tok_out': info.get('gen_output_tokens', 0) or 0,
'tokens': (info.get('gen_input_tokens', 0) or 0) 'tokens': (info.get('gen_input_tokens', 0) or 0)
@ -1006,15 +884,9 @@ def _cmd_eval_run(args) -> int:
'done', _time.time() - t0) 'done', _time.time() - t0)
except Exception as e: except Exception as e:
rows.append({'name': name, 'metric': '-', 'value': None, rows.append({'name': name, 'metric': '-', 'value': None,
'category': bench_category(name),
'secs': round(_time.time() - t0, 1), 'ok': False, 'secs': round(_time.time() - t0, 1), 'ok': False,
'err': f'{type(e).__name__}: {str(e)[:100]}'}) 'err': f'{type(e).__name__}: {str(e)[:100]}'})
print(f'{name}: FAILED {type(e).__name__}: {str(e)[:160]}', file=sys.stderr) print(f'{name}: FAILED {type(e).__name__}: {str(e)[:160]}', file=sys.stderr)
import traceback as _tb
_tb.print_exc() # full stack to stderr: the one-line form hides
# WHERE an error like 'Object of type ellipsis is not JSON
# serializable' actually comes from
from evalharness.hooks import fire as _fire from evalharness.hooks import fire as _fire
_fire('on_benchmark_failed', name=name, error=e, dataset=name) _fire('on_benchmark_failed', name=name, error=e, dataset=name)
@ -1027,17 +899,49 @@ def _cmd_eval_run(args) -> int:
border_style='red', expand=False), justify='center') border_style='red', expand=False), justify='center')
_print_benchmark_result(console, i + 1, total_runs, name, _print_benchmark_result(console, i + 1, total_runs, name,
'failed', _time.time() - t0) 'failed', _time.time() - t0)
# incremental summary: partial results are visible (and survive a
# crash) after EVERY benchmark, not only when the whole run ends
_flush_summaries()
if _shared_reporter is not None: if _shared_reporter is not None:
_shared_reporter.close() _shared_reporter.close()
if all_reports and out_dir: if all_reports and out_dir:
_flush_summaries() # final state (benches already flushed per-bench) try:
print(f'excel -> {out_dir}/summary.xlsx', flush=True) from evalharness.viz import render as _render
xb = _render(all_reports, style='excel',
out=f'{out_dir}/summary.xlsx')
print(f'excel -> {xb}', flush=True)
except Exception as e:
print(f'excel export skipped: {type(e).__name__}: {str(e)[:80]}',
file=sys.stderr)
if rows: if rows:
_print_summary_table(console, rows) _print_summary_table(console, rows)
if out_dir:
import csv as _csv
with open(f'{out_dir}/summary.csv', 'w', newline='', encoding='utf-8') as f:
w = _csv.writer(f)
w.writerow(['benchmark', 'score', 'metric', 'num_samples',
'time_h', 'time_s', 'extract_fail',
'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s',
'categories'])
for r in rows:
perf = (r.get('groups') or {}).get('perf') or {}
cats = '; '.join(f'{g}={_f3(v)}'
for gname, gv in (r.get('groups') or {}).items()
if gname != 'perf'
for g, v in (gv or {}).items()
if isinstance(v, (int, float)))[:2000]
w.writerow([r['name'], _f3(r.get('value')), r['metric'], r.get('n', ''),
r.get('hours', ''), r.get('secs', ''),
r.get('extract_fail', 0)] +
[perf.get(k, '') for k in (
'success_rate', 'latency_mean_s', 'output_tps', 'request_qps',
'input_tokens_mean', 'output_tokens_mean', 'total_tokens',
'ttft_mean_s', 'ttft_p90_s', 'ttft_p99_s',
'tpot_mean_s', 'tpot_p90_s', 'tpot_p99_s')] +
[cats])
# artifacts notice: tell the user where everything landed (or how to save); # artifacts notice: tell the user where everything landed (or how to save);
# rich terminals get clickable file:// links (iTerm2/kitty/WezTerm/WT...) # rich terminals get clickable file:// links (iTerm2/kitty/WezTerm/WT...)
@ -1076,8 +980,7 @@ def _print_summary_table(console, rows):
t = Table(title='Run Summary', header_style='bold cyan', t = Table(title='Run Summary', header_style='bold cyan',
title_style='bold', expand=False) title_style='bold', expand=False)
for col, just in (('benchmark', 'left'), ('category', 'left'), for col, just in (('benchmark', 'left'), ('metric', 'left'),
('metric', 'left'),
('score', 'right'), ('n', 'right'), ('time', 'right'), ('score', 'right'), ('n', 'right'), ('time', 'right'),
('tok in', 'right'), ('tok out', 'right'), ('tok in', 'right'), ('tok out', 'right'),
('in/s', 'right'), ('out/s', 'right')): ('in/s', 'right'), ('out/s', 'right')):
@ -1085,15 +988,13 @@ def _print_summary_table(console, rows):
for r in rows: for r in rows:
v = _fmt_score(r.get('value')) if r['ok'] else 'ERR' v = _fmt_score(r.get('value')) if r['ok'] else 'ERR'
wall = r.get('wall') or r.get('secs') or 0 wall = r.get('wall') or r.get('secs') or 0
tm = 'cached' if r.get('cached') else ( tm = f'{wall / 3600:.2f}h' if wall >= 3600 else f'{wall:.0f}s'
f'{wall / 3600:.2f}h' if wall >= 3600 else f'{wall:.0f}s')
ti, to = r.get('tok_in', 0), r.get('tok_out', 0) ti, to = r.get('tok_in', 0), r.get('tok_out', 0)
tin = f'{ti:,}' if ti else '—' tin = f'{ti:,}' if ti else '—'
tout = f'{to:,}' if to else '—' tout = f'{to:,}' if to else '—'
tis = f'{ti / wall:.0f}' if (wall > 1 and ti) else '—' tis = f'{ti / wall:.0f}' if (wall > 1 and ti) else '—'
tos = f'{to / wall:.0f}' if (wall > 1 and to) else '—' tos = f'{to / wall:.0f}' if (wall > 1 and to) else '—'
t.add_row(r['name'], r.get('category', ''), r['metric'], v, t.add_row(r['name'], r['metric'], v, str(r.get('n', '')), tm,
str(r.get('n', '')), tm,
tin, tout, tis, tos, tin, tout, tis, tos,
style='green' if r['ok'] else 'red') style='green' if r['ok'] else 'red')
wall_all = sum(r.get('wall') or r.get('secs') or 0 for r in rows) wall_all = sum(r.get('wall') or r.get('secs') or 0 for r in rows)
@ -1104,7 +1005,7 @@ def _print_summary_table(console, rows):
t.add_section() t.add_section()
tis_all = f'{ti_all / wall_all:.0f}' if wall_all > 1 else '' tis_all = f'{ti_all / wall_all:.0f}' if wall_all > 1 else ''
tos_all = f'{to_all / wall_all:.0f}' if wall_all > 1 else '' tos_all = f'{to_all / wall_all:.0f}' if wall_all > 1 else ''
t.add_row(f'[bold]{len(rows)} benchmarks[/bold]', '', '', t.add_row(f'[bold]{len(rows)} benchmarks[/bold]', '',
f'{sum(1 for r in rows if r["ok"])}/{len(rows)} ok', f'{sum(1 for r in rows if r["ok"])}/{len(rows)} ok',
str(n_all), tm_all, f'{ti_all:,}', f'{to_all:,}', str(n_all), tm_all, f'{ti_all:,}', f'{to_all:,}',
tis_all, tos_all) tis_all, tos_all)
@ -1194,6 +1095,12 @@ def build_parser() -> argparse.ArgumentParser:
p.add_argument('--provider', default='openai-chat', p.add_argument('--provider', default='openai-chat',
choices=('openai-chat', 'openai-pool'), choices=('openai-chat', 'openai-pool'),
help='API protocol/provider (default: openai-chat)') help='API protocol/provider (default: openai-chat)')
p.add_argument('--auto-concurrency', action='store_true',
help='let the per-endpoint adaptive gate decide concurrency '
'(ramps while healthy, backs off x0.7 on failures, '
'server /metrics when available); current limit shows '
'on the progress bar as "gate N". --concurrency '
'becomes the starting point, not a cap')
p.add_argument('--judge-model', '--judge', dest='judge', default='', p.add_argument('--judge-model', '--judge', dest='judge', default='',
help='judge model name with --judge-api-url, or full spec') help='judge model name with --judge-api-url, or full spec')
p.add_argument('--judge-api-url', default='', p.add_argument('--judge-api-url', default='',
@ -1215,28 +1122,12 @@ def build_parser() -> argparse.ArgumentParser:
"profile.default < profile['<bench>'] < explicit kwargs") "profile.default < profile['<bench>'] < explicit kwargs")
p.add_argument('--disable-thinking', action='store_true', p.add_argument('--disable-thinking', action='store_true',
help='send enable_thinking=false to the OpenAI-compatible model') help='send enable_thinking=false to the OpenAI-compatible model')
p.add_argument('--thinking', default='',
choices=('', 'off', 'low', 'medium', 'high', 'max', 'full'),
help='unified thinking switch: off = --disable-thinking; '
'low/medium/high/max = reasoning_effort (verified on '
'this endpoint: low = ~1/9 tokens); full = default '
'thinking. Takes precedence over --disable-thinking '
'and --reasoning-effort')
p.add_argument('--reasoning-effort', default='',
choices=('', 'minimal', 'low', 'medium', 'high', 'max'),
help="thinking intensity (GLM/Anthropic-style; verified "
"working on the sglang endpoint: low shrinks reasoning "
"~85%%). Overrides the YAML config")
p.add_argument('--perf', action='store_true', p.add_argument('--perf', action='store_true',
help='collect streaming TTFT and ITL metrics') help='collect streaming TTFT and ITL metrics')
p.add_argument('--textools', action='store_true', p.add_argument('--textools', action='store_true',
help='send tools as text instead of native tool calls') help='send tools as text instead of native tool calls')
p.add_argument('--env', default='', help="agent environment (e.g. 'bfcl_mock') -> message pump") p.add_argument('--env', default='', help="agent environment (e.g. 'bfcl_mock') -> message pump")
p.add_argument('--concurrency', default='32', p.add_argument('--concurrency', type=int, default=32, help='parallel model calls (default 32)')
help="parallel model calls (default 32); 'auto' = the "
"adaptive gate decides (starts at 2, ramps +1 while "
"healthy, backs off x0.7 on failures -- see 'gate N' "
"on the progress bar)")
p.add_argument('--progress', action='store_true', default=True, p.add_argument('--progress', action='store_true', default=True,
help='show per-sample progress (default: on)') help='show per-sample progress (default: on)')
p.add_argument('--no-progress', dest='progress', action='store_false', p.add_argument('--no-progress', dest='progress', action='store_false',
@ -1250,9 +1141,6 @@ def build_parser() -> argparse.ArgumentParser:
p.add_argument('--resume', nargs='?', const=True, default=False, p.add_argument('--resume', nargs='?', const=True, default=False,
help='resume from per-sample checkpoint (default path auto-derived; ' help='resume from per-sample checkpoint (default path auto-derived; '
'pass a path to override)') 'pass a path to override)')
p.add_argument('--rescore', action='store_true',
help='force re-scoring even when a matching saved report '
'could be reused (change of recipe/judge, or paranoia)')
p.add_argument('--limit-per-task', type=int, p.add_argument('--limit-per-task', type=int,
help='first N samples PER subset/category (evalscope --limit semantics); ' help='first N samples PER subset/category (evalscope --limit semantics); '
'composable with --limit (intersection)') 'composable with --limit (intersection)')
@ -1301,21 +1189,6 @@ def build_parser() -> argparse.ArgumentParser:
def main(argv=None) -> int: def main(argv=None) -> int:
args = build_parser().parse_args(argv) args = build_parser().parse_args(argv)
# '--concurrency auto' == '--auto-concurrency': normalize once, here,
# so every downstream site (plan display, run_eval, gate initial) sees
# an int + the flag
_th = str(getattr(args, 'thinking', '') or '').strip().lower()
if _th:
if _th == 'off':
args.disable_thinking = True
else:
args.disable_thinking = False
args.reasoning_effort = '' if _th == 'full' else _th
if str(getattr(args, 'concurrency', '32')).strip().lower() == 'auto':
args.auto_concurrency = True
args.concurrency = 1 # gate starts at 1: probe x2, bisect to capacity
else:
args.concurrency = int(args.concurrency)
return args.func(args) return args.func(args)

2
evalharness/config/default.yaml
View File

@ -23,7 +23,6 @@ imo_answerbench:
temperature: 1.0 temperature: 1.0
gpqa_diamond: gpqa_diamond:
temperature: 1.0 temperature: 1.0
repeats: 3
max_tokens: 8192 max_tokens: 8192
mmlu: mmlu:
max_tokens: 8192 max_tokens: 8192
@ -43,7 +42,6 @@ trivia_qa:
max_tokens: 8192 max_tokens: 8192
humaneval: humaneval:
temperature: 1.0 temperature: 1.0
repeats: 3
live_code_bench: live_code_bench:
temperature: 1.0 temperature: 1.0
longbench_v2: longbench_v2:

84
evalharness/config/effort_low.yaml
View File

@ -1,84 +0,0 @@
# 低思考档：全库参数同 default.yaml，仅追加 reasoning_effort: low
# 用法: evalharness eval run ... --config effort_low
default:
temperature: 0.0
top_p: 1.0
stream: true
max_tokens: 32768
reasoning_effort: low
aime24:
temperature: 1.0
repeats: 12
max_tokens: 8192
reasoning_effort: low
aime25:
temperature: 1.0
repeats: 12
max_tokens: 8192
reasoning_effort: low
aime26:
temperature: 1.0
repeats: 12
max_tokens: 8192
reasoning_effort: low
hmmt26:
temperature: 1.0
repeats: 12
max_tokens: 8192
reasoning_effort: low
imo_answerbench:
temperature: 1.0
reasoning_effort: low
gpqa_diamond:
temperature: 1.0
repeats: 3
max_tokens: 8192
reasoning_effort: low
mmlu:
max_tokens: 8192
reasoning_effort: low
mmlu_pro:
max_tokens: 8192
reasoning_effort: low
cmmlu:
max_tokens: 8192
reasoning_effort: low
arc:
max_tokens: 8192
reasoning_effort: low
hellaswag:
max_tokens: 8192
reasoning_effort: low
winogrande:
max_tokens: 8192
reasoning_effort: low
simple_qa:
max_tokens: 8192
reasoning_effort: low
trivia_qa:
max_tokens: 8192
reasoning_effort: low
humaneval:
temperature: 1.0
repeats: 3
reasoning_effort: low
live_code_bench:
temperature: 1.0
reasoning_effort: low
longbench_v2:
max_tokens: 8192
max_input_tokens: 128000
reasoning_effort: low
openai_mrcr:
max_tokens: 8192
max_input_tokens: 128000
reasoning_effort: low
bfcl_v3:
max_tokens: 4096
reasoning_effort: low
general_fc:
max_tokens: 4096
reasoning_effort: low
tau2_bench:
max_tokens: 16384
reasoning_effort: low

29
evalharness/config/sample_counts.json
View File

@ -1,29 +0,0 @@
{
"humaneval": 164,
"aime24": 30,
"aime25": 30,
"aime26": 30,
"hmmt26": 30,
"gpqa_diamond": 198,
"mmlu": 285,
"mmlu_pro": 12032,
"cmmlu": 11528,
"gsm8k": 1319,
"arc": 2376,
"hellaswag": 10042,
"winogrande": 1267,
"drop": 9535,
"longbench_v2": 503,
"live_code_bench": 1055,
"bigcodebench": 1140,
"trivia_qa": 17944,
"simple_qa": 4326,
"hle": 2500,
"imo_answerbench": 89,
"openai_mrcr": 1300,
"bfcl_v3": 2000,
"general_fc": 400,
"tau2_bench": 184,
"competition_math": 5000,
"swe_bench_verified": 500
}

60
evalharness/config/thinking.yaml
View File

@ -1,60 +0,0 @@
default:
temperature: 0.0
top_p: 1.0
stream: true
max_tokens: 200336
aime24:
temperature: 1.0
repeats: 12
max_tokens: 200336
aime25:
temperature: 1.0
repeats: 12
max_tokens: 200336
aime26:
temperature: 1.0
repeats: 12
max_tokens: 200336
hmmt26:
temperature: 1.0
repeats: 12
max_tokens: 200336
imo_answerbench:
temperature: 1.0
gpqa_diamond:
temperature: 1.0
repeats: 3
max_tokens: 200336
mmlu:
max_tokens: 200336
mmlu_pro:
max_tokens: 200336
cmmlu:
max_tokens: 200336
arc:
max_tokens: 200336
hellaswag:
max_tokens: 200336
winogrande:
max_tokens: 200336
simple_qa:
max_tokens: 200336
trivia_qa:
max_tokens: 200336
humaneval:
temperature: 1.0
repeats: 3
live_code_bench:
temperature: 1.0
longbench_v2:
max_tokens: 200336
max_input_tokens: 128000
openai_mrcr:
max_tokens: 200336
max_input_tokens: 128000
bfcl_v3:
max_tokens: 4096
general_fc:
max_tokens: 4096
tau2_bench:
max_tokens: 16384

41
evalharness/eval/checkpoint.py
View File

@ -28,7 +28,6 @@ class CheckpointStore:
self.model = model self.model = model
self.dataset = dataset self.dataset = dataset
self._entries: Dict[str, Dict[str, Any]] = {} self._entries: Dict[str, Dict[str, Any]] = {}
self._scores: Dict[str, Dict[str, Any]] = {}
self._fh = None self._fh = None
@staticmethod @staticmethod
@ -48,7 +47,6 @@ class CheckpointStore:
def load(self) -> Dict[str, Dict[str, Any]]: def load(self) -> Dict[str, Dict[str, Any]]:
"""Read all checkpointed predictions (idempotent).""" """Read all checkpointed predictions (idempotent)."""
self._entries = {} self._entries = {}
self._scores = {}
if not self.path.exists(): if not self.path.exists():
return self._entries return self._entries
with open(self.path, encoding='utf-8') as f: with open(self.path, encoding='utf-8') as f:
@ -59,8 +57,6 @@ class CheckpointStore:
try: try:
rec = json.loads(line) rec = json.loads(line)
self._entries[rec['key']] = rec.get('pred', {}) self._entries[rec['key']] = rec.get('pred', {})
if rec.get('score'):
self._scores[rec['key']] = rec['score']
except (ValueError, KeyError): except (ValueError, KeyError):
continue # torn tail line from a crash -- safe to skip continue # torn tail line from a crash -- safe to skip
return self._entries return self._entries
@ -70,44 +66,9 @@ class CheckpointStore:
worst case loses the last in-flight sample on crash).""" worst case loses the last in-flight sample on crash)."""
rec = {'key': key, 'ts': time.time(), 'pred': pred} rec = {'key': key, 'ts': time.time(), 'pred': pred}
with open(self.path, 'a', encoding='utf-8') as f: with open(self.path, 'a', encoding='utf-8') as f:
f.write(json.dumps(rec, ensure_ascii=False, default=str) + '\n') f.write(json.dumps(rec, ensure_ascii=False) + '\n')
self._entries[key] = pred 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: def __len__(self) -> int:
return len(self._entries) return len(self._entries)

2
evalharness/eval/recipes/agent.py
View File

@ -56,7 +56,7 @@ def bigcodebench():
extract='code_any', extract='code_any',
scorers={'pass': {'name': 'execution', 'harness': _bcb_harness, scorers={'pass': {'name': 'execution', 'harness': _bcb_harness,
# official sandbox image (bundles every task's deps) # official sandbox image (bundles every task's deps)
'image': 'bigcodebench/bigcodebench-evaluate:latest', # official hub image, same as evalscope 'image': 'bigcodebench-sandbox:latest',
'sandbox': 'docker', 'timeout_s': 120}}, 'sandbox': 'docker', 'timeout_s': 120}},
aggregators={'pass': 'pass_at_k'}, aggregators={'pass': 'pass_at_k'},
exec_workers=12, exec_workers=12,

16
evalharness/eval/record.py
View File

@ -56,10 +56,7 @@ class EvalReport(BaseModel):
num_samples: int = 0 num_samples: int = 0
num_failed_extractions: int = 0 num_failed_extractions: int = 0
metrics: Dict[str, float] = Field(default_factory=dict) # {'acc': 0.62} metrics: Dict[str, float] = Field(default_factory=dict) # {'acc': 0.62}
# values may be None (perf stats the adapter couldn't measure), lists metric_groups: Dict[str, Dict[str, float]] = Field(default_factory=dict)
# (repeats.scores) or nested dicts -- a strict float type rejected the
# file on LOAD and silently defeated report reuse
metric_groups: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
# {'by_category': {'algebra': 0.7, ...}, 'pass_at_k': {'pass@1': .., 'pass@8': ..}, # {'by_category': {'algebra': 0.7, ...}, 'pass_at_k': {'pass@1': .., 'pass@8': ..},
# 'by_length_bin': {'8k': .., '32k': ..}} # 'by_length_bin': {'8k': .., '32k': ..}}
@ -68,25 +65,20 @@ class EvalReport(BaseModel):
def save(self, path) -> None: def save(self, path) -> None:
import json import json
# default=str everywhere: score_details/samples carry arbitrary
# scorer output, and ONE exotic object (an Ellipsis sneaked in via
# a scorer's detail dict) must not kill a finished benchmark at
# the save line
if str(path).endswith('.jsonl'): if str(path).endswith('.jsonl'):
# streaming format: first line = report header, then one # streaming format: first line = report header, then one
# sample per line (grep/split/tail friendly) # sample per line (grep/split/tail friendly)
head = self.model_dump(exclude={'samples'}) head = self.model_dump(exclude={'samples'})
head['type'] = 'report' head['type'] = 'report'
with open(path, 'w', encoding='utf-8') as f: with open(path, 'w', encoding='utf-8') as f:
f.write(json.dumps(head, ensure_ascii=False, default=str) + '\n') f.write(json.dumps(head, ensure_ascii=False) + '\n')
for smp in self.samples: for smp in self.samples:
row = smp if isinstance(smp, dict) else smp.model_dump() row = smp if isinstance(smp, dict) else smp.model_dump()
row['type'] = 'sample' row['type'] = 'sample'
f.write(json.dumps(row, ensure_ascii=False, default=str) + '\n') f.write(json.dumps(row, ensure_ascii=False) + '\n')
return return
with open(path, 'w', encoding='utf-8') as f: with open(path, 'w', encoding='utf-8') as f:
json.dump(self.model_dump(), f, ensure_ascii=False, indent=2, json.dump(self.model_dump(), f, ensure_ascii=False, indent=2)
default=str)
@classmethod @classmethod
def load(cls, path) -> 'EvalReport': def load(cls, path) -> 'EvalReport':

124
evalharness/eval/runner.py
View File

@ -51,29 +51,6 @@ def evaluate(
aggregators = recipe.resolve_aggregators() aggregators = recipe.resolve_aggregators()
ctx = ScoreContext(judge=judge, params={}) ctx = ScoreContext(judge=judge, params={})
# fail-fast image preflight: recipe-level AND sample-level images are
# ensured (local or pulled once) BEFORE any container runs -- a missing
# image must kill the bench in seconds with a fix hint, not produce a
# 0.0% after hours of per-sample pull failures
try:
_imgs = set()
for spec in (recipe.scorers or {}).values():
p = spec if isinstance(spec, dict) else {}
if p.get('name') == 'execution' and p.get('sandbox') == 'docker' and p.get('image'):
_imgs.add(p['image'])
for s in samples[:200]:
if getattr(s, 'sandbox', None) and s.sandbox.image:
_imgs.add(s.sandbox.image)
if _imgs:
from ..sandbox.docker import ensure_image
for _img in sorted(_imgs):
ensure_image(_img)
except RuntimeError:
raise
except Exception:
pass # no docker here (local sandbox): the scorer will complain
# If any scorer executes in docker with per-sample images, overlap pulls # If any scorer executes in docker with per-sample images, overlap pulls
# with scoring (run sample N while N+1..N+lookahead images download). # with scoring (run sample N while N+1..N+lookahead images download).
bp = None bp = None
@ -85,10 +62,10 @@ def evaluate(
results: List[SampleResult] = [] results: List[SampleResult] = []
def _shell(sample, pred) -> SampleResult: def judge_one(sample, pred) -> SampleResult:
"""SampleResult with everything derivable from (sample, prediction): """Extract + score ONE sample (thread-safe: everything here is local
identity, raw text, usage, trajectories. Shared by live scoring and except docker/subprocess execution, which parallelizes perfectly --
the checkpoint-score replay path.""" each sample gets its own container/workdir)."""
raw = pred if isinstance(pred, str) else str(pred.get('raw', '')) raw = pred if isinstance(pred, str) else str(pred.get('raw', ''))
override = {} if isinstance(pred, str) else pred override = {} if isinstance(pred, str) else pred
result = SampleResult( result = SampleResult(
@ -112,14 +89,6 @@ def evaluate(
result.env_state = pred['env_state'] result.env_state = pred['env_state']
if isinstance(pred, dict) and pred.get('usage'): if isinstance(pred, dict) and pred.get('usage'):
result.usage = pred['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: try:
if bp is not None and sample.sandbox and sample.sandbox.image: if bp is not None and sample.sandbox and sample.sandbox.image:
bp.ensure(sample.sandbox.image) # wait only if this one still pulling bp.ensure(sample.sandbox.image) # wait only if this one still pulling
@ -196,91 +165,6 @@ def evaluate(
return report 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: def _needs_bg_prefetch(recipe, samples) -> bool:
"""True when the recipe executes in docker AND samples declare images.""" """True when the recipe executes in docker AND samples declare images."""
try: try:

52
evalharness/model/adapter.py
View File

@ -115,17 +115,6 @@ def _key_for(api_base: str) -> str:
def _payload_chars(payload: Dict[str, Any]) -> int:
"""Rough prompt size in characters (~3-4 chars/token). Used to decide
the auto-stream path: gateways that buffer whole requests make a huge
INPUT as hang-prone as a huge output budget."""
n = 0
for m in payload.get('messages') or []:
c = m.get('content')
n += len(c) if isinstance(c, str) else 256
return n
def _parse_text_tool_calls(text: str) -> list: def _parse_text_tool_calls(text: str) -> list:
"""Extract tool calls from a text reply. Handles both shapes: """Extract tool calls from a text reply. Handles both shapes:
- JSON array: [{"name":..,"arguments":{..}}] - JSON array: [{"name":..,"arguments":{..}}]
@ -175,12 +164,8 @@ def _parse_text_tool_calls(text: str) -> list:
args[kw_.arg] = _ast.unparse(kw_.value) args[kw_.arg] = _ast.unparse(kw_.value)
except SyntaxError: except SyntaxError:
return None return None
# literal_eval happily returns Ellipsis (code like `f(key=...)`) and
# other non-JSON constants; default=str keeps the serializer alive
# instead of killing the whole benchmark at parse time
return {'id': '', 'type': 'function', return {'id': '', 'type': 'function',
'function': {'name': name, 'function': {'name': name, 'arguments': json.dumps(args)}}
'arguments': json.dumps(args, default=str)}}
for m_ in _re.finditer(r'([A-Za-z_][A-Za-z0-9_]*)\((.*?)\)', text): for m_ in _re.finditer(r'([A-Za-z_][A-Za-z0-9_]*)\((.*?)\)', text):
c = _py_call(m_) c = _py_call(m_)
@ -220,15 +205,9 @@ class OpenAICompatible(ModelAdapter):
if stream: if stream:
out = await self._post_stream_perf( out = await self._post_stream_perf(
f'{self.api_base}/chat/completions', payload, headers, t0) f'{self.api_base}/chat/completions', payload, headers, t0)
elif (int(payload.get('max_tokens') or 0) > 100000 elif int(payload.get('max_tokens') or 0) > 100000 \
or _payload_chars(payload) > 300_000) \
and not os.environ.get('EVALHARNESS_NO_AUTOSTREAM'): and not os.environ.get('EVALHARNESS_NO_AUTOSTREAM'):
# long generation OR LONG INPUT: stream and aggregate # long generation: stream and aggregate (gateway-safe).
# (gateway-safe). Some gateways buffer the whole request
# before answering on the non-stream path -- a 128k-token
# longbench_v2 prompt sat there past every read timeout;
# streaming starts emitting immediately, so a stuck
# endpoint surfaces in ~60s instead of after 20 minutes.
# Some gateways drop chat_template_kwargs on the STREAM # Some gateways drop chat_template_kwargs on the STREAM
# path only (non-stream honors it) -- append the /no_think # path only (non-stream honors it) -- append the /no_think
# soft switch into the prompt itself as a belt-and-braces # soft switch into the prompt itself as a belt-and-braces
@ -243,10 +222,10 @@ class OpenAICompatible(ModelAdapter):
break break
data = await self._post_stream_aggregate( data = await self._post_stream_aggregate(
f'{self.api_base}/chat/completions', payload, headers) f'{self.api_base}/chat/completions', payload, headers)
out = self._parse(data, allow_text_calls=bool(tools)) out = self._parse(data)
else: else:
data = await self._post(f'{self.api_base}/chat/completions', payload, headers) data = await self._post(f'{self.api_base}/chat/completions', payload, headers)
out = self._parse(data, allow_text_calls=bool(tools)) out = self._parse(data)
out.usage.latency_s = round(_time.time() - t0, 3) out.usage.latency_s = round(_time.time() - t0, 3)
out.usage.retries = attempt out.usage.retries = attempt
return out return out
@ -354,7 +333,7 @@ class OpenAICompatible(ModelAdapter):
'usage': (usage_ev or {}).get('usage') or {}, 'usage': (usage_ev or {}).get('usage') or {},
'model': self.model, 'model': self.model,
} }
out = self._parse(data, allow_text_calls=True) out = self._parse(data)
out.usage.ttft_s = round(ttft, 3) if ttft is not None else None out.usage.ttft_s = round(ttft, 3) if ttft is not None else None
out.usage.itl_mean_s = round(sum(itl_vals) / len(itl_vals), 4) if itl_vals else None out.usage.itl_mean_s = round(sum(itl_vals) / len(itl_vals), 4) if itl_vals else None
out.usage.http_status = status out.usage.http_status = status
@ -380,8 +359,7 @@ class OpenAICompatible(ModelAdapter):
] ]
payload.pop('chat_template_kwargs', None) payload.pop('chat_template_kwargs', None)
for k in ('temperature', 'max_tokens', 'top_p', 'stop', 'seed', for k in ('temperature', 'max_tokens', 'top_p', 'stop', 'seed',
'response_format', 'chat_template_kwargs', 'response_format', 'chat_template_kwargs'):
'thinking', 'reasoning_effort'):
if kw.get(k) is not None: if kw.get(k) is not None:
payload[k] = kw[k] payload[k] = kw[k]
payload.setdefault('max_tokens', self.extra.get('max_tokens', 4096)) # CoT room payload.setdefault('max_tokens', self.extra.get('max_tokens', 4096)) # CoT room
@ -401,17 +379,12 @@ class OpenAICompatible(ModelAdapter):
payload['chat_template_kwargs'] = {'enable_thinking': False} payload['chat_template_kwargs'] = {'enable_thinking': False}
return payload return payload
def _parse(self, data: Dict[str, Any], def _parse(self, data: Dict[str, Any]) -> ModelOutput:
allow_text_calls: bool = True) -> ModelOutput:
choice = (data.get('choices') or [{}])[0] choice = (data.get('choices') or [{}])[0]
msg = choice.get('message') or {} msg = choice.get('message') or {}
calls = [] calls = []
raw_calls = list(msg.get('tool_calls') or []) raw_calls = list(msg.get('tool_calls') or [])
if not raw_calls and allow_text_calls: if not raw_calls:
# text-protocol fallback ONLY for requests that carried tools:
# running it on plain prose/code (humaneval!) regex-matched
# `f(key=...)` style code as "calls", literal_eval'd the `...`
# into an Ellipsis and crashed json.dumps mid-generation
raw_calls = _parse_text_tool_calls(msg.get('content') or '') raw_calls = _parse_text_tool_calls(msg.get('content') or '')
for c in raw_calls: for c in raw_calls:
fn = c.get('function') or {} fn = c.get('function') or {}
@ -451,14 +424,9 @@ class OpenAICompatible(ModelAdapter):
import httpx as _hx import httpx as _hx
# read timeout bounds the wait for the FIRST byte too: a 128k-token
# prompt queued behind other prefills legitimately takes minutes to
# start answering -- 60s killed exactly those (the gate's x0.7 and
# sample-failure containment now handle real hangs)
async with _hx.AsyncClient(timeout=_hx.Timeout( async with _hx.AsyncClient(timeout=_hx.Timeout(
connect=self.extra.get('connect_timeout', 15), connect=self.extra.get('connect_timeout', 15),
read=self.extra.get('stream_read_timeout', 240), read=60, write=30, pool=15)) as client:
write=30, pool=15)) as client:
async with client.stream('POST', url, json=payload, headers=headers) as resp: async with client.stream('POST', url, json=payload, headers=headers) as resp:
if resp.status_code != 200: if resp.status_code != 200:
body = (await resp.aread()).decode('utf-8', 'replace')[:300] body = (await resp.aread()).decode('utf-8', 'replace')[:300]

224
evalharness/model/pool.py
View File

@ -151,25 +151,10 @@ class AdaptiveGate:
Purely additive to PooledAdapter: one gate per backend, no caller change. Purely additive to PooledAdapter: one gate per backend, no caller change.
""" """
LO = 1 # never go below: progress beats perfection LO = 2 # never go below: progress beats perfection
HI = 96 # sane ceiling for one endpoint HI = 96 # sane ceiling for one endpoint
PROBE_S = 5.0 # safety tick (fails/hang detection); ramp decisions use PROBE_S = 5.0 # metrics probe interval
# level statistics, not this interval alone INITIAL = 8.0 # class-level start point (--auto-concurrency rebinds it)
INITIAL = 2.0 # class-level start point ('--concurrency auto' rebinds it)
# ---- exponential-probe + binary-search capacity discovery ----
# probe: 1 -> 2 -> 4 -> ... while throughput keeps IMPROVING (>10%);
# the first level where it plateaus opens a bisect [last_good, bad];
# bisect narrows to the knee; steady holds there. Any failure x0.7s
# immediately and restarts probing from the shrunken level.
GAIN_EPS = 1.1 # rate must beat the previous level by 10% to keep doubling
MIN_OK = 5 # baseline completions needed at a level before judging
# dwell fallback: JUDGE also when the level has been held this long (with
# >= 1 completion) -- scaled by observed inter-completion gap so a bench
# whose single request takes 60s is not judged on one lone sample at t=25s
DWELL_BASE_S = 120.0
# robust judging: sample count scales WITH the level (a 2-completion
# estimate at level 8 is pure quantization noise), plus a minimum dwell
# so one lucky tick cannot speak for the whole level
def __init__(self, adapter: ModelAdapter): def __init__(self, adapter: ModelAdapter):
self.adapter = adapter self.adapter = adapter
@ -181,26 +166,8 @@ class AdaptiveGate:
self._loop = None # loop the cond/probe-task are bound to self._loop = None # loop the cond/probe-task are bound to
self._contended = 0 # acquire() waits this probe interval (demand) self._contended = 0 # acquire() waits this probe interval (demand)
self._interval_fails = 0 # failed releases this probe interval self._interval_fails = 0 # failed releases this probe interval
self._interval_ok = 0 # SUCCESSFUL releases this probe interval:
# zero completions = zero throughput, and a
# gate that ramps on demand alone would pile
# 96 concurrent prefills onto a server whose
# first 2 requests have not even answered
# capacity-discovery state
self._mode = 'probe' # probe | bisect | steady
self._level_t0 = None # when we arrived at the current limit
self._level_ok = 0 # completions at this level
self._prev = (None, None) # (level, rate) we came from / last-good
self._bis = (None, None) # bisect bounds (lo=good, hi=bad)
self._good_rate = 0.0 # throughput at the good bound (baseline)
self._fetch_exec = None # DEDICATED executor for /metrics probes:
# the shared default pool is occupied by
# second-long tokenization jobs, and the
# probe queued behind them never ticked
# (gate frozen at 1 while results flowed)
self._metrics_dead = False # 3 consecutive fetch failures -> stop asking
self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0, self.stats = {'probe': 0, 'ramp': 0, 'hold_queue': 0, 'backoff_fail': 0,
'backoff_queue': 0, 'ramp_demand': 0, 'bisect': 0} 'backoff_queue': 0, 'ramp_demand': 0}
def _push_limit(self) -> None: def _push_limit(self) -> None:
"""Surface the current limit to the progress bar ('gate N').""" """Surface the current limit to the progress bar ('gate N')."""
@ -212,18 +179,6 @@ class AdaptiveGate:
except Exception: except Exception:
pass pass
def push_inflight(self) -> None:
"""Tell the bar how many requests the gate has ACTUALLY admitted
(bar shows 'admitted/held' -- a bare held count with gate 2 read
as 'the gate is broken')."""
rep = (self.adapter.extra or {}).get('progress_reporter')
fn = getattr(rep, 'set_admitted', None)
if fn is not None:
try:
fn(self._inflight)
except Exception:
pass
# ---- gate semantics ---- # ---- gate semantics ----
async def acquire(self) -> None: async def acquire(self) -> None:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
@ -251,40 +206,18 @@ class AdaptiveGate:
finally: finally:
self._cond.release() self._cond.release()
self._inflight += 1 self._inflight += 1
self.push_inflight()
def release(self, ok: bool) -> None: def release(self, ok: bool) -> None:
self._inflight = max(0, self._inflight - 1) self._inflight = max(0, self._inflight - 1)
self.push_inflight() if not ok: # multiplicative decrease -- survival first
if ok:
self._interval_ok += 1
self._level_ok += 1
now = time.monotonic()
last = getattr(self, '_last_ok_t', None)
if last is not None:
gap = now - last
ge = getattr(self, '_gap_ema', None)
self._gap_ema = gap if ge is None else 0.6 * ge + 0.4 * gap
self._last_ok_t = now
else: # multiplicative decrease -- survival first
self._interval_fails += 1 self._interval_fails += 1
# capacity moved (or we overshot): shrink now and restart the
# discovery from the shrunken level
before = self.limit before = self.limit
self.limit = max(self.LO, self.limit * 0.7) self.limit = max(self.LO, self.limit * 0.7)
if before != self.limit: if before != self.limit:
self.stats['backoff_fail'] += 1 self.stats['backoff_fail'] += 1
self._push_limit() self._push_limit()
self._enter_level(mode='probe')
self._wake() self._wake()
def _enter_level(self, mode: str = '') -> None:
"""Arrive at (a new) limit: start measuring this level fresh."""
if mode:
self._mode = mode
self._level_t0 = time.monotonic()
self._level_ok = 0
def _wake(self) -> None: def _wake(self) -> None:
if self._cond is not None: if self._cond is not None:
# fire-and-forget notify (loop may not be ours -- best effort) # fire-and-forget notify (loop may not be ours -- best effort)
@ -300,121 +233,21 @@ class AdaptiveGate:
# ---- server-signal probe ---- # ---- server-signal probe ----
def _no_signal_ramp(self) -> None: def _no_signal_ramp(self) -> None:
"""No server signals (no /metrics, 404, gateway stripped it): """No server signals available (no /metrics, 404/HTTPError, gateway
discover capacity by measuring THROUGHPUT per concurrency level. stripped it, non-sglang backend): fall back to demand-driven AIMD --
ramp while the cap is the binding constraint (callers had to WAIT on
probe: double while completions/s keeps improving (rate > prev x acquire) and the interval was failure-free. Failures still cut x0.7
1.1) -- 1, 2, 4, 8 ... reaches the knee in log time per release, so a drowning backend shrinks the gate immediately."""
bisect: first level where the gain stalls opens [last_good, bad]; if self._interval_fails == 0 and self._contended > 0 \
narrow to the knee with midpoint measurements and int(self.limit) < self.HI:
steady: hold at the converged level; any failure x0.7s (handled in self.limit = min(self.HI, self.limit + 1)
release) and probing restarts from the shrunken level self.stats['ramp_demand'] += 1
self._push_limit()
A level is judged only after MIN_OK completions or MAX_AT_LEVEL_S; self._interval_fails = 0
zero completions so far = hold (hang protection).""" self._contended = 0
try: self._wake()
now = time.monotonic()
if self._level_t0 is None:
self._enter_level()
dt = now - self._level_t0
lvl = max(1, int(self.limit))
# not enough evidence yet at this level: keep measuring.
# evidence = 2 x level completions, no other floor
need_ok = max(2, lvl * 2)
# dwell fallback scales with the OBSERVED completion cadence:
# a 60s-per-request bench needs minutes, not 25s, before a
# single-sample judgment is acceptable
need_dt = max(self.DWELL_BASE_S,
3.0 * (getattr(self, '_gap_ema', None) or 0.0))
if self._level_ok < need_ok and dt < need_dt:
return
# zero completions so far: hang or overloaded -> hold
if self._level_ok == 0:
return
rate = self._level_ok / max(dt, 1e-6)
prev_lvl, prev_rate = self._prev
self._prev = (lvl, rate)
if self._mode == 'probe':
self.stats['probe'] += 1
# continue while NOT WORSE (>= 0.9x): with heterogeneous
# request lengths (lbv2: 4k..2M-token docs) completion-rate
# noise dwarfs a 10% gain threshold, and demanding strict
# improvement bisected [1,2]->1 on the first plateau.
# Only CLEAR degradation (<0.9x) means past the knee.
ok = prev_rate is None or rate >= prev_rate * 0.9
if ok and lvl < self.HI:
self._bis = (lvl, min(lvl * 2, self.HI)) # remember bounds
self.limit = float(min(lvl * 2, self.HI))
self.stats['ramp_demand'] += 1
self._push_limit()
self._enter_level()
elif not ok:
# throughput CLEARLY degraded: knee is in (prev_lvl, lvl]
self._mode = 'bisect'
self._bis = (prev_lvl or max(1, lvl // 2), lvl)
self._good_rate = prev_rate or rate
lo, hi = self._bis
mid = (lo + hi) // 2
if hi - lo <= 1:
self.limit = float(lo) # prev_lvl was the knee
self._push_limit()
self._enter_level('steady')
else:
self.limit = float(mid)
self.stats['bisect'] += 1
self._push_limit()
self._enter_level()
else:
self._enter_level('steady') # hit HI with gains: stay
elif self._mode == 'steady':
# the endpoint is SHARED: other tenants change its capacity
# while we run -- keep judging forever, nudge +-1 against the
# reference rate instead of pinning the converged level
ref = self._good_rate or rate
if rate >= ref * 1.05 and lvl < self.HI:
self.limit = float(min(lvl + 1, self.HI))
self.stats['ramp_demand'] += 1
self._push_limit()
self._enter_level()
elif rate <= ref * 0.85 and lvl > self.LO:
self.limit = float(max(self.LO, lvl - 1))
self.stats['backoff_queue'] += 1
self._push_limit()
self._enter_level()
else:
# reference drifts with fresh measurements (slow EWMA)
self._good_rate = 0.7 * ref + 0.3 * rate
self._enter_level() # restart the measurement window
elif self._mode == 'bisect':
lo, hi = self._bis
if rate >= self._good_rate * 0.9:
lo = lvl # not worse here: knee is at/above
else:
hi = lvl # clearly worse: knee is below
self._bis = (lo, hi)
if hi - lo <= 1:
self.limit = float(lo)
self._push_limit()
self._enter_level('steady')
else:
mid = (lo + hi) // 2
self.limit = float(mid)
self.stats['bisect'] += 1
self._push_limit()
self._enter_level()
finally:
self._interval_fails = 0
self._interval_ok = 0
self._contended = 0
self._wake()
async def _probe_once(self) -> None: async def _probe_once(self) -> None:
if self._metrics_dead:
self._no_signal_ramp() # endpoint said no thrice: stop asking
return
import urllib.request import urllib.request
url = f'{self.adapter.api_base.rstrip("/")}/metrics' url = f'{self.adapter.api_base.rstrip("/")}/metrics'
@ -424,23 +257,12 @@ class AdaptiveGate:
return resp.read().decode('utf-8', 'ignore') return resp.read().decode('utf-8', 'ignore')
try: try:
# DEDICATED single-thread executor: the shared asyncio pool is # thread: the blocking fetch must never stall the event loop
# full of second-long truncation tokenizations, and a probe # (an unreachable host parks urlopen for the full 4s timeout)
# queued behind them never ran (gate appeared frozen) text = await asyncio.to_thread(_fetch)
if self._fetch_exec is None:
import concurrent.futures
self._fetch_exec = concurrent.futures.ThreadPoolExecutor(
max_workers=1, thread_name_prefix='gate-probe')
loop = asyncio.get_running_loop()
text = await loop.run_in_executor(self._fetch_exec, _fetch)
except Exception: except Exception:
self._fetch_fails = getattr(self, '_fetch_fails', 0) + 1
if self._fetch_fails >= 3:
self._metrics_dead = True # 404/unreachable: pure demand mode
self._no_signal_ramp() # no metrics: demand-driven fallback self._no_signal_ramp() # no metrics: demand-driven fallback
return return
self._fetch_fails = 0
running = queue = None running = queue = None
for line in text.splitlines(): for line in text.splitlines():
if line.startswith('sglang:num_running_reqs'): if line.startswith('sglang:num_running_reqs'):
@ -456,10 +278,8 @@ class AdaptiveGate:
self.limit = max(self.LO, self.limit - 1) self.limit = max(self.LO, self.limit - 1)
self.stats['backoff_queue'] += 1 self.stats['backoff_queue'] += 1
self._push_limit() self._push_limit()
elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))) \ elif (queue or 0) == 0 and (running is None or running < max(2, int(self.limit))):
and self._interval_ok > 0:
# underfed: no queue and running below our own cap -> ramp up # underfed: no queue and running below our own cap -> ramp up
# (still requires completions this interval: no throughput, no ramp)
self.limit = min(self.HI, self.limit + 1) self.limit = min(self.HI, self.limit + 1)
self.stats['ramp'] += 1 self.stats['ramp'] += 1
self._push_limit() self._push_limit()

132
evalharness/model/runner.py
View File

@ -317,25 +317,7 @@ async def generate_predictions(
_progress(progress, done_count, len(samples), t0, total_usage) _progress(progress, done_count, len(samples), t0, total_usage)
return pred return pred
# assemble() tokenizes for the max_input_tokens truncation -- on messages = ([ChatMessage(role='user', content=assemble(sample))]
# long-context benches that is SECONDS of CPU per sample (2M-token
# docs). Two failure modes fixed here:
# - inline: froze the whole event loop behind one encode
# - asyncio.to_thread (32-thread default pool): dozens of concurrent
# tokenizers hogged the GIL and starved the progress renderer +
# loop itself (bar froze, then jumped)
# A DEDICATED BOUNDED pool: 8 encodes at a time, remaining workers
# queue -- GIL pressure capped, everything stays responsive.
global _ASSEMBLE_EXEC
if _ASSEMBLE_EXEC is None:
import concurrent.futures
_ASSEMBLE_EXEC = concurrent.futures.ThreadPoolExecutor(
max_workers=8, thread_name_prefix='assemble')
text = await asyncio.get_running_loop().run_in_executor(
_ASSEMBLE_EXEC, assemble, sample) \
if isinstance(sample.input, str) else None
messages = ([ChatMessage(role='user', content=text)]
if isinstance(sample.input, str) else list(sample.input)) if isinstance(sample.input, str) else list(sample.input))
if not system and extra_system[0] and isinstance(sample.input, str): if not system and extra_system[0] and isinstance(sample.input, str):
# renderer-provided SYSTEM contract (es lcb expert-programmer) # renderer-provided SYSTEM contract (es lcb expert-programmer)
@ -438,25 +420,21 @@ async def generate_predictions(
for m_ in members: for m_ in members:
m_.extra['progress_reporter'] = progress_reporter m_.extra['progress_reporter'] = progress_reporter
# terminal (post-retry) sample failures are CONTAINED: one sample that
# never makes it (server queue ate its first byte past every timeout)
# must not kill the remaining hundreds -- it becomes an empty prediction
# (scores as wrong, es-parity for timeouts), is NOT checkpointed (a
# rerun retries it), and only a total wipeout fails the bench
failed_samples: Dict[int, str] = {}
async def run_one(i_s): async def run_one(i_s):
i, s = i_s i, s = i_s
# NO outer retry: the adapter retries internally (and the pool # transient network flaps (cluster routes re-converge): retry with
# fails over per instance); a second loop here multiplied # backoff so one ConnectError burst cannot kill the whole batch --
# worst-case time. One pass, one result or one contained error. # the adapter already retries 5xx/429 and the pool fails over per
# instance; this is the last line of defense around asyncio.gather
# NO outer retry: the adapter retries internally; a second loop
# here multiplied worst-case time (42+ attempts before this fix).
# One pass, one result or one error.
try: try:
pred = await one(s) pred = await one(s)
except Exception as e: except Exception:
if progress_reporter is not None: if progress_reporter is not None:
progress_reporter.advance(success=False) progress_reporter.advance(success=False)
failed_samples[i] = f'{type(e).__name__}: {str(e)[:120]}' raise
return i, None # empty marker: no checkpoint write
if ckpt_store is not None: if ckpt_store is not None:
ckpt_store.append(keys[i], pred) ckpt_store.append(keys[i], pred)
return i, pred return i, pred
@ -468,20 +446,8 @@ async def generate_predictions(
else: else:
status_callback('Generation skipped: the checkpoint already covers every sample') status_callback('Generation skipped: the checkpoint already covers every sample')
fresh = await asyncio.gather(*(run_one((i, s)) for i, s in pending)) fresh = await asyncio.gather(*(run_one((i, s)) for i, s in pending))
if failed_samples and len(failed_samples) >= len(pending):
# every single fresh sample died: the endpoint is down, not flaky
_f = next(iter(failed_samples.values()))
raise RuntimeError(f'all {len(failed_samples)} generations failed '
f'(first: {_f})')
if failed_samples:
print(f'generation: {len(failed_samples)}/{len(pending)} samples '
'failed after all retries (empty predictions, not '
'checkpointed -- rerun to retry them); first: '
f'{next(iter(failed_samples.items()))[1][:100]}', flush=True)
for i, pred in fresh: for i, pred in fresh:
preds_by_key[keys[i]] = pred if pred is not None \ preds_by_key[keys[i]] = pred
else {'raw': '', 'usage': {},
'error': failed_samples.get(i, '')[:200]}
preds = [preds_by_key[k] for k in keys] preds = [preds_by_key[k] for k in keys]
usages = [p.get('usage', {}) for p in preds] usages = [p.get('usage', {}) for p in preds]
# include RESTORED predictions' usage (they carry it in the ckpt); # include RESTORED predictions' usage (they carry it in the ckpt);
@ -500,10 +466,7 @@ async def generate_predictions(
latency_s=float(u.get('latency_s', 0) or 0)) latency_s=float(u.get('latency_s', 0) or 0))
if status_callback and pending: if status_callback and pending:
status_callback(f'Generation complete: {len(preds)} responses collected') status_callback(f'Generation complete: {len(preds)} responses collected')
# ckpt info (store + per-position keys) so run_eval can read/write return preds, usages, total_usage
# 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, len(fresh) - len(failed_samples)
finally: finally:
# reporter lifecycle belongs to the CALLER (CLI reuses one reporter # reporter lifecycle belongs to the CALLER (CLI reuses one reporter
# across benchmarks and closes it after the whole run); only close # across benchmarks and closes it after the whole run); only close
@ -566,7 +529,6 @@ def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) ->
_PROBED_SPECS = set() _PROBED_SPECS = set()
_ASSEMBLE_EXEC = None # bounded truncation pool (lazy)
async def _probe_model(adapter, model_spec: str) -> None: async def _probe_model(adapter, model_spec: str) -> None:
@ -668,7 +630,6 @@ async def run_eval(
gen_profile: str = '', gen_profile: str = '',
repeat: int = 1, repeat: int = 1,
on_scored=None, on_scored=None,
rescore: bool = False,
) -> EvalReport: ) -> EvalReport:
"""Generate + score in one call. Model spec examples: """Generate + score in one call. Model spec examples:
'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'. 'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'.
@ -781,7 +742,7 @@ async def run_eval(
try: try:
from .gen_profiles import merge_gen_kwargs from .gen_profiles import merge_gen_kwargs
preds, _usages, usage, ckpt_info, n_fresh = await generate_predictions( preds, _usages, usage = await generate_predictions(
adapter, list(raw_samples), concurrency, progress=progress, adapter, list(raw_samples), concurrency, progress=progress,
progress_reporter=progress_reporter, progress_reporter=progress_reporter,
status_callback=status_callback, status_callback=status_callback,
@ -801,27 +762,7 @@ async def run_eval(
repeat=repeat) repeat=repeat)
finally: finally:
await adapter.close() 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: if status_callback:
status_callback('loading judge model') status_callback('loading judge model')
judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key) judge_adapter = _make_adapter(judge_spec, api_key=judge_api_key or api_key)
@ -829,38 +770,19 @@ async def run_eval(
if status_callback: if status_callback:
status_callback('Scoring predictions against the benchmark recipe') status_callback('Scoring predictions against the benchmark recipe')
_meta = {'gen_input_tokens': usage.input_tokens, # scoring off the event loop: math_equal/sympy equivalence can chew a
'gen_output_tokens': usage.output_tokens, # single hard problem for minutes (es's checker famously hangs on one) --
'gen_total_tokens': usage.total_tokens, # running it inline froze the progress bar's clock for the whole bench
# fresh=0 means the whole bench replayed from checkpoint: the report = await asyncio.to_thread(
# summary table then shows 'cached' instead of a ~0s time evaluate,
'gen_fresh': n_fresh} samples, preds, recipe,
if _records is not None: model=model_spec,
from ..eval.runner import evaluate_cached judge=judge,
extra_metadata={'gen_input_tokens': usage.input_tokens,
report = evaluate_cached(samples, preds, recipe, _records, 'gen_output_tokens': usage.output_tokens,
model=model_spec, extra_metadata=_meta) 'gen_total_tokens': usage.total_tokens},
else: on_scored=on_scored,
# 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.model = model_spec
report.dataset = name report.dataset = name
if status_callback: if status_callback:

17
evalharness/model/truncation.py
View File

@ -9,34 +9,21 @@ Requires a tokenizer (transformers) at tokenizer_path or auto from the model.
""" """
import os import os
import threading
from functools import lru_cache from functools import lru_cache
from typing import Optional from typing import Optional
DEFAULT_TRUNCATION_TOKENS = 32768 * 4 # 131072, mirrors evalside run.py DEFAULT_TRUNCATION_TOKENS = 32768 * 4 # 131072, mirrors evalside run.py
_TOK_LOCK = threading.Lock()
@lru_cache(maxsize=4) @lru_cache(maxsize=4)
def _get_tokenizer(tokenizer_path: str): def _get_tokenizer(tokenizer_path: str):
if not tokenizer_path or not os.path.exists(tokenizer_path): if not tokenizer_path or not os.path.exists(tokenizer_path):
raise FileNotFoundError( raise FileNotFoundError(
f'tokenizer not found at {tokenizer_path!r} -- token-level truncation ' f'tokenizer not found at {tokenizer_path!r} -- token-level truncation '
'needs a local tokenizer dir (e.g. /data1/models/DeepSeek-V4-Flash-INT8)') 'needs a local tokenizer dir (e.g. /data1/models/DeepSeek-V4-Flash-INT8)')
# serialize the FIRST load: 96 worker threads racing transformers 5.x's from transformers import AutoTokenizer
# lazy imports raised ImportError and silently degraded batches to the
# char approximation; after one success lru_cache serves the rest
with _TOK_LOCK:
from transformers import AutoTokenizer
# the '> model_max_length' warnings are EXPECTED here -- counting a return AutoTokenizer.from_pretrained(tokenizer_path, trust_remote_code=True)
# 2M-token doc before trimming it is the whole point of truncation
import logging
logging.getLogger('transformers').setLevel(logging.ERROR)
return AutoTokenizer.from_pretrained(tokenizer_path, trust_remote_code=True)
def truncate_middle_tokens(text: str, max_tokens: int, tokenizer_path: str) -> str: def truncate_middle_tokens(text: str, max_tokens: int, tokenizer_path: str) -> str:

50
evalharness/progress/rich_terminal.py
View File

@ -19,21 +19,6 @@ def _fmt(sec):
return f'{sec // 60}m{sec % 60:02d}s' if sec >= 60 else f'{sec}s' return f'{sec // 60}m{sec % 60:02d}s' if sec >= 60 else f'{sec}s'
def _fmt_eta(sec):
"""eta: s -> m s -> h m s -> d h m (grows with the unit that matters)."""
sec = int(sec)
if sec < 60:
return f'{sec}s'
m, s = divmod(sec, 60)
if m < 60:
return f'{m}m{s:02d}s'
h, m = divmod(m, 60)
if h < 24:
return f'{h}h{m:02d}m{s:02d}s'
d, h = divmod(h, 24)
return f'{d}d{h:02d}h{m:02d}m'
class RichTerminalProgress: class RichTerminalProgress:
def __init__(self, console=None): def __init__(self, console=None):
# accept an EXTERNAL console: CLI phase messages and the live bar must # accept an EXTERNAL console: CLI phase messages and the live bar must
@ -75,7 +60,6 @@ class RichTerminalProgress:
self.started = 0.0 self.started = 0.0
self.current_started = 0.0 self.current_started = 0.0
self.inflight = 0 self.inflight = 0
self.admitted = None # requests past the pool gate (None = no gate)
self.restored = 0 # checkpoint head start (drives the '+N new' marker) self.restored = 0 # checkpoint head start (drives the '+N new' marker)
self.heartbeat_task = None self.heartbeat_task = None
@ -182,8 +166,8 @@ class RichTerminalProgress:
description=f'[green]{self.bench_tag}{self.bench_name} · scoring[/green]', description=f'[green]{self.bench_tag}{self.bench_name} · scoring[/green]',
total=total, completed=min(done, total), new='', total=total, completed=min(done, total), new='',
rate=f'{done / elapsed:.2f}', inflight=0, cur='0s', rate=f'{done / elapsed:.2f}', inflight=0, cur='0s',
elapsed=_fmt_eta(elapsed), elapsed=_fmt(elapsed),
eta=_fmt_eta((total - done) * elapsed / done) if done and total > done else '-') eta=_fmt((total - done) * elapsed / done) if done and total > done else '-')
def set_bench_tag(self, tag: str): def set_bench_tag(self, tag: str):
if self.disabled: if self.disabled:
@ -211,7 +195,7 @@ class RichTerminalProgress:
return return
self.inflight += 1 self.inflight += 1
self.current_started = time.monotonic() self.current_started = time.monotonic()
self.progress.update(self.task_id, inflight=self._inflight_txt(), cur='0s') self.progress.update(self.task_id, inflight=self.inflight, cur='0s')
def set_retries(self, n: int): def set_retries(self, n: int):
"""Show the retry count on the bar (from the adapter's attempt).""" """Show the retry count on the bar (from the adapter's attempt)."""
@ -223,23 +207,6 @@ class RichTerminalProgress:
if self.task_id is not None: if self.task_id is not None:
self.progress.update(self.task_id, gate=f'gate {n}') self.progress.update(self.task_id, gate=f'gate {n}')
def _inflight_txt(self) -> str:
# 'held' counts workers past the global semaphore; when a pool gate
# is active most of them are QUEUED on it -- 'admitted' is what
# actually hits the server. Showing a bare 96 with gate 2 read as
# 'the gate is not working'
if self.admitted is None:
return str(self.inflight)
# 'N gen · M wait': N admitted by the gate (real server load),
# M held by the pipeline (tokenizing or gate-queued)
return f'{self.admitted} gen · {max(0, self.inflight - self.admitted)} wait'
def set_admitted(self, n: int):
"""Pooled runs: requests actually admitted by the gate."""
self.admitted = n
if self.task_id is not None:
self.progress.update(self.task_id, inflight=self._inflight_txt())
def rollback(self): def rollback(self):
if self.disabled: if self.disabled:
return return
@ -247,7 +214,7 @@ class RichTerminalProgress:
just decrement the in-flight count, no success/fail bookkeeping.""" just decrement the in-flight count, no success/fail bookkeeping."""
self.inflight = max(0, self.inflight - 1) self.inflight = max(0, self.inflight - 1)
if self.task_id is not None: if self.task_id is not None:
self.progress.update(self.task_id, inflight=self._inflight_txt(), cur='0s') self.progress.update(self.task_id, inflight=self.inflight, cur='0s')
def advance(self, success: bool = True): def advance(self, success: bool = True):
if self.disabled: if self.disabled:
@ -266,9 +233,9 @@ class RichTerminalProgress:
advance=1, advance=1,
new=self._new_txt(completed), new=self._new_txt(completed),
rate=f"{fresh / elapsed:.2f}", rate=f"{fresh / elapsed:.2f}",
inflight=self._inflight_txt(), cur='0s', inflight=self.inflight, cur='0s',
elapsed=_fmt_eta(elapsed), elapsed=_fmt(elapsed),
eta=_fmt_eta((task.total - completed) * elapsed / fresh) eta=_fmt((task.total - completed) * elapsed / fresh)
if fresh and task.total and task.total > completed else '-', if fresh and task.total and task.total > completed else '-',
) )
@ -279,7 +246,8 @@ class RichTerminalProgress:
try: try:
while self.task_id is not None: while self.task_id is not None:
e = time.monotonic() - self.started e = time.monotonic() - self.started
upd = {'elapsed': _fmt_eta(e)} upd = {'elapsed': f'{int(e) // 60}m{int(e) % 60:02d}s' if e >= 60
else f'{int(e)}s'}
if self.inflight: if self.inflight:
secs = int(time.monotonic() - self.current_started) secs = int(time.monotonic() - self.current_started)
upd['cur'] = (f'{secs // 60}m{secs % 60:02d}s' upd['cur'] = (f'{secs // 60}m{secs % 60:02d}s'

34
evalharness/sandbox/docker.py
View File

@ -26,32 +26,6 @@ def docker_available() -> bool:
return _run(['docker', 'info']).returncode == 0 return _run(['docker', 'info']).returncode == 0
def ensure_image(img: str) -> None:
"""Fail-fast sandbox image preflight: present locally, else pull ONCE.
Without this, every sample's `docker run` tries its own pull at scoring
time -- a missing image burned 1140 x 3 retries x ~70s on bigcodebench
before anyone saw a 0.0%."""
if not img:
return
if _run(['docker', 'image', 'inspect', img]).returncode == 0:
return
# CN-mirror fallback chain (daocloud -> 1ms.run -> baidubce -> sjtu ->
# rat.dev), mirroring the SWE prefetch path; a hit is retagged to the
# canonical name so the recipe never knows which mirror answered
from .prefetch import _pull_one
try:
_pull_one(img)
return
except RuntimeError as e:
raise RuntimeError(
f'sandbox image {img!r} is not available: not local, and every '
f'mirror failed. {str(e)[:200]}. '
'Fix: pull/build it manually (for bigcodebench the official image '
'is bigcodebench/bigcodebench-evaluate:latest), then re-run with '
'--rescore to score the cached predictions.') from e
@register_sandbox('docker') @register_sandbox('docker')
class DockerSandbox(Sandbox): class DockerSandbox(Sandbox):
name = 'docker' name = 'docker'
@ -116,14 +90,6 @@ class DockerSandbox(Sandbox):
raise raise
if proc.returncode != 125 or attempt == 2: if proc.returncode != 125 or attempt == 2:
break break
# image-not-found is PERMANENT: retrying it 3x per sample
# burned 1140 x ~200s on a nonexistent bigcodebench image
_nf = 'Unable to find image' in (proc.stderr or '') \
or 'failed to resolve' in (proc.stderr or '') \
or 'manifest unknown' in (proc.stderr or '') \
or 'pull access denied' in (proc.stderr or '')
if _nf:
break
# clear any husk; fresh name next try. Bounded: an rm against # clear any husk; fresh name next try. Bounded: an rm against
# a bloated daemon can hang for minutes and silently eat the # a bloated daemon can hang for minutes and silently eat the
# whole worker pool (7 of 8 workers were observed stuck here) # whole worker pool (7 of 8 workers were observed stuck here)

1
evalharness/sandbox/prefetch.py
View File

@ -21,7 +21,6 @@ from ..data.dataset import Dataset
# fall through: daemon default -> 1ms.run -> baidubce -> sjtug. # fall through: daemon default -> 1ms.run -> baidubce -> sjtug.
_CN_MIRROR_FALLBACKS = [ _CN_MIRROR_FALLBACKS = [
'{img}', # daemon default (uses its own registry-mirrors config) '{img}', # daemon default (uses its own registry-mirrors config)
'docker.m.daocloud.io/{img}',
'docker.1ms.run/{img}', 'docker.1ms.run/{img}',
'mirror.baidubce.com/{img}', 'mirror.baidubce.com/{img}',
'docker.mirrors.sjtug.sjtu.edu.cn/{img}', 'docker.mirrors.sjtug.sjtu.edu.cn/{img}',