latent.flows.failure_mode_flow

failure_mode_flow — Classify failure modes for evaluation records.

Functions

classify_failure_modes

classify_failure_modes(records: list[RecordResult], taxonomy: dict[str, str], filter_fn: Callable[[RecordResult], bool] | None = None, model: str = 'claude-sonnet-4-5-20250929', concurrency: int = 1, per_task_timeout_s: float | None = None) -> tuple[list[RecordResult], list[FailureModeSummary]]

Classify failure modes for filtered records using LLM.

Selected records are classified concurrently via asyncio.gather bounded by an asyncio.Semaphore of size concurrency, the same shape judge_flow uses for per-row scoring.

Args: records: Evaluation records to classify. taxonomy: {"mode_name": "human-readable description"} dict. filter_fn: Predicate selecting which records to classify. Default: records where failure_mode is None. model: LiteLLM model string for classification. concurrency: Number of records classified concurrently. Defaults to 1 (fully serial). per_task_timeout_s: Deadline for a single record's classification (excluding time spent waiting for a concurrency slot). Covers all of that record's internal retry attempts, not one provider call, so size it from the retry budget rather than from one call's p99. On expiry that record's failure_mode is left as it arrived, as for any other classification failure; the remaining records are unaffected, and the count of unclassified records is logged.

Returns: Tuple of: - Updated records (copies with failure_mode set where classified; a record whose classification failed keeps the mode it arrived with). - FailureModeSummary list with counts and proportions.