Find What's
Actually Broken.

A failure record keeps everything needed to diagnose it later without re-running the agent.

from __future__ import annotations
from dataclasses import dataclass


@dataclass(frozen=True)
class Failure:
    """A single failed evaluation captured for analysis.

    Args:
        example_id: The golden example that failed.
        input: The input handed to the agent.
        expected: The expected outcome.
        actual: What the agent actually produced.
        score: The numeric score assigned by the evaluator.
    """
    example_id: str
    input: str
    expected: str
    actual: str
    score: float


def collect_failures(results, examples, threshold: float = 1.0) -> list[Failure]:
    """Extract failing items from an evaluation run.

    Args:
        results: Per-item results keyed by example id.
        examples: The golden examples that were evaluated.
        threshold: Scores below this are treated as failures.

    Returns:
        A list of :class:`Failure` records ready for annotation.
    """
    by_id = {ex.id: ex for ex in examples}
    return [
        Failure(r.id, by_id[r.id].input, str(by_id[r.id].expected), r.actual, r.score)
        for r in results if r.score < threshold
    ]

An LLM can propose a failure-mode label for each case; a human then confirms or corrects it. This scales the first pass without sacrificing trust.

from collections import Counter
from .models import Failure

CLASSIFY_PROMPT = """You are triaging an AI agent failure.
Choose exactly one failure_mode from: {modes}.
Input: {input}
Expected: {expected}
Actual: {actual}
Respond with only the failure_mode."""


def classify_failure(failure: Failure, modes: list[str], judge) -> str:
    """Assign a failure mode to a single failure using an LLM judge.

    Args:
        failure: The failure to classify.
        modes: The allowed controlled vocabulary of failure modes.
        judge: Callable that maps a prompt to a model response.

    Returns:
        A failure-mode label; falls back to ``"unclassified"`` if invalid.
    """
    label = judge(CLASSIFY_PROMPT.format(
        modes=", ".join(modes),
        input=failure.input, expected=failure.expected, actual=failure.actual,
    )).strip()
    return label if label in modes else "unclassified"


def summarize(failures: list[Failure], modes: list[str], judge) -> Counter:
    """Return a frequency count of failure modes across all failures."""
    return Counter(classify_failure(f, modes, judge) for f in failures)

Weight each mode by severity so the top of the list is the highest-value fix.

SEVERITY_WEIGHT = {"low": 1, "medium": 3, "high": 9}


def prioritize(counts: dict[str, int], severity: dict[str, str]) -> list[tuple]:
    """Rank failure modes by ``frequency × severity weight``.

    Args:
        counts: Number of occurrences per failure mode.
        severity: Mapping of failure mode to a severity level.

    Returns:
        Modes sorted from highest to lowest priority score.
    """
    scored = [
        (mode, n, n * SEVERITY_WEIGHT.get(severity.get(mode, "medium"), 3))
        for mode, n in counts.items()
    ]
    return sorted(scored, key=lambda row: row[2], reverse=True)