"""EvalHarness CLI. Zero third-party deps beyond the data layer (pydantic).""" import argparse import os import time import json import sys from concurrent.futures import ThreadPoolExecutor, as_completed def _overrides(args): """Optional DatasetSpec field overrides shared by fetch/stats/show.""" if getattr(args, 'hf_endpoint', None): os.environ['HF_ENDPOINT'] = args.hf_endpoint if getattr(args, 'cache_dir', None): from evalharness.data.dataset import set_cache_root set_cache_root(args.cache_dir) ov = {} for k in ('source', 'split', 'subset'): v = getattr(args, k, None) if v is not None: ov[k] = v return ov def _cmd_data_list(_args) -> int: from evalharness.data import list_datasets specs = list_datasets() if not specs: print('no datasets registered') return 0 name_w = max(len(s.name) for s in specs) type_w = max(len(s.task_type) for s in specs) for s in specs: print(f'{s.name:<{name_w}} {s.task_type:<{type_w}} {s.source} {s.description}') print(f'\n{len(specs)} dataset(s) registered') return 0 def _fetch_one(name: str, force: bool, overrides) -> str: from evalharness.data import get_dataset ds = get_dataset(name, **overrides) ds.materialize(force=force) origin = 'cache' if ds.lineage.get('from') == 'cache' else 'source' return f'{name}: {len(ds)} sample(s) [{origin}] -> {ds.cache_dir}' def _cmd_data_fetch(args) -> int: names = args.names if len(names) == 1: print(_fetch_one(names[0], args.force, _overrides(args))) return 0 # Concurrent prefetch: downloads are I/O-bound, threads suffice. # Per-dataset file locks inside materialize() guard shared cache entries. ok = True with ThreadPoolExecutor(max_workers=args.workers) as pool: futures = {pool.submit(_fetch_one, n, args.force, _overrides(args)): n for n in names} for fut in as_completed(futures): try: print(fut.result()) except Exception as e: # one failure must not block the rest ok = False print(f'{futures[fut]}: FAILED ({e})', file=sys.stderr) return 0 if ok else 1 def _cmd_data_stats(args) -> int: from evalharness.data import get_dataset stats = get_dataset(args.name, **_overrides(args)).stats() print(json.dumps(stats, ensure_ascii=False, indent=2)) return 0 def _cmd_data_show(args) -> int: from evalharness.data import get_dataset ds = get_dataset(args.name, **_overrides(args)) for s in ds[: args.n]: print(json.dumps(s.model_dump(), ensure_ascii=False, indent=2)) print('---') return 0 def _cmd_data_unload(args) -> int: from evalharness.data import get_dataset for name in args.names: ds = get_dataset(name, **_overrides(args)) removed = ds.unload() print(f'{name}: cache {"removed" if removed else "not present (nothing to do)"} -> {ds.cache_dir}') return 0 def _cmd_sandbox_prefetch(args) -> int: from evalharness.data import get_dataset from evalharness.sandbox import docker_available, images_for_dataset, prefetch_images if not docker_available(): print('docker is not available on this host', file=sys.stderr) return 1 ds = get_dataset(args.dataset, **_overrides(args)) images = images_for_dataset(ds, limit=args.limit) if not images: print(f'{args.dataset}: no sandbox images declared by its samples') return 0 prefetch_images(images, workers=args.workers) return 0 def _add_override_flags(p: argparse.ArgumentParser) -> None: p.add_argument('--hf-endpoint', default='', help='HuggingFace endpoint override, e.g. https://hf-mirror.com ' '(sets HF_ENDPOINT before any dataset download)') p.add_argument('--source', help='override DatasetSpec.source (e.g. a local dir)') p.add_argument('--split', help='override DatasetSpec.split') p.add_argument('--subset', help='override DatasetSpec.subset') p.add_argument('--cache-dir', help='cache root (default: $EVALHARNESS_CACHE or ~/.cache/evalharness)') def _cmd_eval_list(_args) -> int: from evalharness.eval import list_evals names = list_evals() print('\n'.join(names) if names else 'no eval recipes registered') print(f'\n{len(names)} eval recipe(s) registered') return 0 def _print_run_progress(done, total, name='', status='running', started=None): """Print one live progress line for a multi-benchmark run.""" import time width = 28 filled = int(width * done / max(total, 1)) bar = '#' * filled + '-' * (width - filled) elapsed = time.time() - started if started else 0 label = f'{done}/{total} [{bar}] {status}: {name}' print(f'\r{label} ({elapsed:.0f}s)', end='\n' if done >= total else '', flush=True) def _rich_console(): """Return a Rich console when installed; keep the CLI dependency-free.""" try: from rich.console import Console return Console() except ImportError: return None def _print_run_plan(console, args, model_spec): """Print the important run facts before any dataset work starts.""" title = 'EvalHarness · Run Plan' provider = getattr(args, 'provider', 'openai-chat') if model_spec else '—' api_url = getattr(args, 'api_url', '') or '—' model_name = getattr(args, 'model', '') or 'predictions file' # sampling summary: what the run will actually evaluate if getattr(args, 'limit', None): samples = f'up to {args.limit} total (--limit)' elif getattr(args, 'limit_per_task', None): samples = f'up to {args.limit_per_task} per subject (--limit-per-task)' else: samples = 'full dataset (counted when each loads)' if console is None: print(f'=== {title} ===') print(f'Provider: {provider}') print(f'API URL: {api_url}') print(f'Model: {model_name}') print(f'Benchmarks: {len(args.datasets)} -> {", ".join(args.datasets)}') print(f'Samples: {samples}') print(f'Concurrency: {args.concurrency} | Thinking: ' f'{"enabled" if not args.disable_thinking else "disabled"} | ' f'Performance: {"on" if args.perf else "off"}') print(f'Resume: {"on" if args.resume else "off"} | Output: {args.out_dir or "(none)"}') return from rich.panel import Panel from rich.table import Table table = Table(show_header=False, box=None, padding=(0, 1)) table.add_column('Item', style='cyan', no_wrap=True) table.add_column('Value', style='white') table.add_row('Provider', provider) table.add_row('API URL', api_url) table.add_row('Model', model_name) table.add_row('Benchmarks', f'{len(args.datasets)} · {", ".join(args.datasets)}') table.add_row('Samples', samples) table.add_row('Concurrency', str(args.concurrency)) table.add_row('Thinking', '[red]disabled[/red]' if args.disable_thinking else '[green]enabled[/green]') 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('Output', args.out_dir or '[dim](not specified)[/dim]') console.print(Panel(table, title=title, border_style='blue', expand=False), justify='center') def _narration(msg: str) -> str: """Icon + path-highlighting for narration lines (visual separation at a glance; paths in cyan).""" icon = '' m = msg.lower() if m.startswith('loading/'): icon = '⬇ ' elif 'dataset ready' in m or m.startswith('dataset ready'): icon = '📦 ' elif 'few-shot' in m: icon = '✳ ' elif 'checkpoint' in m or m.endswith('samples') or 'to generate' in m: icon = '◷ ' elif 'generation skipped' in m: icon = '⏭ ' elif 'generating' in m: icon = '🤖 ' elif 'generation complete' in m: icon = '✓ ' elif 'scoring' in m: icon = '★ ' elif 'writing' in m: icon = '📝 ' elif 'endpoint ok' in m: icon = '🔗 ' # THE key fact: the score in 'scoring complete · %' # gets bold green -- it is what the eye should find first import re as _re1 m = _re1.search(r'[:·] ([a-zA-Z_@]+ [0-9.]+%)(?=\s|$)', msg) if m: head = f'{icon}{msg[:m.start()]}: [bold green]{m.group(1)}[/bold green]' tail = msg[m.end():] if tail: import re as _re2 tail = _re2.sub(r'(?' or 'to' (only absolute paths) for sep in ('-> ', 'to '): head, _, tail = msg.rpartition(sep) if head and tail.startswith('/'): return f'{icon}{head}{sep}[cyan]{tail}[/cyan]' return f'{icon}{msg}' def _phase_color(message: str) -> str: """Narration line color. Currently PLAIN WHITE for everything (user preference); flip the returns to 'yellow'/'green' to restore the start/finish coloring scheme.""" return '' m = message.lower() # unreachable, kept for quick restore if any(k in m for k in ('complete', 'ready', 'downloaded', 'parsed', 'restored', 'ok (')): return 'green' return 'yellow' def _print_phase(console, index, total, name, message): # single-benchmark runs: the [1/1] tag is noise, drop it prefix = f'[{index}/{total}] ' if total > 1 else '' text = f'{prefix}{name}: {_narration(message)}' color = _phase_color(message) if console is not None: if color: console.print(f'[{color}]{text}[/{color}]', highlight=False) else: console.print(text, highlight=False) else: import re as _re0 print(_re0.sub(r'\[/?[a-z ]+\]', '', text), flush=True) def _print_benchmark_result(console, index, total, name, status, elapsed): if console is not None: color = 'green' if status == 'done' else 'red' icon = '✓' if status == 'done' else '✗' console.print(f'[{color}]{icon}[/{color}] benchmark {index}/{total} ' f'{name} · {status} · elapsed={elapsed:.0f}s') else: _print_run_progress(index, total, name, status, time.time() - elapsed) def _model_with_flags(model, args): """Translate explicit CLI flags to the adapter's internal options.""" for enabled, flag in ((getattr(args, 'disable_thinking', False), '!nothink'), (getattr(args, 'perf', False), '!perf'), (getattr(args, 'textools', False), '!textools')): if enabled and not model.endswith(flag): model += flag return model def _compose_model_spec(args): """Build the internal model spec from separate provider fields.""" model = args.model or '' api_url = getattr(args, 'api_url', '') or '' provider = getattr(args, 'provider', 'openai-chat') or 'openai-chat' # openai-chat is the public name; the current client implementation # remains registered as openai internally. internal_provider = {'openai-chat': 'openai', 'openai-pool': 'openai-pool'}.get(provider, provider) if api_url: if not model: raise SystemExit('error: --api-url requires --model (model name)') model = f'{internal_provider}/{api_url.rstrip("/")}?{model}' return _model_with_flags(model, args) def _print_result_panel(console, rep, wall_s: float = 0.0): """Rich result panel: metrics + throughput + health + top/bottom groups.""" if console is None: return False from rich.panel import Panel from rich.table import Table from rich.text import Text def color(v): return 'green' if v >= 0.8 else ('yellow' if v >= 0.6 else 'red') metrics = [(m, v) for m, v in rep.metrics.items() if isinstance(v, (int, float))] info = rep.metric_groups.get('run_info', {}) or {} model_lat = sum(float((s.usage or {}).get('latency_s', 0) or 0) for s in rep.samples) tok_in = info.get('gen_input_tokens', 0) or 0 tok_out = info.get('gen_output_tokens', 0) or 0 tput = rep.num_samples / wall_s if wall_s > 0.5 else 0.0 tps = tok_out / wall_s if (wall_s > 0.5 and tok_out) else 0.0 import re as _re m = str(rep.model or '?').split('?')[-1] m = _re.sub(r'![a-z]+$', '', m).strip('/') m = m.rsplit('/', 1)[-1] if '/' in m else m title = f'{rep.dataset} · {m}' body = Table(show_header=False, box=None, padding=(0, 2)) body.add_column('k', style='dim', no_wrap=True) body.add_column('v', overflow='fold') def score_bar(v, width=22): # rich-Bar-style glyphs with half-cell precision: ━━━╸╌╌ filled = max(0.0, min(1.0, v)) * width full = int(filled) half = '╸' if filled - full >= 0.5 else '' return '━' * full + half + '╌' * (width - full - len(half)) for m, v in metrics: if m == 'extraction_failure_rate': continue body.add_row(m, Text.assemble( (f'{v * 100:6.1f}% ', f'bold {color(v)}'), (score_bar(v), color(v)))) stats = [] stats.append(f'n={rep.num_samples}') if wall_s >= 1: stats.append(f'wall={wall_s:.0f}s') if model_lat >= 1: stats.append(f'model-time={model_lat:.0f}s') if tput: stats.append(f'{tput:.1f} samples/s') if tok_in or tok_out: stats.append(f'tokens in={tok_in:,} out={tok_out:,}') if tps: stats.append(f'{tps:.0f} tok/s out') if rep.num_failed_extractions: stats.append(f'[red]extract-fail={rep.num_failed_extractions}[/red]') body.add_row('run', ' '.join(stats)) for gname, groups in rep.metric_groups.items(): if gname in ('run_info', 'perf') or gname.startswith('agg_error'): continue numeric = {g: v for g, v in groups.items() if isinstance(v, (int, float))} if len(numeric) < 2: continue rank = sorted(numeric.items(), key=lambda kv: -kv[1]) show = [f'[green]{g} {v * 100:.0f}%[/green]' for g, v in rank[:3]] show += [f'[red]{g} {v * 100:.0f}%[/red]' for g, v in rank[-2:]] body.add_row(gname, ' '.join(show) + f' (+{len(numeric) - 5} more)' if len(numeric) > 5 else ' '.join(show)) perf = rep.metric_groups.get('perf') or {} if perf: keys = ('ttft_mean_s', 'ttft_p90_s', 'latency_mean_s', 'latency_p90_s', 'output_tps', 'success_rate', 'retry_rate') pstats = [f'{k}={perf[k]:.2f}' if isinstance(perf.get(k), float) and perf[k] < 10 else f'{k}={perf[k]}' for k in keys if perf.get(k) is not None] if pstats: body.add_row('perf', ' '.join(pstats)) console.print(Panel(body, title=title, border_style='blue', expand=False), justify='center') return True def _compose_judge_spec(args): """--judge accepts a bare model name (with --judge-api-url) or a full legacy spec; keep both working like the main model flags.""" judge = getattr(args, 'judge', '') or '' url = getattr(args, 'judge_api_url', '') or '' provider = getattr(args, 'judge_provider', 'openai-chat') or 'openai-chat' internal = {'openai-chat': 'openai', 'openai-pool': 'openai-pool'}.get(provider, provider) if url and judge and '/' not in judge: judge = f'{internal}/{url.rstrip("/")}?{judge}' return judge or None def _print_result_panel(console, rep, wall_s: float = 0.0): """Rich result panel: metrics + throughput + health + top/bottom groups.""" if console is None: return False from rich.panel import Panel from rich.table import Table from rich.text import Text def color(v): return 'green' if v >= 0.8 else ('yellow' if v >= 0.6 else 'red') metrics = [(m, v) for m, v in rep.metrics.items() if isinstance(v, (int, float))] info = rep.metric_groups.get('run_info', {}) or {} model_lat = sum(float((s.usage or {}).get('latency_s', 0) or 0) for s in rep.samples) tok_in = info.get('gen_input_tokens', 0) or 0 tok_out = info.get('gen_output_tokens', 0) or 0 tput = rep.num_samples / wall_s if wall_s > 0.5 else 0.0 tps = tok_out / wall_s if (wall_s > 0.5 and tok_out) else 0.0 import re as _re m = str(rep.model or '?').split('?')[-1] m = _re.sub(r'![a-z]+$', '', m).strip('/') m = m.rsplit('/', 1)[-1] if '/' in m else m title = f'{rep.dataset} · {m}' body = Table(show_header=False, box=None, padding=(0, 2)) body.add_column('k', style='dim', no_wrap=True) body.add_column('v', overflow='fold') def score_bar(v, width=22): # rich-Bar-style glyphs with half-cell precision: ━━━╸╌╌ filled = max(0.0, min(1.0, v)) * width full = int(filled) half = '╸' if filled - full >= 0.5 else '' return '━' * full + half + '╌' * (width - full - len(half)) for m, v in metrics: if m == 'extraction_failure_rate': continue body.add_row(m, Text.assemble( (f'{v * 100:6.1f}% ', f'bold {color(v)}'), (score_bar(v), color(v)))) stats = [] stats.append(f'n={rep.num_samples}') if wall_s >= 1: stats.append(f'wall={wall_s:.0f}s') if model_lat >= 1: stats.append(f'model-time={model_lat:.0f}s') if tput: stats.append(f'{tput:.1f} samples/s') if tok_in or tok_out: stats.append(f'tokens in={tok_in:,} out={tok_out:,}') if tps: stats.append(f'{tps:.0f} tok/s out') if rep.num_failed_extractions: stats.append(f'[red]extract-fail={rep.num_failed_extractions}[/red]') body.add_row('run', ' '.join(stats)) for gname, groups in rep.metric_groups.items(): if gname in ('run_info', 'perf') or gname.startswith('agg_error'): continue numeric = {g: v for g, v in groups.items() if isinstance(v, (int, float))} if len(numeric) < 2: continue rank = sorted(numeric.items(), key=lambda kv: -kv[1]) show = [f'[green]{g} {v * 100:.0f}%[/green]' for g, v in rank[:3]] show += [f'[red]{g} {v * 100:.0f}%[/red]' for g, v in rank[-2:]] body.add_row(gname, ' '.join(show) + f' (+{len(numeric) - 5} more)' if len(numeric) > 5 else ' '.join(show)) perf = rep.metric_groups.get('perf') or {} if perf: keys = ('ttft_mean_s', 'ttft_p90_s', 'latency_mean_s', 'latency_p90_s', 'output_tps', 'success_rate', 'retry_rate') pstats = [f'{k}={perf[k]:.2f}' if isinstance(perf.get(k), float) and perf[k] < 10 else f'{k}={perf[k]}' for k in keys if perf.get(k) is not None] if pstats: body.add_row('perf', ' '.join(pstats)) console.print(Panel(body, title=title, border_style='blue', expand=False), justify='center') return True def _compose_judge_spec(args): """--judge accepts a bare model name (with --judge-api-url) or a full legacy spec; both keep working, mirroring the main model flags.""" judge = getattr(args, 'judge', '') or '' url = getattr(args, 'judge_api_url', '') or '' provider = getattr(args, 'judge_provider', 'openai-chat') or 'openai-chat' internal = {'openai-chat': 'openai', 'openai-pool': 'openai-pool'}.get(provider, provider) if url and judge and '/' not in judge: judge = f'{internal}/{url.rstrip("/")}?{judge}' return judge or None def _cmd_eval_run(args) -> int: import asyncio import time as _time from evalharness.data import get_dataset from evalharness.viz import render comma = [d for d in getattr(args, 'datasets', []) if ',' in d] if comma: tip = ', '.join(comma) fixed = ' '.join(tip.split(',')) raise SystemExit( f'error: benchmark names must be SPACE-separated.\n' f' got: evalharness eval run {tip}\n' f' expected: evalharness eval run {fixed}') overrides = _overrides(args) out_dir = args.out_dir if out_dir: from pathlib import Path Path(out_dir).mkdir(parents=True, exist_ok=True) rows = [] all_reports = [] run_started = _time.time() total_runs = len(args.datasets) model_spec = _compose_model_spec(args) if not out_dir and model_spec and not args.out: # always persist results: default dir = evalharness-results/-/ import re as _re stamp = _time.strftime('%Y%m%d-%H%M%S') tag = _re.sub(r'[^A-Za-z0-9._-]+', '-', args.model or 'run')[:40].strip('-') out_dir = f'evalharness-results/{stamp}-{tag}' from pathlib import Path Path(out_dir).mkdir(parents=True, exist_ok=True) console = _rich_console() _print_run_plan(console, args, model_spec) _shared_reporter = None for i, name in enumerate(args.datasets): t0 = _time.time() try: if _shared_reporter is not None and total_runs > 1: _shared_reporter.set_bench_tag(f'[{i + 1}/{total_runs}]') def _emit(msg, _i=i, _n=name): if _shared_reporter is not None: _shared_reporter.log( f'[{_i + 1}/{total_runs}] {_n}: {_narration(msg)}') else: _print_phase(console, _i + 1, total_runs, _n, msg) _emit('Loading dataset (downloads on first use, cached afterwards)') ds = get_dataset(name, **overrides) if _shared_reporter is not None: _shared_reporter.pause() # let hub tqdm print cleanly sample_count = len(ds) if _shared_reporter is not None: _shared_reporter.resume() origin = ds.lineage.get('from', 'unknown') _emit(f'Dataset ready: {sample_count} samples from {origin}') if model_spec: # generate + score in one go from evalharness.model import run_eval progress_reporter = None if args.progress: from evalharness.progress import RichTerminalProgress if RichTerminalProgress 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 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 = RichTerminalProgress(console=console) _shared_reporter.owned_externally = True progress_reporter = _shared_reporter def status_callback(msg, _idx=i + 1, _name=name, _reporter=progress_reporter, _console=console): _tag = f'[{_idx}/{total_runs}] ' if total_runs > 1 else '' if _reporter is not None: _reporter.log(f'{_tag}{_name}: {_narration(msg)}') if 'scoring' in msg: _reporter.set_phase('scoring') elif 'generating model responses' in msg: _reporter.set_phase('generating') elif 'writing' in msg: _reporter.set_phase('writing') else: _print_phase(_console, _idx, total_runs, _name, msg) report = asyncio.run(run_eval( ds, model_spec, concurrency=args.concurrency, limit=args.limit, limit_per_task=args.limit_per_task, checkpoint=args.resume, judge_spec=_compose_judge_spec(args), env=args.env, api_key=getattr(args, 'api_key', ''), judge_api_key=getattr(args, 'judge_api_key', ''), gen_profile=getattr(args, 'profile', ''), progress_reporter=progress_reporter, status_callback=status_callback)) else: from evalharness.eval import evaluate if not args.predictions: raise SystemExit('error: provide --model or a predictions file') _print_phase(console, i + 1, total_runs, name, 'loading predictions') preds_path = args.predictions[i] if len(args.predictions) > i else args.predictions[0] preds = [json.loads(line) for line in open(preds_path, encoding='utf-8') if line.strip()] preds = [p.get('raw', p.get('prediction', '')) if isinstance(p, dict) else p for p in preds] report = evaluate(ds, preds, model=model_spec or 'preds') _print_phase(console, i + 1, total_runs, name, 'scoring complete') if args.out: report.save(args.out) if out_dir: _emit(f'Writing results to {out_dir}/{name}/' if out_dir else 'Writing results') from pathlib import Path as _P bench_dir = _P(out_dir) / name bench_dir.mkdir(parents=True, exist_ok=True) report.save(str(bench_dir / 'report.jsonl')) with open(bench_dir / 'detail.md', 'w', encoding='utf-8') as f: f.write(render(report, style='md')) if args.verbose: if not _print_result_panel(console, report, _time.time() - t0): print(render(report, style=args.style)) elif len(args.datasets) == 1: if console is None: print(render(report, style=args.style)) all_reports.append(report) 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() if isinstance(v, dict) and k not in ('run_info',) and not k.startswith('agg_error')} info = report.metric_groups.get('run_info', {}) or {} lats = sorted(float((s.usage or {}).get('latency_s', 0) or 0) for s in report.samples if float((s.usage or {}).get('latency_s', 0) or 0) > 0) def _pct(q): return lats[min(int(len(lats) * q), len(lats) - 1)] if lats else 0.0 fins = [(s.usage or {}).get('finish_reason', '') for s in report.samples] rows.append({'name': name, 'metric': primary, 'value': report.metrics.get(primary), 'n': report.num_samples, 'extract_fail': report.num_failed_extractions, 'secs': round(secs_total, 1), 'wall': round(_time.time() - t0, 1), 'hours': round(secs_total / 3600, 2), 'tok_in': info.get('gen_input_tokens', 0) or 0, 'tok_out': info.get('gen_output_tokens', 0) or 0, 'tokens': (info.get('gen_input_tokens', 0) or 0) + (info.get('gen_output_tokens', 0) or 0), 'lat_p50': _pct(0.50), 'lat_p90': _pct(0.90), 'trunc': sum(1 for f in fins if f == 'length'), 'groups': groups, 'ok': True}) if progress_reporter is not None: progress_reporter.advance_overall() _print_benchmark_result(console, i + 1, total_runs, name, 'done', _time.time() - t0) except Exception as e: rows.append({'name': name, 'metric': '-', 'value': None, 'secs': round(_time.time() - t0, 1), 'ok': False, 'err': f'{type(e).__name__}: {str(e)[:100]}'}) print(f'{name}: FAILED {type(e).__name__}: {str(e)[:160]}', file=sys.stderr) if console is not None: from rich.panel import Panel console.print(Panel( f'{type(e).__name__}: {e}', title=f'[bold red]✗ {name} FAILED[/bold red]', border_style='red', expand=False), justify='center') _print_benchmark_result(console, i + 1, total_runs, name, 'failed', _time.time() - t0) if _shared_reporter is not None: _shared_reporter.close() if all_reports and out_dir: try: 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: _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); # rich terminals get clickable file:// links (iTerm2/kitty/WezTerm/WT...) def _notice(label, *paths): print(f'\n{label} -> ' + ' · '.join(os.path.abspath(p) for p in paths)) ok_n = sum(1 for r in rows if r['ok']) if out_dir: ap = os.path.abspath(out_dir) mark = '[green]✓[/green]' if ok_n == len(rows) else '[yellow]◐[/yellow]' if console is not None: console.print( f'\n{mark} [bold]运行结束[/bold] · {ok_n}/{len(rows)} benchmarks ok' f' · 结果 {ap}') else: print(f'\n运行结束 · {ok_n}/{len(rows)} benchmarks ok · 结果 {ap}') elif rows and rows[0]['ok'] and args.out: _notice('运行结束 · 结果已保存', args.out) print(f'{ok_n}/{len(rows)} benchmarks ok') return 0 if all(r['ok'] for r in rows) else 1 def _fmt_score(v): """Unified score format: fractions render as percentages everywhere.""" try: v = float(v) except (TypeError, ValueError): return 'ERR' return f'{v * 100:.1f}%' if 0.0 <= v <= 1.0 else f'{v:g}' def _print_summary_table(console, rows): """Rich multi-benchmark summary (falls back to aligned plain text).""" if console is not None: from rich.table import Table t = Table(title='Run Summary', header_style='bold cyan', title_style='bold', expand=False) for col, just in (('benchmark', 'left'), ('metric', 'left'), ('score', 'right'), ('n', 'right'), ('time', 'right'), ('tok in', 'right'), ('tok out', 'right'), ('in/s', 'right'), ('out/s', 'right')): t.add_column(col, justify=just) for r in rows: v = _fmt_score(r.get('value')) if r['ok'] else 'ERR' wall = r.get('wall') or r.get('secs') or 0 tm = 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) tin = f'{ti:,}' if ti else '—' tout = f'{to:,}' if to else '—' tis = f'{ti / wall:.0f}' if (wall > 1 and ti) else '—' tos = f'{to / wall:.0f}' if (wall > 1 and to) else '—' t.add_row(r['name'], r['metric'], v, str(r.get('n', '')), tm, tin, tout, tis, tos, style='green' if r['ok'] else 'red') wall_all = sum(r.get('wall') or r.get('secs') or 0 for r in rows) ti_all = sum(r.get('tok_in', 0) for r in rows) to_all = sum(r.get('tok_out', 0) for r in rows) n_all = sum(r.get('n', 0) or 0 for r in rows if isinstance(r.get('n'), int)) tm_all = f'{wall_all / 3600:.2f}h' if wall_all >= 3600 else f'{wall_all:.0f}s' t.add_section() 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 '' t.add_row(f'[bold]{len(rows)} benchmarks[/bold]', '', f'{sum(1 for r in rows if r["ok"])}/{len(rows)} ok', str(n_all), tm_all, f'{ti_all:,}', f'{to_all:,}', tis_all, tos_all) console.print(t, justify='center') return print(f'\n{"benchmark":<20} {"metric":<16} {"score":>8} {"n":>6} {"time":>8}') print('-' * 64) for r in rows: v = _fmt_score(r.get('value')) if r['ok'] else 'ERR' h = r.get('hours') or 0 tm = f'{h:.2f}h' if h >= 0.995 else f"{r.get('secs', 0):.0f}s" err = f" {r.get('err', '')}" if not r['ok'] else '' print(f"{r['name']:<20} {r['metric']:<16} {v:>8} " f"{str(r.get('n', '')):>6} {tm:>8}{err}") def _f3(v): try: return round(float(v), 4) except (TypeError, ValueError): return v def _cmd_viz_show(args) -> int: from evalharness.viz import render if args.style == 'excel': from evalharness.eval.record import EvalReport reps = [EvalReport.load(p) for p in args.reports] out = render(reps, style='excel', **({'n': args.n} if args.n else {}), **({'out': args.out} if args.out else {})) print(f'excel -> {out}') return 0 print(render([*args.reports], style=args.style, **({'n': args.n} if args.n else {}))) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog='evalharness', description='EvalHarness CLI') sub = parser.add_subparsers(dest='command', required=True) data = sub.add_parser('data', help='dataset plugin commands') dsub = data.add_subparsers(dest='data_command', required=True) p = dsub.add_parser('list', help='list registered datasets (no download)') p.set_defaults(func=_cmd_data_list) p = dsub.add_parser('fetch', help='materialize dataset(s) into the cache') p.add_argument('names', nargs='+') p.add_argument('--force', action='store_true', help='re-download and rebuild the cache') p.add_argument('--workers', type=int, default=8, help='concurrent downloads (default 8)') _add_override_flags(p) p.set_defaults(func=_cmd_data_fetch) p = dsub.add_parser('unload', help='drop cache entries (raw + samples); images belong to the sandbox layer') p.add_argument('names', nargs='+') _add_override_flags(p) p.set_defaults(func=_cmd_data_unload) p = dsub.add_parser('stats', help='materialize and show dataset statistics') p.add_argument('name') _add_override_flags(p) p.set_defaults(func=_cmd_data_stats) p = dsub.add_parser('show', help='print the first N samples') p.add_argument('name') p.add_argument('-n', type=int, default=2) _add_override_flags(p) p.set_defaults(func=_cmd_data_show) # ---- eval ---- ev = sub.add_parser('eval', help='evaluation recipes & runs') esub = ev.add_subparsers(dest='eval_command', required=True) p = esub.add_parser('list', help='list registered eval recipes') p.set_defaults(func=_cmd_eval_list) p = esub.add_parser('run', help='score predictions (file) or generate+score (--model); multiple datasets OK') p.add_argument('datasets', nargs='+', help='dataset name(s) (recipe auto-resolved)') p.add_argument('predictions', nargs='?', help='jsonl: one raw string or {"raw": ...} per sample') p.add_argument('--model', default='', help='served model name when --api-url is used; or full legacy model spec') p.add_argument('--api-url', default='', help='API base URL when --model is only the served model name') p.add_argument('--provider', default='openai-chat', choices=('openai-chat', 'openai-pool'), help='API protocol/provider (default: openai-chat)') p.add_argument('--judge-model', '--judge', dest='judge', default='', help='judge model name with --judge-api-url, or full spec') p.add_argument('--judge-api-url', default='', help='judge API base URL when --judge is only the model name') p.add_argument('--api-key', default='', help='explicit API key for the model endpoint (overrides ' 'env-based resolution; never written into reports)') p.add_argument('--judge-api-key', default='', help='explicit API key for the judge endpoint') p.add_argument('--judge-provider', default='openai-chat', help='judge protocol/provider (default openai-chat; ' 'openai-pool for multi-endpoint judges)') p.add_argument('--profile', default='', help='named gen-params profile (dp4-nothink | qwen3-es-parity | t1-short ' 'or any @register_gen_profile name); layers: plugin default < ' "profile.default < profile[''] < explicit kwargs") p.add_argument('--disable-thinking', action='store_true', help='send enable_thinking=false to the OpenAI-compatible model') p.add_argument('--perf', action='store_true', help='collect streaming TTFT and ITL metrics') p.add_argument('--textools', action='store_true', 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('--concurrency', type=int, default=32, help='parallel model calls (default 32)') p.add_argument('--progress', action='store_true', default=True, help='show per-sample Rich terminal progress (default: on)') p.add_argument('--no-progress', dest='progress', action='store_false', help='disable per-sample Rich terminal progress') p.add_argument('--limit', type=int, help='evaluate only the first N samples total') p.add_argument('--resume', nargs='?', const=True, default=False, help='resume from per-sample checkpoint (default path auto-derived; ' 'pass a path to override)') p.add_argument('--limit-per-task', type=int, help='first N samples PER subset/category (evalscope --limit semantics); ' 'composable with --limit (intersection)') p.add_argument('--out', help='save the EvalReport json here (single dataset)') p.add_argument('--out-dir', help='output directory: summary.xlsx/csv + one dir per benchmark') p.add_argument('--style', default='text', help='result render style (text/md/radar/errors)') p.add_argument('--verbose', action='store_true', help='print full render for every dataset') _add_override_flags(p) p.set_defaults(func=_cmd_eval_run) # ---- sandbox ---- sb = sub.add_parser('sandbox', help='execution environment management') bsub = sb.add_subparsers(dest='sandbox_command', required=True) p = bsub.add_parser('prefetch', help='parallel docker pull of a dataset\'s sandbox images') p.add_argument('dataset', help='dataset whose samples declare images (e.g. swe_bench_verified)') p.add_argument('--workers', type=int, default=8, help='concurrent pulls (default 8)') p.add_argument('--limit', type=int, default=0, help='only first N samples (0=all)') _add_override_flags(p) p.set_defaults(func=_cmd_sandbox_prefetch) # ---- viz ---- vz = sub.add_parser('viz', help='render saved EvalReport artifacts') zsub = vz.add_subparsers(dest='viz_command', required=True) p = zsub.add_parser('show', help='render report file(s)') p.add_argument('reports', nargs='+') p.add_argument('--style', default='text', help='text | md | md_compare | radar | errors | excel (writes .xlsx)') p.add_argument('--out', help='excel output path (default ./evalharness_report.xlsx)') p.add_argument('-n', type=int, help='for errors style: how many samples') p.set_defaults(func=_cmd_viz_show) return parser def main(argv=None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == '__main__': sys.exit(main())