File size: 4,012 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 | import importlib.util
import tempfile
import unittest
from pathlib import Path
from vons.data import smoke_examples, write_jsonl
from vons.evaluation import Prediction, write_report
def _load_summary_tool():
path = Path(__file__).parents[1] / "tools" / "summarize_quality.py"
spec = importlib.util.spec_from_file_location("vons_quality_summary", path)
if spec is None or spec.loader is None:
raise RuntimeError("could not load quality summary tool")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
summary_tool = _load_summary_tool()
class QualitySummaryTests(unittest.TestCase):
def test_recomputes_saved_direct_and_diffusion_quality(self) -> None:
root = Path(__file__).parents[1]
data_path = root / "data/generated/pilot-test.jsonl"
reports = [
root / "reports/pilot-direct-test.json",
root / "reports/pilot-diffusion-test.json",
]
if not all(path.is_file() for path in [data_path, *reports]):
self.skipTest("Historical pilot inputs are intentionally excluded from public source")
result = summary_tool.build_quality_summary(data_path, reports)
self.assertEqual(result["measurement_label"], "measured_synthetic_pilot")
self.assertEqual(result["dataset"]["rows"], 200)
self.assertEqual(result["reports"][0]["metrics"]["accuracy_answerable"], 1.0)
self.assertEqual(result["reports"][1]["metrics"]["coverage"], 0.155)
self.assertIsNone(result["external_evaluation"]["mind2web"]["candidate_recall"])
self.assertEqual(result["external_evaluation"]["mind2web"]["status"], "not_evaluated")
def test_recomputes_metrics_without_private_pilot_files(self) -> None:
rows = smoke_examples()[:2]
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
data_path = root / "synthetic.jsonl"
write_jsonl(data_path, rows)
reports = []
for backend in ("direct", "diffusion"):
predictions = [
Prediction(
row.id, row.label,
{option: float(option == row.label) for option in row.options},
1.0, backend == "diffusion" and index == 1,
)
for index, row in enumerate(rows)
]
report_path = root / f"{backend}.json"
write_report(report_path, config={"backend": backend, "seed": 7},
summary={"coverage": -1}, predictions=predictions)
reports.append(report_path)
result = summary_tool.build_quality_summary(data_path, reports, version="v2")
self.assertEqual(result["dataset"]["rows"], 2)
self.assertEqual(result["metric_version"], "vons.metrics/v2")
direct, diffusion = result["reports"]
self.assertEqual(direct["metrics"]["accuracy_answerable"], 1.0)
self.assertEqual(direct["metrics"]["coverage"], 1.0)
self.assertEqual(diffusion["metrics"]["coverage"], 0.5)
self.assertEqual(diffusion["metrics"]["selective_risk"], 0.0)
self.assertEqual(diffusion["abstained_rows"], 1)
self.assertIsNone(result["external_evaluation"]["mind2web"]["candidate_recall"])
def test_descriptive_stats_preserve_undefined_seed_metrics_as_null(self) -> None:
rows = [
{"backend": "direct", "metrics": {"accuracy_answerable": 1.0}},
{"backend": "direct", "metrics": {"accuracy_answerable": None}},
]
result = summary_tool.descriptive_seed_stats(rows)
metric = result["direct"]["accuracy_answerable"]
self.assertEqual(metric["mean"], 1.0)
self.assertEqual(metric["defined_seed_count"], 1)
self.assertEqual(metric["null_seed_count"], 1)
self.assertEqual(metric["seed_count"], 2)
if __name__ == "__main__":
unittest.main()
|