From 8780b321d726643b64f10269ac4d38e842ab2874 Mon Sep 17 00:00:00 2001 From: Chathurangi Shyalika Date: Mon, 14 Sep 2026 21:59:10 -0400 Subject: [PATCH 1/2] feat(evaluation): add FMEA scorer Signed-off-by: Chathurangi Shyalika --- docs/fmea-evaluation.md | 40 ++++ src/evaluation/loader.py | 29 ++- src/evaluation/scorers/__init__.py | 4 +- src/evaluation/scorers/fmea.py | 250 +++++++++++++++++++++++ src/evaluation/tests/test_cli.py | 8 + src/evaluation/tests/test_fmea_scorer.py | 134 ++++++++++++ src/evaluation/tests/test_loader.py | 26 +++ 7 files changed, 485 insertions(+), 6 deletions(-) create mode 100644 docs/fmea-evaluation.md create mode 100644 src/evaluation/scorers/fmea.py create mode 100644 src/evaluation/tests/test_fmea_scorer.py diff --git a/docs/fmea-evaluation.md b/docs/fmea-evaluation.md new file mode 100644 index 000000000..0741a7e8a --- /dev/null +++ b/docs/fmea-evaluation.md @@ -0,0 +1,40 @@ +# FMEA evaluation + +The `fmea` scorer evaluates structured FMEA catalogue scenarios without an +LLM call. Its primary score is semantic F1, and a scenario passes when its F1 +is strictly greater than `0.70`. The rubric-weighted composite and strict exact +success remain available as secondary diagnostics. + +For `asset_modes` scenarios, the scorer measures failure-mode set recall, +failure-mode set precision, output-contract compliance, and exact +mode/description pairing. Asset names are compared case-insensitively, but +failure-mode names remain exact because the task requires catalogue spellings. + +For catalogue-analysis scenarios, nested objects are decomposed into atomic +key-value facts. Mappings are order-independent, while arrays such as rankings +are compared positionally. Numeric JSON values are compared by value, so `7` +and `7.0` are equivalent. + +The scorer also reports whether the trajectory used both required catalogue +tools and whether it consulted out-of-scope domain servers. These process +fields are diagnostics and do not change the current answer score. + +The aggregate `score_avg` is semantic macro-F1. Per-scenario details include +`semantic_f1`, `f1_pass_threshold`, `rubric_weighted_score`, and +`strict_success`. + +Run the MiniMax FMEA evaluation with: + +```bash +uv run evaluate \ + --trajectories /Users/chathurangishyalika/Documents/AssetOpsBenchRuns/trajectories/stirrup_agent/tokenrouter-MiniMax-M3 \ + --scenarios /Users/chathurangishyalika/Documents/AssetOpsBenchScenarioGeneration/scenarios_data \ + --scenario-ids fmea_all \ + --scorer-default fmea \ + --reports-dir /Users/chathurangishyalika/Documents/AssetOpsBenchRuns/reports/stirrup_agent/tokenrouter-MiniMax-M3-fmea +``` + +`reference_answer.json` currently stores summary baseline metrics rather than +the baseline output. The scorer therefore records its reported F1 but does not +apply the stated RACE normalization; that requires running the actual baseline +answer through the same scorer. diff --git a/src/evaluation/loader.py b/src/evaluation/loader.py index c8fbf6dd9..c799aaee3 100644 --- a/src/evaluation/loader.py +++ b/src/evaluation/loader.py @@ -101,12 +101,31 @@ def _load_scenario_dir(path: Path) -> list[Scenario]: expected_answer = groundtruth_path.read_text(encoding="utf-8").strip() eval_metadata_path = child / "groundtruth_eval.json" - evaluation_metadata = None + evaluation_metadata: dict = {} if eval_metadata_path.exists(): - evaluation_metadata = json.loads( - eval_metadata_path.read_text(encoding="utf-8") + evaluation_metadata.update( + json.loads(eval_metadata_path.read_text(encoding="utf-8")) ) + # Keep scenario-specific scoring inputs separate from the canonical + # answer while making them available to custom deterministic scorers. + scoring_method = None + for filename, metadata_key in ( + ("scenario_meta.json", "scenario_meta"), + ("rubric.json", "rubric"), + ("reference_answer.json", "reference_answer"), + ): + metadata_path = child / filename + if metadata_path.exists(): + metadata_value = json.loads( + metadata_path.read_text(encoding="utf-8") + ) + evaluation_metadata[metadata_key] = metadata_value + if metadata_key == "scenario_meta" and isinstance( + metadata_value, dict + ): + scoring_method = metadata_value.get("scoring_method") + question_path = child / "question.txt" text = ( question_path.read_text(encoding="utf-8").strip() @@ -121,8 +140,8 @@ def _load_scenario_dir(path: Path) -> list[Scenario]: "text": text, "type": "structured", "expected_answer": expected_answer, - "evaluation_metadata": evaluation_metadata, - "scoring_method": "static_json", + "evaluation_metadata": evaluation_metadata or None, + "scoring_method": scoring_method or "static_json", } ) ) diff --git a/src/evaluation/scorers/__init__.py b/src/evaluation/scorers/__init__.py index f681844a8..ee333b9eb 100644 --- a/src/evaluation/scorers/__init__.py +++ b/src/evaluation/scorers/__init__.py @@ -47,5 +47,7 @@ def names() -> list[str]: from . import code_based # noqa: E402,F401 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 -_install_static_json() \ No newline at end of file +_install_static_json() +_install_fmea() diff --git a/src/evaluation/scorers/fmea.py b/src/evaluation/scorers/fmea.py new file mode 100644 index 000000000..3d6c418b8 --- /dev/null +++ b/src/evaluation/scorers/fmea.py @@ -0,0 +1,250 @@ +"""Deterministic scoring for structured FMEA catalogue scenarios.""" + +from __future__ import annotations + +import json +from collections import Counter +from typing import Any + +from ..models import Scenario, ScorerResult +from . import register +from .static_json import flatten_answer, normalize_value, parse_structured_answer + +_REQUIRED_CATALOG_TOOLS = { + "utilities__get_asset_catalog", + "utilities__get_failure_mode_catalog", +} +_DOMAIN_PREFIXES = ("fmsr__", "iot__", "tsfm__", "vibration__", "wo__") +_DEFAULT_F1_PASS_THRESHOLD = 0.70 + + +def _strict_json_object(answer: str) -> tuple[dict[str, Any] | None, list[str]]: + """Parse one JSON object and record duplicate keys at any nesting level.""" + duplicates: list[str] = [] + + def object_pairs_hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + counts = Counter(key for key, _ in pairs) + duplicates.extend(sorted(key for key, count in counts.items() if count > 1)) + return dict(pairs) + + try: + parsed = json.loads(answer.strip(), object_pairs_hook=object_pairs_hook) + except (json.JSONDecodeError, TypeError): + return None, [] + return (parsed if isinstance(parsed, dict) else None), duplicates + + +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 _weights(scenario: Scenario) -> dict[str, float]: + defaults = { + "coverage": 0.35, + "precision": 0.35, + "instruction_following": 0.20, + "grounding": 0.10, + } + dimensions = _metadata(scenario, "rubric").get("dimensions", []) + if not isinstance(dimensions, list): + return defaults + + weights = dict(defaults) + for dimension in dimensions: + if not isinstance(dimension, dict): + continue + name, weight = dimension.get("name"), dimension.get("weight") + if name in weights and isinstance(weight, (int, float)): + weights[name] = float(weight) + + total = sum(weights.values()) + return {name: value / total for name, value in weights.items()} if total else defaults + + +def _same_scalar(left: Any, right: Any) -> bool: + return normalize_value(left) == normalize_value(right) + + +def _instruction_score( + strict_object: dict[str, Any] | None, + duplicates: list[str], + expected_top_keys: set[str], +) -> tuple[float, bool, bool]: + json_only = strict_object is not None and not duplicates + actual_top_keys = set(strict_object) if strict_object is not None else set() + exact_top_keys = actual_top_keys == expected_top_keys + return (float(json_only) + float(exact_top_keys)) / 2.0, json_only, exact_top_keys + + +def _tool_diagnostics(trajectory_text: str) -> dict[str, Any]: + try: + trajectory = json.loads(trajectory_text) if trajectory_text else None + except json.JSONDecodeError: + trajectory = None + + names: list[str] = [] + + def visit(value: Any) -> None: + if isinstance(value, dict): + name = value.get("name") + if isinstance(name, str) and "__" in name: + names.append(name) + for nested in value.values(): + visit(nested) + elif isinstance(value, list): + for nested in value: + visit(nested) + + visit(trajectory) + unique = sorted(set(names)) + used_required = sorted(_REQUIRED_CATALOG_TOOLS.intersection(unique)) + missing_required = sorted(_REQUIRED_CATALOG_TOOLS.difference(unique)) + out_of_scope = sorted(name for name in unique if name.startswith(_DOMAIN_PREFIXES)) + return { + "catalog_tools_used": used_required, + "missing_catalog_tools": missing_required, + "out_of_scope_domain_tools": out_of_scope, + "catalog_process_compliant": not missing_required and not out_of_scope, + } + + +def _score_asset_modes( + gold: dict[str, Any], model: dict[str, Any] +) -> tuple[dict[str, float], dict[str, Any], bool]: + gold_raw, model_raw = gold.get("failure_modes"), model.get("failure_modes") + gold_modes = gold_raw if isinstance(gold_raw, dict) else {} + model_modes = model_raw if isinstance(model_raw, dict) else {} + gold_keys, model_keys = set(gold_modes), set(model_modes) + correct_keys = gold_keys & model_keys + correct_pairs = { + key for key in correct_keys if _same_scalar(gold_modes[key], model_modes[key]) + } + + coverage = len(correct_keys) / len(gold_keys) if gold_keys else 1.0 + precision = len(correct_keys) / len(model_keys) if model_keys else float(not gold_keys) + grounding = len(correct_pairs) / len(model_keys) if model_keys else float(not gold_keys) + asset_match = ( + isinstance(model.get("asset"), str) + and isinstance(gold.get("asset"), str) + and model["asset"].strip().casefold() == gold["asset"].strip().casefold() + ) + dimensions = {"coverage": coverage, "precision": precision, "grounding": grounding} + details = { + "asset_match": asset_match, + "gold_modes": sorted(gold_keys), + "returned_modes": sorted(model_keys), + "correct_modes": sorted(correct_keys), + "missing_modes": sorted(gold_keys - model_keys), + "extra_modes": sorted(model_keys - gold_keys), + "description_matches": sorted(correct_pairs), + } + semantic_exact = asset_match and gold_keys == model_keys and len(correct_pairs) == len(gold_keys) + return dimensions, details, semantic_exact + + +def _score_analytics( + gold: dict[str, Any], model: dict[str, Any] +) -> tuple[dict[str, float], dict[str, Any], bool]: + gold_flat, model_flat = flatten_answer(gold), flatten_answer(model) + gold_keys, model_keys = set(gold_flat), set(model_flat) + shared_keys = gold_keys & model_keys + correct_keys = {key for key in shared_keys if gold_flat[key] == model_flat[key]} + coverage = len(correct_keys) / len(gold_keys) if gold_keys else 1.0 + precision = len(correct_keys) / len(model_keys) if model_keys else float(not gold_keys) + grounding = len(correct_keys) / len(shared_keys) if shared_keys else 0.0 + dimensions = {"coverage": coverage, "precision": precision, "grounding": grounding} + details = { + "gold_facts": gold_flat, + "returned_facts": model_flat, + "correct_fact_keys": sorted(correct_keys), + "missing_fact_keys": sorted(gold_keys - model_keys), + "extra_fact_keys": sorted(model_keys - gold_keys), + "wrong_value_keys": sorted(shared_keys - correct_keys), + } + return dimensions, details, gold_flat == model_flat + + +def fmea_score(scenario: Scenario, answer: str, trajectory_text: str) -> ScorerResult: + """Score FMEA content by semantic F1 and retain strict/rubric diagnostics.""" + gold = parse_structured_answer(scenario.expected_answer) + tolerant_model = parse_structured_answer(answer) + strict_model, duplicates = _strict_json_object(answer) + + if not isinstance(gold, dict): + return ScorerResult( + scorer="fmea", + passed=False, + score=0.0, + rationale="FMEA ground truth is not a JSON object.", + details={"parse_error": "invalid_ground_truth"}, + ) + + model = tolerant_model if isinstance(tolerant_model, dict) else {} + meta = _metadata(scenario, "scenario_meta") + tag = str(meta.get("tag", "")) + is_asset_modes = tag == "asset_modes" or ( + set(gold) == {"asset", "failure_modes"} + and isinstance(gold.get("failure_modes"), dict) + ) + if is_asset_modes: + dimensions, content_details, semantic_exact = _score_asset_modes(gold, model) + scoring_family = "asset_modes" + else: + dimensions, content_details, semantic_exact = _score_analytics(gold, model) + scoring_family = "catalogue_analytics" + + instruction, json_only, exact_top_keys = _instruction_score(strict_model, duplicates, set(gold)) + dimensions["instruction_following"] = instruction + weights = _weights(scenario) + rubric_weighted_score = sum(weights[name] * dimensions[name] for name in weights) + precision, recall = dimensions["precision"], dimensions["coverage"] + semantic_f1 = 2.0 * precision * recall / (precision + recall) if precision + recall else 0.0 + threshold_value = meta.get("f1_pass_threshold", _DEFAULT_F1_PASS_THRESHOLD) + f1_pass_threshold = ( + float(threshold_value) + if isinstance(threshold_value, (int, float)) + else _DEFAULT_F1_PASS_THRESHOLD + ) + passed = semantic_f1 > f1_pass_threshold + strict_success = semantic_exact and json_only and exact_top_keys + baseline_f1 = _metadata(scenario, "reference_answer").get("f1") + details: dict[str, Any] = { + "family": scoring_family, + "scenario_tag": tag or None, + "dimensions": dimensions, + "weights": weights, + "semantic_f1": semantic_f1, + "f1_pass_threshold": f1_pass_threshold, + "f1_threshold_operator": ">", + "rubric_weighted_score": rubric_weighted_score, + "raw_weighted_score": rubric_weighted_score, + "strict_success": strict_success, + "semantic_exact": semantic_exact, + "json_only": json_only, + "exact_top_level_keys": exact_top_keys, + "duplicate_keys": duplicates, + "baseline_f1_reported": baseline_f1, + "baseline_normalization_applied": False, + **content_details, + **_tool_diagnostics(trajectory_text), + } + rationale = ( + f"Semantic F1={semantic_f1:.3f} " + f"({'>' if passed else '<='} {f1_pass_threshold:.2f}); " + f"coverage={recall:.3f}, precision={precision:.3f}, " + f"rubric_weighted_score={rubric_weighted_score:.3f}." + ) + return ScorerResult( + scorer="fmea", + passed=passed, + score=round(semantic_f1, 6), + rationale=rationale, + details=details, + ) + + +def install(name: str = "fmea") -> None: + """Register the deterministic FMEA scorer.""" + register(name, fmea_score) diff --git a/src/evaluation/tests/test_cli.py b/src/evaluation/tests/test_cli.py index dc050bbd8..a0072d95b 100644 --- a/src/evaluation/tests/test_cli.py +++ b/src/evaluation/tests/test_cli.py @@ -48,3 +48,11 @@ def test_resolve_scenario_ids_loads_lite_yaml_category() -> None: def test_resolve_scenario_ids_is_optional() -> None: assert _resolve_scenario_ids(None) is None + + +def test_resolve_scenario_ids_loads_fmea_category() -> None: + assert _resolve_scenario_ids("fmea_all") == { + "9001", "9003", "9005", "9007", "9009", "9011", + "9013", "9015", "9017", "9019", "9021", "9023", + "9025", "9027", "9029", "9031", "9033", + } diff --git a/src/evaluation/tests/test_fmea_scorer.py b/src/evaluation/tests/test_fmea_scorer.py new file mode 100644 index 000000000..98e89c52a --- /dev/null +++ b/src/evaluation/tests/test_fmea_scorer.py @@ -0,0 +1,134 @@ +import json + +import pytest + +from evaluation.models import Scenario +from evaluation.scorers.fmea import fmea_score + + +def _scenario(gold, *, tag="asset_modes", weights=None): + dimensions = weights or [ + {"name": "coverage", "weight": 0.35}, + {"name": "precision", "weight": 0.35}, + {"name": "instruction_following", "weight": 0.2}, + {"name": "grounding", "weight": 0.1}, + ] + return Scenario( + id="9001", + text="", + expected_answer=json.dumps(gold), + evaluation_metadata={ + "scenario_meta": {"tag": tag}, + "rubric": {"dimensions": dimensions}, + }, + ) + + +def test_asset_modes_awards_set_precision_and_recall(): + scenario = _scenario({"asset": "Electric Motor", "failure_modes": {"A": "", "B": ""}}) + answer = json.dumps({"asset": "electric motor", "failure_modes": {"A": "", "C": ""}}) + + result = fmea_score(scenario, answer, "") + + assert result.passed is False + assert result.details["dimensions"]["coverage"] == 0.5 + assert result.details["dimensions"]["precision"] == 0.5 + assert result.details["asset_match"] is True + assert result.details["missing_modes"] == ["B"] + assert result.details["extra_modes"] == ["C"] + assert result.score == 0.5 + + +def test_asset_modes_perfect_answer_passes_case_insensitive_asset(): + gold = {"asset": "Electric Motor", "failure_modes": {"A": ""}} + result = fmea_score( + _scenario(gold), '{"asset":"electric motor","failure_modes":{"A":""}}', "" + ) + assert result.passed is True + assert result.score == 1.0 + + +def test_markdown_fence_gets_content_credit_but_fails_contract(): + gold = {"asset": "Electric Motor", "failure_modes": {"A": ""}} + result = fmea_score(_scenario(gold), f"```json\n{json.dumps(gold)}\n```", "") + assert result.passed is True + assert result.details["semantic_exact"] is True + assert result.details["strict_success"] is False + assert result.details["json_only"] is False + assert result.details["dimensions"]["instruction_following"] == 0.0 + + +def test_analytics_scores_atomic_facts_and_ordered_arrays(): + scenario = _scenario( + { + "ranking": ["Seal", "Bearing", "Motor"], + "counts": {"Seal": 16, "Bearing": 13, "Motor": 7}, + }, + tag="ranking_top3", + ) + answer = json.dumps( + { + "ranking": ["Bearing", "Seal", "Motor"], + "counts": {"Seal": 16, "Bearing": 13, "Motor": 7}, + } + ) + result = fmea_score(scenario, answer, "") + assert result.passed is False + assert result.details["dimensions"]["coverage"] == pytest.approx(4 / 6) + assert result.details["wrong_value_keys"] == ["answer.ranking[0]", "answer.ranking[1]"] + + +def test_duplicate_nested_mode_key_fails_json_contract(): + scenario = _scenario({"asset": "Volute", "failure_modes": {"Corrosion": ""}}) + answer = '{"asset":"Volute","failure_modes":{"Corrosion":"","Corrosion":""}}' + result = fmea_score(scenario, answer, "") + assert result.passed is True + assert result.details["semantic_exact"] is True + assert result.details["strict_success"] is False + assert result.details["duplicate_keys"] == ["Corrosion"] + + +def test_trajectory_reports_catalog_and_out_of_scope_tool_use(): + scenario = _scenario({"asset": "Electric Motor", "failure_modes": {"A": ""}}) + trajectory = json.dumps( + {"turns": [{"tool_calls": [ + {"name": "utilities__get_asset_catalog"}, + {"name": "utilities__get_failure_mode_catalog"}, + {"name": "fmsr__get_failure_modes"}, + ]}]} + ) + result = fmea_score(scenario, scenario.expected_answer, trajectory) + assert result.details["missing_catalog_tools"] == [] + assert result.details["out_of_scope_domain_tools"] == ["fmsr__get_failure_modes"] + assert result.details["catalog_process_compliant"] is False + + +def test_pass_uses_strict_greater_than_point_seven_f1(): + scenario = _scenario( + {"asset": "Electric Motor", "failure_modes": {str(i): "" for i in range(7)}} + ) + below = { + "asset": "Electric Motor", + "failure_modes": { + **{str(i): "" for i in range(4)}, + **{f"extra-{i}": "" for i in range(3)}, + }, + } + result = fmea_score(scenario, json.dumps(below), "") + assert result.score == pytest.approx(4 / 7, abs=1e-6) + assert result.passed is False + assert result.details["f1_pass_threshold"] == 0.7 + + +def test_f1_above_point_seven_passes_without_requiring_exact_match(): + scenario = _scenario( + {"asset": "Electric Motor", "failure_modes": {str(i): "" for i in range(8)}} + ) + partial = { + "asset": "Electric Motor", + "failure_modes": {str(i): "" for i in range(6)}, + } + result = fmea_score(scenario, json.dumps(partial), "") + assert result.score == pytest.approx(6 / 7, abs=1e-6) + assert result.passed is True + assert result.details["strict_success"] is False diff --git a/src/evaluation/tests/test_loader.py b/src/evaluation/tests/test_loader.py index b7841530f..eaf6a85f3 100644 --- a/src/evaluation/tests/test_loader.py +++ b/src/evaluation/tests/test_loader.py @@ -145,3 +145,29 @@ def test_load_scenarios_reads_groundtruth_eval_metadata(tmp_path): "mode": "clarification", "required_terms": ["main unit"], } + + +def test_load_scenarios_reads_custom_scorer_and_fmea_metadata(tmp_path): + scenario_dir = tmp_path / "scenario_9001" + scenario_dir.mkdir() + (scenario_dir / "groundtruth.txt").write_text( + '{"asset": "Electric Motor", "failure_modes": {}}', encoding="utf-8" + ) + (scenario_dir / "scenario_meta.json").write_text( + '{"tag": "asset_modes", "scoring_method": "fmea"}', encoding="utf-8" + ) + (scenario_dir / "rubric.json").write_text( + '{"dimensions": []}', encoding="utf-8" + ) + (scenario_dir / "reference_answer.json").write_text( + '{"f1": 0.2}', encoding="utf-8" + ) + + scenario = load_scenarios(tmp_path)[0] + + assert scenario.scoring_method == "fmea" + assert scenario.evaluation_metadata == { + "scenario_meta": {"tag": "asset_modes", "scoring_method": "fmea"}, + "rubric": {"dimensions": []}, + "reference_answer": {"f1": 0.2}, + } From b8f2f9e845704c1fe26698cf34b59fd77bf9110b Mon Sep 17 00:00:00 2001 From: Chathurangi Shyalika Date: Mon, 14 Sep 2026 22:03:14 -0400 Subject: [PATCH 2/2] Updating README.md Signed-off-by: Chathurangi Shyalika --- docs/fmea-evaluation.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/fmea-evaluation.md b/docs/fmea-evaluation.md index 0741a7e8a..119954a23 100644 --- a/docs/fmea-evaluation.md +++ b/docs/fmea-evaluation.md @@ -27,11 +27,11 @@ Run the MiniMax FMEA evaluation with: ```bash uv run evaluate \ - --trajectories /Users/chathurangishyalika/Documents/AssetOpsBenchRuns/trajectories/stirrup_agent/tokenrouter-MiniMax-M3 \ - --scenarios /Users/chathurangishyalika/Documents/AssetOpsBenchScenarioGeneration/scenarios_data \ + --trajectories "../stirrup_agent/tokenrouter-MiniMax-M3 \ + --scenarios "../scenarios_data" \ --scenario-ids fmea_all \ --scorer-default fmea \ - --reports-dir /Users/chathurangishyalika/Documents/AssetOpsBenchRuns/reports/stirrup_agent/tokenrouter-MiniMax-M3-fmea + --reports-dir "../reports/stirrup_agent/tokenrouter-MiniMax-M3-fmea" ``` `reference_answer.json` currently stores summary baseline metrics rather than