diff --git a/src/evaluation/scorers/__init__.py b/src/evaluation/scorers/__init__.py index ee333b9e..a50f5508 100644 --- a/src/evaluation/scorers/__init__.py +++ b/src/evaluation/scorers/__init__.py @@ -48,6 +48,8 @@ def names() -> list[str]: from . import semantic # noqa: E402,F401 from .static_json import install as _install_static_json # noqa: E402 from .fmea import install as _install_fmea # noqa: E402 +from .fcc import install as _install_fcc # noqa: E402 _install_static_json() _install_fmea() +_install_fcc() diff --git a/src/evaluation/scorers/fcc.py b/src/evaluation/scorers/fcc.py new file mode 100644 index 00000000..ece75c97 --- /dev/null +++ b/src/evaluation/scorers/fcc.py @@ -0,0 +1,279 @@ +"""Deterministic scoring for failure-code classification (FCC) scenarios. + +A scenario opts in through ``scenario_meta.json`` with one of two shapes:: + + {"scoring_method": "fcc", "match": "numeric", "numeric_tolerance": 1} + {"scoring_method": "fcc", "match": "set_f1", "f1_pass_threshold": 0.8} + +``numeric`` passes when the answer is within ``numeric_tolerance`` of the +gold number: a count, a percentage such as ``40.0%``, or a list of numbers +compared element-wise. ``set_f1`` scores a list of failure codes, or of +``{"wo_id": ..., "failure_code": ...}`` objects, by F1 over distinct items +and passes at ``f1_pass_threshold``. Both comparisons are inclusive and are +made on values rounded to 6 decimals, so float error such as +``43.3 - 40.0 = 3.299999999999997`` cannot move an answer across a boundary. + +Bad metadata or ground truth fails the scenario with a rationale instead of +raising, so one malformed file cannot abort an evaluation run. +""" + +from __future__ import annotations + +import math +import re +from typing import Any + +from ..models import Scenario, ScorerResult +from . import register +from .static_json import normalize_value, parse_structured_answer + +_NAME = "fcc" +_NUMBER_RE = re.compile(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)") +# A first line that is only the value: a number, a percentage, or a list. +_BARE_VALUE_RE = re.compile(r"\[.*\]|[+-]?(?:\d+(?:\.\d*)?|\.\d+)\s*%?") +_PARSE_ERRORS = (TypeError, ValueError, RecursionError) + + +def _metadata(scenario: Scenario, key: str) -> dict[str, Any]: + metadata = scenario.evaluation_metadata or {} + value = metadata.get(key, {}) + return value if isinstance(value, dict) else {} + + +def _parse(value: Any) -> Any: + # ast.literal_eval raises TypeError on literals such as "{[1]: 2}". + try: + return parse_structured_answer(value) + except _PARSE_ERRORS: + return value + + +def _normalize(value: Any) -> str: + try: + return normalize_value(value) + except _PARSE_ERRORS: + return str(value).strip().lower() + + +def _number(value: Any) -> float | None: + """Read one finite number; a string may carry a trailing ``%``.""" + if isinstance(value, str): + value = value.strip().removesuffix("%").strip() + if not _NUMBER_RE.fullmatch(value): + return None + elif isinstance(value, bool) or not isinstance(value, (int, float)): + return None + try: + number = float(value) + except OverflowError: + return None + return number if math.isfinite(number) else None + + +def _numbers(value: Any, want_list: bool) -> float | list[float] | None: + if not want_list: + return _number(value) + if not isinstance(value, (list, tuple)): + return None + numbers = [_number(item) for item in value] + return None if any(number is None for number in numbers) else numbers + + +def _setting(meta: dict[str, Any], key: str) -> float | None: + value = meta.get(key) + return _number(value) if isinstance(value, (int, float)) else None + + +def _read_model_numbers( + answer: str, want_list: bool +) -> tuple[float | list[float] | None, str]: + """Return the answer's number(s) and where they were read from.""" + lines = [line.strip() for line in (answer or "").splitlines() if line.strip()] + first_line = lines[0] if lines else "" + # The questions put the value alone on the first line, and some allow a + # free-text line after it. Parsing the whole answer would read "(3)" + # out of "12\nFlag: A (3) + B (9) = 12 is above C (10)". + if _BARE_VALUE_RE.fullmatch(first_line): + value = _numbers(_parse(first_line), want_list) + if value is not None: + return value, "first_line" + value = _numbers(_parse(answer), want_list) + if value is not None: + return value, "answer" + value = _numbers(_parse(first_line), want_list) + if value is not None: + return value, "first_line" + return None, "unreadable" + + +def _format(value: float | list[float] | None) -> str: + if value is None: + return "none" + if isinstance(value, list): + return "[" + ", ".join(f"{item:g}" for item in value) + "]" + return f"{value:g}" + + +def _failed(rationale: str, **details: Any) -> ScorerResult: + return ScorerResult( + scorer=_NAME, passed=False, score=0.0, rationale=rationale, details=details + ) + + +def _score_numeric( + scenario: Scenario, answer: str, meta: dict[str, Any] +) -> ScorerResult: + tolerance = _setting(meta, "numeric_tolerance") + if tolerance is None or tolerance < 0: + return _failed( + "scenario_meta.json needs a non-negative numeric_tolerance for " + f"match 'numeric'; got {meta.get('numeric_tolerance')!r}.", + error="invalid_scenario_meta", + ) + + parsed_gold = _parse(scenario.expected_answer) + want_list = isinstance(parsed_gold, (list, tuple)) + gold = _numbers(parsed_gold, want_list) + if gold is None or gold == []: + return _failed( + "FCC ground truth is not a number, percentage, or non-empty list of numbers.", + error="invalid_ground_truth", + ) + + model, source = _read_model_numbers(answer, want_list) + details: dict[str, Any] = { + "match": "numeric", + "gold": gold, + "model": model, + "model_source": source, + "deltas": [], + "numeric_tolerance": tolerance, + "tolerance_operator": "<=", + } + gold_values = gold if isinstance(gold, list) else [gold] + model_values = model if isinstance(model, list) else [model] + if model is None: + expected = "a list of numbers" if want_list else "a number" + return _failed(f"Could not read {expected} from the answer.", **details) + if len(model_values) != len(gold_values): + return _failed( + f"Answer has {len(model_values)} values, gold has {len(gold_values)}; " + f"answer={_format(model)}, gold={_format(gold)}.", + **details, + ) + + deltas = [round(abs(m - g), 6) for m, g in zip(model_values, gold_values)] + passed = all(delta <= tolerance for delta in deltas) + details["deltas"] = deltas + rationale = ( + f"Max |answer - gold| = {max(deltas):g} " + f"({'<=' if passed else '>'} tolerance {tolerance:g}); " + f"answer={_format(model)}, gold={_format(gold)}." + ) + return ScorerResult( + scorer=_NAME, + passed=passed, + score=1.0 if passed else 0.0, + rationale=rationale, + details=details, + ) + + +def _canonical(item: Any) -> Any: + # An object matches only as a whole, so a right wo_id with a wrong + # failure_code is one missing item plus one extra item. + if isinstance(item, dict): + return tuple( + sorted((str(key).strip().lower(), _normalize(value)) for key, value in item.items()) + ) + return _normalize(item) + + +def _distinct(items: list[Any] | tuple[Any, ...]) -> dict[Any, Any]: + """Map each canonical item to its first original form, in answer order.""" + out: dict[Any, Any] = {} + for item in items: + out.setdefault(_canonical(item), item) + return out + + +def _score_set_f1( + scenario: Scenario, answer: str, meta: dict[str, Any] +) -> ScorerResult: + threshold = _setting(meta, "f1_pass_threshold") + if threshold is None or not 0.0 <= threshold <= 1.0: + return _failed( + "scenario_meta.json needs an f1_pass_threshold in [0, 1] for " + f"match 'set_f1'; got {meta.get('f1_pass_threshold')!r}.", + error="invalid_scenario_meta", + ) + + gold = _parse(scenario.expected_answer) + if not isinstance(gold, (list, tuple)): + return _failed("FCC ground truth is not a list.", error="invalid_ground_truth") + + gold_items = _distinct(gold) + model = _parse(answer) + is_list = isinstance(model, (list, tuple)) + model_items = _distinct(model) if is_list else {} + matched = gold_items.keys() & model_items.keys() + if not is_list: + precision = recall = f1 = 0.0 + else: + precision = len(matched) / len(model_items) if model_items else float(not gold_items) + recall = len(matched) / len(gold_items) if gold_items else 1.0 + total = len(gold_items) + len(model_items) + f1 = 2.0 * len(matched) / total if total else 1.0 + missing = [item for key, item in gold_items.items() if key not in model_items] + extra = [item for key, item in model_items.items() if key not in gold_items] + + # Compare at the reported precision, as the fmea scorer does. + score = round(f1, 6) + passed = score >= threshold + details: dict[str, Any] = { + "match": "set_f1", + "answer_is_list": is_list, + "precision": precision, + "recall": recall, + "f1": f1, + "f1_pass_threshold": threshold, + "f1_threshold_operator": ">=", + "gold_count": len(gold_items), + "model_count": len(model_items), + "matched_count": len(matched), + "missing_items": missing, + "extra_items": extra, + } + if not is_list: + rationale = f"Answer is not a list; F1=0.000 (< {threshold:g})." + else: + rationale = ( + f"Set F1={f1:.3f} ({'>=' if passed else '<'} {threshold:g}); " + f"precision={precision:.3f}, recall={recall:.3f}, " + f"{len(missing)} missing, {len(extra)} extra." + ) + return ScorerResult( + scorer=_NAME, passed=passed, score=score, rationale=rationale, details=details + ) + + +_MATCHERS = {"numeric": _score_numeric, "set_f1": _score_set_f1} + + +def fcc_score(scenario: Scenario, answer: str, trajectory_text: str) -> ScorerResult: + """Score an FCC answer by numeric tolerance or set F1, per scenario_meta.json.""" + meta = _metadata(scenario, "scenario_meta") + match = meta.get("match") + matcher = _MATCHERS.get(match) if isinstance(match, str) else None + if matcher is None: + return _failed( + f"scenario_meta.json match must be one of {sorted(_MATCHERS)}; " + f"got {match!r}.", + error="invalid_scenario_meta", + ) + return matcher(scenario, answer, meta) + + +def install(name: str = _NAME) -> None: + """Register the deterministic FCC scorer.""" + register(name, fcc_score) diff --git a/src/evaluation/tests/test_fcc_scorer.py b/src/evaluation/tests/test_fcc_scorer.py new file mode 100644 index 00000000..a21ed17e --- /dev/null +++ b/src/evaluation/tests/test_fcc_scorer.py @@ -0,0 +1,215 @@ +import json + +import pytest + +from evaluation import scorers as registry +from evaluation.models import Scenario +from evaluation.scorers.fcc import fcc_score + + +def _scenario(gold, meta): + return Scenario( + id="9101", + text="", + expected_answer=gold if isinstance(gold, str) else json.dumps(gold), + evaluation_metadata={"scenario_meta": {"scoring_method": "fcc", **meta}}, + ) + + +def _numeric(gold, tolerance): + return _scenario(gold, {"match": "numeric", "numeric_tolerance": tolerance}) + + +def _set_f1(gold, threshold): + return _scenario(gold, {"match": "set_f1", "f1_pass_threshold": threshold}) + + +def _pairs(codes): + return [ + {"wo_id": f"WO-{index}", "failure_code": code} + for index, code in enumerate(codes, start=1) + ] + + +def test_registered_as_fcc(): + assert registry.get("fcc") is fcc_score + + +@pytest.mark.parametrize(("answer", "passed"), [("11", True), ("13", True), ("10", False)]) +def test_count_passes_within_tolerance_inclusive(answer, passed): + result = fcc_score(_numeric("12", 1), answer, "") + assert result.passed is passed + assert result.score == (1.0 if passed else 0.0) + assert result.details["gold"] == 12 + assert result.details["model"] == float(answer) + assert result.details["deltas"] == [abs(float(answer) - 12)] + assert result.details["numeric_tolerance"] == 1 + assert result.details["tolerance_operator"] == "<=" + + +def test_prose_with_final_bare_number_passes(): + answer = "After filling the blank records I counted them.\n12" + result = fcc_score(_numeric("12", 1), answer, "") + assert result.passed is True + assert result.details["model"] == 12 + assert result.details["model_source"] == "answer" + + +def test_bare_first_line_wins_over_numbers_in_a_flag_line(): + # Parsing the whole answer would pick up the parenthesised "(3)". + answer = "12\nFlag: A (3) + B (9) = 12 is above C (10)" + result = fcc_score(_numeric("12", 1), answer, "") + assert result.passed is True + assert result.details["model"] == 12 + assert result.details["model_source"] == "first_line" + + +def test_first_line_fallback_when_the_whole_answer_is_ambiguous(): + answer = "12 records\nThat is above the 10 and 11 seen last time" + result = fcc_score(_numeric("12", 1), answer, "") + assert result.passed is True + assert result.details["model"] == 12 + assert result.details["model_source"] == "first_line" + + +def test_boolean_is_never_a_number(): + result = fcc_score(_numeric("1", 1), "true", "") + assert result.passed is False + assert result.details["model"] is None + + +@pytest.mark.parametrize( + ("answer", "passed"), + [ + ("43.2%", True), + ("43.3%", True), # 43.3 - 40.0 is 3.299999999999997 in floats + ("36.7%", True), + ("44.0%", False), + ("40.0%\nFlag: the descriptions are too thin to trust", True), + ], +) +def test_percentage_within_tolerance(answer, passed): + result = fcc_score(_numeric("40.0%", 3.3), answer, "") + assert result.passed is passed + assert result.details["gold"] == 40.0 + + +@pytest.mark.parametrize( + ("answer", "passed"), + [ + ("[52.0, 48.0]", True), + ("[51.7, 48.3]", True), + ('["55.0%", "45.0%"]', True), + ("[51.0, 49.0]", False), + ("[55.0]", False), + ("[55.0, 45.0, 0.0]", False), + ("55.0", False), + ], +) +def test_number_list_compares_element_wise(answer, passed): + result = fcc_score(_numeric([55.0, 45.0], 3.3), answer, "") + assert result.passed is passed + + +def test_number_list_reports_each_delta(): + result = fcc_score(_numeric([55.0, 45.0], 3.3), "[51.0, 49.0]", "") + assert result.details["deltas"] == [4.0, 4.0] + assert result.details["model"] == [51.0, 49.0] + + +def test_set_f1_exactly_at_threshold_passes(): + result = fcc_score(_set_f1(["CODE_A", "CODE_B", "CODE_C"], 0.8), '["CODE_A", "CODE_B"]', "") + assert result.score == 0.8 + assert result.passed is True + assert result.details["precision"] == 1.0 + assert result.details["recall"] == pytest.approx(2 / 3) + assert result.details["f1_pass_threshold"] == 0.8 + assert result.details["f1_threshold_operator"] == ">=" + assert result.details["missing_items"] == ["CODE_C"] + assert result.details["extra_items"] == [] + + +def test_set_f1_below_threshold_fails(): + result = fcc_score(_set_f1(["CODE_A", "CODE_B", "CODE_C"], 0.8), '["CODE_A"]', "") + assert result.score == 0.5 + assert result.passed is False + + +def test_set_f1_ignores_order_case_and_duplicates(): + scenario = _set_f1(["CODE_A", "CODE_B", "CODE_C"], 0.8) + result = fcc_score(scenario, '["code_c", "CODE_B", "Code_A", "CODE_B"]', "") + assert result.score == 1.0 + assert result.details["model_count"] == 3 + + +def test_set_f1_non_list_answer_fails(): + result = fcc_score(_set_f1(["CODE_A", "CODE_B"], 0.8), "CODE_A, CODE_B", "") + assert result.passed is False + assert result.score == 0.0 + assert result.details["answer_is_list"] is False + + +def test_empty_gold_and_empty_answer_is_a_perfect_match(): + result = fcc_score(_set_f1([], 0.8), "[]", "") + assert result.passed is True + assert result.score == 1.0 + + +def test_object_pairs_eight_of_ten_passes_at_point_eight(): + gold = _pairs(["CODE_A", "CODE_B", "CODE_C"] * 3 + ["CODE_A"]) + model = [dict(item) for item in gold] + model[0]["failure_code"] = "CODE_C" + model[1]["failure_code"] = "CODE_C" + model[2] = {"WO_ID": " wo-3 ", "failure_code": "code_c"} + result = fcc_score(_set_f1(gold, 0.8), json.dumps(model), "") + assert result.details["matched_count"] == 8 + assert result.score == 0.8 + assert result.passed is True + + +def test_object_with_right_wo_id_and_wrong_code_is_a_miss_and_an_extra(): + gold = _pairs(["CODE_A", "CODE_B"]) + model = _pairs(["CODE_A", "CODE_C"]) + result = fcc_score(_set_f1(gold, 0.8), json.dumps(model), "") + assert result.details["missing_items"] == [gold[1]] + assert result.details["extra_items"] == [model[1]] + assert result.score == 0.5 + assert result.passed is False + + +@pytest.mark.parametrize( + ("gold", "meta"), + [ + ("12", {}), + ("12", {"match": "fuzzy"}), + ("12", {"match": "numeric"}), + ("12", {"match": "numeric", "numeric_tolerance": -1}), + ("12", {"match": "numeric", "numeric_tolerance": True}), + ("12", {"match": "numeric", "numeric_tolerance": "1"}), + ("not a number", {"match": "numeric", "numeric_tolerance": 1}), + ("[]", {"match": "numeric", "numeric_tolerance": 1}), + ('["CODE_A"]', {"match": "set_f1"}), + ('["CODE_A"]', {"match": "set_f1", "f1_pass_threshold": 1.5}), + ('["CODE_A"]', {"match": "set_f1", "f1_pass_threshold": "0.8"}), + ("CODE_A", {"match": "set_f1", "f1_pass_threshold": 0.8}), + ], +) +def test_invalid_meta_or_gold_fails_without_raising(gold, meta): + result = fcc_score(_scenario(gold, meta), "12", "") + assert result.passed is False + assert result.score == 0.0 + assert result.details["error"] in {"invalid_scenario_meta", "invalid_ground_truth"} + assert result.rationale + + +def test_missing_scenario_meta_fails_without_raising(): + scenario = Scenario(id="9101", text="", expected_answer="12") + result = fcc_score(scenario, "12", "") + assert result.passed is False + assert result.details["error"] == "invalid_scenario_meta" + + +@pytest.mark.parametrize("answer", ["{[1]: 2}", '[{"wo_id": "{[1]: 2}"}]', ""]) +def test_unparseable_answer_fails_without_raising(answer): + assert fcc_score(_numeric("12", 1), answer, "").passed is False + assert fcc_score(_set_f1(["CODE_A"], 0.8), answer, "").passed is False