Skip to content

agentix.resilience

resilience

Resilient model wrappers.

These wrap a :class:~agentix.model.ModelFn and are themselves ModelFns, so they drop into Agent(model=...) and compose (FallbackModel([RetryModel(primary), secondary])).

  • :class:RetryModel — retry transient errors with exponential backoff.
  • :class:FallbackModel — try each model in order; fall back on failure.

By default they catch Exception; narrow retry_on / fallback_on to your provider's transient error types (e.g. anthropic.APIStatusError, anthropic.RateLimitError) so you don't retry/mask real bugs.

Note: these are non-streaming wrappers (no stream method). Wrapping a streaming model with them disables streaming — the agent loop transparently falls back to a single __call__ per turn. Retrying a partially-streamed response is unsafe, so v1 keeps these one-shot.

RetryModel 

RetryModel(
    model: ModelFn,
    *,
    retries: int = 2,
    backoff: float = 0.5,
    retry_on: Sequence[type[BaseException]] = (Exception,),
    retry_after: RetryAfterFn = default_retry_after,
    max_sleep: float = 60.0,
    on_retry: OnRetry | None = None,
)

Retry a model on transient errors.

Backoff is exponential by default, but rate-limit aware: when the error carries a Retry-After (via retry_after), that server-requested delay is honored instead of blind backoff (capped at max_sleep). Wire on_retry to surface/log waits. Set retry_after=lambda _e: None to disable and always use exponential backoff.

with_response_format 

with_response_format(schema: dict[str, Any]) -> RetryModel

Delegate structured-output binding to the wrapped model (for Agent(response_model=…) composed with retries).

FallbackModel 

FallbackModel(
    models: Sequence[ModelFn],
    *,
    fallback_on: Sequence[type[BaseException]] = (
        Exception,
    ),
)

Try models in order; on a matching error, fall back to the next.

Use to escalate (small → big model) or to survive a provider outage. Falls back on exceptions — a model that returns a (refusal) response is not an error here; handle refusals separately.

with_response_format 

with_response_format(
    schema: dict[str, Any],
) -> FallbackModel

Bind structured-output to every wrapped model that supports it.

default_retry_after 

default_retry_after(exc: BaseException) -> float | None

Best-effort Retry-After extraction across SDK/HTTP error shapes.

Checks exc.retry_after (some SDKs) then a Retry-After response header (exc.response.headers). Returns seconds, or None if absent/unparseable (the caller then falls back to exponential backoff). Only the delta-seconds form is honored; an HTTP-date Retry-After is ignored.