def failure_attribution(
config: ModelConfig | None = None,
*,
key: str = "failure",
agent_name: str | None = None,
taxonomy: Iterable[str] = ERROR_MODES,
retries: int = 1,
middleware: Iterable[MiddlewareFactory] = (),
) -> Scorer:
"""Build a failure-attribution :class:`Scorer`.
Args:
config: Optional judge model for the LLM attributor. Without it the
scorer is deterministic-only (mechanical failures + ``none``).
key: Result key; its ``value_counts`` is the failure-mode distribution.
agent_name: Label used for ``responsible_agent`` (single-agent today).
taxonomy: Allowed error modes shown to the LLM attributor.
retries: ``content()`` re-asks on schema-validation failure.
middleware: Middleware for the attributor agent (e.g. ``TelemetryMiddleware``).
"""
modes = tuple(taxonomy)
attributor = (
Agent(
f"attributor_{key}",
_system_prompt(modes),
config=config,
response_schema=_AttributionVerdict,
middleware=middleware,
)
if config is not None
else None
)
async def _attribute(
inputs: dict[str, Any],
outputs: dict[str, Any],
reference_outputs: dict[str, Any] | None,
trace: Trace,
) -> Feedback:
attribution = _detect_mechanical(trace, agent_name)
if attribution is None:
if attributor is not None:
attribution = await _llm_attribute(
attributor, inputs, outputs, reference_outputs, trace, retries, agent_name
)
else:
attribution = Attribution(
failed=False,
error_mode="none",
responsible_agent=agent_name,
reasoning="no mechanical failure detected",
)
return Feedback(
key=key, value=attribution.error_mode, comment=attribution.reasoning, detail=attribution.model_dump()
)
return Scorer(_attribute, key=key)