Classifiers¶

All classifiers satisfy the Classifier protocol. Agent uses RulesClassifier by default; pass classifier= to swap it out.

Classifier protocol¶

Classifier ¶

Bases: Protocol

Synchronous failure classifier. Must not make any API calls.

classify() remains the required, synchronous contract (agent.py calls it via anyio.to_thread.run_sync on the failure path). Classifiers that talk to an LLM API may additionally define an optional async def aclassify(self, trajectory, task) -> FailureType method — when present, agent.py awaits it directly instead of running classify() in a thread, avoiding that hop. This is not part of the Classifier protocol itself (kept structural/duck-typed) since most classifiers, like RulesClassifier, have no I/O to make async.

ClassificationResult¶

ClassificationResult `dataclass` ¶

A FailureType plus the classifier's confidence in it (0.0-1.0).

Returned by the optional, duck-typed classify_with_confidence() / aclassify_with_confidence() methods a classifier may define — not part of the Classifier protocol itself, same reasoning as aclassify(): RulesClassifier has no meaningful confidence to report (its answers are 100%-precision-by-construction or UNKNOWN), so this isn't forced on every classifier. HybridClassifier(confidence_threshold=...) checks for these methods via getattr and, when present, only trusts a non-UNKNOWN answer whose confidence meets the threshold — see docs/known-limitations.md's "close the recall gap, but not the precision gap" section for the measurement this exists to address, and LLMClassifier.classify_with_confidence() for the reference implementation.

RulesClassifier¶

RulesClassifier ¶

Pattern-based classifier. Instantiate with optional constraint strings.

Parameters:

Name	Type	Description	Default
`constraints`	`list[str] \| None`	Forbidden strings to detect in step `llm_output`. If any of these strings appear verbatim in a step's LLM output, the failure is classified as `CONSTRAINT_IGNORED`. Pass the forbidden content itself, not the rule description:: `# Correct: flag if the word "markdown" appears in output RulesClassifier(constraints=["markdown"]) # Correct: flag if a specific phrase leaks into output RulesClassifier(constraints=["<script>", "DROP TABLE"]) # Wrong: this passes the rule text, not the forbidden content RulesClassifier(constraints=["no markdown allowed"])`	`None`
`loop_window`	`int`	Number of consecutive steps required to declare a loop. Default 3. Set higher (e.g. 4–5) if your agent legitimately repeats the same tool call twice in a row.	`3`
`loop_similarity_threshold`	`float \| None`	If set (e.g. `0.9`), loop detection additionally matches steps whose canonical `tool_input` strings are similar rather than identical — catching loops where the agent reworded a query slightly on each retry (e.g. `{"q": "revenue Q1"}` vs `{"q": "revenue for Q1"}`). Similarity is computed with `difflib.SequenceMatcher.ratio()` on the canonical JSON string form of `tool_input`, compared consecutively within the window (each step vs. the previous one) rather than all pairs against the first — this matches loops that drift gradually, not just loops identical to the very first step. `tool_called` must still match exactly across the whole window; only `tool_input` gets the fuzzy comparison. Default `None` disables fuzzy matching — behavior is unchanged from pre-v0.12 (exact match only).	`None`
`framework`	`str \| None`	Optional SDK/framework name. When set, per-framework error patterns are checked in addition to the generic patterns. Supported values: `"openai"`, `"anthropic"`, `"langgraph"`. Case-insensitive. Unknown values (including `"langchain"` — covered by generic patterns) are silently ignored; generic patterns still apply.	`None`

init ¶

__init__(constraints: list[str] | None = None, loop_window: int = 3, loop_similarity_threshold: float | None = None, framework: str | None = None) -> None

classify ¶

classify(trajectory: Trajectory, task: str) -> FailureType

fit ¶

fit(corrections_path: str = 'corrections.jsonl') -> dict[str, dict[str, int]]

Read a corrections JSONL file and report classifier coverage.

For each correction where the rules classifier would have returned a different type than expected, a warning is logged. Use this to identify systematic misclassifications after calling agent.report_misclassification().

Returns a coverage dict keyed by FailureType value::

{
    "wrong_tool_called": {"correct": 5, "wrong": 1},
    "external_fault":    {"correct": 3, "wrong": 0},
}

Does not modify the classifier's rules or thresholds — purely diagnostic.

Rule priority¶

Rules fire in order; first match wins:

LOOP_DETECTED — last loop_window steps share identical tool_called and equal (or fuzzy-similar) tool_input
WRONG_TOOL_CALLED — error matches tool-not-found patterns across OpenAI / Anthropic / LangGraph SDKs
SCHEMA_MISMATCH — error matches validation / JSON parse patterns
EXTERNAL_FAULT — error contains an HTTP status code (429, 500, 502, 503) as a whole token, not in a quantity context
TIMEOUT — error matches timeout / deadline patterns
CONSTRAINT_IGNORED — llm_output contains a forbidden constraint string
UNKNOWN — default

