Agent¶

triage.Agent is the core wrapper. It runs your async callable in a retry loop, classifying failures and dispatching recovery actions.

Agent class¶

Agent ¶

Wraps any async callable and adds failure classification + recovery.

The wrapped function must accept record_step and optionally update_state keyword arguments::

async def my_agent(task: str, *, record_step, update_state, **kwargs) -> Any:
    data = fetch_something()
    record_step(Step(index=0, action="fetch", tool_output=data))
    update_state({"data": data})   # persisted into checkpoints
    return result

Synchronous callables are also accepted — triage runs them via anyio.to_thread.run_sync() so the event loop is never blocked::

def my_sync_agent(task: str, *, record_step, **kwargs) -> Any:
    ...

Alternatively, use triage.agent.get_recorder() inside the agent body to avoid changing the function signature.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	The agent callable to wrap — either `async def` or plain `def`.	required
`policy`	`FailurePolicy`	Maps each `FailureType` to a recovery strategy.	required
`classifier`	`Classifier \| None`	Classifies failures from the trajectory. Defaults to `RulesClassifier`.	`None`
`checkpoint_store`	`CheckpointStore \| None`	Stores and loads checkpoints. Defaults to `InMemoryCheckpointStore`.	`None`
`max_recovery_attempts`	`int`	Maximum number of recovery loop iterations per `run()` call (default 3). Each failed attempt + dispatch counts as one iteration.	`3`
`max_total_attempts`	`int \| None`	Hard cap on `len(attempt_history)` across all failure types. When reached, triage escalates regardless of the active strategy. `None` (default) disables this cap and defers entirely to `max_recovery_attempts`.	`None`
`max_recovery_seconds`	`float \| None`	Wall-clock budget for the entire recovery process. If recovery has been ongoing for more than this many seconds, triage escalates. `None` (default) disables this cap. The timer starts after the first failure.	`None`
`auto_checkpoint`	`bool`	If `True`, saves a checkpoint after every `record_step()` call.	`False`
`strict_idempotency`	`bool`	If `True`, triage will escalate instead of retrying whenever any step in the trajectory has `idempotent=False`. Default `False`. Use this when your agent has steps that send emails, charge cards, or perform other non-reversible side effects.	`False`
`tracer`	`Any`	Optional OpenTelemetry `Tracer` instance. When provided, triage emits `triage.run`, `triage.classify`, and `triage.dispatch` spans for every `run()` call. When `None` (default), triage auto-detects: if `opentelemetry-sdk` is installed and a real tracer provider has been configured via `trace.set_tracer_provider(...)`, the global tracer is used automatically. If OTel is not installed or no provider is configured, tracing is a silent no-op with zero overhead.	`None`
`meter`	`Any`	Optional OpenTelemetry `Meter` instance for recording metrics (`triage.runs`, `triage.failures`, `triage.recoveries`, `triage.run.duration`, `triage.recovery.attempts`). When `None` (default), triage auto-detects a configured `MeterProvider` the same way `tracer` auto-detects a `TracerProvider`. If OTel metrics is not installed or no provider is configured, metrics are a silent no-op.	`None`
`circuit_breakers`	`list[CircuitBreaker] \| None`	Optional list of `CircuitBreaker` instances to notify on every successful `run()` completion. After a clean return (no failures), triage calls `breaker.record_success()` on each breaker in order. This closes any breaker that was in HALF_OPEN state after a probe, without requiring manual success-signalling at the call site. Pass the same breaker instances that are wired into `circuit_breaker()` strategy wrappers in the policy.	`None`
`suspension_store`	`SuspensionStore \| None`	Store for serializing paused runs when `RecoveryAction.SUSPEND()` is returned. Defaults to `InMemorySuspensionStore` (not durable across restarts). Swap for a Redis-backed store in production so tokens survive process restarts.	`None`
`on_escalate`	`Callable[[FailureContext], Awaitable[RecoveryAction \| None]] \| None`	Optional async callback invoked just before `TriageEscalationError` is raised. Signature:: `async def on_escalate(ctx: FailureContext) -> RecoveryAction \| None: ...` Return a `RecoveryAction` to override the escalation — triage will execute that action instead of raising. Return `None` (or omit a return) to proceed with the escalation as normal. This is the human-in-the-loop hook: pause autonomous execution, notify a human, wait for a decision, and return an action (or `None` to surface the error). Exceptions raised inside `on_escalate` are propagated directly (unlike lifecycle hooks such as `on_failure`, which swallow errors).	`None`
`max_tokens`	`int \| None`	Maximum total tokens (input + output) allowed per `run()` call across all LLM calls recorded via `record_usage()`. When exceeded, triage raises `TriageEscalationError` before the next recovery attempt. `None` (default) disables this cap. Important: enforcement is checked at each failure point, not preemptively. An agent that burns tokens but never raises will complete regardless of this cap. The check fires after the first failure — meaning up to one failure's worth of overage is always possible before triage intervenes. This is intentional: triage only intercepts at the failure boundary, so the cap is a "do not keep retrying after I'm already over budget" guard, not a hard token ceiling.	`None`
`max_cost_usd`	`float \| None`	Maximum total cost in USD allowed per `run()` call. Subject to the same failure-point enforcement as `max_tokens` above — not a preemptive cap. When exceeded, triage raises `TriageEscalationError`. `None` (default) disables this cap. Use alongside `record_usage(Usage(cost_usd=...))` calls in the agent body or via the `LLMClassifier` auto-reporting.	`None`
`on_compensator_error`	`Callable[[int, Exception], None] \| None`	Optional callback invoked when a saga compensator raises. Signature:: `def on_compensator_error(step_index: int, exc: Exception) -> None: notify_ops(f"refund for step {step_index} failed: {exc}")` Called before the warning is logged. The compensator error is still swallowed after the callback — compensation is best-effort and the checkpoint restore always proceeds. Use this hook to alert, record to a dead-letter queue, or increment a metric. Exceptions raised inside the hook are themselves swallowed (same as other lifecycle hooks).	`None`

