diff --git a/bash/collect_results.py b/bash/collect_results.py index 8536c78..636d347 100644 --- a/bash/collect_results.py +++ b/bash/collect_results.py @@ -10,6 +10,9 @@ Rules: - Perf metrics (latency, TTFT, TPOT, TPS, tokens) are recomputed from raw predictions across all seeds / multi-runs, so resuming from a checkpoint no longer resets cumulative statistics. +- Writing CSV/Excel upserts by Benchmark name: existing rows for other + benchmarks are kept, the finished benchmark overwrites its own row, and + a new benchmark is appended. The 总计 row is always recomputed. """ import argparse @@ -62,6 +65,8 @@ BENCHMARK_DOMAIN = { 'fp_fusion': '模型安全与指纹', } +TOTAL_CATEGORY = '总计' + # Column order matching the reference CSV OUTPUT_COLUMNS = [ '分类', @@ -99,6 +104,73 @@ def percentile(values, q): return float(np.percentile(values, q)) +def _usage_block(summary: dict) -> dict: + """Prefer ``usage`` (current reports) and fall back to legacy ``tokens``.""" + if not isinstance(summary, dict): + return {} + usage = summary.get('usage') + if isinstance(usage, dict) and usage: + return usage + tokens = summary.get('tokens') + return tokens if isinstance(tokens, dict) else {} + + +def _as_num(value): + return np.nan if value is None else value + + +def request_perf_from_summary(summary: Optional[dict]) -> Optional[dict]: + """Map per-request ``perf_metrics.summary`` onto CSV column values. + + Agent/sandbox reports record TTFT/TPOT/latency per LLM request here, even + when prediction jsonl rows have no ``perf_metrics``. + """ + if not isinstance(summary, dict) or not summary: + return None + latency = summary.get('latency') if isinstance(summary.get('latency'), dict) else {} + throughput = summary.get('throughput') if isinstance(summary.get('throughput'), dict) else {} + ttft = summary.get('ttft') if isinstance(summary.get('ttft'), dict) else {} + tpot = summary.get('tpot') if isinstance(summary.get('tpot'), dict) else {} + usage = _usage_block(summary) + if not any((latency, ttft, tpot, usage, throughput)): + return None + in_tok = usage.get('input_tokens') if isinstance(usage.get('input_tokens'), dict) else {} + out_tok = usage.get('output_tokens') if isinstance(usage.get('output_tokens'), dict) else {} + return { + 'latency_mean': _as_num(latency.get('mean')), + 'avg_output_tps': _as_num(throughput.get('avg_output_tps')), + 'avg_req_ps': _as_num(throughput.get('avg_req_ps')), + 'input_tok_mean': _as_num(in_tok.get('mean')), + 'output_tok_mean': _as_num(out_tok.get('mean')), + 'total_tokens': _as_num(usage.get('total_tokens_count')), + 'ttft_mean': _as_num(ttft.get('mean')), + 'ttft_p90': _as_num(ttft.get('90%')), + 'ttft_p99': _as_num(ttft.get('99%')), + 'tpot_mean': _as_num(tpot.get('mean')), + 'tpot_p90': _as_num(tpot.get('90%')), + 'tpot_p99': _as_num(tpot.get('99%')), + 'n_samples': summary.get('n_samples'), + } + + +def _assign_request_perf(fields: dict): + """Unpack ``request_perf_from_summary`` into the collect_benchmark locals.""" + return ( + fields.get('latency_mean', np.nan), + fields.get('avg_output_tps', np.nan), + fields.get('avg_req_ps', np.nan), + fields.get('input_tok_mean', np.nan), + fields.get('output_tok_mean', np.nan), + fields.get('total_tokens', np.nan), + fields.get('ttft_mean', np.nan), + fields.get('ttft_p90', np.nan), + fields.get('ttft_p99', np.nan), + fields.get('tpot_mean', np.nan), + fields.get('tpot_p90', np.nan), + fields.get('tpot_p99', np.nan), + ) + + def read_predictions(pred_file: Path): """Yield perf_metrics dicts from a predictions JSONL file.""" for obj in read_predictions_with_index(pred_file): @@ -486,34 +558,13 @@ def collect_benchmark(output_dir: Path, benchmark: str, model_name: str): if otok is not None: output_tokens.append(int(otok)) - # Agent/sandbox benchmarks do not expose per-call perf metrics through - # EvalScope's model wrapper. Fall back to parsing the trial trajectory - # files (timestamps + final token counts) to get approximate latency and - # token usage. - if not latencies and pred_files: - for obj in read_agent_perf_from_trajectory(pred_files[0]): - idx = obj['index'] - key = ('trajectory', idx) - if idx is not None: - if key in seen_keys: - continue - seen_keys.add(key) - sample_indexes.append(idx) - pm = obj['perf_metrics'] - if pm.get('latency') is not None: - latencies.append(float(pm['latency'])) - if pm.get('input_tokens') is not None: - input_tokens.append(int(pm['input_tokens'])) - if pm.get('output_tokens') is not None: - output_tokens.append(int(pm['output_tokens'])) - - # If we don't have raw predictions but have a perf_stats backup, that - # represents a known-good summary captured right after a clean run — - # preferable to summary0 (which may be the just-reset run). - backup_summary = None - if not latencies: - backup_summary = load_backup_summary(output_dir, benchmark, model_name) + report_perf = request_perf_from_summary(summary0) + backup_summary = load_backup_summary(output_dir, benchmark, model_name) if not latencies else None + backup_perf = request_perf_from_summary(backup_summary) + # Per-call jsonl metrics (typical MCQ / generation benches). Agent jsonl + # usually has no perf_metrics; prefer the report's per-request summary so + # TTFT/TPOT/latency share the same request-count口径 as ``n_samples``. if latencies: latency_mean = float(np.mean(latencies)) total_compute_time = float(np.sum(latencies)) @@ -529,86 +580,86 @@ def collect_benchmark(output_dir: Path, benchmark: str, model_name: str): tpot_mean = float(np.mean(tpots)) if tpots else np.nan tpot_p90 = percentile(tpots, 90) tpot_p99 = percentile(tpots, 99) - # The actual sample count we just rebuilt from raw predictions is more - # reliable than the (possibly reset) report summary's n_samples. - # For multi-seed / multi-run benchmarks we report the total number of - # evaluated predictions (all seeds/runs) rather than unique problem IDs. if n_samples_unique < len(latencies): n_samples_unique = len(latencies) - elif summary0: - # Fallback to report summary if raw predictions are unavailable - latency_mean = summary0.get('latency', {}).get('mean', np.nan) - avg_output_tps = summary0.get('throughput', {}).get('avg_output_tps', np.nan) - avg_req_ps = summary0.get('throughput', {}).get('avg_req_ps', np.nan) - input_tok_mean = summary0.get('tokens', {}).get('input_tokens', {}).get('mean', np.nan) - output_tok_mean = summary0.get('tokens', {}).get('output_tokens', {}).get('mean', np.nan) - total_tokens = summary0.get('tokens', {}).get('total_tokens_count', np.nan) - ttft_mean = summary0.get('ttft', {}).get('mean', np.nan) - ttft_p90 = summary0.get('ttft', {}).get('90%', np.nan) - ttft_p99 = summary0.get('ttft', {}).get('99%', np.nan) - tpot_mean = summary0.get('tpot', {}).get('mean', np.nan) - tpot_p90 = summary0.get('tpot', {}).get('90%', np.nan) - tpot_p99 = summary0.get('tpot', {}).get('99%', np.nan) - # If the just-read report looks like a freshly-reset run (smaller - # n_samples than the durable backup), prefer the backup's summary so - # the cumulative numbers are not lost. - try: - from perf_backup import get_backup_paths - backup_path, _ = get_backup_paths(output_dir, benchmark, model_name) - if backup_path.exists(): - payload = json.loads(backup_path.read_text(encoding='utf-8')) - if (payload.get('n_samples') or 0) > (summary0.get('n_samples') or 0): - backup_summary = payload.get('summary') or {} - latency_mean = backup_summary.get('latency', {}).get('mean', latency_mean) - avg_output_tps = backup_summary.get('throughput', {}).get('avg_output_tps', avg_output_tps) - avg_req_ps = backup_summary.get('throughput', {}).get('avg_req_ps', avg_req_ps) - input_tok_mean = backup_summary.get('usage', {}).get('input_tokens', {}).get('mean', input_tok_mean) - output_tok_mean = backup_summary.get('usage', {}).get('output_tokens', {}).get('mean', output_tok_mean) - total_tokens = backup_summary.get('usage', {}).get('total_tokens_count', total_tokens) - ttft_mean = backup_summary.get('ttft', {}).get('mean', ttft_mean) - ttft_p90 = backup_summary.get('ttft', {}).get('90%', ttft_p90) - ttft_p99 = backup_summary.get('ttft', {}).get('99%', ttft_p99) - tpot_mean = backup_summary.get('tpot', {}).get('mean', tpot_mean) - tpot_p90 = backup_summary.get('tpot', {}).get('90%', tpot_p90) - tpot_p99 = backup_summary.get('tpot', {}).get('99%', tpot_p99) - n_samples_unique = payload.get('n_samples') or n_samples_unique - except Exception: - pass - elif backup_summary: - latency_mean = backup_summary.get('latency', {}).get('mean', np.nan) - avg_output_tps = backup_summary.get('throughput', {}).get('avg_output_tps', np.nan) - avg_req_ps = backup_summary.get('throughput', {}).get('avg_req_ps', np.nan) - input_tok_mean = backup_summary.get('usage', {}).get('input_tokens', {}).get('mean', np.nan) - output_tok_mean = backup_summary.get('usage', {}).get('output_tokens', {}).get('mean', np.nan) - total_tokens = backup_summary.get('usage', {}).get('total_tokens_count', np.nan) - ttft_mean = backup_summary.get('ttft', {}).get('mean', np.nan) - ttft_p90 = backup_summary.get('ttft', {}).get('90%', np.nan) - ttft_p99 = backup_summary.get('ttft', {}).get('99%', np.nan) - tpot_mean = backup_summary.get('tpot', {}).get('mean', np.nan) - tpot_p90 = backup_summary.get('tpot', {}).get('90%', np.nan) - tpot_p99 = backup_summary.get('tpot', {}).get('99%', np.nan) + elif report_perf or backup_perf: + report_n = int((summary0 or {}).get('n_samples') or 0) + backup_n = int((backup_summary or {}).get('n_samples') or 0) + if backup_perf is not None and backup_n > report_n: + chosen = backup_perf + n_samples_unique = backup_n + else: + chosen = report_perf or backup_perf + if chosen.get('n_samples') is not None: + n_samples_unique = chosen['n_samples'] + ( + latency_mean, + avg_output_tps, + avg_req_ps, + input_tok_mean, + output_tok_mean, + total_tokens, + ttft_mean, + ttft_p90, + ttft_p99, + tpot_mean, + tpot_p90, + tpot_p99, + ) = _assign_request_perf(chosen) else: - # Agent / sandbox benchmarks (e.g. terminal_bench_v2_1) provide a score - # but do not record per-sample perf metrics. Keep the score and leave - # perf columns empty rather than dropping the benchmark entirely. - latency_mean = np.nan - avg_output_tps = np.nan - avg_req_ps = np.nan - input_tok_mean = np.nan - output_tok_mean = np.nan - total_tokens = np.nan - ttft_mean = np.nan - ttft_p90 = np.nan - ttft_p99 = np.nan - tpot_mean = np.nan - tpot_p90 = np.nan - tpot_p99 = np.nan - if not n_samples_unique and reports: - try: - data = json.loads(reports[0].read_text(encoding='utf-8')) - n_samples_unique = data.get('num', 0) - except Exception: - pass + # Last resort: Harbor trajectory wall-clock (no TTFT/TPOT). + if pred_files: + for obj in read_agent_perf_from_trajectory(pred_files[0]): + idx = obj['index'] + key = ('trajectory', idx) + if idx is not None: + if key in seen_keys: + continue + seen_keys.add(key) + sample_indexes.append(idx) + pm = obj['perf_metrics'] + if pm.get('latency') is not None: + latencies.append(float(pm['latency'])) + if pm.get('input_tokens') is not None: + input_tokens.append(int(pm['input_tokens'])) + if pm.get('output_tokens') is not None: + output_tokens.append(int(pm['output_tokens'])) + if latencies: + latency_mean = float(np.mean(latencies)) + total_compute_time = float(np.sum(latencies)) + total_output_tokens = sum(output_tokens) + avg_output_tps = total_output_tokens / total_compute_time if total_compute_time > 0 else np.nan + avg_req_ps = len(latencies) / total_compute_time if total_compute_time > 0 else np.nan + input_tok_mean = float(np.mean(input_tokens)) if input_tokens else np.nan + output_tok_mean = float(np.mean(output_tokens)) if output_tokens else np.nan + total_tokens = sum(input_tokens) + sum(output_tokens) + ttft_mean = np.nan + ttft_p90 = np.nan + ttft_p99 = np.nan + tpot_mean = np.nan + tpot_p90 = np.nan + tpot_p99 = np.nan + if n_samples_unique < len(latencies): + n_samples_unique = len(latencies) + else: + latency_mean = np.nan + avg_output_tps = np.nan + avg_req_ps = np.nan + input_tok_mean = np.nan + output_tok_mean = np.nan + total_tokens = np.nan + ttft_mean = np.nan + ttft_p90 = np.nan + ttft_p99 = np.nan + tpot_mean = np.nan + tpot_p90 = np.nan + tpot_p99 = np.nan + if not n_samples_unique and reports: + try: + data = json.loads(reports[0].read_text(encoding='utf-8')) + n_samples_unique = data.get('num', 0) + except Exception: + pass # Duration: prefer the durable active timer maintained by perf_backup.py, # which only counts time when run_task() is actually executing. This avoids @@ -679,6 +730,116 @@ def collect_benchmark(output_dir: Path, benchmark: str, model_name: str): } +def _canonical_benchmark_label(name: Optional[str]) -> str: + """Normalize a summary-table benchmark label (including directory aliases).""" + try: + if name is None or pd.isna(name): + return '' + except (TypeError, ValueError): + if name is None: + return '' + label = str(name).strip() + if not label or label.lower() == 'nan': + return '' + return BENCHMARK_NAME_ALIAS.get(label, label) + + +def _coerce_summary_columns(df: pd.DataFrame) -> pd.DataFrame: + """Align an existing table to ``OUTPUT_COLUMNS``, filling missing fields.""" + out = pd.DataFrame(index=df.index, columns=OUTPUT_COLUMNS) + for col in OUTPUT_COLUMNS: + if col in df.columns: + out[col] = df[col] + return out.reset_index(drop=True) + + +def _summary_body(df: pd.DataFrame) -> pd.DataFrame: + """Drop the 总计 row from a previously written summary.""" + body = _coerce_summary_columns(df) + category = body['分类'].astype(str).str.strip() + return body.loc[category != TOTAL_CATEGORY].reset_index(drop=True) + + +def _with_total_row(body: pd.DataFrame) -> pd.DataFrame: + """Append a recomputed 总计 row. ``body`` must not already contain one.""" + if body is None or body.empty: + return pd.DataFrame(columns=OUTPUT_COLUMNS) + + total_score = body['得分'].mean(skipna=True) + total_time = body['实测时间(h)'].sum(skipna=True) + total_samples = body['总样本数'].sum(skipna=True) + total_tokens = body['累计总tokens'].sum(skipna=True) + total_row = {col: np.nan for col in OUTPUT_COLUMNS} + total_row.update({ + '分类': TOTAL_CATEGORY, + 'Benchmark': '', + '得分': round(float(total_score), 4) if pd.notna(total_score) else np.nan, + '实测时间(h)': round(float(total_time), 4) if pd.notna(total_time) else np.nan, + '总样本数': total_samples if pd.notna(total_samples) else np.nan, + '累计总tokens': total_tokens if pd.notna(total_tokens) else np.nan, + }) + return pd.concat([body, pd.DataFrame([total_row], columns=OUTPUT_COLUMNS)], ignore_index=True) + + +def _load_existing_summary(csv_path: Path, xlsx_path: Path) -> Optional[pd.DataFrame]: + """Load the current summary table, preferring CSV then Excel.""" + readers = ( + (csv_path, lambda p: pd.read_csv(p, encoding='utf-8-sig')), + (xlsx_path, lambda p: pd.read_excel(p)), + ) + for path, reader in readers: + if not path.exists(): + continue + try: + df = reader(path) + except Exception as e: + print(f'WARNING: failed to read existing summary {path}: {e}') + continue + if df is None or df.empty: + continue + return df + return None + + +def upsert_summary_rows(existing: Optional[pd.DataFrame], new_rows: list) -> pd.DataFrame: + """Overwrite matching Benchmark rows, append unseen ones, then recompute 总计. + + Existing row order is preserved. New benchmarks are appended just before + the 总计 row. + """ + incoming = pd.DataFrame(new_rows, columns=OUTPUT_COLUMNS) + incoming = incoming.drop_duplicates(subset=['Benchmark'], keep='last') + incoming['Benchmark'] = incoming['Benchmark'].map(_canonical_benchmark_label) + + if existing is None or existing.empty: + body_records = [] + else: + body_records = _summary_body(existing).to_dict('records') + for rec in body_records: + rec['Benchmark'] = _canonical_benchmark_label(rec.get('Benchmark')) + + index_by_name = {} + for i, rec in enumerate(body_records): + name = str(rec.get('Benchmark') or '') + if name: + index_by_name[name] = i + + for row in incoming.to_dict('records'): + name = str(row.get('Benchmark') or '') + if not name: + continue + if name in index_by_name: + body_records[index_by_name[name]] = row + else: + index_by_name[name] = len(body_records) + body_records.append(row) + + body = pd.DataFrame(body_records, columns=OUTPUT_COLUMNS) if body_records else incoming + if not body.empty: + body = body.drop_duplicates(subset=['Benchmark'], keep='first') + return _with_total_row(body) + + def eval_benchmark(benchmark_names: list, output_dir: Path, model_name: str, out_name: str = None, excel_output_dir: Path = None): """Collect results for a specific list of benchmarks. @@ -688,8 +849,8 @@ def eval_benchmark(benchmark_names: list, output_dir: Path, model_name: str, Args: benchmark_names: List of canonical benchmark names to aggregate (e.g. - ``['aime24', 'gsm8k', 'arc']``). Only benchmarks with reports on - disk will appear in the summary. + ``['aime24', 'gsm8k', 'arc']``). Only these benchmarks are + refreshed from disk; other rows already in the summary file are kept. output_dir: EvalScope output root directory. model_name: Model name to look up reports/predictions for. out_name: Output file name (without extension). Defaults to the safe @@ -709,7 +870,12 @@ def eval_benchmark(benchmark_names: list, output_dir: Path, model_name: str, def collect_all(output_dir: Path, model_name: str, out_name: str = None, include_benchmarks: list = None, excel_output_dir: Path = None): - """Collect benchmarks under ``output_dir`` and write summary Excel/CSV.""" + """Collect benchmarks under ``output_dir`` and upsert them into summary Excel/CSV. + + Existing rows for other benchmarks are preserved. Matching ``Benchmark`` + names are overwritten; unseen names are appended. The 总计 row is rebuilt + from the merged table. + """ if not output_dir.exists(): raise FileNotFoundError(f'Output directory not found: {output_dir}') @@ -758,27 +924,6 @@ def collect_all(output_dir: Path, model_name: str, out_name: str = None, print(f'No results found for model {model_name} under {output_dir}') return None, None - df = pd.DataFrame(rows, columns=OUTPUT_COLUMNS) - df = df.drop_duplicates(subset=['Benchmark'], keep='first') - - # Add total row - total_score = df['得分'].mean() - total_time = df['实测时间(h)'].sum() - total_samples = df['总样本数'].sum() if '总样本数' in df.columns else np.nan - total_tokens = df['累计总tokens'].sum() if '累计总tokens' in df.columns else np.nan - total_row = { - '分类': '总计', - 'Benchmark': '', - '得分': round(total_score, 4), - '实测时间(h)': round(total_time, 4), - '总样本数': total_samples if not np.isnan(total_samples) else np.nan, - '累计总tokens': total_tokens if not np.isnan(total_tokens) else np.nan, - } - for col in OUTPUT_COLUMNS: - if col not in total_row: - total_row[col] = np.nan - df = pd.concat([df, pd.DataFrame([total_row], columns=OUTPUT_COLUMNS)], ignore_index=True) - if out_name is None: out_name = safe_model @@ -795,8 +940,12 @@ def collect_all(output_dir: Path, model_name: str, out_name: str = None, csv_path = summary_dir / f'{out_name}.csv' xlsx_path = summary_dir / f'{out_name}.xlsx' + df = upsert_summary_rows(_load_existing_summary(csv_path, xlsx_path), rows) df.to_csv(csv_path, index=False, encoding='utf-8-sig') - df.to_excel(xlsx_path, index=False) + try: + df.to_excel(xlsx_path, index=False) + except Exception as e: + print(f'WARNING: failed to write Excel {xlsx_path}: {e}') print(f'Summary written to:') print(f' CSV: {csv_path}') diff --git a/bash/run.py b/bash/run.py index a92212c..a5d49ba 100644 --- a/bash/run.py +++ b/bash/run.py @@ -456,9 +456,24 @@ def truncate_middle(text: str, max_tokens: int, tokenizer_path: str) -> str: def _patch_adapters_for_truncation(tokenizer_path: str, truncation_tokens: int): + from evalscope.benchmarks.aa_lcr.aa_lcr_adapter import AALCRAdapter from evalscope.benchmarks.longbench_v2.longbench_v2_adapter import LongBenchV2Adapter from evalscope.benchmarks.openai_mrcr.openai_mrcr_adapter import OpenAIMRCRAdapter + _orig_aa_lcr_sample = AALCRAdapter.record_to_sample + + def _patched_aa_lcr_sample(self, record): + sample = _orig_aa_lcr_sample(self, record) + if not sample.input: + return sample + msg = sample.input[0] + content = getattr(msg, 'content', None) + if isinstance(content, str) and content: + msg.content = truncate_middle(content, truncation_tokens, tokenizer_path) + return sample + + AALCRAdapter.record_to_sample = _patched_aa_lcr_sample + _orig_longbench_format = LongBenchV2Adapter.format_prompt_template def _patched_longbench_format(self, sample): @@ -695,9 +710,8 @@ def run_and_summarize(task_cfg, write_summary_flag: bool, output_dir: str, """Run a benchmark task and optionally refresh the summary table. ``benchmark_names`` is the running list of canonical benchmark names that - were scheduled for this invocation. The summary table will be limited to - these benchmarks so it does not pick up older results left in the output - directory from previous runs. + were scheduled for this invocation. Those rows are refreshed in the + summary table; other benchmarks already written to CSV/Excel are kept. After a successful ``run_task()`` we also snapshot the cumulative ``perf_metrics.summary`` and the per-sample predictions into the durable @@ -988,8 +1002,8 @@ def main(): except Exception as e: print(f'ERROR in {dataset_name} (run {run_idx + 1 if run_idx else 1}): {e}') - # Keep the ordered list of benchmarks actually run in this invocation so - # the summary table only includes them, not stale results from earlier runs. + # Benchmarks scheduled in this invocation: their summary rows are upserted; + # other existing CSV/Excel rows are left in place. benchmark_names = [] # Build a list of benchmark execution units. Each unit runs one complete diff --git a/bash/tests/test_summary_upsert.py b/bash/tests/test_summary_upsert.py new file mode 100644 index 0000000..2718a89 --- /dev/null +++ b/bash/tests/test_summary_upsert.py @@ -0,0 +1,159 @@ +"""Upsert behaviour for the project-level CSV/Excel summary.""" + +from pathlib import Path +import sys + +import numpy as np +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from collect_results import ( # noqa: E402 + OUTPUT_COLUMNS, + TOTAL_CATEGORY, + collect_all, + collect_benchmark, + upsert_summary_rows, +) + + +def _row(benchmark: str, score: float, n: int = 1, category: str = '其他') -> dict: + row = {col: np.nan for col in OUTPUT_COLUMNS} + row.update({ + '分类': category, + 'Benchmark': benchmark, + '得分': score, + '实测时间(h)': 1.0, + '总样本数': n, + '累计总tokens': 10, + }) + return row + + +def test_upsert_appends_new_benchmark_and_keeps_existing(): + existing = pd.DataFrame([_row('gpqa_diamond', 0.5, category='知识与语言理解')], columns=OUTPUT_COLUMNS) + existing = pd.concat( + [existing, pd.DataFrame([{col: np.nan for col in OUTPUT_COLUMNS} | { + '分类': TOTAL_CATEGORY, + 'Benchmark': '', + '得分': 0.5, + }], columns=OUTPUT_COLUMNS)], + ignore_index=True, + ) + + out = upsert_summary_rows(existing, [_row('terminal_bench_v2_1', 0.0, category='智能体与工具')]) + names = [n for n in out['Benchmark'].tolist() if str(n).strip()] + assert names == ['gpqa_diamond', 'terminal_bench_v2_1'] + assert out.iloc[-1]['分类'] == TOTAL_CATEGORY + assert out.iloc[-1]['总样本数'] == 2 + + +def test_upsert_overwrites_matching_benchmark_in_place(): + existing = pd.DataFrame( + [ + _row('gpqa_diamond', 0.1, n=1, category='知识与语言理解'), + _row('hle', 0.2, n=2, category='知识与语言理解'), + ], + columns=OUTPUT_COLUMNS, + ) + + out = upsert_summary_rows(existing, [_row('gpqa_diamond', 0.9, n=100, category='知识与语言理解')]) + gpqa = out.loc[out['Benchmark'] == 'gpqa_diamond'].iloc[0] + assert gpqa['得分'] == 0.9 + assert gpqa['总样本数'] == 100 + assert list(out['Benchmark'].tolist()[:-1]) == ['gpqa_diamond', 'hle'] + assert (out['Benchmark'] == 'gpqa_diamond').sum() == 1 + + +def test_upsert_aliases_hle_low_to_hle(): + existing = pd.DataFrame([_row('hle_low', 0.3)], columns=OUTPUT_COLUMNS) + out = upsert_summary_rows(existing, [_row('hle', 0.8)]) + assert list(out['Benchmark'].tolist()[:-1]) == ['hle'] + assert out.loc[out['Benchmark'] == 'hle'].iloc[0]['得分'] == 0.8 + + +def test_collect_all_merges_into_existing_csv(tmp_path: Path): + summary_dir = tmp_path / 'results' + summary_dir.mkdir() + prior = pd.DataFrame([_row('gpqa_diamond', 0.4, category='知识与语言理解')], columns=OUTPUT_COLUMNS) + prior = upsert_summary_rows(None, [_row('gpqa_diamond', 0.4, category='知识与语言理解')]) + csv_path = summary_dir / 'mock-model.csv' + prior.to_csv(csv_path, index=False, encoding='utf-8-sig') + + output_dir = tmp_path / 'output' + reports = output_dir / 'terminal_bench_v2_1' / 'seed_42' / 'reports' + reports.mkdir(parents=True) + (reports / 'terminal_bench_v2_1.json').write_text( + '{"score": 0.0, "num": 2, "metrics": [{"identity": {"name": "accuracy", "aggregation": "mean", "dimensions": {}}, "score": 0.0, "num": 2}]}', + encoding='utf-8', + ) + + collect_all( + output_dir, + 'mock-model', + out_name='mock-model', + include_benchmarks=['terminal_bench_v2_1'], + excel_output_dir=summary_dir, + ) + + df = pd.read_csv(csv_path, encoding='utf-8-sig') + names = [n for n in df['Benchmark'].fillna('').tolist() if str(n).strip()] + assert names == ['gpqa_diamond', 'terminal_bench_v2_1'] + tb = df.loc[df['Benchmark'] == 'terminal_bench_v2_1'].iloc[0] + assert tb['得分'] == 0.0 + assert df.iloc[-1]['分类'] == TOTAL_CATEGORY + + +def test_agent_bench_uses_report_per_request_perf(tmp_path: Path): + import json + + output_dir = tmp_path / 'output' + bench = 'terminal_bench_v2_1' + reports = output_dir / bench / 'seed_42' / 'reports' + preds = output_dir / bench / 'seed_42' / 'predictions' + reports.mkdir(parents=True) + preds.mkdir(parents=True) + (reports / f'{bench}.json').write_text( + json.dumps({ + 'score': 0.0, + 'num': 2, + 'metrics': [{ + 'identity': {'name': 'accuracy', 'aggregation': 'mean', 'dimensions': {}}, + 'score': 0.0, + 'num': 2, + }], + 'perf_metrics': { + 'summary': { + 'n_samples': 22, + 'latency': {'mean': 12.80038}, + 'throughput': {'avg_output_tps': 140.06, 'avg_req_ps': 0.0781}, + 'usage': { + 'input_tokens': {'mean': 6238.136364}, + 'output_tokens': {'mean': 1792.818182}, + 'total_tokens_count': 176681, + }, + 'ttft': {'mean': 0.583404, '90%': 1.482557, '99%': 1.721449}, + 'tpot': {'mean': 0.006934, '90%': 0.008364, '99%': 0.008729}, + } + }, + }), + encoding='utf-8', + ) + (preds / f'{bench}__m.jsonl').write_text( + json.dumps({ + 'index': 0, + 'model_output': {'choices': [{'message': {'content': 'file:///tmp/missing-trial'}}]}, + }) + '\n', + encoding='utf-8', + ) + + row = collect_benchmark(output_dir, bench, 'm') + assert row['总样本数'] == 22 + assert row['延迟_mean(s)'] == 12.80038 + assert row['TTFT_mean(s)'] == 0.5834 + assert row['TTFT P90'] == 1.48256 + assert row['TPOT_mean(s)'] == 0.00693 + assert row['输入tokens_mean'] == 6238.14 + assert row['累计总tokens'] == 176681 + assert row['输出TPS'] == 140.06 + diff --git a/evalscope/evalscope/evaluation_versioning.py b/evalscope/evalscope/evaluation_versioning.py index 49ccb79..442cc14 100644 --- a/evalscope/evalscope/evaluation_versioning.py +++ b/evalscope/evalscope/evaluation_versioning.py @@ -162,17 +162,7 @@ def build_benchmark_identity( task_config: 'TaskConfig', ) -> BenchmarkEvaluationIdentity: """Build a stable cache identity from effective evaluation semantics.""" - validate_evaluation_version(evaluation_version) - payload = { - 'evaluation_version': evaluation_version, - 'benchmark': spec.model_dump(mode='json'), - 'task': _fingerprint_task_config(task_config), - } - encoded = _canonical_json(_scrub_secrets(payload)).encode('utf-8') - return BenchmarkEvaluationIdentity( - evaluation_version=evaluation_version, - fingerprint=f'sha256:{hashlib.sha256(encoded).hexdigest()}', - ) + return _identity_from_spec(spec, evaluation_version, task_config.to_dict()) def build_evaluation_identity( @@ -206,6 +196,10 @@ def validate_cached_evaluation_identity( ``previous_config`` is deliberately the raw output snapshot: generated identity is audit data, not input that alters the current adapter defaults. + + Matching uses a fingerprint recomputed from the snapshot with the current + rules, so caches hashed under an older field set (for example when ``limit`` + was still included) remain valid when only ignored fields changed. """ previous_identity = _identity_from_config(previous_config) sources: Dict[str, CacheSource] = {} @@ -213,7 +207,8 @@ def validate_cached_evaluation_identity( previous = None inferred_legacy = False if previous_identity is not None: - previous = previous_identity.benchmarks.get(benchmark_name) + stored = previous_identity.benchmarks.get(benchmark_name) + previous = _recompute_cached_identity(previous_config, benchmark_name, stored) elif previous_config is not None: previous = legacy_identity_from_config(previous_config, benchmark_name) inferred_legacy = previous is not None @@ -242,16 +237,7 @@ def legacy_identity_from_config( ) except Exception: return None - payload = { - 'evaluation_version': 'v1.0', - 'benchmark': spec.model_dump(mode='json'), - 'task': _fingerprint_task_mapping(task_config), - } - encoded = _canonical_json(_scrub_secrets(payload)).encode('utf-8') - return BenchmarkEvaluationIdentity( - evaluation_version='v1.0', - fingerprint=f'sha256:{hashlib.sha256(encoded).hexdigest()}', - ) + return _identity_from_spec(spec, 'v1.0', task_config) def _identity_from_config(config: Optional[Dict[str, Any]]) -> Optional[EvaluationIdentity]: @@ -264,11 +250,63 @@ def _identity_from_config(config: Optional[Dict[str, Any]]) -> Optional[Evaluati return None -def _fingerprint_task_config(task_config: 'TaskConfig') -> Dict[str, Any]: - return _fingerprint_task_mapping(task_config.to_dict()) +def _identity_from_spec( + spec: ResolvedBenchmarkSpec, + evaluation_version: str, + task_config: Dict[str, Any], +) -> BenchmarkEvaluationIdentity: + validate_evaluation_version(evaluation_version) + payload = { + 'evaluation_version': evaluation_version, + 'benchmark': spec.model_dump(mode='json'), + 'task': _fingerprint_task_mapping(task_config), + } + encoded = _canonical_json(_scrub_secrets(payload)).encode('utf-8') + return BenchmarkEvaluationIdentity( + evaluation_version=evaluation_version, + fingerprint=f'sha256:{hashlib.sha256(encoded).hexdigest()}', + ) + + +def _spec_from_snapshot(config: Optional[Dict[str, Any]], benchmark_name: str) -> Optional[ResolvedBenchmarkSpec]: + if not config: + return None + resolved = config.get('resolved_benchmarks') + if not isinstance(resolved, dict): + return None + raw_spec = resolved.get(benchmark_name) + if not isinstance(raw_spec, dict): + return None + try: + return ResolvedBenchmarkSpec.model_validate(raw_spec) + except Exception: + return None + + +def _recompute_cached_identity( + previous_config: Optional[Dict[str, Any]], + benchmark_name: str, + stored: Optional[BenchmarkEvaluationIdentity], +) -> Optional[BenchmarkEvaluationIdentity]: + """Rebuild a stored identity with the current fingerprint field set. + + ``limit`` is ignored, so an older dump whose hash still included ``limit`` + can match a later run that only changes how many samples are requested. + """ + if stored is None: + return None + spec = _spec_from_snapshot(previous_config, benchmark_name) + if spec is None or previous_config is None: + return stored + try: + return _identity_from_spec(spec, stored.evaluation_version, previous_config) + except Exception: + return stored def _fingerprint_task_mapping(task_config: Dict[str, Any]) -> Dict[str, Any]: + # ``limit`` is intentionally omitted: sample count is a run-size knob, not + # evaluation semantics. Cache reuse still filters by sample id. keys = ( 'model', 'model_id', @@ -278,7 +316,6 @@ def _fingerprint_task_mapping(task_config: Dict[str, Any]) -> Dict[str, Any]: 'generation_config', 'eval_type', 'api_url', - 'limit', 'repeats', 'seed', 'judge', diff --git a/evalscope/tests/api/test_evaluation_versioning.py b/evalscope/tests/api/test_evaluation_versioning.py index c2b1d54..90504d4 100644 --- a/evalscope/tests/api/test_evaluation_versioning.py +++ b/evalscope/tests/api/test_evaluation_versioning.py @@ -101,7 +101,18 @@ def test_identity_changes_only_for_evaluation_semantics() -> None: assert first.fingerprint != build_benchmark_identity(spec, 'v1.1', config).fingerprint assert first.fingerprint != build_benchmark_identity(spec, 'v1.0', _task_config(seed=7)).fingerprint - assert first.fingerprint != build_benchmark_identity(spec, 'v1.0', _task_config(limit=5)).fingerprint + same_except_limit = _task_config( + api_key='secret', + model_args={'api_key': 'nested-secret', 'headers': {'X-API-Key': 'header-secret'}}, + limit=100, + ) + assert first.fingerprint == build_benchmark_identity(spec, 'v1.0', same_except_limit).fingerprint + warmer = _task_config( + api_key='secret', + model_args={'api_key': 'nested-secret', 'headers': {'X-API-Key': 'header-secret'}}, + generation_config={'temperature': 0.7}, + ) + assert first.fingerprint != build_benchmark_identity(spec, 'v1.0', warmer).fingerprint changed_prompt = spec.model_copy(update={'prompt_template': 'Changed prompt'}) assert first.fingerprint != build_benchmark_identity(changed_prompt, 'v1.0', config).fingerprint changed_revision = spec.model_copy(update={'dataset_revision': '2026-08-21'}) @@ -190,6 +201,23 @@ def test_native_cache_identity_blocks_mismatched_run_before_snapshot_overwrite(t assert snapshot_path.read_text() == snapshot +def test_cached_identity_accepts_limit_only_change_from_legacy_hash() -> None: + previous_config = _task_config(limit=1) + spec = ResolvedBenchmarkSpec.from_meta(_meta(), previous_config) + previous = build_evaluation_identity({'demo': spec}, {'demo': 'v1.0'}, previous_config) + snapshot = previous_config.to_dict() + snapshot['resolved_benchmarks'] = {'demo': spec.model_dump(mode='json')} + snapshot['evaluation_identity'] = previous.model_dump(mode='json') + snapshot['evaluation_identity']['benchmarks']['demo']['fingerprint'] = _fingerprint_including_limit( + spec, 'v1.0', previous_config + ) + snapshot['limit'] = 1 + + current = build_evaluation_identity({'demo': spec}, {'demo': 'v1.0'}, _task_config(limit=100)) + assert previous.benchmarks['demo'].fingerprint != snapshot['evaluation_identity']['benchmarks']['demo']['fingerprint'] + assert validate_cached_evaluation_identity(snapshot, current, rerun_review=False) == {} + + def test_native_rerun_review_records_the_prediction_source(tmp_path) -> None: base = { 'model': 'mock-model', @@ -291,3 +319,38 @@ def test_analysis_uses_compact_context_without_full_meta_or_perf(monkeypatch) -> def _identity(benchmarks: dict[str, BenchmarkEvaluationIdentity]) -> EvaluationIdentity: return EvaluationIdentity(benchmarks=benchmarks) + + +def _fingerprint_including_limit(spec: ResolvedBenchmarkSpec, evaluation_version: str, config: TaskConfig) -> str: + """Reproduce the pre-change task fingerprint that still hashed ``limit``.""" + import hashlib + import json + + from evalscope.evaluation_versioning import _scrub_secrets + + mapping = config.to_dict() + keys = ( + 'model', + 'model_id', + 'model_args', + 'model_task', + 'chat_template', + 'generation_config', + 'eval_type', + 'api_url', + 'limit', + 'repeats', + 'seed', + 'judge', + 'sandbox', + 'agent_config', + ) + payload = { + 'evaluation_version': evaluation_version, + 'benchmark': spec.model_dump(mode='json'), + 'task': {key: mapping.get(key) for key in keys}, + } + encoded = json.dumps( + _scrub_secrets(payload), ensure_ascii=False, sort_keys=True, separators=(',', ':'), default=str + ).encode('utf-8') + return f'sha256:{hashlib.sha256(encoded).hexdigest()}'