diff --git a/every_eval_ever/adapters/README.md b/every_eval_ever/adapters/README.md index a6477c15d..eeb562b04 100644 --- a/every_eval_ever/adapters/README.md +++ b/every_eval_ever/adapters/README.md @@ -36,6 +36,7 @@ Each adapter is run with `uv run python -m every_eval_ever.adapters..adapt | `mmlu_pro` | TIGER-Lab leaderboard CSV | Converts the MMLU-Pro leaderboard (`TIGER-Lab/mmlu_pro_leaderboard_submission`) into `data/mmlu-pro/`. Emits per-model overall + 14 per-subject accuracies. | | `lexam` | LEXam project website | Converts the LEXam legal-reasoning leaderboard (open-question judge scores + 4-choice MCQ accuracy) into `data/lexam/`. | | `vectara_hallucination_leaderboard` | HuggingFace (`vectara/results`) | Converts the Vectara Hallucination Leaderboard result files, pinned to a source commit, into `data/vectara-hallucination-leaderboard/`. Emits 4 aggregate metrics plus per-category and per-text-complexity breakdowns (40 scores per model). | +| `wild` | HuggingFace (`kensho/WILD-raw`) | Converts the WILD-raw item-level eval responses (65 models × 27 benchmarks, run with Inspect AI) into `data/wild/`: aggregate accuracy per model×benchmark and per subtask, with optional per-item `_samples.jsonl` sidecars (`--include-instances`). See [`wild/README.md`](wild/README.md). | ### Mercor Evaluation Exports diff --git a/every_eval_ever/adapters/wild/README.md b/every_eval_ever/adapters/wild/README.md new file mode 100644 index 000000000..3d4b410fb --- /dev/null +++ b/every_eval_ever/adapters/wild/README.md @@ -0,0 +1,143 @@ +# WILD-raw adapter + +Converts **WILD-raw** (`kensho/WILD-raw`, [arXiv:2604.01418](https://arxiv.org/abs/2604.01418)) +into Every Eval Ever records. + +WILD-raw is **item-level** evaluation data: ~7.5M `(model, item)` rows for 65 +models across 27 benchmarks (109,566 items), each row a single item response — +conversation, model answer, target, binary score, token usage, and scorer output. +Kensho ran the evaluations, so it's an `evaluation_run` source with +`evaluator_relationship = third_party`. + +## Mapping +- **Aggregate** — one `EvaluationLog` per (model, benchmark). `evaluation_results` + = the benchmark overall accuracy (`wild.`) plus one per subtask + (`wild..`); each is a `continuous` `[0,1]` accuracy (mean of the + binary item scores), with item counts + mean token usage in + `metric_config.additional_details`. `model_info.id` is the dataset's HF-form model + id as-is (e.g. `01-ai/Yi-1.5-34B-Chat`). +- **Instances** (`--include-instances`) — the raw per-item rows become an + instance-level `_samples.jsonl` sidecar (single-turn: `input` = the prompt + + `target`, `output` = the model's full generation, `evaluation` = score/is_correct, + `token_usage`), referenced by the aggregate's `detailed_evaluation_results`. This + is the faithful use of the *raw* dataset; it is off by default because it is large. + +## Usage + +Reads parquet straight from HuggingFace in **bounded record batches** (never the +full ~7GB, and never a whole 500,000-row row group). The output directory must be a +`data/wild` path, because records are published to +`data/wild///.json`. Smoke run over the first shard, writing +outside the repo: + +```bash +uv run python -m every_eval_ever.adapters.wild.adapter \ + --output-dir /tmp/eee-wild/data/wild --limit-shards 1 +uv run python -m every_eval_ever validate '/tmp/eee-wild/data/wild/*/*/*.json*' +``` + +Filter models / include a capped instance sample: + +```bash +uv run python -m every_eval_ever.adapters.wild.adapter \ + --output-dir /tmp/eee-wild-inst/data/wild \ + --limit-shards 1 --models 01-ai/Yi-1.5-34B-Chat \ + --include-instances --max-instances 2000 +``` + +Full run (all 15 shards; heavy — hours + lots of I/O for instances): + +```bash +uv run python -m every_eval_ever.adapters.wild.adapter --output-dir data/wild # aggregates +uv run python -m every_eval_ever.adapters.wild.adapter --output-dir data/wild --include-instances +``` + +Local parquet instead of fetching. A local file carries no source commit date, so +`--evaluation-timestamp` is required: + +```bash +uv run python -m every_eval_ever.adapters.wild.adapter --parquet data-000*.parquet \ + --output-dir /tmp/eee-wild/data/wild --evaluation-timestamp 1780000000.0 +``` + +Record filenames are fresh uuid4s, so a rerun into a populated output directory is +an error rather than a second copy of every record; pass `--replace-existing` to +replace what is there. Replacement goes by identity rather than by directory — only +the (model, benchmark) pairs this run rewrites, and only once their replacements are +published — so a run whose input covers just some of a model's benchmarks (a +`--limit-shards` smoke run, a subset of local `--parquet` files) leaves the rest of +that model's directory alone, and a run that fails partway leaves the previous +refresh whole. Everything is staged and preflighted before any file is created, and a +failure removes whatever the run created. + +A row whose `score` is not a usable binary correctness value is left out of the +aggregate (rather than counted as wrong) and named in +`adapter_reports/wild_failures.json`; the command then exits non-zero so a partial +refresh is distinguishable from a complete one. + +## Notes +- Timestamps: `retrieved_timestamp` = when this record was built (**now**); + `evaluation_timestamp` = when WILD ran the eval, proxied by the HF dataset's + `lastModified` (overrides: `--retrieved-timestamp` / `--evaluation-timestamp`). + `evaluation_id` is keyed on the evaluation time so reruns are idempotent — there is + no `now()` fallback, which would give identical reruns different identities. +- Remote runs pin one concrete commit and read it in both passes. If that commit + cannot be resolved the run stops rather than reading the mutable `main` ref; pass + `--revision ` to pin a snapshot yourself. `--revision` may name a branch or + tag, but only while the lookup can resolve it to a commit — if the lookup itself + fails, only a 40-character SHA is accepted, because a ref that moves between the + two passes would have them read different data. A local `--parquet` run records a + local marker instead of a revision it cannot know. +- `eval_library` = `inspect_ai` — the WILD paper states the evals were run with the + Inspect AI framework; the scorer (`match`) is in `additional_details`. +- `source_data` points at each **benchmark's own dataset repo** (all verified on + HF; see `WILD_DATASET_REPO` + `AIME_REPO_BY_SUBTASK`) — e.g. `mmlu`→`cais/mmlu`, + `arc_*`→`allenai/ai2_arc`, `squad`→`rajpurkar/squad`, `finance_fundamentals`→ + `kensho/bizbench`, `pre_flight`→`AirsideLabs/pre-flight-06`, `bbeh`→`BBEH/bbeh`, + `aime`→`Maxwell-Jia/AIME_2024`+`math-ai/aime25` (per subtask). It is **not** + `kensho/WILD-raw` — WILD-raw is the *results* source, recorded in `source_metadata`. +- `source_type=evaluation_run`, `evaluator_relationship=third_party`, + `interaction_type=single_turn` — all confirmed from the paper (short-horizon QA; + multi-turn/agentic explicitly out of scope). Generation settings (temp/sampling) + are not documented → omitted. +- **Instances:** `input.raw` = the user/system turns only, so the answer never leaks + into the input. `output.raw` = the assistant turn(s), i.e. the model's *full* + generation — a row with no assistant turn gets an empty list rather than the + scorer's parsed answer. The parsed answer goes in + `answer_attribution.extracted_value`, with `source` naming where it came from + (`answer`, or `scores..answer` when that column is empty) and + `extraction_method` = the real Inspect scorer (the `scores` key: `match`/`choice`/…). + `sample_hash` = sha256 over canonical JSON of `{"raw", "reference"}` — the shared + cross-adapter recipe, so the same item joins across adapters. + Instances attach to the finest-grain result only: every item belongs to exactly one + subtask, so re-emitting them under the overall would duplicate ~7.5M rows for no + new information. +- **Aggregation:** per-item binary `score` → `accuracy` (`continuous [0,1]`) with an + analytic proportion `standard_error` + `num_samples`. Token means cover only the + rows that carried complete token usage. A benchmark with ≤1 distinct subtask emits only the overall + `wild.` result (no duplicate leaf); multi-subtask benchmarks emit the + overall + one `wild..` per subtask (`metric_parameters` marks + overall vs subtask; `micro` pooling). +- Requires `pyarrow` (parquet reads), declared as the `wild` extra — install it + first with `uv sync --extra wild` (or `uv sync --all-extras`); a fresh env without + it fails at import. +- Aggregation reads only the small columns (fast); `--include-instances` reads the + full rows and is the expensive path — use `--limit-shards` / `--max-instances` + for smoke runs. + +## Benchmark canonicalization (eval-card-registry follow-up) + +`evaluation_name` is `wild.[.]` and `source_data` carries the dataset +repo; canonicalizing the benchmark id is the registry's job. Of WILD's 27 +benchmarks, 20 resolve today; the rest are follow-ups for the registry: +- **Add aliases** to existing canonicals: `arc_easy`, `arc_challenge` → + `ai2-reasoning-challenge-arc` (AI2 ARC, *not* ARC-AGI); `race_h` → `race`. +- **New canonicals** (genuinely absent from the registry; all sources now + resolved & public): `squad` (`rajpurkar/squad`), `paws` + (`google-research-datasets/paws`), `chembench` (`jablonkagroup/ChemBench`), + `finance_fundamentals` (`kensho/bizbench` — a curated subset), `pre_flight` + (`AirsideLabs/pre-flight-06`, an Inspect Evals task). Also `bbeh` (`BBEH/bbeh`) + and `aime` (`Maxwell-Jia/AIME_2024` / `math-ai/aime25`) resolve today but their + dataset repos are worth recording. +- The AI2-ARC canonical currently has no `dataset_repo`; `allenai/ai2_arc` should be + added, and easy/challenge are collapsed into one canonical upstream. diff --git a/every_eval_ever/adapters/wild/__init__.py b/every_eval_ever/adapters/wild/__init__.py new file mode 100644 index 000000000..04e6ef1c2 --- /dev/null +++ b/every_eval_ever/adapters/wild/__init__.py @@ -0,0 +1 @@ +"""WILD-raw adapter package.""" diff --git a/every_eval_ever/adapters/wild/adapter.py b/every_eval_ever/adapters/wild/adapter.py new file mode 100644 index 000000000..fa95d3e6b --- /dev/null +++ b/every_eval_ever/adapters/wild/adapter.py @@ -0,0 +1,786 @@ +#!/usr/bin/env python3 +"""Convert kensho/WILD-raw (arXiv:2604.01418) into Every Eval Ever records. + +WILD-raw is item-level evaluation data: ~7.5M (model, item) rows for 65 models +across 27 benchmarks, run by Kensho with Inspect AI. One aggregate log per +(model, benchmark) holds the overall accuracy plus one result per subtask; +`--include-instances` also writes the per-item instance sidecar. See README.md. + +Run: + uv run python -m every_eval_ever.adapters.wild.adapter --output-dir /tmp/eee-wild/data/wild --limit-shards 1 + uv run python -m every_eval_ever validate '/tmp/eee-wild/data/wild/*/*/*.json*' +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import re +import tempfile +import time +import uuid +from collections import defaultdict +from dataclasses import dataclass +from datetime import timezone +from pathlib import Path +from typing import Iterator + +import pyarrow.parquet as pq + +from every_eval_ever.converters.common.publication import ( + publish_evaluation_logs, +) +from every_eval_ever.eval_types import ( + DetailedEvaluationResults, + EvalLibrary, + EvaluationLog, + EvaluationResult, + EvaluatorRelationship, + Format, + HashAlgorithm, + MetricConfig, + ModelInfo, + ScoreDetails, + ScoreType, + SourceDataHf, + SourceDataPrivate, + SourceMetadata, +) +from every_eval_ever.helpers import SCHEMA_VERSION +from every_eval_ever.helpers.developer import get_developer +from every_eval_ever.helpers.io import ( + SourceConversionResult, + SourceRecordFailure, + datastore_output_dir, + datastore_repo_file_path, + default_failure_report_path, + save_failure_report, +) +from every_eval_ever.instance_level_types import ( + AnswerAttributionItem, + Evaluation, + Input, + InstanceLevelEvaluationLog, + InteractionType, + Output, + TokenUsage, +) + +HF_REPO_ID = 'kensho/WILD-raw' +HF_REVISION = 'main' +N_SHARDS = 15 +BATCH_ROWS = 20_000 # rows held in memory per read; see iter_batches +COLLECTION = 'wild' +DEFAULT_OUTPUT_DIR = f'data/{COLLECTION}' +SOURCE_NAME = 'WILD-raw' +SOURCE_ORGANIZATION = 'Kensho' +HF_DATASET_URL = f'https://huggingface.co/datasets/{HF_REPO_ID}' +PAPER_URL = 'https://arxiv.org/abs/2604.01418' + +# WILD task -> the dataset the eval ran on (each verified to exist on HF), NOT +# kensho/WILD-raw, which holds the results. Tasks with no clean public repo use the +# `other` variant. Canonicalizing the benchmark id is the eval-card-registry's job. +WILD_DATASET_REPO = { + 'arc_easy': 'allenai/ai2_arc', 'arc_challenge': 'allenai/ai2_arc', + 'bbh': 'lukaemon/bbh', 'bigcodebench': 'bigcode/bigcodebench', + 'boolq': 'google/boolq', 'chembench': 'jablonkagroup/ChemBench', + 'commonsense_qa': 'tau/commonsense_qa', 'drop': 'ucinlp/drop', + 'gsm8k': 'openai/gsm8k', 'gsm_symbolic': 'apple/GSM-Symbolic', + 'hellaswag': 'Rowan/hellaswag', 'ifeval': 'google/IFEval', + 'math': 'hendrycks/competition_math', 'medqa': 'bigbio/med_qa', + 'mmlu': 'cais/mmlu', 'mmlu_pro': 'TIGER-Lab/MMLU-Pro', 'musr': 'TAUR-Lab/MuSR', + 'paws': 'google-research-datasets/paws', 'piqa': 'ybisk/piqa', + 'race_h': 'ehovy/race', 'squad': 'rajpurkar/squad', + 'truthfulqa': 'truthfulqa/truthful_qa', 'winogrande': 'allenai/winogrande', + # provenance resolved from the WILD paper + Inspect Evals loaders: + 'finance_fundamentals': 'kensho/bizbench', 'pre_flight': 'AirsideLabs/pre-flight-06', + 'bbeh': 'BBEH/bbeh', +} +# aime's two subtasks come from different repos (the exact ones Inspect Evals loads). +AIME_REPO_BY_SUBTASK = {'2024': 'Maxwell-Jia/AIME_2024', '2025': 'math-ai/aime25'} + +# item_id is read in the aggregate pass too, to name an unusable row in the report. +AGG_COLUMNS = ['model', 'task', 'subtask', 'item_id', 'score', + 'input_tokens', 'output_tokens'] +INSTANCE_COLUMNS = AGG_COLUMNS + ['conversation', 'target', 'answer', + 'scores', 'stop_reason'] + + +# parquet streaming (HF or local), batched, column-projected + +def _shard_handles(parquet: list[str] | None, limit_shards: int | None, + revision: str | None = None): + """Yield (label, opener) for each parquet source. opener() -> file-like. + `revision` pins the HF commit for remote reads (see resolve_source_revision).""" + if parquet: + sources = parquet + else: + sources = [ + f'datasets/{HF_REPO_ID}/data-{i:05d}-of-{N_SHARDS:05d}.parquet' + for i in range(N_SHARDS) + ] + if limit_shards is not None: + sources = sources[:limit_shards] + for src in sources: + if parquet: # local path + yield src, (lambda s=src: open(s, 'rb')) + else: # HuggingFace + from huggingface_hub import HfFileSystem + fs = HfFileSystem() + rev = revision or HF_REVISION + yield src, (lambda s=src: fs.open(s, revision=rev)) + + +def iter_batches(parquet: list[str] | None, columns: list[str], + limit_shards: int | None = None, + revision: str | None = None, + batch_size: int = BATCH_ROWS) -> Iterator[tuple[str, dict[str, list]]]: + """Yield ``(shard_label, {col: [values]})`` in batches of ``batch_size`` rows. + + A WILD shard is a single row group of 500,000 rows, so reading whole row groups + would hold every selected column for all of them at once — several GB for the + instance columns, even when ``--max-instances`` stops after the first few. + """ + for label, opener in _shard_handles(parquet, limit_shards, revision): + with opener() as fh: + pf = pq.ParquetFile(fh) + for batch in pf.iter_batches(batch_size=batch_size, columns=columns): + yield label, {c: batch.column(c).to_pylist() for c in columns} + + +# aggregation + +@dataclass +class Agg: + n: int = 0 + correct: float = 0.0 + in_tok: int = 0 + out_tok: int = 0 + tok_n: int = 0 # rows with complete token usage — the token-mean divisor + + def add(self, score: float, in_t, out_t): + self.n += 1 + self.correct += score + if in_t is not None and out_t is not None: + self.in_tok += int(in_t) + self.out_tok += int(out_t) + self.tok_n += 1 + + +def item_score(raw) -> float | None: + """The row's binary correctness, or ``None`` when it carries no usable one — a + missing score is not a wrong answer, so it must not be counted as 0.""" + if raw is None: + return None + try: + score = float(raw) + except (TypeError, ValueError): + return None + if not math.isfinite(score) or score not in (0.0, 1.0): + return None + return score + + +def aggregate(parquet, limit_shards, models: set[str] | None, + revision: str | None = None): + """Aggregate item rows into ``{(model, task): {subtask|None: Agg}}``, the ``None`` + key being the benchmark overall. Returns ``(groups, total_rows, failures)``: a row + without a usable 0/1 score is reported, never counted into a denominator.""" + groups: dict[tuple[str, str], dict[str | None, Agg]] = defaultdict( + lambda: defaultdict(Agg)) + failures: list[SourceRecordFailure] = [] + total = 0 + for label, batch in iter_batches(parquet, AGG_COLUMNS, limit_shards, revision): + for model, task, subtask, item_id, raw, in_t, out_t in zip( + batch['model'], batch['task'], batch['subtask'], + batch['item_id'], batch['score'], + batch['input_tokens'], batch['output_tokens']): + if models and model not in models: + continue + total += 1 + score = item_score(raw) + if score is None: + failures.append(SourceRecordFailure( + source_ref=f'{label}#{model}/{task}/{item_id}', + reason=f'score {raw!r} is not a usable binary correctness value', + )) + continue + g = groups[(model, task)] + g[None].add(score, in_t, out_t) # benchmark overall + g[subtask if subtask not in (None, '') else '_'].add(score, in_t, out_t) + return groups, total, failures + + +# record construction + +def _source_data(task: str, n: int, subtask: str | None = None): + """The dataset the eval ran on (not WILD-raw, which holds the results).""" + repo = WILD_DATASET_REPO.get(task) + if task == 'aime' and subtask in AIME_REPO_BY_SUBTASK: + repo = AIME_REPO_BY_SUBTASK[subtask] + if repo: + return SourceDataHf(dataset_name=task, source_type='hf_dataset', + hf_repo=repo, samples_number=n) + return SourceDataPrivate( + dataset_name=task, source_type='other', + additional_details={'note': 'no single public HF dataset repo for this WILD ' + 'task/subtask; results are in ' + HF_REPO_ID}) + + +def metric_details(agg: Agg) -> dict[str, str]: + """Item counts, plus token means over the rows that carried token usage — a row + without token counts leaves the mean rather than averaging in as a zero.""" + details = {'n_items': str(agg.n), 'n_correct': str(int(agg.correct))} + if agg.tok_n: + details['n_items_with_token_usage'] = str(agg.tok_n) + details['mean_input_tokens'] = f'{agg.in_tok / agg.tok_n:.1f}' + details['mean_output_tokens'] = f'{agg.out_tok / agg.tok_n:.1f}' + return details + + +def _result(task: str, subtask: str | None, agg: Agg) -> EvaluationResult: + name = f'wild.{task}' if subtask is None else f'wild.{task}.{subtask}' + rid = task if subtask is None else f'{task}::{subtask}' + accuracy = agg.correct / agg.n if agg.n else 0.0 + # score is binary per item (verified), so accuracy = mean and the analytic + # standard error of a proportion is sqrt(p(1-p)/n). + se = math.sqrt(accuracy * (1 - accuracy) / agg.n) if agg.n else 0.0 + level = 'overall' if subtask is None else 'subtask' + return EvaluationResult( + evaluation_result_id=rid, + evaluation_name=name, + source_data=_source_data(task, agg.n, subtask), + metric_config=MetricConfig( + evaluation_description=( + f'Mean binary item correctness on {name} (WILD-raw).'), + # The registry's canonical global metric: `evaluation_name` keeps the + # tasks apart, so the cross-source accuracy join stays whole. + metric_id='accuracy', + metric_name='accuracy', + metric_kind='accuracy', + metric_unit='proportion', + lower_is_better=False, + score_type=ScoreType.continuous, + min_score=0.0, + max_score=1.0, + metric_parameters={'aggregation_level': level, 'aggregation': 'micro'}, + additional_details=metric_details(agg), + ), + score_details=ScoreDetails( + score=accuracy, + details={'n_items': str(agg.n), 'n_correct': str(int(agg.correct))}, + uncertainty={'standard_error': {'value': se, 'method': 'analytic'}, + 'num_samples': agg.n}, + ), + ) + + +def build_log(model: str, task: str, subs: dict[str | None, Agg], + eval_ts: str, retrieved_ts: str, + revision: str | None = None) -> tuple[EvaluationLog, str, str]: + developer = get_developer(model) + model_slug = model.split('/')[-1] + sanitized = model.replace('/', '_') + real_subs = sorted(k for k in subs if k is not None) + # With ≤1 distinct subtask the overall IS that subtask, so emit only the overall + # (17 WILD tasks have "general" as their only subtask). + results = [_result(task, None, subs[None])] + if len(real_subs) > 1: + for sub in real_subs: + results.append(_result(task, sub, subs[sub])) + # Only claim a dataset_revision for remote reads (a pinned commit). A local + # --parquet run's revision is unknown, so it gets a local marker instead of a + # false remote-provenance claim. + source_details = { + 'dataset_url': HF_DATASET_URL, + 'paper_url': PAPER_URL, + 'note': 'Item-level evals run by Kensho with the Inspect AI framework (WILD paper).', + } + if revision: + source_details['dataset_revision'] = revision + else: + source_details['dataset_source'] = ( + 'local parquet file(s); source WILD-raw revision unknown (not stamped)') + log = EvaluationLog( + schema_version=SCHEMA_VERSION, + # keyed on the stable evaluation time, so reruns are idempotent + evaluation_id=f'wild/{sanitized}/{task}/{eval_ts}', + retrieved_timestamp=retrieved_ts, + evaluation_timestamp=eval_ts, + source_metadata=SourceMetadata( + source_name=SOURCE_NAME, + source_type='evaluation_run', + source_organization_name=SOURCE_ORGANIZATION, + source_organization_url=HF_DATASET_URL, + evaluator_relationship=EvaluatorRelationship.third_party, + additional_details=source_details, + ), + eval_library=EvalLibrary( + name='inspect_ai', version='unknown', + additional_details={'note': 'Run with the Inspect AI framework (WILD paper).'}, + ), + model_info=ModelInfo( + name=model, id=model, developer=developer, + additional_details={'wild_model_id': model}, + ), + evaluation_results=results, + ) + return log, developer, model_slug + + +# instance-level (--include-instances) + +def _sample_hash(raw: str, reference: list[str]) -> str: + """Canonical cross-adapter sample hash: sha256 over canonical JSON of + {"raw", "reference"}. Any other spelling stops joining with the other adapters' + instances for the same item (`every_eval_ever/adapters/openeval`).""" + payload = json.dumps({'raw': raw, 'reference': reference}, + sort_keys=True, separators=(',', ':')) + return hashlib.sha256(payload.encode('utf-8')).hexdigest() + + +def _split_conversation(raw: str | None) -> tuple[str, list[str]]: + """Split the `conversation` column into (prompt, generation_turns). + + prompt = the user + system turns only -> input.raw: answer-free and + model-independent, so it hashes identically across models. generation_turns = the + assistant turn content(s) -> output.raw, the model's full generation.""" + if not raw: + return '', [] + try: + msgs = json.loads(raw) + except (ValueError, TypeError): + return str(raw), [] + if not isinstance(msgs, list): + return str(raw), [] + prompt_parts: list[str] = [] + gen_parts: list[str] = [] + for m in msgs: + if not isinstance(m, dict): + continue + content = m.get('content') + if not content: + continue + text = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False) + role = m.get('role') + if role in ('user', 'system'): + prompt_parts.append(text) + elif role == 'assistant': + gen_parts.append(text) + return '\n\n'.join(prompt_parts), gen_parts + + +def _primary_scorer(scores_json: str | None) -> tuple[str, str]: + """Return (scorer_name, scored_answer) from the SAME scorer entry, so the name + and value can't point at different scorers. The Inspect `scores` map is keyed by + scorer name ('match', 'choice', 'model_graded_qa', …); WILD emits one scorer per + item, and the first key is taken if several ever appear. `scored_answer` is the + scorer's parsed answer, not the model's generation.""" + if scores_json: + try: + scores = json.loads(scores_json) + if scores: + name = str(next(iter(scores))) # the scorer we attribute to + val = scores[name] + ans = val.get('answer') if isinstance(val, dict) else None + return name, (str(ans) if ans else '') + except (ValueError, TypeError, AttributeError, KeyError): + pass + return 'unknown', '' + + +def make_instance(row: dict, evaluation_id: str, model: str, + multi_subtask: bool) -> InstanceLevelEvaluationLog | None: + """One instance record, or ``None`` for a row the aggregate also excluded.""" + task = row['task'] + subtask = row['subtask'] if row['subtask'] not in (None, '') else '_' + # Attach to the finest-grain result: the leaf subtask when the benchmark is + # split, else the lone overall (matching build_log's dedup, so the FK resolves). + # Leaf-only is intentional: every item belongs to exactly one subtask, so linking + # the overall as well would duplicate all ~7.5M rows for no new information. + if multi_subtask: + name, rid = f'wild.{task}.{subtask}', f'{task}::{subtask}' + else: + name, rid = f'wild.{task}', task + score = item_score(row['score']) + if score is None: + return None # reported by the aggregate pass; not scored here + in_t, out_t = row['input_tokens'], row['output_tokens'] + scorer, scored_answer = _primary_scorer(row.get('scores')) + # `source` names where the parsed answer came from, so it is never attributed + # to output.raw, which holds the full generation. + column_answer = str(row.get('answer') or '') + extracted = column_answer or scored_answer + if column_answer: + attribution_source = 'answer' + elif scored_answer: + attribution_source = f'scores.{scorer}.answer' + else: + attribution_source = 'unavailable' + prompt, generation = _split_conversation(row.get('conversation')) + reference = [str(row.get('target') or '')] + # A row with no assistant turn gets an empty list: substituting the parsed + # answer would label scorer data as the model's output. + output_raw = generation + sample_hash = _sample_hash(prompt, reference) + return InstanceLevelEvaluationLog( + schema_version=SCHEMA_VERSION, + evaluation_id=evaluation_id, + model_id=model, + evaluation_name=name, + evaluation_result_id=rid, + sample_id=str(row['item_id']), + sample_hash=sample_hash, + interaction_type=InteractionType.single_turn, + input=Input(raw=prompt, reference=reference), + output=Output(raw=output_raw), + answer_attribution=[AnswerAttributionItem( + turn_idx=0, source=attribution_source, + extracted_value=extracted, + extraction_method=scorer, is_terminal=True)], + evaluation=Evaluation(score=score, is_correct=score == 1.0), + # Omitted rather than zeroed when the row carries no usage. + token_usage=( + TokenUsage(input_tokens=int(in_t), output_tokens=int(out_t), + total_tokens=int(in_t) + int(out_t)) + if in_t is not None and out_t is not None else None), + metadata={'stop_reason': str(row.get('stop_reason') or ''), + 'subtask': str(subtask), 'scorer': scorer}, + ) + + +def write_instances(parquet, limit_shards, models, staged_paths: dict, + eval_ids: dict, multi: set, max_instances: int | None, + revision: str | None = None + ) -> dict[tuple[str, str], tuple[str, int]]: + """Stream item rows into the staged sidecars. Returns ``{key: (sha256, rows)}``. + + The digest is accumulated as the bytes are appended, so no sidecar is ever + re-read or held whole in memory — WILD writes ~7.5M instance rows.""" + digests: dict[tuple[str, str], object] = {} + counts: dict[tuple[str, str], int] = defaultdict(int) + written = 0 + reached_cap = False + for _label, batch in iter_batches(parquet, INSTANCE_COLUMNS, limit_shards, + revision): + # group this batch's rows by (model, task) to bound open handles + buckets: dict[tuple[str, str], list[str]] = defaultdict(list) + for i in range(len(batch['model'])): + key = (batch['model'][i], batch['task'][i]) + if models and key[0] not in models: + continue + if key not in staged_paths: + continue + if max_instances is not None and written >= max_instances: + reached_cap = True + break + row = {c: batch[c][i] for c in INSTANCE_COLUMNS} + inst = make_instance(row, eval_ids[key], key[0], key in multi) + if inst is None: # unusable score: the aggregate skipped it too + continue + buckets[key].append( + json.dumps(inst.model_dump(mode='json', exclude_none=True), + ensure_ascii=False)) + written += 1 + for key, lines in buckets.items(): + payload = ('\n'.join(lines) + '\n').encode('utf-8') + staged_paths[key].parent.mkdir(parents=True, exist_ok=True) + with staged_paths[key].open('ab') as fh: + fh.write(payload) + digests.setdefault(key, hashlib.sha256()).update(payload) + counts[key] += len(lines) + if reached_cap: + break + return {key: (digest.hexdigest(), counts[key]) + for key, digest in digests.items()} + + +# driver + +FULL_SHA_RE = re.compile(r'[0-9a-f]{40}') + + +def resolve_base_output_dir(output_dir: Path) -> Path: + """The datastore root above ``output_dir``, which has to be the collection dir. + + Publication derives ``/wild///`` itself, so it takes the + root rather than the leaf. A path whose last component is not the collection + would silently write beside the one asked for, and the replacement scan would + read that other directory too.""" + if output_dir.name != COLLECTION: + raise SystemExit( + f'--output-dir must end in {COLLECTION!r}, the collection directory ' + f'publication writes into; got {output_dir}. Pass /{COLLECTION} ' + f'(default {DEFAULT_OUTPUT_DIR}).' + ) + return output_dir.parent + + +def resolve_source_revision(override: str | None, + parquet: list[str] | None) -> tuple[str | None, str | None]: + """Pin a concrete commit as ``(revision, commit_timestamp)``, so both passes and + any rerun read one snapshot. Local `--parquet` runs have no remote revision.""" + if parquet: # local files: no remote revision to pin + return None, None + try: + from huggingface_hub import HfApi + info = HfApi().dataset_info(HF_REPO_ID, revision=override or HF_REVISION) + commit_ts = None + if info.lastModified: + dt = info.lastModified + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + commit_ts = repr(dt.timestamp()) + return info.sha, commit_ts # info.sha is the concrete commit SHA + except Exception as exc: # noqa: BLE001 + if override and FULL_SHA_RE.fullmatch(override): + # A commit SHA is already the pin; the lookup only adds the commit + # date, so losing it costs provenance, not reproducibility. Without + # that date resolve_eval_timestamp requires --evaluation-timestamp. + print(f'WARNING: could not read metadata for {HF_REPO_ID}@{override} ' + f'({exc!r}); using it as given, with no commit date.') + return override, None + if override: + raise SystemExit( + f'could not resolve {HF_REPO_ID}@{override} to a commit ({exc!r}), ' + f'and {override!r} is not a commit SHA. A branch or tag can move ' + 'between the aggregate pass and the instance pass, so it cannot ' + 'stand in for the pin the lookup failed to produce. Pass the ' + '40-character commit SHA.' + ) + raise SystemExit( + f'could not resolve {HF_REPO_ID}@{HF_REVISION} to a concrete commit ' + f'({exc!r}). Reading the mutable {HF_REVISION!r} ref would make both ' + 'passes and any rerun read possibly different data. Pass --revision ' + ' to pin a snapshot explicitly.' + ) + + +def resolve_eval_timestamp(override: str | None, + commit_ts: str | None = None) -> str: + """When the evaluation was RUN: an explicit override, else the pinned commit date. + + ``evaluation_id`` is keyed on this, so there is no now() fallback — it would give + the same data a different identity on every run.""" + if override: + return str(override) + if commit_ts: + return commit_ts + raise SystemExit( + 'no evaluation timestamp is available: local --parquet runs carry no source ' + 'commit date, and evaluation_id is keyed on this value, so falling back to ' + 'now() would give identical reruns different logical identities. Pass ' + '--evaluation-timestamp .' + ) + + +def logical_identity(evaluation_id: str) -> str: + """The (model, benchmark) an ``evaluation_id`` is about, without its timestamp. + + ``evaluation_id`` ends in the source commit date, so re-pinning the dataset gives + the same model and benchmark a new id. Replacement keys on this prefix instead, so + a refresh supersedes its own earlier copy however the snapshot moved.""" + return evaluation_id.rsplit('/', 1)[0] + + +def superseded_records(base_output_dir: Path, + logs: list[EvaluationLog]) -> list[Path]: + """Files a previous run published for the (model, benchmark) pairs in ``logs``. + + Filenames are fresh uuid4s, so publishing into a populated target adds a second + copy of a record rather than replacing it. Each candidate is read for its own + ``evaluation_id`` rather than matched on its path, because a path names only the + model: the same directory holds the benchmarks this run does not cover, and a + partial run must leave those alone. A sidecar travels with its aggregate.""" + wanted = {logical_identity(log.evaluation_id) for log in logs} + directories = { + datastore_output_dir(base_output_dir, COLLECTION, log.model_info.id, + log.model_info.developer) + for log in logs + } + found: set[Path] = set() + for directory in sorted(directories): + for path in sorted(directory.glob('*.json')): + try: + published = json.loads(path.read_text())['evaluation_id'] + except (OSError, ValueError, KeyError, TypeError) as exc: + print(f'WARNING: {path} carries no readable evaluation_id ' + f'({exc!r}); leaving it in place. If this run writes the same ' + 'model and benchmark, the directory will hold both.') + continue + if logical_identity(str(published)) not in wanted: + continue + found.add(path) + sidecar = path.with_name(f'{path.stem}_samples.jsonl') + if sidecar.exists(): + found.add(sidecar) + return sorted(found) + + +def publish(logs: list[EvaluationLog], file_uuids: list[str], + base_output_dir: Path, staging_root: Path) -> list[Path]: + """Publish each aggregate together with its sidecar, one log at a time. + + The shared publisher buffers a batch's bytes before creating any file, so one call + over WILD's ~1,700 logs would hold the whole sidecar corpus in memory. Anything + already created is removed if a later log fails.""" + published: list[Path] = [] + try: + for log, file_uuid in zip(logs, file_uuids): + published.extend(publish_evaluation_logs( + [log], base_output_dir, [file_uuid], + staged_output_dir=staging_root, collection_override=COLLECTION)) + except Exception: + for path in reversed(published): + path.with_name(f'{path.stem}_samples.jsonl').unlink(missing_ok=True) + path.unlink(missing_ok=True) + raise + return published + + +def run(args: argparse.Namespace) -> int: + models = set(args.models) if args.models else None + # Checked before any lookup or read, so a mistyped destination costs nothing. + base_output_dir = resolve_base_output_dir(args.output_dir) + revision, commit_ts = resolve_source_revision(args.revision, args.parquet) + # retrieved = when this record was created (now); evaluation = when WILD ran it. + eval_ts = resolve_eval_timestamp(args.evaluation_timestamp, commit_ts) + retrieved_ts = str(args.retrieved_timestamp) if args.retrieved_timestamp else str(time.time()) + print(f'dataset_revision = {revision} | evaluation_timestamp = {eval_ts} ' + f'| retrieved_timestamp = {retrieved_ts}') + + groups, total_rows, failures = aggregate(args.parquet, args.limit_shards, + models, revision) + print(f'aggregated {len(groups)} (model, benchmark) groups ' + f'from {total_rows} item rows') + if models: + matched = {model for model, _task in groups} + if not matched: + raise SystemExit( + f'--models selected {len(models)} model(s) and the source has none ' + f'of them: {", ".join(sorted(models))}. Nothing would be published, ' + 'so the selection is treated as a mistake rather than an empty ' + 'refresh.' + ) + if missing := models - matched: + print(f'WARNING: no source rows for {len(missing)} selected model(s): ' + f'{", ".join(sorted(missing))}') + + keys = sorted(groups) + logs: list[EvaluationLog] = [] + file_uuids: list[str] = [] + for model, task in keys: + log, _developer, _model_slug = build_log(model, task, groups[(model, task)], + eval_ts, retrieved_ts, revision) + logs.append(log) + file_uuids.append(str(uuid.uuid4())) + + unresolved = sorted({log.model_info.id for log in logs + if log.model_info.developer == 'unknown'}) + if unresolved: + raise SystemExit( + f'{len(unresolved)} model id(s) name no publisher, and the datastore ' + f'path needs one: {", ".join(unresolved)}. A flat id is resolved by ' + 'every_eval_ever.helpers.developer; add the model family there rather ' + 'than filing these under a placeholder.' + ) + + # Checked before the instance pass so a rejected rerun costs nothing. + superseded = superseded_records(base_output_dir, logs) + if superseded and not args.replace_existing: + raise SystemExit( + f'{len(superseded)} file(s) under {args.output_dir} already hold the ' + f'model and benchmark pairs this run writes, e.g. {superseded[0]}. ' + 'Filenames are fresh uuid4s, so writing now would add a second copy of ' + 'each rather than replace it. Pass --replace-existing to replace them.' + ) + + with tempfile.TemporaryDirectory(prefix='eee-wild-publication-') as staging: + staging_root = Path(staging) + if args.include_instances: + print('staging instance sidecars…') + multi = {k for k, subs in groups.items() + if len([s for s in subs if s is not None]) > 1} + staged_paths = { + key: datastore_output_dir(staging_root, COLLECTION, + log.model_info.id, + log.model_info.developer) + / f'{file_uuid}_samples.jsonl' + for key, log, file_uuid in zip(keys, logs, file_uuids)} + eval_ids = {key: log.evaluation_id for key, log in zip(keys, logs)} + staged = write_instances(args.parquet, args.limit_shards, models, + staged_paths, eval_ids, multi, + args.max_instances, revision) + for key, log, file_uuid in zip(keys, logs, file_uuids): + if key not in staged: + continue + checksum, rows = staged[key] + log.detailed_evaluation_results = DetailedEvaluationResults( + format=Format.jsonl, + # The full repository-relative path, not the basename: it is what + # the schema, the publisher and the datastore gate all check. + file_path=datastore_repo_file_path( + COLLECTION, log.model_info.id, log.model_info.developer, + f'{file_uuid}_samples.jsonl'), + hash_algorithm=HashAlgorithm.sha256, checksum=checksum, + total_rows=rows) + print(f'staged {sum(rows for _, rows in staged.values())} ' + 'instance records') + + result = SourceConversionResult( + source_name=SOURCE_NAME, total_records=total_rows, + records=logs, failures=failures) + # Written before publication: it accounts for the conversion, so a + # publication that raises must not take the record of what failed with it. + if failures: + print('Unconverted source rows: ' + f'{save_failure_report(result, default_failure_report_path(args.output_dir))}') + + published = publish(logs, file_uuids, base_output_dir, staging_root) + # Removed only once the replacement is in place, so an aborted run leaves + # the previous refresh whole rather than a hole where it used to be. + for path in superseded: + path.unlink(missing_ok=True) + + print(f'wrote {len(published)} aggregate EvaluationLog(s) -> {args.output_dir}') + result.raise_if_incomplete() + return len(published) + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description='Convert kensho/WILD-raw to Every Eval Ever.') + # nargs='+' so a bare --parquet errors instead of silently converting all 15 + # remote shards. + p.add_argument('--parquet', nargs='+', default=None, + help='Local parquet path(s); default fetches the HF shards.') + p.add_argument('--output-dir', type=Path, default=Path(DEFAULT_OUTPUT_DIR)) + p.add_argument('--limit-shards', type=int, default=None, + help='Only read the first N shards (for smoke runs).') + # nargs='+' for the same reason as --parquet: a bare --models must error, not + # parse to [] and quietly convert every model. + p.add_argument('--models', nargs='+', default=None, + help='Filter to these model ids.') + p.add_argument('--include-instances', action='store_true', + help='Also write per-item `_samples.jsonl` instance sidecars.') + p.add_argument('--max-instances', type=int, default=None, + help='Cap total instance rows written (smoke runs).') + p.add_argument('--retrieved-timestamp', default=None, + help='Override the record-creation epoch (default: now).') + p.add_argument('--evaluation-timestamp', default=None, + help='Override when the eval ran (default: the pinned commit date).') + p.add_argument('--replace-existing', action='store_true', + help='Replace the files already published for the model and ' + 'benchmark pairs this run writes; anything else in the ' + 'output directory is left alone. Without it their presence ' + 'is an error, because a rerun would otherwise add a second ' + 'copy of each record rather than replace it.') + p.add_argument('--revision', default=None, + help='Pin a specific kensho/WILD-raw commit SHA/tag for reproducible ' + 'reruns (default: resolve the current main commit and pin that).') + return p.parse_args() + + +if __name__ == '__main__': + written = run(parse_args()) + print(f'Wrote {written} WILD model×benchmark log(s).') diff --git a/pyproject.toml b/pyproject.toml index e252f670f..05bbd38cb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,9 +42,11 @@ helm = [ # because crfm-helm is frozen. "nltk<3.10.1", ] +wild = ["pyarrow>=14.0"] # every_eval_ever/adapters/wild reads WILD-raw parquet via pyarrow all = [ "every-eval-ever[inspect]", "every-eval-ever[helm]", + "every-eval-ever[wild]", ] [project.scripts] diff --git a/tests/test_wild_adapter.py b/tests/test_wild_adapter.py new file mode 100644 index 000000000..fc2c85be3 --- /dev/null +++ b/tests/test_wild_adapter.py @@ -0,0 +1,425 @@ +"""Tests for the WILD-raw adapter (every_eval_ever/adapters/wild/adapter.py). No network — builds a +tiny local parquet and runs the adapter over it.""" +import argparse +import hashlib +import json +import math +import sys + +import pytest + +pytest.importorskip( + 'pyarrow', + reason='pyarrow not installed; the wild adapter needs it (uv sync --extra wild)', +) + +import pyarrow as pa # noqa: E402 +import pyarrow.parquet as pq # noqa: E402 + +from every_eval_ever.adapters.wild import adapter # noqa: E402 +from every_eval_ever.helpers.io import SourceRecordsError # noqa: E402 +from every_eval_ever.validate import validate_file # noqa: E402 + + +def _synth_parquet(path): + rows = [] + for model in ["openai/gpt-x", "01-ai/Yi-1.5-34B-Chat"]: + for subtask in ["algebra", "logic"]: + for i in range(3): + score = 1 if i % 2 == 0 else 0 + # the assistant turn is the model's FULL generation (chain-of-thought); + # it deliberately DIFFERS from the extracted `answer` ("4"/"5") and the + # scorer's parsed answer so tests can prove output.raw != extracted_value. + gen = f"Step by step: two plus two is four. Final answer: {'4' if score else '5'}." + convo = json.dumps([{"role": "user", "content": "What is 2+2?"}, + {"role": "assistant", "content": gen}]) + rows.append(dict( + model=model, task="mmlu", subtask=subtask, + item_id=f"{model[:3]}{subtask[:2]}{i}", score=score, + input_tokens=100 + i, output_tokens=20 + i, conversation=convo, + stop_reason="stop", target="4", answer="4" if score else "5", + scores=json.dumps({"match": {"value": "C" if score else "I", + "answer": "the answer is 4"}}))) + pq.write_table(pa.Table.from_pylist(rows), str(path)) + + +def _out(tmp_path): + # the adapter publishes into /wild//, deriving + # from the output dir's parent, so the output dir must be a `data/`. + return tmp_path / "data" / "wild" + + +def _args(parquet, out, **kw): + base = dict(parquet=[str(parquet)], output_dir=out, limit_shards=None, + models=None, include_instances=False, max_instances=None, + retrieved_timestamp="1700000000.0", evaluation_timestamp="1780000000.0", + revision=None, replace_existing=False) + base.update(kw) + return argparse.Namespace(**base) + + +def test_aggregates(tmp_path): + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + n = adapter.run(_args(pqt, out)) + assert n == 2 # 2 models x 1 benchmark + files = list(out.rglob("*.json")) + assert len(files) == 2 + for f in files: + report = validate_file(f) + assert report.valid, report.errors + log = json.loads(next((out / "openai" / "gpt-x").glob("*.json")).read_text()) + names = {r["evaluation_name"] for r in log["evaluation_results"]} + assert names == {"wild.mmlu", "wild.mmlu.algebra", "wild.mmlu.logic"} + overall = next(r for r in log["evaluation_results"] if r["evaluation_name"] == "wild.mmlu") + # the registry's canonical global metric on every result; the task lives in + # evaluation_name, so a cross-source accuracy join stays joinable + assert {r["metric_config"]["metric_id"] for r in log["evaluation_results"]} == {"accuracy"} + assert overall["metric_config"]["score_type"] == "continuous" + assert (overall["metric_config"]["min_score"], overall["metric_config"]["max_score"]) == (0.0, 1.0) + assert abs(overall["score_details"]["score"] - 2 / 3) < 1e-9 + # analytic proportion SE = sqrt(p(1-p)/n), p=2/3 over n=6 items — regression guard + unc = overall["score_details"]["uncertainty"] + assert abs(unc["standard_error"]["value"] - math.sqrt((2 / 3) * (1 / 3) / 6)) < 1e-9 + assert unc["num_samples"] == 6 + assert log["source_metadata"]["source_type"] == "evaluation_run" + assert log["model_info"]["id"] == "openai/gpt-x" + assert log["evaluation_id"] == "wild/openai_gpt-x/mmlu/1780000000.0" # keyed on eval time + assert log["retrieved_timestamp"] == "1700000000.0" # record-creation time + assert log["evaluation_timestamp"] == "1780000000.0" # when the eval ran + assert log["eval_library"]["name"] == "inspect_ai" + # source_data points at the benchmark's dataset repo, not WILD-raw + assert overall["source_data"]["source_type"] == "hf_dataset" + assert overall["source_data"]["hf_repo"] == "cais/mmlu" + + +def test_single_subtask_dedup(tmp_path): + # a task whose only subtask is "general" must emit ONLY wild. (no dup leaf), + # and instances must attach to the overall result id (task), not task::general. + convo = json.dumps([{"role": "user", "content": "Q?"}, + {"role": "assistant", "content": "ANSWER: C"}]) + rows = [dict(model="openai/gpt-x", task="arc_challenge", subtask="general", + item_id=f"i{i}", score=i % 2, input_tokens=10, output_tokens=2, + conversation=convo, stop_reason="stop", target="C", answer="C", + scores=json.dumps({"choice": {"value": "C", "answer": "ANSWER: C"}})) + for i in range(4)] + pqt = tmp_path / "w.parquet" + pq.write_table(pa.Table.from_pylist(rows), str(pqt)) + out = _out(tmp_path) + adapter.run(_args(pqt, out, include_instances=True)) + log = json.loads(next(out.rglob("*.json")).read_text()) + names = [r["evaluation_name"] for r in log["evaluation_results"]] + assert names == ["wild.arc_challenge"] # deduped: no wild.arc_challenge.general + inst = json.loads(next(out.rglob("*_samples.jsonl")).read_text().splitlines()[0]) + assert inst["evaluation_result_id"] == "arc_challenge" # FK resolves to overall + assert inst["input"]["raw"] == "Q?" # answer NOT leaked in + assert inst["output"]["raw"] == ["ANSWER: C"] # full generation = assistant turn + assert inst["answer_attribution"][0]["extraction_method"] == "choice" # real scorer + assert "sample_hash" in inst + # source_data for arc_challenge -> the AI2 ARC dataset + assert log["evaluation_results"][0]["source_data"]["hf_repo"] == "allenai/ai2_arc" + + +def test_instances(tmp_path): + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + adapter.run(_args(pqt, out, include_instances=True)) + samples = list(out.rglob("*_samples.jsonl")) + assert len(samples) == 2 + for s in samples: + report = validate_file(s) + assert report.valid, report.errors + # aggregate points at its sidecar + agg = next((out / "openai" / "gpt-x").glob("*.json")) + log = json.loads(agg.read_text()) + det = log["detailed_evaluation_results"] + assert det["format"] == "jsonl" and det["total_rows"] == 6 + inst = json.loads(next((out / "openai" / "gpt-x").glob("*_samples.jsonl")).read_text().splitlines()[0]) + assert inst["interaction_type"] == "single_turn" + assert inst["evaluation"]["is_correct"] in (True, False) + assert inst["token_usage"]["total_tokens"] == inst["token_usage"]["input_tokens"] + inst["token_usage"]["output_tokens"] + assert inst["evaluation_name"].startswith("wild.mmlu.") + # output.raw is the model's FULL generation (assistant turn), NOT the parsed answer + full = inst["output"]["raw"] + assert len(full) == 1 and full[0].startswith("Step by step") + ev = inst["answer_attribution"][0]["extracted_value"] + assert ev in ("4", "5") + assert ev != full[0] # generation != extracted answer (regression guard) + # sample_hash uses the canonical cross-adapter recipe over (input.raw, reference) + assert inst["sample_hash"] == adapter._sample_hash(inst["input"]["raw"], inst["input"]["reference"]) + # the sidecar link is the full repository-relative path, not the basename + assert det["file_path"] == ( + f"data/wild/openai/gpt-x/{agg.stem}_samples.jsonl") + assert det["checksum"] == hashlib.sha256( + agg.with_name(f"{agg.stem}_samples.jsonl").read_bytes()).hexdigest() + + +def test_rerun_needs_replace_existing(tmp_path): + # filenames are fresh uuid4s, so a second run into the same directory would add a + # duplicate copy of every evaluation_id instead of replacing it + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + adapter.run(_args(pqt, out, include_instances=True)) + published = sorted(p.name for p in out.rglob("*.json*")) + with pytest.raises(SystemExit, match="--replace-existing"): + adapter.run(_args(pqt, out, include_instances=True)) + assert sorted(p.name for p in out.rglob("*.json*")) == published # untouched + adapter.run(_args(pqt, out, include_instances=True, replace_existing=True)) + replaced = sorted(p.name for p in out.rglob("*.json*")) + assert len(replaced) == len(published) # replaced, not accumulated + assert replaced != published # fresh uuids + + +def test_a_failed_publication_leaves_no_partial_output(tmp_path, monkeypatch): + # a mid-batch failure must not leave half a refresh behind + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + real = adapter.publish_evaluation_logs + calls = [] + + def flaky(*a, **kw): + calls.append(1) + if len(calls) == 2: + raise RuntimeError("boom") + return real(*a, **kw) + + monkeypatch.setattr(adapter, "publish_evaluation_logs", flaky) + with pytest.raises(RuntimeError): + adapter.run(_args(pqt, out, include_instances=True)) + assert not list(out.rglob("*.json*")) + + +def _task_rows(model, task): + convo = json.dumps([{"role": "user", "content": "Q?"}, + {"role": "assistant", "content": "ANSWER: C"}]) + return [dict(model=model, task=task, subtask="main", item_id=f"i{i}", + score=i % 2, input_tokens=10, output_tokens=2, conversation=convo, + stop_reason="stop", target="C", answer="C", + scores=json.dumps({"choice": {"value": "C", "answer": "ANSWER: C"}})) + for i in range(4)] + + +def _one_task_parquet(path, model, task): + pq.write_table(pa.Table.from_pylist(_task_rows(model, task)), str(path)) + + +def test_replacement_supersedes_only_the_benchmarks_it_rewrites(tmp_path): + # one model directory holds every benchmark that model was evaluated on, so a run + # covering one of them must replace its own prior copy and leave the rest alone + mmlu, arc = tmp_path / "mmlu.parquet", tmp_path / "arc.parquet" + _one_task_parquet(mmlu, "openai/gpt-x", "mmlu") + _one_task_parquet(arc, "openai/gpt-x", "arc_challenge") + out = _out(tmp_path) + adapter.run(_args(mmlu, out, include_instances=True)) + kept = {p.name for p in out.rglob("*.json*")} + # a benchmark the target does not hold yet supersedes nothing, so it needs no flag + adapter.run(_args(arc, out, include_instances=True)) + adapter.run(_args(arc, out, include_instances=True, replace_existing=True)) + after = {p.name for p in out.rglob("*.json*")} + assert kept < after # the mmlu aggregate and sidecar survive + assert len(after) == 2 * len(kept) # arc replaced rather than accumulated + assert sorted(json.loads(p.read_text())["evaluation_id"].split("/")[2] + for p in out.rglob("*.json")) == ["arc_challenge", "mmlu"] + + +def test_a_failed_replacement_leaves_the_previous_refresh_in_place(tmp_path, monkeypatch): + # the prior records are removed only after the new ones exist, so a refresh that + # dies mid-publication cannot leave the directory emptier than it started + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + adapter.run(_args(pqt, out, include_instances=True)) + before = {p.name for p in out.rglob("*.json*")} + real = adapter.publish_evaluation_logs + calls = [] + + def flaky(*a, **kw): + calls.append(1) + if len(calls) == 2: + raise RuntimeError("boom") + return real(*a, **kw) + + monkeypatch.setattr(adapter, "publish_evaluation_logs", flaky) + with pytest.raises(RuntimeError): + adapter.run(_args(pqt, out, include_instances=True, replace_existing=True)) + assert {p.name for p in out.rglob("*.json*")} == before + + +def test_output_dir_must_be_the_collection_directory(tmp_path): + # publication derives /wild// itself, so any other leaf + # would write beside the directory asked for — and be scanned for replacement too + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + with pytest.raises(SystemExit, match="must end in 'wild'"): + adapter.run(_args(pqt, tmp_path / "data" / "wild-v2")) + assert not list((tmp_path / "data").rglob("*.json*")) + assert adapter.resolve_base_output_dir(_out(tmp_path)) == tmp_path / "data" + + +def test_models_filter_matching_nothing_publishes_nothing(tmp_path, capsys): + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + with pytest.raises(SystemExit, match="the source has none of them"): + adapter.run(_args(pqt, out, models=["openai/gpt-y"])) + assert not list(out.rglob("*.json*")) + # a partly-matching selection is a warning, not an error: what matched is real + adapter.run(_args(pqt, out, models=["openai/gpt-x", "openai/gpt-y"])) + assert len(list(out.rglob("*.json"))) == 1 + assert "no source rows for 1 selected model(s): openai/gpt-y" in capsys.readouterr().out + + +def _flat_id_parquet(path, models): + rows = [row for model in models for row in _task_rows(model, "mmlu")] + pq.write_table(pa.Table.from_pylist(rows), str(path)) + + +def test_model_ids_without_a_namespace_still_name_a_publisher(tmp_path): + # 15 of WILD's 65 models are named without one ("gpt-4o", "nova-pro"), and the + # datastore path needs a publisher for every record. + pqt = tmp_path / "w.parquet" + _flat_id_parquet(pqt, ["gpt-4o", "claude-3-haiku", "llama-3.1-8b", "nova-pro"]) + out = _out(tmp_path) + adapter.run(_args(pqt, out)) + assert {p.relative_to(out).parts[:2] for p in out.rglob("*.json")} == { + ("openai", "gpt-4o"), ("anthropic", "claude-3-haiku"), + ("meta", "llama-3.1-8b"), ("amazon", "nova-pro")} + for path in out.rglob("*.json"): + assert validate_file(path).valid + + +def test_a_model_that_names_no_publisher_is_named_in_the_error(tmp_path): + # the path helper's own refusal says only "model_info.developer must be known", + # which does not say which of the run's models it was + pqt = tmp_path / "w.parquet" + _flat_id_parquet(pqt, ["entirely-novel-model"]) + out = _out(tmp_path) + with pytest.raises(SystemExit, match="entirely-novel-model"): + adapter.run(_args(pqt, out)) + assert not list(out.rglob("*.json*")) + + +def test_bare_models_flag_is_an_error(monkeypatch): + # nargs='+', so `--models` with nothing after it cannot parse to [] and then + # convert every model in the source + monkeypatch.setattr(sys, "argv", ["adapter", "--models"]) + with pytest.raises(SystemExit): + adapter.parse_args() + + +def test_symbolic_revision_cannot_stand_in_for_a_failed_pin(monkeypatch, capsys): + # the lookup is what turns 'main' into a commit; if it fails, 'main' still moves + # between the aggregate pass and the instance pass, so it is not a pin + monkeypatch.setitem(sys.modules, "huggingface_hub", None) + with pytest.raises(SystemExit, match="40-character commit SHA"): + adapter.resolve_source_revision("main", None) + with pytest.raises(SystemExit, match="--revision"): + adapter.resolve_source_revision(None, None) + sha = "a" * 40 + # a SHA is already the pin; only the commit date is lost, and without it + # resolve_eval_timestamp demands --evaluation-timestamp rather than guessing + assert adapter.resolve_source_revision(sha, None) == (sha, None) + assert "no commit date" in capsys.readouterr().out + + +def _unusable_score_parquet(path): + convo = json.dumps([{"role": "user", "content": "Q?"}, + {"role": "assistant", "content": "A"}]) + rows = [dict(model="openai/gpt-x", task="gsm8k", subtask="main", + item_id=f"i{i}", score=score, input_tokens=in_tok, + output_tokens=out_tok, conversation=convo, stop_reason="stop", + target="4", answer="4", + scores=json.dumps({"match": {"value": "C", "answer": "4"}})) + for i, (score, in_tok, out_tok) in enumerate( + [(1, 10, 2), (0, 20, 4), (None, 30, 6), (1, None, 8)])] + pq.write_table(pa.Table.from_pylist(rows), str(path)) + + +def test_unusable_score_is_reported_not_counted(tmp_path): + # a missing score is not a wrong answer: it must leave the denominator, be named + # in the failure report, skip the sidecar, and make the run exit non-zero + pqt = tmp_path / "w.parquet" + _unusable_score_parquet(pqt) + out = _out(tmp_path) + with pytest.raises(SourceRecordsError): + adapter.run(_args(pqt, out, include_instances=True)) + log = json.loads(next(out.rglob("*.json")).read_text()) + overall = log["evaluation_results"][0] + assert overall["score_details"]["uncertainty"]["num_samples"] == 3 # not 4 + assert abs(overall["score_details"]["score"] - 2 / 3) < 1e-9 + report = json.loads( + adapter.default_failure_report_path(out).read_text()) + assert report["total_source_records"] == 4 + assert len(report["failed_records"]) == 1 + assert report["failed_records"][0]["source_ref"].endswith("gpt-x/gsm8k/i2") + lines = next(out.rglob("*_samples.jsonl")).read_text().splitlines() + assert [json.loads(line)["sample_id"] for line in lines] == ["i0", "i1", "i3"] + assert log["detailed_evaluation_results"]["total_rows"] == 3 + + +def test_incomplete_token_usage_is_omitted_not_zeroed(tmp_path): + pqt = tmp_path / "w.parquet" + _unusable_score_parquet(pqt) + out = _out(tmp_path) + with pytest.raises(SourceRecordsError): + adapter.run(_args(pqt, out, include_instances=True)) + details = json.loads(next(out.rglob("*.json")).read_text())[ + "evaluation_results"][0]["metric_config"]["additional_details"] + # i3 carries no input_tokens, so it is out of the mean rather than a zero in it + assert details["n_items_with_token_usage"] == "2" + assert details["mean_input_tokens"] == "15.0" + rows = {json.loads(line)["sample_id"]: json.loads(line) + for line in next(out.rglob("*_samples.jsonl")).read_text().splitlines()} + assert "token_usage" not in rows["i3"] + assert rows["i0"]["token_usage"]["total_tokens"] == 12 + + +def test_iter_batches_bounds_rows_per_read(tmp_path): + # the cap must bound the allocation too: a WILD shard is one 500k-row row group, + # so reads are batched rather than per row group + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + sizes = [len(batch["model"]) for _, batch in adapter.iter_batches( + [str(pqt)], ["model"], batch_size=4)] + assert sizes == [4, 4, 4] + + +def test_missing_evaluation_timestamp_is_an_error(): + # evaluation_id is keyed on it, so a now() fallback would give identical reruns + # different logical identities + with pytest.raises(SystemExit, match="--evaluation-timestamp"): + adapter.resolve_eval_timestamp(None, None) + assert adapter.resolve_eval_timestamp(None, "1780000000.0") == "1780000000.0" + + +def test_sample_hash_is_canonical(): + # locks the recipe to the skill's templates/instance_sidecar._sample_hash + expected = hashlib.sha256( + json.dumps({"raw": "Q?", "reference": ["C"]}, sort_keys=True, + separators=(",", ":")).encode("utf-8")).hexdigest() + assert adapter._sample_hash("Q?", ["C"]) == expected + + +def test_split_conversation_separates_prompt_and_generation(): + convo = json.dumps([{"role": "system", "content": "sys"}, + {"role": "user", "content": "Q?"}, + {"role": "assistant", "content": "the full model answer"}]) + prompt, generation = adapter._split_conversation(convo) + assert prompt == "sys\n\nQ?" # user + system only, no assistant + assert generation == ["the full model answer"] # assistant turn -> output.raw + + +def test_local_run_provenance_no_false_revision(tmp_path): + # a local --parquet run must NOT stamp dataset_revision='main' (false remote provenance) + pqt = tmp_path / "w.parquet" + _synth_parquet(pqt) + out = _out(tmp_path) + adapter.run(_args(pqt, out)) + log = json.loads(next((out / "openai" / "gpt-x").glob("*.json")).read_text()) + ad = log["source_metadata"]["additional_details"] + assert "dataset_revision" not in ad # unknown for a local file + assert "local" in ad.get("dataset_source", "").lower() diff --git a/uv.lock b/uv.lock index 4065c6532..e6895b88a 100644 --- a/uv.lock +++ b/uv.lock @@ -866,6 +866,7 @@ all = [ { name = "crfm-helm" }, { name = "inspect-ai" }, { name = "nltk" }, + { name = "pyarrow" }, { name = "typer" }, ] helm = [ @@ -876,6 +877,9 @@ helm = [ inspect = [ { name = "inspect-ai" }, ] +wild = [ + { name = "pyarrow" }, +] [package.dev-dependencies] dev = [ @@ -891,6 +895,7 @@ requires-dist = [ { name = "duckdb", specifier = ">=1.5.2" }, { name = "every-eval-ever", extras = ["helm"], marker = "extra == 'all'" }, { name = "every-eval-ever", extras = ["inspect"], marker = "extra == 'all'" }, + { name = "every-eval-ever", extras = ["wild"], marker = "extra == 'all'" }, { name = "huggingface-hub", specifier = ">=0.36.0,<1.0.0" }, { name = "inspect-ai", marker = "extra == 'inspect'", specifier = ">=0.3.160,<0.4.0" }, { name = "jsonschema", specifier = ">=4.26.0,<5.0.0" }, @@ -898,6 +903,7 @@ requires-dist = [ { name = "nltk", marker = "extra == 'helm'", specifier = "<3.10.1" }, { name = "numpy", specifier = ">=2.4.1" }, { name = "pandas", specifier = ">=2.3.3" }, + { name = "pyarrow", marker = "extra == 'wild'", specifier = ">=14.0" }, { name = "pydantic", specifier = ">=2.12.5,<3.0.0" }, { name = "pyyaml", specifier = ">=6.0.3" }, { name = "requests", specifier = ">=2.32.5,<3.0.0" }, @@ -905,7 +911,7 @@ requires-dist = [ { name = "seaborn", specifier = ">=0.13.2" }, { name = "typer", marker = "extra == 'helm'", specifier = ">=0.12,<1.0" }, ] -provides-extras = ["inspect", "helm", "all"] +provides-extras = ["inspect", "helm", "wild", "all"] [package.metadata.requires-dev] dev = [