File size: 7,967 Bytes
49ad2ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | """Run the restricted Mind2Web candidate/selection evaluation.
The input rows must be normalized JSONL rows accepted by
``Mind2WebExample.from_mapping``. Predictions are intentionally a separate
JSONL file so retrieval misses and missing selections remain observable.
"""
from __future__ import annotations
import argparse
import hashlib
import json
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any, NamedTuple
from vons.mind2web import Mind2WebExample, evaluate_mind2web, task_bootstrap_intervals
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
with path.open(encoding="utf-8") as handle:
for line_number, line in enumerate(handle, start=1):
if not line.strip():
continue
value = json.loads(line)
if not isinstance(value, Mapping):
raise TypeError(f"{path}:{line_number}: expected a JSON object")
rows.append(dict(value))
return rows
def _file_sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _string_sequence(value: Any, label: str) -> tuple[str, ...]:
if isinstance(value, (str, bytes, bytearray)) or not isinstance(value, Sequence):
raise TypeError(f"{label} must be a JSON list")
values: list[str] = []
for item in value:
candidate = item.get("id") if isinstance(item, Mapping) else item
if not isinstance(candidate, str) or not candidate:
raise TypeError(f"{label} ids must be non-empty strings")
values.append(candidate)
if len(values) != len(set(values)):
raise ValueError(f"{label} ids must be unique")
return tuple(values)
class PredictionBundle(NamedTuple):
generated_candidates: dict[str, tuple[str, ...]]
selections: dict[str, str | None]
declared_k: int | None
def load_predictions(path: Path) -> PredictionBundle:
generated_candidates: dict[str, tuple[str, ...]] = {}
selections: dict[str, str | None] = {}
declared_ks: set[int] = set()
missing_k = False
for row in _read_jsonl(path):
example_id = row.get("id")
if not isinstance(example_id, str) or not example_id:
raise ValueError(f"{path}: prediction requires id")
key = example_id
if key in generated_candidates:
raise ValueError(f"{path}: duplicate prediction id {key}")
if "generated_candidates" not in row:
raise ValueError(f"{path}: prediction requires generated_candidates")
generated_candidates[key] = _string_sequence(row["generated_candidates"], "generated_candidates")
selection = row.get("selection", row.get("selected_id"))
selections[key] = None if selection in (None, "") else str(selection)
row_k = row.get("k")
if row_k is None:
missing_k = True
elif isinstance(row_k, bool) or not isinstance(row_k, int) or row_k <= 0:
raise ValueError(f"{path}: prediction k must be a positive integer")
else:
declared_ks.add(row_k)
if declared_ks and missing_k:
raise ValueError(f"{path}: every prediction row must declare k when any row does")
if len(declared_ks) > 1:
raise ValueError(f"{path}: prediction rows must use one k, found {sorted(declared_ks)}")
return PredictionBundle(
generated_candidates=generated_candidates,
selections=selections,
declared_k=next(iter(declared_ks), None),
)
def evaluate_files(
rows_path: Path,
predictions_path: Path,
ks: Sequence[int],
split: str | None = None,
) -> dict[str, Any]:
raw_rows = _read_jsonl(rows_path)
if split is not None:
raw_rows = [row for row in raw_rows if row.get("split") == split]
examples = [Mind2WebExample.from_mapping(row) for row in raw_rows]
if len({example.example_id for example in examples}) != len(examples):
raise ValueError(f"{rows_path}: duplicate gold row ids")
predictions = load_predictions(predictions_path)
generated_candidates = predictions.generated_candidates
example_ids = {row.example_id for row in examples}
unknown_prediction_ids = sorted(set(generated_candidates) - example_ids)
if unknown_prediction_ids:
raise ValueError(f"predictions contain unknown example ids: {unknown_prediction_ids[:5]}")
if not ks or any(k <= 0 for k in ks):
raise ValueError("ks must contain positive integers")
if len(set(ks)) != len(ks):
raise ValueError("ks must not contain duplicates")
if predictions.declared_k is not None and tuple(ks) != (predictions.declared_k,):
raise ValueError(
f"predictions are k={predictions.declared_k}; evaluate exactly that single k, not {list(ks)}"
)
selection_is_k_specific = predictions.declared_k is not None or len(ks) == 1
selections = predictions.selections if selection_is_k_specific else {}
metric_mappings: list[dict[str, Any]] = []
bootstrap_by_k: dict[str, dict[str, Any]] = {}
for k in ks:
metric = evaluate_mind2web(examples, generated_candidates, selections, k=k)
mapping = metric.to_mapping()
if not selection_is_k_specific:
mapping["selection_accuracy_given_recall"] = None
mapping["complete_case_selection_accuracy_given_recall"] = None
mapping["selection_metrics_status"] = "not_evaluated_recall_only"
metric_mappings.append(mapping)
bootstrap = task_bootstrap_intervals(examples, generated_candidates, selections, k=k)
if not selection_is_k_specific:
bootstrap["selection_accuracy_given_recall_task_macro_ci95"] = None
bootstrap_by_k[str(k)] = bootstrap
return {
"schema": "vons.mind2web-evaluation/v1",
"rows_path": str(rows_path),
"predictions_path": str(predictions_path),
"rows": len(examples),
"prediction_rows": len(generated_candidates),
"split": split,
"input_sha256": {
"rows": _file_sha256(rows_path),
"predictions": _file_sha256(predictions_path),
},
"prediction_k": predictions.declared_k,
"metrics_by_k": metric_mappings,
"task_bootstrap_by_k": bootstrap_by_k,
"selection_evaluation": {
"ranking_and_selection_predictions_supplied_once": True,
"selection_reused_across_k": False,
"selection_metrics_valid": selection_is_k_specific,
"interpretation": (
"Selection is evaluated only at the declared single prediction k. "
"When multiple k values are requested without a declared k, metrics are recall-only; "
"separate k-specific prediction files are required for selection."
),
},
"scope": "candidate recall and candidate-in-set selection only; no browser task success",
}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--rows", type=Path, required=True, help="normalized Mind2Web rows JSONL")
parser.add_argument("--predictions", type=Path, required=True, help="retriever/selector predictions JSONL")
parser.add_argument("--output", type=Path, required=True, help="evaluation report JSON")
parser.add_argument("--split", help="evaluate only rows whose normalized split matches this value")
parser.add_argument("--k", type=int, nargs="+", default=[5, 10, 20, 32])
args = parser.parse_args()
report = evaluate_files(args.rows, args.predictions, args.k, split=args.split)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
if __name__ == "__main__":
main()
|