Why this matters

LLM providers go down. Rate limits get hit. Models return unexpected outputs. A production AI feature that works 95% of the time and completely fails the other 5% delivers a terrible user experience. Reliability engineering for AI systems requires the same patterns as general distributed systems — circuit breakers, fallbacks, retry with backoff — plus AI-specific additions like quality-aware fallback chains and graceful degradation to non-AI responses.

Core concepts

Circuit breakers

A circuit breaker monitors failure rates and, after a threshold is exceeded, stops sending requests to the failing service for a cooldown period. This prevents a slow or failed dependency from causing cascading failures in your own service.

States: Closed (normal operation), Open (failures exceeded threshold, requests blocked), Half-open (cooldown passed, sending probe requests to test recovery).

import time
from enum import Enum
from dataclasses import dataclass, field


class CircuitState(Enum):
    closed = "closed"
    open = "open"
    half_open = "half_open"


@dataclass
class CircuitBreaker:
    name: str
    failure_threshold: int = 5
    recovery_timeout_seconds: float = 60.0
    success_threshold: int = 2  # successes needed to close from half-open

    _state: CircuitState = field(default=CircuitState.closed, init=False)
    _failure_count: int = field(default=0, init=False)
    _success_count: int = field(default=0, init=False)
    _opened_at: float | None = field(default=None, init=False)

    @property
    def is_open(self) -> bool:
        if self._state == CircuitState.open:
            if time.monotonic() - self._opened_at >= self.recovery_timeout_seconds:
                self._state = CircuitState.half_open
                self._success_count = 0
                return False
            return True
        return False

    def record_success(self) -> None:
        if self._state == CircuitState.half_open:
            self._success_count += 1
            if self._success_count >= self.success_threshold:
                self._state = CircuitState.closed
                self._failure_count = 0
        elif self._state == CircuitState.closed:
            self._failure_count = max(0, self._failure_count - 1)

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

    @property
    def state(self) -> CircuitState:
        _ = self.is_open  # trigger state transition check
        return self._state

Fallback chains

A fallback chain tries a sequence of providers in order. If the primary fails (or the circuit is open), it falls through to the next. The chain should be ordered from fastest/cheapest to slowest/most capable, or from primary to backup depending on your goals.

import asyncio
import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class ProviderConfig:
    name: str
    client: object
    model: str
    circuit_breaker: CircuitBreaker


class FallbackChain:
    def __init__(self, providers: list[ProviderConfig]):
        self.providers = providers

    async def complete(self, messages: list[dict]) -> tuple[str, str]:
        '''Returns (content, provider_name_used).'''
        last_error = None

        for provider in self.providers:
            if provider.circuit_breaker.is_open:
                logger.info("circuit_open_skipping", extra={"provider": provider.name})
                continue

            try:
                content = await self._call(provider, messages)
                provider.circuit_breaker.record_success()
                return content, provider.name
            except Exception as exc:
                logger.warning(
                    "provider_failed_trying_next",
                    extra={"provider": provider.name, "error": str(exc)},
                )
                provider.circuit_breaker.record_failure()
                last_error = exc

        raise RuntimeError(f"All providers exhausted. Last error: {last_error}")

    async def _call(self, provider: ProviderConfig, messages: list[dict]) -> str:
        # Provider-specific call implementation
        raise NotImplementedError

Graceful degradation

When all AI providers are unavailable, you have a choice: fail completely or return a degraded but useful response. Graceful degradation keeps your product functional in reduced-capability mode.

Common degradation strategies:

1. Cached fallback: return the most recent cached response for similar inputs
2. Template fallback: return a structured response that tells the user the AI is temporarily unavailable
3. Non-AI fallback: route to a simpler rule-based or search-based implementation
4. Queue for later: accept the request, process it when the provider recovers, notify the user

from enum import Enum


class DegradationLevel(Enum):
    full = "full"          # AI provider available
    cached = "cached"      # using cached AI responses
    template = "template"  # template responses only
    unavailable = "unavailable"  # feature disabled


async def handle_with_degradation(
    query: str,
    chain: FallbackChain,
    cache: object,
    degradation_level: DegradationLevel,
) -> dict:

    if degradation_level == DegradationLevel.full:
        try:
            content, provider = await chain.complete([{"role": "user", "content": query}])
            return {"content": content, "source": "ai", "provider": provider}
        except RuntimeError:
            degradation_level = DegradationLevel.cached

    if degradation_level == DegradationLevel.cached:
        cached = cache.get_best_match(query)
        if cached:
            return {"content": cached.content, "source": "cache", "note": "AI temporarily unavailable"}

    return {
        "content": "Our AI assistant is temporarily unavailable. Please try again in a few minutes.",
        "source": "fallback",
    }