PLAN_INCOMPLETE and CONTEXT_OVERFLOW require semantic understanding and always return UNKNOWN from RulesClassifier. Use LLMClassifier or HybridClassifier for those.

LLMClassifier¶

LLMClassifier ¶

Semantic failure classifier backed by an LLM.

Satisfies the Classifier protocol (synchronous classify method). Also defines the optional, duck-typed classify_with_confidence() / aclassify_with_confidence() methods, returning a ClassificationResult (failure type + a 0.0-1.0 self-reported confidence) instead of a bare FailureType — see that method's docstring, and HybridClassifier(confidence_threshold=...) for the primary consumer.

When base_url is None (default), uses anthropic.Anthropic (requires pip install triage-agent[anthropic]).

When base_url is set, uses openai.OpenAI pointed at that base URL — compatible with Ollama, Groq, OpenAI, and any OpenAI-compatible provider (requires pip install triage-agent[openai] or pip install openai).

max_tokens (default 32, or TRIAGE_LLM_MAX_TOKENS) bounds the classification call's output. 32 suffices for a plain instruct model's one-word answer; a reasoning model (gpt-oss, o1/o3-style, DeepSeek-R1, Qwen3 "thinking" mode, ...) needs far more or the response is truncated to empty content before the answer is ever emitted — silently, since that's not an error. Falls back to FailureType.UNKNOWN on any error (network, parse, rate limit) — an empty response from a starved reasoning model fails the same way, indistinguishably, unless you've set max_tokens high enough for that model.

init ¶

__init__(api_key: str | None = None, model: str | None = None, max_trajectory_steps: int = 10, base_url: str | None = None, max_retries: int = 1, retry_backoff_base: float = 0.5, max_tokens: int | None = None) -> None

classify ¶

classify(trajectory: Trajectory, task: str) -> FailureType

aclassify `async` ¶

aclassify(trajectory: Trajectory, task: str) -> FailureType

Async counterpart to classify() using the native async SDK client.

Prefer this over classify() when calling from async code — it awaits the HTTP call directly instead of running the sync client in a thread. Same fallback-to-UNKNOWN behavior on any error, and the same retry budget for transient errors (429/5xx/timeout/connection).

classify_with_confidence ¶

classify_with_confidence(trajectory: Trajectory, task: str) -> ClassificationResult

Like classify(), but also asks the model to self-report a confidence score (0.0-1.0) and returns both as a ClassificationResult.

Not part of the Classifier protocol — duck-typed, same pattern as aclassify(). Exists specifically so a caller (HybridClassifier with confidence_threshold= set, most directly) can decline to trust a low-confidence guess instead of returning it as fact. See the comment above _CONFIDENCE_SYSTEM_PROMPT for why this exists instead of more prompt tuning on classify() itself, and note this model's confidence has not been independently validated as calibrated — calibrate any threshold against your own labeled data before trusting it in production.

Falls back to ClassificationResult(FailureType.UNKNOWN, 0.0) on any error, same as classify() falls back to FailureType.UNKNOWN — zero confidence signals "don't trust this" just as clearly as UNKNOWN does, so a threshold-gated caller declines correctly either way.

aclassify_with_confidence `async` ¶

aclassify_with_confidence(trajectory: Trajectory, task: str) -> ClassificationResult

Async counterpart to classify_with_confidence() — see that method's docstring. Same native-async-client preference as aclassify() over classify().

HybridClassifier¶

HybridClassifier ¶

Rules-first classifier with LLM fallback on UNKNOWN.

Satisfies the Classifier protocol (synchronous classify method). LLMClassifier.classify() is called in a thread by agent.py (via anyio.to_thread.run_sync), so blocking HTTP is safe here.

max_llm_calls_per_run bounds how many times the wrapped LLM classifier is actually called within one Agent.run() call — protecting against an agent that fails repeatedly from burning an LLM call on every recovery attempt. Once the cap is reached, ambiguous (rules-UNKNOWN) failures fall straight to UNKNOWN without touching the LLM. None (default) means unlimited.

The call counter is a plain, threading.Lock-guarded instance attribute rather than a ContextVar — deliberately, because classify() may run inside anyio.to_thread.run_sync() (when a classifier has no aclassify(), or is called directly outside Agent), and a ContextVar mutated inside a thread dispatch is invisible to the caller once that dispatch returns (each to_thread.run_sync call runs in its own copy of the current context). A real shared counter is required for the cap to work through both the sync (classify()-in-a-thread) and async (aclassify()) dispatch paths.

