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.

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).

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).

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.

init ¶

__init__(llm: Any, max_llm_calls_per_run: int | 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.

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

Classifiers¶

Classifier protocol¶

Classifier ¶

RulesClassifier¶

RulesClassifier ¶

__init__ ¶

classify ¶

fit ¶

Rule priority¶

LLMClassifier¶

LLMClassifier ¶

__init__ ¶

classify ¶

aclassify async ¶

HybridClassifier¶

HybridClassifier ¶

__init__ ¶

classify ¶

aclassify async ¶

reset_call_count ¶

Custom classifier¶

init ¶

init ¶

aclassify `async` ¶

init ¶

aclassify `async` ¶