File size: 6,384 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 | import importlib.util
import json
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
def _load_runner():
path = Path(__file__).parents[1] / "tools" / "evaluate_mind2web.py"
spec = importlib.util.spec_from_file_location("vons_mind2web_runner", path)
if spec is None or spec.loader is None:
raise RuntimeError("could not load Mind2Web runner")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
runner = _load_runner()
class Mind2WebRunnerTests(unittest.TestCase):
def test_runner_preserves_missing_prediction_rows(self) -> None:
with TemporaryDirectory() as directory:
root = Path(directory)
rows_path = root / "rows.jsonl"
predictions_path = root / "predictions.jsonl"
rows_path.write_text(
"\n".join(
[
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}),
json.dumps({"id": "b", "candidate_ids": [], "positive_ids": ["target"]}),
]
)
+ "\n",
encoding="utf-8",
)
predictions_path.write_text(
json.dumps(
{"id": "a", "generated_candidates": ["target"], "selection": "target"}
)
+ "\n",
encoding="utf-8",
)
report = runner.evaluate_files(rows_path, predictions_path, [1, 5])
self.assertEqual(report["rows"], 2)
self.assertEqual(report["prediction_rows"], 1)
self.assertEqual(report["metrics_by_k"][0]["candidate_recall"], 0.5)
self.assertIsNone(report["metrics_by_k"][0]["selection_accuracy_given_recall"])
self.assertFalse(report["selection_evaluation"]["selection_metrics_valid"])
self.assertEqual(len(report["input_sha256"]["rows"]), 64)
self.assertEqual(len(report["input_sha256"]["predictions"]), 64)
self.assertIn("task_bootstrap_by_k", report)
single_k_report = runner.evaluate_files(rows_path, predictions_path, [1])
self.assertEqual(single_k_report["metrics_by_k"][0]["selection_accuracy_given_recall"], 1.0)
def test_runner_requires_prediction_k_to_match_single_requested_k(self) -> None:
with TemporaryDirectory() as directory:
root = Path(directory)
rows_path = root / "rows.jsonl"
predictions_path = root / "predictions.jsonl"
rows_path.write_text(
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}) + "\n",
encoding="utf-8",
)
predictions_path.write_text(
json.dumps({"id": "a", "k": 5, "generated_candidates": ["target"], "selection": "target"})
+ "\n",
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "exactly that single k"):
runner.evaluate_files(rows_path, predictions_path, [10])
def test_runner_filters_normalized_rows_by_split_before_matching(self) -> None:
with TemporaryDirectory() as directory:
root = Path(directory)
rows_path = root / "rows.jsonl"
predictions_path = root / "predictions.jsonl"
rows_path.write_text(
"\n".join(
[
json.dumps({"id": "keep", "split": "test_task", "candidate_ids": [], "positive_ids": ["target"]}),
json.dumps({"id": "drop", "split": "test_domain", "candidate_ids": [], "positive_ids": ["target"]}),
]
)
+ "\n",
encoding="utf-8",
)
predictions_path.write_text(
json.dumps({"id": "keep", "generated_candidates": ["target"], "selection": "target"})
+ "\n",
encoding="utf-8",
)
report = runner.evaluate_files(rows_path, predictions_path, [1], split="test_task")
self.assertEqual(report["split"], "test_task")
self.assertEqual(report["rows"], 1)
self.assertEqual(report["metrics_by_k"][0]["candidate_recall"], 1.0)
def test_runner_rejects_unknown_prediction_ids(self) -> None:
with TemporaryDirectory() as directory:
root = Path(directory)
rows_path = root / "rows.jsonl"
predictions_path = root / "predictions.jsonl"
rows_path.write_text(
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}) + "\n",
encoding="utf-8",
)
predictions_path.write_text(
json.dumps({"id": "unknown", "generated_candidates": []}) + "\n",
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "unknown example ids"):
runner.evaluate_files(rows_path, predictions_path, [5])
def test_runner_rejects_duplicate_gold_ids_and_missing_generated_candidates(self) -> None:
with TemporaryDirectory() as directory:
root = Path(directory)
rows_path = root / "rows.jsonl"
predictions_path = root / "predictions.jsonl"
rows_path.write_text(
"\n".join(
[
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}),
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}),
]
)
+ "\n",
encoding="utf-8",
)
predictions_path.write_text(json.dumps({"id": "a"}) + "\n", encoding="utf-8")
with self.assertRaisesRegex(ValueError, "duplicate gold"):
runner.evaluate_files(rows_path, predictions_path, [5])
rows_path.write_text(
json.dumps({"id": "a", "candidate_ids": [], "positive_ids": ["target"]}) + "\n",
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "generated_candidates"):
runner.evaluate_files(rows_path, predictions_path, [5])
if __name__ == "__main__":
unittest.main()
|