init ¶

__init__(fn: Callable[..., Any], policy: FailurePolicy, classifier: Classifier | None = None, checkpoint_store: CheckpointStore | None = None, max_recovery_attempts: int = 3, max_total_attempts: int | None = None, max_recovery_seconds: float | None = None, auto_checkpoint: bool = False, on_step: Callable[[Step], None] | None = None, on_failure: Callable[[FailureContext], None] | None = None, on_recovery: Callable[[FailureContext, RecoveryAction], None] | None = None, strict_idempotency: bool = False, risk_scorer: StepRiskScorer | None = None, risk_threshold: float = 0.9, tracer: Any = None, meter: Any = None, circuit_breakers: list[CircuitBreaker] | None = None, on_escalate: Callable[[FailureContext], Awaitable[RecoveryAction | None]] | None = None, suspension_store: SuspensionStore | None = None, max_tokens: int | None = None, max_cost_usd: float | None = None, on_compensator_error: Callable[[int, Exception], None] | None = None) -> None

run `async` ¶

run(task: str, **kwargs: Any) -> Any

Run the wrapped agent, recovering from failures per the policy.

stream `async` ¶

stream(task: str, **kwargs: Any) -> AsyncGenerator[Any, None]

Run the wrapped async-generator callable, yielding its output.

On failure, classifies the exception, dispatches a recovery action, yields a StreamRetryEvent, and re-starts the generator with the updated kwargs. The caller should discard accumulated output on receiving a StreamRetryEvent.

All caps (max_recovery_attempts, max_recovery_seconds, max_tokens, circuit breakers, lifecycle hooks, OTel spans) apply exactly as in run(). Raises TriageEscalationError, TriageAbortError, or TriageSuspendedError on terminal outcomes.

The wrapped callable must be an async def generator function (uses yield). Plain coroutine callables should use run() instead — calling stream() on a non-generator raises TypeError.

resume `async` ¶

resume(token: str, *, action: RecoveryAction) -> Any

Resume a previously suspended run.

Loads the SuspendedRun from the suspension store, executes action as the human's decision, then continues the recovery loop from where it left off.

The suspended run is deleted from the store after a successful load; tokens are single-use.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token from `TriageSuspendedError.token`.	required
`action`	`RecoveryAction`	The `RecoveryAction` chosen by the human. Common choices: `RecoveryAction.RETRY()` to try again, `RecoveryAction.REPLAN(hint="...")` to generate a new plan, or `RecoveryAction.ABORT(reason="...")` to stop.	required

Raises:

Type	Description
`KeyError`	If `token` is not found in the suspension store.
`TriageSuspendedError`	If the resumed run is immediately suspended again by the policy.
`TriageEscalationError / TriageAbortError`	If the recovery loop exhausts attempts or the action is ABORT.

clone ¶

clone() -> Agent

Return a new Agent sharing the same policy, classifier, and checkpoint store but with fresh per-run state, independent lifecycle hooks, and its own _last_ctx.

Concurrent run() calls on a single Agent instance are already safe (per-run state is isolated via contextvars) — use clone() when you want a task to have independent hooks or a dedicated checkpoint store instead::

agents = [agent.clone() for _ in tasks]
results = await asyncio.gather(*[ag.run(t) for ag, t in zip(agents, tasks)])

report_misclassification ¶

report_misclassification(expected_type: FailureType, *, store_path: str = 'corrections.jsonl') -> None

Record that the last classification was wrong.

Call this after run() raises, passing the correct failure type::

try:
    await agent.run("task")
except TriageEscalationError:
    agent.report_misclassification(
        FailureType.EXTERNAL_FAULT,
        store_path="corrections.jsonl",
    )

Appends a labeled entry to corrections.jsonl. Use RulesClassifier.fit(path) to review coverage.

