How to build a circuit breaker for LLM API calls

A circuit breaker for LLM API calls stops a single degraded model endpoint from cascading failures into your whole inference path. Treating the circuit breaker llm api pattern as a first-class infrastructure component saves you from retry storms and wasted token spend. Most teams wire retries but skip the breaker, then wonder why a 5-minute provider hiccup becomes a full outage.

Step 1: Define what counts as a failure

Not every non-200 from an LLM endpoint should trip a breaker. A 400 with invalid_request_error is a bug in your prompt construction, not a provider fault. Trip only on conditions that indicate the upstream is unhealthy or unreachable:

• TCP connection timeouts and TLS handshake failures
• HTTP 502/503/504 from the gateway or origin model server
• HTTP 429 where Retry-After exceeds your remaining latency budget
• Read timeouts where the connection opens but tokens stop arriving
• Streaming stalls detected by a per-token deadline

Set an explicit latency threshold. A 25-second cold start on a self-hosted Llama instance is fine if your user-facing SLA is 60 seconds. The breaker should enforce the budget, not guess.

from dataclasses import dataclass
from typing import tuple

@dataclass
class BreakerConfig:
    failure_threshold: int = 5          # consecutive failures to open
    reset_timeout: float = 30.0         # seconds spent in open state
    latency_threshold: float = 20.0     # max seconds before counting as timeout
    expected_exceptions: tuple = (TimeoutError, ConnectionError)
    model: str = "unknown"

Step 2: Implement the circuit breaker state machine

The classic three states—closed, open, half-open—are enough. In closed, you count failures. After failure_threshold, you move to open and reject calls locally. After reset_timeout, you move to half-open and allow exactly one probe. Success closes the breaker; failure reopens it.

A naive consecutive-counter breaks under bursty traffic. Prefer a rolling window keyed by timestamp so a single slow minute doesn’t permanently open the breaker at low throughput.

import time
import threading
from collections import deque

class CircuitBreaker:
    def __init__(self, config: BreakerConfig):
        self.cfg = config
        self._fail_ts = deque()          # timestamps of recent failures
        self._state = "closed"
        self._opened_at = 0.0
        self._lock = threading.Lock()

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            with self._lock:
                self._maybe_half_open()
                if self._state == "open":
                    raise RuntimeError(f"circuit open for {self.cfg.model}")
            try:
                return func(*args, **kwargs)
            except self.cfg.expected_exceptions:
                self._record_failure()
                raise
        return wrapper

    def _maybe_half_open(self):
        if self._state == "open" and time.monotonic() - self._opened_at >= self.cfg.reset_timeout:
            self._state = "half-open"

    def _record_failure(self):
        now = time.monotonic()
        with self._lock:
            self._fail_ts.append(now)
            # drop failures outside a 60s window
            while self._fail_ts and now - self._fail_ts[0] > 60.0:
                self._fail_ts.popleft()
            if self._state == "half-open":
                self._state = "open"
                self._opened_at = now
            elif len(self._fail_ts) >= self.cfg.failure_threshold:
                self._state = "open"
                self._opened_at = now

    def _record_success(self):
        with self._lock:
            if self._state == "half-open":
                self._state = "closed"
                self._fail_ts.clear()

For asyncio services, swap threading.Lock for asyncio.Lock and await func(*args, **kwargs) inside the wrapper. The state logic is identical.

Step 3: Wrap your LLM client calls

Decorate the function that actually hits the network. Below we wrap the synchronous OpenAI client, but the pattern is the same for aiohttp or the async AsyncOpenAI client.

from openai import OpenAI, APITimeoutError, APIConnectionError

client = OpenAI(base_url="https://api.example.com/v1", api_key="sk-...")

breaker = CircuitBreaker(BreakerConfig(model="gpt-4o-mini", failure_threshold=4))

@breaker
def complete(prompt: str, model: str = "gpt-4o-mini") -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        timeout=breaker.cfg.latency_threshold,
    )
    return resp.choices[0].message.content

If you stream, wrap the iteration loop and enforce a per-token deadline:

import sys

@breaker
def stream_complete(prompt: str):
    stream = client.chat.completions.create(
        model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}],
        stream=True, timeout=10.0
    )
    last = time.monotonic()
    for chunk in stream:
        if time.monotonic() - last > 5.0:
            raise TimeoutError("token stall")
        last = time.monotonic()
        yield chunk

