File size: 9,935 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 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | import json
import math
import tempfile
import unittest
from pathlib import Path
from vons.data import Example, smoke_examples
from vons.evaluation import (
METRIC_VERSION,
Prediction,
accuracy,
brier_score,
coverage,
ece,
macro_f1,
nll,
selective_risk,
summarize,
validate_distribution,
write_report,
)
def _row(identifier: str, label: str | None, *, answerable: bool = True, options: tuple[str, ...] = ("a", "b")) -> Example:
return Example(
id=identifier,
task_group="evaluation",
state="state",
question="choose",
options=options,
label=label,
answerable=answerable,
split="test",
provenance={},
metadata={},
)
def _reject_constant(value: str) -> float:
raise ValueError(f"non-strict JSON constant {value}")
class EvaluationTests(unittest.TestCase):
def test_metrics_include_abstention_and_probability_checks(self) -> None:
rows = smoke_examples()[:3]
predictions = [
Prediction(row.id, row.label, {option: 1.0 if option == row.label else 0.0 for option in row.options}, 1.0, False)
for row in rows
]
self.assertEqual(macro_f1(rows, predictions), 1.0)
self.assertEqual(nll(rows, predictions), 0.0)
self.assertEqual(brier_score(rows, predictions), 0.0)
self.assertEqual(selective_risk(rows, predictions), 0.0)
summary = summarize(rows, predictions)
self.assertIn("macro_f1_answerable", summary)
self.assertEqual(summary["metric_version"], METRIC_VERSION)
self.assertIn("brier_answerable", summary["metric_definitions"])
self.assertEqual(summary["probability_population"], "complete")
def test_missing_predictions_stay_in_accuracy_and_coverage_denominators(self) -> None:
rows = [_row(f"r{index}", "a") for index in range(10)]
predictions = [Prediction("r0", "a", {"a": 0.99, "b": 0.01}, 0.99, False)]
summary = summarize(rows, predictions)
self.assertEqual(summary["accuracy_answerable"], 0.1)
self.assertEqual(summary["coverage"], 0.1)
self.assertEqual(summary["selective_risk"], 0.0)
self.assertEqual(summary["counts"]["missing_prediction_rows"], 9)
self.assertEqual(summary["counts"]["probability_rows"], 1)
self.assertEqual(summary["counts"]["probability_unavailable_answerable_rows"], 9)
self.assertEqual(summary["probability_population"], "incomplete")
self.assertAlmostEqual(summary["nll_answerable"], -math.log(0.99))
def test_abstained_and_error_predictions_are_never_correct_or_covered(self) -> None:
rows = [_row("abstained", "a"), _row("failed", "a")]
predictions = [
Prediction("abstained", "a", {"a": 0.9, "b": 0.1}, 0.9, True),
Prediction("failed", "a", {}, 0.0, False, error="timeout"),
]
summary = summarize(rows, predictions)
self.assertEqual(accuracy(rows, predictions), 0.0)
self.assertEqual(macro_f1(rows, predictions), 0.0)
self.assertEqual(coverage(rows, predictions), 0.0)
self.assertIsNone(selective_risk(rows, predictions))
self.assertEqual(summary["counts"]["abstained_prediction_rows"], 1)
self.assertEqual(summary["counts"]["error_prediction_rows"], 1)
self.assertEqual(summary["counts"]["covered_rows"], 0)
self.assertEqual(summary["counts"]["probability_rows"], 1)
def test_covered_unanswerable_prediction_is_a_selective_risk_error(self) -> None:
rows = [_row("answerable", "a"), _row("unanswerable", None, answerable=False)]
predictions = [
Prediction("answerable", "a", {"a": 0.9, "b": 0.1}, 0.9, False),
Prediction("unanswerable", "a", {"a": 0.9, "b": 0.1}, 0.9, False),
]
self.assertEqual(selective_risk(rows, predictions), 0.5)
self.assertEqual(coverage(rows, predictions), 1.0)
self.assertAlmostEqual(ece(rows, predictions), 0.4)
self.assertEqual(summarize(rows, predictions)["counts"]["covered_unanswerable_rows"], 1)
def test_undefined_metrics_are_none(self) -> None:
rows = [_row("a", "a"), _row("b", "b")]
abstained = [
Prediction("a", None, {"a": 0.2, "b": 0.8}, 0.2, True),
Prediction("b", None, {"a": 0.9, "b": 0.1}, 0.1, True),
]
self.assertIsNone(selective_risk(rows, abstained))
self.assertIsNone(ece(rows, abstained))
self.assertEqual(coverage(rows, abstained), 0.0)
unanswerable = [_row("u", None, answerable=False)]
answered = [Prediction("u", "a", {"a": 0.5, "b": 0.5}, 0.5, False)]
for metric in (accuracy, macro_f1, nll, brier_score):
self.assertIsNone(metric(unanswerable, answered))
self.assertIsNone(coverage([], []))
def test_macro_f1_ignores_choices_outside_answerable_rows(self) -> None:
rows = [_row("a", "a", options=("a", "b", "zzz")), _row("b", "b", options=("a", "b", "zzz")), _row("u", None, answerable=False, options=("a", "b", "zzz"))]
predictions = [
Prediction("a", "a", {}, 0.9, False),
Prediction("b", "b", {}, 0.9, False),
Prediction("u", "zzz", {}, 0.9, False),
]
self.assertEqual(macro_f1(rows, predictions), 1.0)
def test_unknown_and_duplicate_ids_fail_closed(self) -> None:
rows = [_row("a", "a")]
with self.assertRaisesRegex(ValueError, "unknown example id"):
summarize(rows, [Prediction("ghost", "a", {}, 0.9, False)])
with self.assertRaisesRegex(ValueError, "duplicate prediction"):
summarize(rows, [Prediction("a", "a", {}, 0.9, False), Prediction("a", "b", {}, 0.9, False)])
with self.assertRaisesRegex(ValueError, "duplicate example id"):
summarize([_row("a", "a"), _row("a", "a")], [])
def test_invalid_confidence_and_probability_vectors_fail_closed(self) -> None:
rows = [_row("a", "a")]
for confidence in (math.nan, -0.1, 1.8):
with self.subTest(confidence=confidence), self.assertRaisesRegex(ValueError, "confidence"):
ece(rows, [Prediction("a", "a", {"a": 1.0}, confidence, False)])
for probabilities in ({"a": math.nan, "b": 0.5}, {"a": -0.1, "b": 1.1}, {"a": 0.5, "b": 0.2}, {"a": 0.5, "c": 0.5}):
with self.subTest(probabilities=probabilities), self.assertRaises(ValueError):
nll(rows, [Prediction("a", "a", probabilities, 0.5, False)])
with self.assertRaisesRegex(ValueError, "no choice"):
coverage(rows, [Prediction("a", None, {"a": 1.0}, 1.0, False)])
def test_answerable_row_without_label_fails_closed(self) -> None:
rows = [_row("a", None)]
with self.assertRaisesRegex(ValueError, "needs a label"):
accuracy(rows, [Prediction("a", None, {}, 0.1, True)])
def test_brier_score_is_the_standard_multiclass_sum(self) -> None:
for options, expected in ((("a", "b"), 0.5), (("a", "b", "c", "d"), 0.75)):
rows = [_row("r", "a", options=options)]
uniform = {option: 1.0 / len(options) for option in options}
self.assertAlmostEqual(brier_score(rows, [Prediction("r", "a", uniform, 0.5, False)]), expected)
def test_probability_metrics_are_conditional_on_valid_distributions(self) -> None:
rows = [_row("valid", "a"), _row("failed", "a"), _row("absent", "a"), _row("zero", "a")]
predictions = [
Prediction("valid", "a", {"a": 0.8, "b": 0.2}, 0.8, False),
Prediction("failed", None, {}, 0.0, False, error="crash"),
Prediction("absent", "a", {}, 0.7, False),
Prediction("zero", "b", {"b": 1.0}, 1.0, False),
]
summary = summarize(rows, predictions)
self.assertEqual(summary["counts"]["probability_rows"], 2)
self.assertEqual(summary["counts"]["probability_unavailable_answerable_rows"], 2)
self.assertEqual(summary["counts"]["nll_floor_clipped_rows"], 1)
self.assertEqual(summary["probability_population"], "incomplete")
self.assertAlmostEqual(summary["nll_answerable"], (-math.log(0.8) - math.log(1e-12)) / 2)
self.assertEqual(validate_distribution({"b": 1.0}, ("a", "b")), {"b": 1.0})
def test_ece_uses_last_bin_for_confidence_one(self) -> None:
rows = [_row("a", "a"), _row("b", "b"), _row("c", "a"), _row("u", None, answerable=False)]
predictions = [
Prediction("a", "a", {"a": 0.9, "b": 0.1}, 0.9, False),
Prediction("b", "a", {"a": 0.9, "b": 0.1}, 0.9, False),
Prediction("c", "a", {"a": 1.0}, 1.0, False),
Prediction("u", "b", {"a": 0.3, "b": 0.7}, 0.3, False),
]
self.assertAlmostEqual(ece(rows, predictions), 0.275)
with self.assertRaisesRegex(ValueError, "bins"):
ece(rows, predictions, bins=0)
def test_write_report_is_strict_json(self) -> None:
rows = [_row("a", "a")]
predictions = [Prediction("a", "a", {"a": 1.0}, 1.0, False, latency_ms=1.5)]
with tempfile.TemporaryDirectory() as directory:
invalid = Path(directory) / "invalid.json"
with self.assertRaises(ValueError):
write_report(invalid, config={}, summary={"nll_answerable": math.nan}, predictions=predictions)
self.assertFalse(invalid.exists())
valid = Path(directory) / "valid.json"
write_report(valid, config={"model_id": "m"}, summary=summarize(rows, predictions), predictions=predictions)
payload = json.loads(valid.read_text(encoding="utf-8"), parse_constant=_reject_constant)
self.assertEqual(payload["summary"]["metric_version"], METRIC_VERSION)
self.assertEqual(payload["summary"]["latency_p50_ms"], 1.5)
if __name__ == "__main__":
unittest.main()
|