Decorator form¶

agent ¶

agent(policy: FailurePolicy, **kwargs: Any) -> Callable[[Callable[..., Any]], Agent]

Decorator factory. Wraps an async or sync function with triage recovery.

Usage::

@triage.agent(policy=my_policy)
async def my_agent(task: str, *, record_step, update_state, **kwargs) -> str:
    ...

@triage.agent(policy=my_policy)
def my_sync_agent(task: str, *, record_step, **kwargs) -> str:
    ...

Exceptions¶

TriageEscalationError ¶

Bases: Exception

Raised when a strategy returns RecoveryAction.ESCALATE or max attempts exceeded.

TriageAbortError ¶

Bases: Exception

Raised when a strategy returns RecoveryAction.ABORT.

ContextVar helpers¶

Use these inside an agent body to avoid changing its signature:

get_recorder ¶

get_recorder() -> Callable[[Step], None]

Return the record_step callback for the current triage run.

For use by agents that cannot or do not want to accept record_step as a keyword argument::

from triage.agent import get_recorder
from triage.taxonomy import Step

async def my_agent(task: str, **kwargs) -> str:
    record = get_recorder()
    record(Step(index=0, action="fetch", tool_output=data))
    return result

Raises RuntimeError if called outside a triage Agent.run() context.

get_state_updater ¶

get_state_updater() -> Callable[[dict[str, Any]], None]

Return the update_state callback for the current triage run.

Raises RuntimeError if called outside a triage Agent.run() context.

get_usage_recorder ¶

get_usage_recorder() -> Callable[[Usage], None]

Return the record_usage callback for the current triage run.

Call this inside a wrapped agent to report token and cost usage::

from triage.agent import get_usage_recorder
from triage.usage import Usage

async def my_agent(task: str, *, record_step, **kwargs) -> str:
    result = await call_llm(prompt)
    get_usage_recorder()(Usage(
        input_tokens=result.usage.input_tokens,
        output_tokens=result.usage.output_tokens,
    ))
    return result.content

Raises RuntimeError if called outside a triage Agent.run() context.

Conceptual notes¶

max_total_attempts vs max_recovery_attempts¶

max_recovery_attempts counts loop iterations within one run() call. max_total_attempts counts the total len(attempt_history) across all failure types and fires first when it is lower. Use both together to bound cross-type accumulation:

agent = triage.Agent(
    my_agent, policy=policy,
    max_recovery_attempts=5,  # per-loop guard
    max_total_attempts=3,     # global guard — fires first if reached
)

Concurrent run() calls¶

A single Agent instance is safe for concurrent run() calls. Each call's trajectory, state, and checkpoint bookkeeping is isolated per-task via ContextVar:

import anyio

async def run_all(tasks: list[str]) -> dict[str, Any]:
    results = {}

    async def go(t: str) -> None:
        results[t] = await agent.run(t)

    async with anyio.create_task_group() as tg:
        for t in tasks:
            tg.start_soon(go, t)
    return results

Wrapped function contract¶

async def my_agent(
    task: str,
    *,
    record_step: Callable[[Step], None],
    update_state: Callable[[dict], None],
    record_usage: Callable[[Usage], None],
    _triage_context: TriageContext | None = None,
    _triage_hint: str | None = None,        # backward-compat
    _triage_subgoal: str | None = None,     # backward-compat
    _triage_state: dict | None = None,      # backward-compat
    **kwargs,
) -> Any: ...

_triage_context is the canonical form — a typed object with failure_type, attempt_number, hint, subgoal, and state. The individual _triage_* kwargs remain for backward compatibility.

Observability¶

Install the optional extra:

pip install triage-agent[otel]

When opentelemetry-sdk is installed and a real TracerProvider is configured, triage emits three span types per run() call with no code change required:

Span	When	Key attributes
`triage.run`	Wraps the entire call including retries	`triage.run_id`, `triage.task`
`triage.classify`	Wraps each failure classification	`triage.failure_type`
`triage.dispatch`	Wraps each strategy dispatch	`triage.action_kind`, `triage.attempt`

All spans share the same trace_id and triage.run_id. Span status is OK on success, ERROR on escalate/abort, and UNSET (incomplete) on cancellation. The six structured log events (failure_classified, action_dispatched, etc.) emit regardless of whether OTel is configured.

Agent¶

Agent class¶

Agent ¶

__init__ ¶

run async ¶

stream async ¶

resume async ¶

clone ¶

report_misclassification ¶

Decorator form¶

agent ¶

Exceptions¶

TriageEscalationError ¶

TriageAbortError ¶

ContextVar helpers¶

get_recorder ¶

get_state_updater ¶

get_usage_recorder ¶

Conceptual notes¶

max_total_attempts vs max_recovery_attempts¶

Concurrent run() calls¶

Wrapped function contract¶

Observability¶

init ¶

run `async` ¶

stream `async` ¶

resume `async` ¶