The breaker catches the TimeoutError and records a failure without waiting for the full request to die.

Step 4: Add fallback and degradation

An open breaker should not return an error to the user if a degraded answer is acceptable. Route to a smaller model, return a cached prior response, or emit a queued acknowledgement.

cache = {}

def safe_complete(prompt: str) -> str:
    try:
        return complete(prompt)
    except RuntimeError:
        # primary circuit open
        return fallback_complete(prompt)

fallback_breaker = CircuitBreaker(BreakerConfig(model="gpt-3.5-turbo", failure_threshold=8))

@fallback_breaker
def fallback_complete(prompt: str) -> str:
    if prompt in cache:
        return cache[prompt]
    resp = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        timeout=10.0,
    )
    cache[prompt] = resp.choices[0].message.content
    return cache[prompt]

Keep the fallback on its own breaker. A failing cheap model should not trip the primary, and vice versa.

Step 5: Leverage gateway-level fallback

If you front your inference traffic with a gateway that already does provider failover, your local breaker still earns its keep. For example, n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models and applies automatic fallback when a provider is rate-limited or degraded. That handles cross-provider outages transparently, but a local circuit breaker llm api layer prevents your service from flooding the gateway with doomed requests during a partial degradation window.

When using such a gateway, honor its client routing directives and forward provider cache-control hints. Don’t wrap the gateway call in a blind ten-retry loop; the gateway already retries internally. Your breaker should sit above the client and trip only on hard signals.

Step 6: Test with fault injection

Write a test that forces timeouts and asserts the breaker opens, blocks, and half-opens. Use monkeypatch to throw the exact exception your client raises.

import pytest
import time
from openai import APITimeoutError

def test_breaker_lifecycle(monkeypatch):
    br = CircuitBreaker(BreakerConfig(model="test", failure_threshold=3, reset_timeout=0.1))
    calls = {"n": 0}

    def fake(*a, **k):
        calls["n"] += 1
        raise APITimeoutError("timeout")

    wrapped = br(fake)
    for _ in range(3):
        with pytest.raises(APITimeoutError):
            wrapped()
    with pytest.raises(RuntimeError):
        wrapped()           # open state, no network call
    assert calls["n"] == 3  # proof we blocked locally

    time.sleep(0.15)        # wait past reset_timeout
    with pytest.raises(APITimeoutError):
        wrapped()           # half-open probe allowed
    assert calls["n"] == 4
    with pytest.raises(RuntimeError):
        wrapped()           # probe failed, back to open

Run this in CI on every change to the breaker logic. Add a chaos test that points the client at http://127.0.0.1:9/ (a closed port) to confirm connection errors trip the breaker in production-like conditions.

Step 7: Monitor and tune thresholds

Export state transitions as metrics. At minimum, count breaker_opened_total{model="..."}, breaker_half_open_total, and breaker_closed_total. Alert when open events exceed one per minute for a given model.

import logging

def _record_failure(self):
    # inside the existing method, after state change to open:
    if self._state == "open":
        logging.info("breaker_opened model=%s", self.cfg.model)

Tune failure_threshold to traffic shape. At 200 req/s, five failures is 25 ms of pain; at 0.2 req/s, it is 25 seconds. Use the rolling window from Step 2 and set the threshold relative to a 60-second interval. For high-throughput services, track p95 latency and open the breaker when error rate exceeds 20% over the window, not just raw counts.

Verify success

You have a working circuit breaker llm api integration when the following hold:

1. A scripted fault injection causes the breaker to open after the configured threshold.
2. Calls during open state raise RuntimeError locally with zero outbound HTTP attempts.
3. After reset_timeout, exactly one probe call executes (confirmed by call counter).
4. A successful probe closes the breaker; a failing probe reopens it.
5. The fallback path returns a degraded but valid response to the caller instead of an exception.
6. Metrics show breaker_opened_total increments only during induced faults, not during normal operation.

Run the pytest suite from Step 6 in CI and add a load test that injects 10% artificial latency to confirm the breaker trips and recovers without manual intervention. The pattern is small, but skipping it turns a blip into an outage.