Multi-provider failover

Failover differs from fallback chains in that it is about switching to an equivalent provider, not a degraded alternative. Your primary is OpenAI; your failover is Anthropic or a self-hosted model. The responses should be comparable in quality.

from abc import ABC, abstractmethod
import asyncio


class LLMProvider(ABC):
    @abstractmethod
    async def complete(self, messages: list[dict]) -> str: ...

    @abstractmethod
    async def health_check(self) -> bool: ...


class MultiProviderClient:
    def __init__(self, providers: list[LLMProvider], timeout: float = 10.0):
        self.providers = providers
        self.timeout = timeout

    async def complete_with_fallback(self, messages: list[dict]) -> str:
        for provider in self.providers:
            try:
                return await asyncio.wait_for(
                    provider.complete(messages),
                    timeout=self.timeout
                )
            except (asyncio.TimeoutError, Exception) as exc:
                continue  # circuit breaker would be applied here in production

        raise RuntimeError("All providers failed")

SLAs for AI features

AI features need two kinds of SLOs: system SLOs (latency, availability) and quality SLOs (output quality thresholds). Document both before launch.

SLO type	Metric	Typical target
System	P95 response latency	< 3 seconds
System	API availability	> 99.5%
System	Error rate	< 1% over 5 min
Quality	Pass rate (eval)	> 90% rolling 24h
Quality	Toxicity rate	< 0.1% sampled
Quality	Provider fallback rate	< 5%

When a quality SLO is breached, the response is a product decision: degraded mode, feature rollback, or user communication. Unlike system SLOs, quality SLO breaches require human judgment about whether the degradation is temporary or structural.

Working example

A production-hardened request handler combining circuit breaker, fallback chain, and quality threshold check:

import asyncio
import logging

logger = logging.getLogger(__name__)


async def reliable_complete(
    messages: list[dict],
    chain: FallbackChain,
    quality_threshold: float = 0.8,
) -> dict:
    try:
        content, provider = await asyncio.wait_for(
            chain.complete(messages),
            timeout=30.0
        )
    except asyncio.TimeoutError:
        logger.error("all_providers_timed_out")
        return {"content": None, "error": "timeout", "degraded": True}
    except RuntimeError as exc:
        logger.error("all_providers_failed", extra={"error": str(exc)})
        return {"content": None, "error": "provider_failure", "degraded": True}

    # Optional quality gate — only apply for high-stakes features
    quality_score = assess_quality(content)
    if quality_score < quality_threshold:
        logger.warning(
            "quality_below_threshold",
            extra={"provider": provider, "score": quality_score}
        )
        return {
            "content": content,
            "provider": provider,
            "quality_score": quality_score,
            "flagged_for_review": True,
        }

    return {"content": content, "provider": provider, "quality_score": quality_score}


def assess_quality(content: str) -> float:
    '''Placeholder — replace with real quality assessment.'''
    if not content or len(content.strip()) < 10:
        return 0.0
    return 1.0

Common mistakes

No circuit breaker, only retry. Retrying a consistently failing provider adds latency and burns your retry budget. A circuit breaker stops retrying after the threshold and recovers gracefully.

Fallback to a much weaker model without user communication. If your fallback is a significantly worse model, users notice. Either communicate the degradation or ensure the fallback quality is acceptable for your feature.

SLOs defined only for system metrics. An AI feature with 100% uptime and consistently low-quality responses is failing its users. Quality SLOs are as important as availability SLOs.

Circuit breaker with no monitoring. A circuit breaker that opens silently is invisible. Alert when a circuit opens so the on-call engineer knows a dependency is degraded.

No graceful degradation path. Discovering your degradation strategy during an incident is too late. Test your fallback paths in staging, including the circuit-open state.

Try it yourself

1. Implement the CircuitBreaker above and write a test that drives it through the full state cycle: closed -> open (after N failures) -> half-open (after recovery timeout) -> closed (after M successes).
2. Build a FallbackChain with two mock providers where the first always raises an exception. Verify that the second provider's response is returned.
3. Define SLOs for an AI feature you are building or have built. Write down the metric, target value, measurement window, and the response procedure when the SLO is breached.