From 02351dc22255d9b31723c220c38f7b6aa208e079 Mon Sep 17 00:00:00 2001 From: Max Rattray Date: Mon, 10 Aug 2026 16:55:58 +0000 Subject: [PATCH 1/2] feat: add failure cohort analysis for evaluation reports Add strands_evals.analysis module with analyze_failure_cohorts() that groups failed cases by evaluator name, sorted largest-first. This helps identify systemic problems (e.g. 14 Faithfulness failures out of 20 total) vs scattered one-off edge cases. New types: - FailureCohort: a group of cases that failed the same evaluator - CohortAnalysis: sorted list of cohorts with summary counts Also includes a print_cohort_summary() Rich display helper. --- src/strands_evals/__init__.py | 3 +- src/strands_evals/analysis/__init__.py | 14 + src/strands_evals/analysis/failure_cohorts.py | 112 +++++ tests/strands_evals/analysis/__init__.py | 0 .../analysis/test_failure_cohorts.py | 418 ++++++++++++++++++ 5 files changed, 546 insertions(+), 1 deletion(-) create mode 100644 src/strands_evals/analysis/__init__.py create mode 100644 src/strands_evals/analysis/failure_cohorts.py create mode 100644 tests/strands_evals/analysis/__init__.py create mode 100644 tests/strands_evals/analysis/test_failure_cohorts.py diff --git a/src/strands_evals/__init__.py b/src/strands_evals/__init__.py index 5dbaee36..b080b12d 100644 --- a/src/strands_evals/__init__.py +++ b/src/strands_evals/__init__.py @@ -1,4 +1,4 @@ -from . import chaos, detectors, evaluators, extractors, generators, providers, simulation, telemetry, types +from . import analysis, chaos, detectors, evaluators, extractors, generators, providers, simulation, telemetry, types from .case import Case from .eval_task_handler import EvalTaskHandler, TracedHandler, eval_task from .evaluation_data_store import EvaluationDataStore @@ -19,6 +19,7 @@ "EvalTaskHandler", "TracedHandler", "eval_task", + "analysis", "chaos", "detectors", "evaluators", diff --git a/src/strands_evals/analysis/__init__.py b/src/strands_evals/analysis/__init__.py new file mode 100644 index 00000000..c6c197ae --- /dev/null +++ b/src/strands_evals/analysis/__init__.py @@ -0,0 +1,14 @@ +"""Analysis utilities for evaluation reports. + +This module provides tools for analyzing evaluation results at the report +level, complementing the per-session detectors module. +""" + +from .failure_cohorts import CohortAnalysis, FailureCohort, analyze_failure_cohorts, print_cohort_summary + +__all__ = [ + "FailureCohort", + "CohortAnalysis", + "analyze_failure_cohorts", + "print_cohort_summary", +] diff --git a/src/strands_evals/analysis/failure_cohorts.py b/src/strands_evals/analysis/failure_cohorts.py new file mode 100644 index 00000000..ce27f4ef --- /dev/null +++ b/src/strands_evals/analysis/failure_cohorts.py @@ -0,0 +1,112 @@ +"""Group failed cases by evaluator for failure cohort analysis. + +Given an EvaluationReport, this module buckets failed cases by the evaluator +that produced them, sorted largest-first. A large bucket indicates a systemic +problem worth investigating; scattered one-off failures are noise. +""" + +from __future__ import annotations + +from pydantic import BaseModel + +from ..types.evaluation_report import EvaluationReport + + +class FailureCohort(BaseModel): + """A group of test cases that all failed the same evaluator.""" + + evaluator_name: str + failed_case_indices: list[int] + failed_case_names: list[str] + count: int + + @property + def is_systemic(self) -> bool: + """Two or more failures suggests a shared root cause.""" + return self.count >= 2 + + +class CohortAnalysis(BaseModel): + """Result of grouping failed cases by evaluator.""" + + cohorts: list[FailureCohort] + total_failures: int + total_cases: int + + @property + def systemic_cohorts(self) -> list[FailureCohort]: + """Return only cohorts with two or more failures.""" + return [c for c in self.cohorts if c.is_systemic] + + @property + def one_off_failures(self) -> list[FailureCohort]: + """Return only cohorts with exactly one failure.""" + return [c for c in self.cohorts if c.count == 1] + + +def analyze_failure_cohorts(report: EvaluationReport) -> CohortAnalysis: + """Group failed cases by evaluator, sorted largest-first. + + Each case in the report has an "evaluator" key identifying which evaluator + produced that row. Failed cases get bucketed by that key. + + Args: + report: An EvaluationReport (typically from Experiment.run_evaluations or + EvaluationReport.from_file). + + Returns: + A CohortAnalysis containing sorted failure cohorts and summary counts. + """ + failures_by_evaluator: dict[str, list[tuple[int, str]]] = {} + + for i, (case, passed) in enumerate(zip(report.cases, report.test_passes, strict=False)): + if passed: + continue + eval_name = case.get("evaluator", "unknown") + case_name = case.get("name", f"case_{i}") + failures_by_evaluator.setdefault(eval_name, []).append((i, case_name)) + + cohorts = [] + for eval_name, members in failures_by_evaluator.items(): + indices = [m[0] for m in members] + names = [m[1] for m in members] + cohorts.append( + FailureCohort( + evaluator_name=eval_name, + failed_case_indices=indices, + failed_case_names=names, + count=len(members), + ) + ) + + cohorts.sort(key=lambda c: (-c.count, c.evaluator_name)) + + return CohortAnalysis( + cohorts=cohorts, + total_failures=sum(1 for p in report.test_passes if not p), + total_cases=len(report.test_passes), + ) + + +def print_cohort_summary(analysis: CohortAnalysis) -> None: + """Print a Rich table summarizing the failure cohorts. + + Args: + analysis: A CohortAnalysis returned by analyze_failure_cohorts. + """ + from rich.console import Console + from rich.table import Table + + console = Console() + table = Table(title=f"Failure Cohorts ({analysis.total_failures}/{analysis.total_cases} failed)") + table.add_column("Evaluator", style="bold") + table.add_column("Count", justify="right") + table.add_column("Cases") + + for cohort in analysis.cohorts: + names = ", ".join(cohort.failed_case_names[:5]) + if cohort.count > 5: + names += f" (+{cohort.count - 5} more)" + table.add_row(cohort.evaluator_name, str(cohort.count), names) + + console.print(table) diff --git a/tests/strands_evals/analysis/__init__.py b/tests/strands_evals/analysis/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/strands_evals/analysis/test_failure_cohorts.py b/tests/strands_evals/analysis/test_failure_cohorts.py new file mode 100644 index 00000000..a41bee79 --- /dev/null +++ b/tests/strands_evals/analysis/test_failure_cohorts.py @@ -0,0 +1,418 @@ +"""Tests for strands_evals.analysis.failure_cohorts module.""" + +from strands_evals.analysis import CohortAnalysis, FailureCohort, analyze_failure_cohorts, print_cohort_summary +from strands_evals.types.evaluation_report import EvaluationReport + + +class TestFailureCohort: + """Tests for the FailureCohort model.""" + + def test_basic_construction(self): + cohort = FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 2, 5], + failed_case_names=["case_0", "case_2", "case_5"], + count=3, + ) + assert cohort.evaluator_name == "Faithfulness" + assert cohort.failed_case_indices == [0, 2, 5] + assert cohort.failed_case_names == ["case_0", "case_2", "case_5"] + assert cohort.count == 3 + + def test_is_systemic_with_multiple_failures(self): + cohort = FailureCohort( + evaluator_name="Correctness", + failed_case_indices=[1, 3], + failed_case_names=["a", "b"], + count=2, + ) + assert cohort.is_systemic is True + + def test_is_systemic_with_single_failure(self): + cohort = FailureCohort( + evaluator_name="Harmfulness", + failed_case_indices=[4], + failed_case_names=["edge_case"], + count=1, + ) + assert cohort.is_systemic is False + + def test_is_systemic_boundary(self): + cohort = FailureCohort( + evaluator_name="X", + failed_case_indices=[0, 1], + failed_case_names=["a", "b"], + count=2, + ) + assert cohort.is_systemic is True + + def test_serialization_roundtrip(self): + cohort = FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 2], + failed_case_names=["case_0", "case_2"], + count=2, + ) + data = cohort.model_dump() + restored = FailureCohort.model_validate(data) + assert restored == cohort + + +class TestCohortAnalysis: + """Tests for the CohortAnalysis model.""" + + def test_systemic_cohorts_property(self): + analysis = CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 1, 2], + failed_case_names=["a", "b", "c"], + count=3, + ), + FailureCohort( + evaluator_name="Harmfulness", + failed_case_indices=[5], + failed_case_names=["e"], + count=1, + ), + ], + total_failures=4, + total_cases=10, + ) + systemic = analysis.systemic_cohorts + assert len(systemic) == 1 + assert systemic[0].evaluator_name == "Faithfulness" + + def test_one_off_failures_property(self): + analysis = CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 1, 2], + failed_case_names=["a", "b", "c"], + count=3, + ), + FailureCohort( + evaluator_name="Harmfulness", + failed_case_indices=[5], + failed_case_names=["e"], + count=1, + ), + ], + total_failures=4, + total_cases=10, + ) + one_offs = analysis.one_off_failures + assert len(one_offs) == 1 + assert one_offs[0].evaluator_name == "Harmfulness" + + def test_empty_cohorts(self): + analysis = CohortAnalysis(cohorts=[], total_failures=0, total_cases=5) + assert analysis.systemic_cohorts == [] + assert analysis.one_off_failures == [] + + def test_serialization_roundtrip(self): + analysis = CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Correctness", + failed_case_indices=[1], + failed_case_names=["x"], + count=1, + ), + ], + total_failures=1, + total_cases=3, + ) + data = analysis.model_dump() + restored = CohortAnalysis.model_validate(data) + assert restored.cohorts == analysis.cohorts + assert restored.total_failures == 1 + assert restored.total_cases == 3 + + +class TestAnalyzeFailureCohorts: + """Tests for the analyze_failure_cohorts function.""" + + def test_all_passing(self): + report = EvaluationReport( + overall_score=1.0, + scores=[1.0, 1.0, 1.0], + cases=[ + {"name": "case-1", "evaluator": "Correctness"}, + {"name": "case-2", "evaluator": "Correctness"}, + {"name": "case-3", "evaluator": "Faithfulness"}, + ], + test_passes=[True, True, True], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == 0 + assert analysis.total_cases == 3 + assert analysis.cohorts == [] + + def test_all_failing_single_evaluator(self): + report = EvaluationReport( + overall_score=0.0, + scores=[0.0, 0.0, 0.0], + cases=[ + {"name": "case-1", "evaluator": "Faithfulness"}, + {"name": "case-2", "evaluator": "Faithfulness"}, + {"name": "case-3", "evaluator": "Faithfulness"}, + ], + test_passes=[False, False, False], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == 3 + assert analysis.total_cases == 3 + assert len(analysis.cohorts) == 1 + assert analysis.cohorts[0].evaluator_name == "Faithfulness" + assert analysis.cohorts[0].count == 3 + assert analysis.cohorts[0].failed_case_indices == [0, 1, 2] + assert analysis.cohorts[0].failed_case_names == ["case-1", "case-2", "case-3"] + + def test_multiple_evaluators_sorted_by_count(self): + report = EvaluationReport( + overall_score=0.5, + scores=[0.0, 0.0, 0.0, 0.0, 1.0, 1.0], + cases=[ + {"name": "c1", "evaluator": "Correctness"}, + {"name": "c2", "evaluator": "Faithfulness"}, + {"name": "c3", "evaluator": "Faithfulness"}, + {"name": "c4", "evaluator": "Faithfulness"}, + {"name": "c5", "evaluator": "Correctness"}, + {"name": "c6", "evaluator": "Faithfulness"}, + ], + test_passes=[False, False, False, False, True, True], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == 4 + assert len(analysis.cohorts) == 2 + # Faithfulness has 3 failures, Correctness has 1 + assert analysis.cohorts[0].evaluator_name == "Faithfulness" + assert analysis.cohorts[0].count == 3 + assert analysis.cohorts[1].evaluator_name == "Correctness" + assert analysis.cohorts[1].count == 1 + + def test_alphabetical_tiebreaker(self): + report = EvaluationReport( + overall_score=0.0, + scores=[0.0, 0.0], + cases=[ + {"name": "c1", "evaluator": "Zebra"}, + {"name": "c2", "evaluator": "Alpha"}, + ], + test_passes=[False, False], + ) + analysis = analyze_failure_cohorts(report) + assert len(analysis.cohorts) == 2 + # Same count (1 each), sorted alphabetically + assert analysis.cohorts[0].evaluator_name == "Alpha" + assert analysis.cohorts[1].evaluator_name == "Zebra" + + def test_missing_evaluator_key_defaults_to_unknown(self): + report = EvaluationReport( + overall_score=0.0, + scores=[0.0, 0.0], + cases=[ + {"name": "c1"}, + {"name": "c2"}, + ], + test_passes=[False, False], + ) + analysis = analyze_failure_cohorts(report) + assert len(analysis.cohorts) == 1 + assert analysis.cohorts[0].evaluator_name == "unknown" + assert analysis.cohorts[0].count == 2 + + def test_missing_name_key_defaults_to_case_index(self): + report = EvaluationReport( + overall_score=0.0, + scores=[0.0, 0.0], + cases=[ + {"evaluator": "X"}, + {"evaluator": "X"}, + ], + test_passes=[False, False], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.cohorts[0].failed_case_names == ["case_0", "case_1"] + + def test_mixed_pass_fail(self): + report = EvaluationReport( + overall_score=0.6, + scores=[1.0, 0.0, 1.0, 0.0, 0.0], + cases=[ + {"name": "pass1", "evaluator": "Correctness"}, + {"name": "fail1", "evaluator": "Correctness"}, + {"name": "pass2", "evaluator": "Faithfulness"}, + {"name": "fail2", "evaluator": "Faithfulness"}, + {"name": "fail3", "evaluator": "Harmfulness"}, + ], + test_passes=[True, False, True, False, False], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == 3 + assert analysis.total_cases == 5 + assert len(analysis.cohorts) == 3 + # Each evaluator has 1 failure, so sorted alphabetically + evaluator_names = [c.evaluator_name for c in analysis.cohorts] + assert evaluator_names == ["Correctness", "Faithfulness", "Harmfulness"] + + def test_empty_report(self): + report = EvaluationReport( + overall_score=0.0, + scores=[], + cases=[], + test_passes=[], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == 0 + assert analysis.total_cases == 0 + assert analysis.cohorts == [] + + def test_indices_reflect_original_position(self): + report = EvaluationReport( + overall_score=0.5, + scores=[1.0, 0.0, 1.0, 1.0, 0.0], + cases=[ + {"name": "a", "evaluator": "E1"}, + {"name": "b", "evaluator": "E1"}, + {"name": "c", "evaluator": "E1"}, + {"name": "d", "evaluator": "E1"}, + {"name": "e", "evaluator": "E1"}, + ], + test_passes=[True, False, True, True, False], + ) + analysis = analyze_failure_cohorts(report) + assert analysis.cohorts[0].failed_case_indices == [1, 4] + assert analysis.cohorts[0].failed_case_names == ["b", "e"] + + def test_works_with_report_from_file(self, tmp_path): + report = EvaluationReport( + overall_score=0.5, + scores=[1.0, 0.0, 0.0, 1.0], + cases=[ + {"name": "c1", "evaluator": "Eval1"}, + {"name": "c2", "evaluator": "Eval1"}, + {"name": "c3", "evaluator": "Eval2"}, + {"name": "c4", "evaluator": "Eval2"}, + ], + test_passes=[True, False, False, True], + ) + filepath = str(tmp_path / "report.json") + report.to_file(filepath) + + loaded = EvaluationReport.from_file(filepath) + analysis = analyze_failure_cohorts(loaded) + assert analysis.total_failures == 2 + assert len(analysis.cohorts) == 2 + + def test_works_with_flattened_report(self): + report1 = EvaluationReport( + overall_score=0.5, + scores=[1.0, 0.0], + cases=[ + {"name": "c1", "evaluator": "Correctness"}, + {"name": "c2", "evaluator": "Correctness"}, + ], + test_passes=[True, False], + ) + report2 = EvaluationReport( + overall_score=0.0, + scores=[0.0, 0.0], + cases=[ + {"name": "c1", "evaluator": "Faithfulness"}, + {"name": "c2", "evaluator": "Faithfulness"}, + ], + test_passes=[False, False], + ) + flattened = EvaluationReport.flatten([report1, report2]) + analysis = analyze_failure_cohorts(flattened) + assert analysis.total_failures == 3 + assert analysis.total_cases == 4 + assert analysis.cohorts[0].evaluator_name == "Faithfulness" + assert analysis.cohorts[0].count == 2 + assert analysis.cohorts[1].evaluator_name == "Correctness" + assert analysis.cohorts[1].count == 1 + + def test_large_cohort(self): + n = 100 + cases = [{"name": f"case_{i}", "evaluator": "BigEval"} for i in range(n)] + report = EvaluationReport( + overall_score=0.0, + scores=[0.0] * n, + cases=cases, + test_passes=[False] * n, + ) + analysis = analyze_failure_cohorts(report) + assert analysis.total_failures == n + assert len(analysis.cohorts) == 1 + assert analysis.cohorts[0].count == n + assert len(analysis.cohorts[0].failed_case_indices) == n + + +class TestPrintCohortSummary: + """Tests for the print_cohort_summary display helper.""" + + def test_prints_table_without_error(self, capsys): + analysis = CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 1, 2], + failed_case_names=["a", "b", "c"], + count=3, + ), + FailureCohort( + evaluator_name="Correctness", + failed_case_indices=[5], + failed_case_names=["f"], + count=1, + ), + ], + total_failures=4, + total_cases=10, + ) + # Should not raise + print_cohort_summary(analysis) + + def test_truncates_case_names_beyond_five(self, capsys): + analysis = CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="BigEval", + failed_case_indices=list(range(8)), + failed_case_names=[f"case_{i}" for i in range(8)], + count=8, + ), + ], + total_failures=8, + total_cases=20, + ) + print_cohort_summary(analysis) + captured = capsys.readouterr() + assert "+3 more" in captured.out + + def test_empty_analysis(self, capsys): + analysis = CohortAnalysis(cohorts=[], total_failures=0, total_cases=5) + # Should not raise on empty + print_cohort_summary(analysis) + + +class TestModuleImports: + """Tests that the module is accessible from the top-level package.""" + + def test_import_from_analysis_submodule(self): + from strands_evals.analysis import analyze_failure_cohorts as fn + + assert callable(fn) + + def test_import_from_top_level(self): + import strands_evals + + assert hasattr(strands_evals, "analysis") + assert hasattr(strands_evals.analysis, "analyze_failure_cohorts") + assert hasattr(strands_evals.analysis, "FailureCohort") + assert hasattr(strands_evals.analysis, "CohortAnalysis") + assert hasattr(strands_evals.analysis, "print_cohort_summary") From 69bea41ca83ef86c5a0ebff6ce48d0f022e64c59 Mon Sep 17 00:00:00 2001 From: Max Rattray Date: Mon, 10 Aug 2026 18:07:25 +0000 Subject: [PATCH 2/2] fix: address review feedback --- src/strands_evals/analysis/failure_cohorts.py | 7 +- .../analysis/test_failure_cohorts.py | 150 +++++++++++++----- 2 files changed, 114 insertions(+), 43 deletions(-) diff --git a/src/strands_evals/analysis/failure_cohorts.py b/src/strands_evals/analysis/failure_cohorts.py index ce27f4ef..e1bc63e9 100644 --- a/src/strands_evals/analysis/failure_cohorts.py +++ b/src/strands_evals/analysis/failure_cohorts.py @@ -8,6 +8,8 @@ from __future__ import annotations from pydantic import BaseModel +from rich.console import Console +from rich.table import Table from ..types.evaluation_report import EvaluationReport @@ -59,7 +61,7 @@ def analyze_failure_cohorts(report: EvaluationReport) -> CohortAnalysis: """ failures_by_evaluator: dict[str, list[tuple[int, str]]] = {} - for i, (case, passed) in enumerate(zip(report.cases, report.test_passes, strict=False)): + for i, (case, passed) in enumerate(zip(report.cases, report.test_passes, strict=True)): if passed: continue eval_name = case.get("evaluator", "unknown") @@ -94,9 +96,6 @@ def print_cohort_summary(analysis: CohortAnalysis) -> None: Args: analysis: A CohortAnalysis returned by analyze_failure_cohorts. """ - from rich.console import Console - from rich.table import Table - console = Console() table = Table(title=f"Failure Cohorts ({analysis.total_failures}/{analysis.total_cases} failed)") table.add_column("Evaluator", style="bold") diff --git a/tests/strands_evals/analysis/test_failure_cohorts.py b/tests/strands_evals/analysis/test_failure_cohorts.py index a41bee79..5f01b5ca 100644 --- a/tests/strands_evals/analysis/test_failure_cohorts.py +++ b/tests/strands_evals/analysis/test_failure_cohorts.py @@ -127,9 +127,7 @@ def test_serialization_roundtrip(self): ) data = analysis.model_dump() restored = CohortAnalysis.model_validate(data) - assert restored.cohorts == analysis.cohorts - assert restored.total_failures == 1 - assert restored.total_cases == 3 + assert restored == analysis class TestAnalyzeFailureCohorts: @@ -147,9 +145,7 @@ def test_all_passing(self): test_passes=[True, True, True], ) analysis = analyze_failure_cohorts(report) - assert analysis.total_failures == 0 - assert analysis.total_cases == 3 - assert analysis.cohorts == [] + assert analysis == CohortAnalysis(cohorts=[], total_failures=0, total_cases=3) def test_all_failing_single_evaluator(self): report = EvaluationReport( @@ -163,13 +159,18 @@ def test_all_failing_single_evaluator(self): test_passes=[False, False, False], ) analysis = analyze_failure_cohorts(report) - assert analysis.total_failures == 3 - assert analysis.total_cases == 3 - assert len(analysis.cohorts) == 1 - assert analysis.cohorts[0].evaluator_name == "Faithfulness" - assert analysis.cohorts[0].count == 3 - assert analysis.cohorts[0].failed_case_indices == [0, 1, 2] - assert analysis.cohorts[0].failed_case_names == ["case-1", "case-2", "case-3"] + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[0, 1, 2], + failed_case_names=["case-1", "case-2", "case-3"], + count=3, + ), + ], + total_failures=3, + total_cases=3, + ) def test_multiple_evaluators_sorted_by_count(self): report = EvaluationReport( @@ -186,13 +187,24 @@ def test_multiple_evaluators_sorted_by_count(self): test_passes=[False, False, False, False, True, True], ) analysis = analyze_failure_cohorts(report) - assert analysis.total_failures == 4 - assert len(analysis.cohorts) == 2 - # Faithfulness has 3 failures, Correctness has 1 - assert analysis.cohorts[0].evaluator_name == "Faithfulness" - assert analysis.cohorts[0].count == 3 - assert analysis.cohorts[1].evaluator_name == "Correctness" - assert analysis.cohorts[1].count == 1 + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[1, 2, 3], + failed_case_names=["c2", "c3", "c4"], + count=3, + ), + FailureCohort( + evaluator_name="Correctness", + failed_case_indices=[0], + failed_case_names=["c1"], + count=1, + ), + ], + total_failures=4, + total_cases=6, + ) def test_alphabetical_tiebreaker(self): report = EvaluationReport( @@ -205,10 +217,24 @@ def test_alphabetical_tiebreaker(self): test_passes=[False, False], ) analysis = analyze_failure_cohorts(report) - assert len(analysis.cohorts) == 2 - # Same count (1 each), sorted alphabetically - assert analysis.cohorts[0].evaluator_name == "Alpha" - assert analysis.cohorts[1].evaluator_name == "Zebra" + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Alpha", + failed_case_indices=[1], + failed_case_names=["c2"], + count=1, + ), + FailureCohort( + evaluator_name="Zebra", + failed_case_indices=[0], + failed_case_names=["c1"], + count=1, + ), + ], + total_failures=2, + total_cases=2, + ) def test_missing_evaluator_key_defaults_to_unknown(self): report = EvaluationReport( @@ -221,9 +247,18 @@ def test_missing_evaluator_key_defaults_to_unknown(self): test_passes=[False, False], ) analysis = analyze_failure_cohorts(report) - assert len(analysis.cohorts) == 1 - assert analysis.cohorts[0].evaluator_name == "unknown" - assert analysis.cohorts[0].count == 2 + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="unknown", + failed_case_indices=[0, 1], + failed_case_names=["c1", "c2"], + count=2, + ), + ], + total_failures=2, + total_cases=2, + ) def test_missing_name_key_defaults_to_case_index(self): report = EvaluationReport( @@ -236,7 +271,18 @@ def test_missing_name_key_defaults_to_case_index(self): test_passes=[False, False], ) analysis = analyze_failure_cohorts(report) - assert analysis.cohorts[0].failed_case_names == ["case_0", "case_1"] + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="X", + failed_case_indices=[0, 1], + failed_case_names=["case_0", "case_1"], + count=2, + ), + ], + total_failures=2, + total_cases=2, + ) def test_mixed_pass_fail(self): report = EvaluationReport( @@ -252,12 +298,30 @@ def test_mixed_pass_fail(self): test_passes=[True, False, True, False, False], ) analysis = analyze_failure_cohorts(report) - assert analysis.total_failures == 3 - assert analysis.total_cases == 5 - assert len(analysis.cohorts) == 3 - # Each evaluator has 1 failure, so sorted alphabetically - evaluator_names = [c.evaluator_name for c in analysis.cohorts] - assert evaluator_names == ["Correctness", "Faithfulness", "Harmfulness"] + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="Correctness", + failed_case_indices=[1], + failed_case_names=["fail1"], + count=1, + ), + FailureCohort( + evaluator_name="Faithfulness", + failed_case_indices=[3], + failed_case_names=["fail2"], + count=1, + ), + FailureCohort( + evaluator_name="Harmfulness", + failed_case_indices=[4], + failed_case_names=["fail3"], + count=1, + ), + ], + total_failures=3, + total_cases=5, + ) def test_empty_report(self): report = EvaluationReport( @@ -267,9 +331,7 @@ def test_empty_report(self): test_passes=[], ) analysis = analyze_failure_cohorts(report) - assert analysis.total_failures == 0 - assert analysis.total_cases == 0 - assert analysis.cohorts == [] + assert analysis == CohortAnalysis(cohorts=[], total_failures=0, total_cases=0) def test_indices_reflect_original_position(self): report = EvaluationReport( @@ -285,8 +347,18 @@ def test_indices_reflect_original_position(self): test_passes=[True, False, True, True, False], ) analysis = analyze_failure_cohorts(report) - assert analysis.cohorts[0].failed_case_indices == [1, 4] - assert analysis.cohorts[0].failed_case_names == ["b", "e"] + assert analysis == CohortAnalysis( + cohorts=[ + FailureCohort( + evaluator_name="E1", + failed_case_indices=[1, 4], + failed_case_names=["b", "e"], + count=2, + ), + ], + total_failures=2, + total_cases=5, + ) def test_works_with_report_from_file(self, tmp_path): report = EvaluationReport(