Why this matters

AI providers are not databases. They rate-limit, return 529s during high traffic, timeout on long generations, return malformed JSON on edge-case prompts, and occasionally go fully offline. A pipeline that propagates every provider error to the user is fragile. A pipeline that silently swallows errors is worse — you end up with corrupted outputs and no signal.

The patterns in this lesson — retry with backoff, circuit breakers, timeout cascading, and partial failure handling — are what separate toy AI integrations from systems that run reliably in production.

Core concepts

Retry with exponential backoff and jitter

Most provider failures are transient: rate limits, brief 5xx errors, network blips. A retry with exponential backoff recovers from these without manual intervention:

import asyncio
import logging
import random
from typing import TypeVar, Callable, Awaitable

logger = logging.getLogger(__name__)
T = TypeVar("T")


class RetryableError(Exception):
    # Raised for errors that should trigger a retry (rate limits, 5xx, timeouts).


class PermanentError(Exception):
    # Raised for errors that should not be retried (4xx client errors).


async def with_retry(
    fn: Callable[[], Awaitable[T]],
    max_attempts: int = 3,
    base_delay_s: float = 1.0,
    max_delay_s: float = 60.0,
    retryable_exceptions: tuple = (RetryableError, TimeoutError),
) -> T:
    # Retry an async callable with exponential backoff and full jitter.
    last_exc: Exception | None = None

    for attempt in range(max_attempts):
        try:
            return await fn()
        except PermanentError:
            raise
        except retryable_exceptions as exc:
            last_exc = exc
            if attempt < max_attempts - 1:
                cap = min(base_delay_s * (2 ** attempt), max_delay_s)
                delay = random.uniform(0, cap)
                logger.warning(
                    "retry_scheduled",
                    extra={"attempt": attempt + 1, "max_attempts": max_attempts, "delay_s": round(delay, 2), "error": str(exc)},
                )
                await asyncio.sleep(delay)

    raise last_exc  # type: ignore[misc]


def classify_http_error(status_code: int) -> type[Exception]:
    if status_code == 429 or status_code >= 500:
        return RetryableError
    return PermanentError

Full jitter (random.uniform(0, cap)) spreads retries across the full delay range, preventing thundering herds when many clients retry simultaneously.

Circuit breaker pattern

A circuit breaker stops hammering a provider that is clearly down, giving it time to recover:

import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Awaitable, TypeVar

T = TypeVar("T")


class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout_s: float = 60.0
    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    _failure_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0.0, init=False)

    @property
    def state(self) -> CircuitState:
        if self._state == CircuitState.OPEN:
            if time.monotonic() - self._last_failure_time > self.recovery_timeout_s:
                self._state = CircuitState.HALF_OPEN
        return self._state

    def record_success(self) -> None:
        self._failure_count = 0
        self._state = CircuitState.CLOSED

    def record_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = time.monotonic()
        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN

    async def call(self, fn: Callable[[], Awaitable[T]]) -> T:
        if self.state == CircuitState.OPEN:
            raise RuntimeError("Circuit breaker is open — provider unavailable")
        try:
            result = await fn()
            self.record_success()
            return result
        except Exception:
            self.record_failure()
            raise

After failure_threshold consecutive failures, the circuit opens and all calls fail immediately. After recovery_timeout_s, it moves to HALF_OPEN and allows one test call through.

Resilient provider wrapper combining retry and circuit breaker

import asyncio
import logging
import time

logger = logging.getLogger(__name__)


class ResilientProvider:
    def __init__(self, client, name: str, max_retries: int = 3, circuit_threshold: int = 5, circuit_timeout_s: float = 60.0) -> None:
        self._client = client
        self._name = name
        self._max_retries = max_retries
        self._circuit = CircuitBreaker(failure_threshold=circuit_threshold, recovery_timeout_s=circuit_timeout_s)

    async def generate(self, prompt: str, timeout_s: float = 30.0) -> dict:
        start = time.monotonic()
        try:
            result = await self._circuit.call(
                lambda: with_retry(
                    lambda: asyncio.wait_for(self._client.generate(prompt), timeout=timeout_s),
                    max_attempts=self._max_retries,
                )
            )
            latency = int((time.monotonic() - start) * 1000)
            logger.info("provider_success", extra={"provider": self._name, "latency_ms": latency})
            return result
        except RuntimeError as exc:
            if "Circuit breaker" in str(exc):
                logger.error("circuit_open_rejection", extra={"provider": self._name})
            raise
        except Exception as exc:
            latency = int((time.monotonic() - start) * 1000)
            logger.error("provider_failed", extra={"provider": self._name, "latency_ms": latency, "error": str(exc)})
            raise

Partial failure handling in multi-step pipelines

When a pipeline has multiple steps and one step fails for some items, continue with the items that succeeded:

from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable


@dataclass
class PipelineResult:
    item_id: str
    success: bool
    data: Any = None
    failed_at_step: str | None = None
    error: str | None = None


async def run_pipeline_with_partial_failures(
    items: list[dict],
    steps: list[tuple[str, Callable[[Any], Awaitable[Any]]]],
) -> list[PipelineResult]:
    results = []

    for item in items:
        current_data = item
        failed = False

        for step_name, step_fn in steps:
            try:
                current_data = await step_fn(current_data)
            except Exception as exc:
                results.append(PipelineResult(
                    item_id=item.get("id", "unknown"),
                    success=False,
                    failed_at_step=step_name,
                    error=str(exc),
                ))
                failed = True
                break

        if not failed:
            results.append(PipelineResult(item_id=item.get("id", "unknown"), success=True, data=current_data))

    successful = [r for r in results if r.success]
    failed_results = [r for r in results if not r.success]
    logger.info("pipeline_complete", extra={"total": len(results), "successful": len(successful), "failed": len(failed_results)})
    return results

Common mistakes

Retrying non-retryable errors. A 400 Bad Request caused by a malformed prompt will never succeed on retry. Classify errors before retrying. Only retry transient errors (429, 5xx, timeout).

No jitter on retries. Without jitter, all clients that hit the same rate limit retry at the same intervals, creating thundering herds. Always add randomness to retry delays.

Circuit breaker threshold too low. A threshold of 1 or 2 opens the circuit on the first transient error. Set it high enough that a brief flap does not open the circuit.

Catching all exceptions at the top level without logging. except Exception: return None hides every error. Log the exception type and message before swallowing.

No total timeout on the pipeline. A multi-step pipeline where each step has a 30s timeout can run for 150 seconds if all five steps timeout. Set a total wall-clock budget on the pipeline as well.

Try it yourself

1. Implement with_retry with exponential backoff and jitter. Write a test that injects failures for the first two attempts and verifies it succeeds on the third. Check that the delay between retries is randomized and increasing.
2. Implement a CircuitBreaker class. Write a test that opens the circuit after N consecutive failures, verifies it rejects calls immediately when open, and allows calls again after the recovery timeout.
3. Build a ResilientProvider wrapper for an LLM client you use. Add logging for each retry, circuit state change, and final failure. Run it against a mock that fails the first two calls and succeeds on the third.