Agent.run() calls reset_call_count() (if present, duck-typed) once at the very start of every run() call, so the cap is scoped to a single run rather than this instance's whole lifetime. If you share one HybridClassifier across multiple concurrently running Agent instances (or concurrent run() calls sharing one Agent), the reset is best-effort, not isolated — one run's reset can zero out a budget another concurrent run was still counting against. Use a separate HybridClassifier (or agent.clone(), which does not share this counter) per concurrent task if you need a precise, independent budget.

confidence_threshold (default None) gates the LLM fallback on a confidence score instead of trusting any non-UNKNOWN answer outright. None (the default) is the original, unconditional behavior — every existing caller's behavior is unchanged unless this is explicitly set. When set, the wrapped classifier's classify_with_confidence() / aclassify_with_confidence() method (duck-typed — see triage.classifier.base.ClassificationResult) is used instead of classify() / aclassify(), and the fallback only returns a non-UNKNOWN answer whose confidence is >= confidence_threshold; otherwise it returns UNKNOWN, same as a rules miss with no LLM fallback at all. If the wrapped classifier doesn't define a *_with_confidence method, this silently falls back to the unconditional behavior — setting confidence_threshold on a classifier that can't report confidence is a no-op, not an error, since Classifier implementations aren't required to support it.

This exists because prompt-level guidance alone, tried first, measurably failed to reduce HybridClassifier's override rate on genuinely out-of-taxonomy trajectories — see docs/known-limitations.md's "close the recall gap, but not the precision gap" section for the measurement. Pick confidence_threshold by calibrating against your own labeled data, not a default guessed here — a wrapped classifier's self-reported confidence has not been independently validated as calibrated for your traffic distribution. scripts/hybrid_ambiguity_accuracy.py --confidence- threshold scores this trade-off against the ambiguous corpus.

init ¶

__init__(llm: Any, max_llm_calls_per_run: int | None = None, confidence_threshold: float | None = None) -> None

classify ¶

classify(trajectory: Trajectory, task: str) -> FailureType

aclassify `async` ¶

aclassify(trajectory: Trajectory, task: str) -> FailureType

Async counterpart to classify(). Uses self._llm.aclassify() when the configured LLM classifier defines one (e.g. LLMClassifier), avoiding the sync-client-in-a-thread hop on the failure path. When confidence_threshold is set, prefers aclassify_with_confidence() over aclassify()/classify() — see __init__'s docstring.

reset_call_count ¶

reset_call_count() -> None

Reset the per-run LLM call counter. Called by Agent.run() at the start of every run — see the class docstring for the concurrency caveat when sharing one instance across concurrent runs.

Custom classifier¶

Any object with a synchronous classify(trajectory, task) -> FailureType method satisfies the protocol:

from triage.taxonomy import FailureType
from triage.trajectory import Trajectory

class MyClassifier:
    def classify(self, trajectory: Trajectory, task: str) -> FailureType: ...

agent = triage.Agent(my_agent, policy=policy, classifier=MyClassifier())

For async classifiers, add an aclassify method (duck-typed, not part of the protocol):

class MyAsyncClassifier:
    def classify(self, trajectory: Trajectory, task: str) -> FailureType:
        ...  # sync fallback

    async def aclassify(self, trajectory: Trajectory, task: str) -> FailureType:
        ...  # async path — used by Agent when present

A classifier may additionally define classify_with_confidence(trajectory, task) -> ClassificationResult (and/or its async counterpart aclassify_with_confidence) — also duck-typed, not part of the protocol. HybridClassifier(llm=my_classifier, confidence_threshold=0.7) checks for these methods and, when present, only trusts a non-UNKNOWN answer whose confidence meets the threshold; a classifier without them (like RulesClassifier, which has nothing meaningful to report beyond its 100%-precision-by-construction rules) is unaffected — confidence_threshold on a classifier that can't report confidence is a no-op, not an error:

from triage.classifier.base import ClassificationResult

class MyConfidentClassifier:
    def classify(self, trajectory: Trajectory, task: str) -> FailureType:
        ...

    def classify_with_confidence(self, trajectory: Trajectory, task: str) -> ClassificationResult:
        ...  # e.g. ask the model for a confidence score alongside the category

Classifiers¶

Classifier protocol¶

Classifier ¶

ClassificationResult¶

ClassificationResult dataclass ¶

RulesClassifier¶

RulesClassifier ¶

__init__ ¶

classify ¶

fit ¶

Rule priority¶

LLMClassifier¶

LLMClassifier ¶

__init__ ¶

classify ¶

aclassify async ¶

classify_with_confidence ¶

aclassify_with_confidence async ¶

HybridClassifier¶

HybridClassifier ¶

__init__ ¶

classify ¶

aclassify async ¶

reset_call_count ¶

Custom classifier¶

ClassificationResult `dataclass` ¶

init ¶

init ¶

aclassify `async` ¶

aclassify_with_confidence `async` ¶

init ¶

aclassify `async` ¶