How to Implement Circuit Breakers for External LLM APIs

How to Implement Circuit Breakers for External LLM APIs

Large Language Model (LLM) APIs are powerful, but they are also fragile dependencies. Rate limits, transient network failures, provider outages, and slow token streaming can all turn a healthy application into a cascading failure. A circuit breaker is a resilience pattern that stops your system from repeatedly calling a failing service, giving it time to recover and protecting your own resources. In this tutorial, we'll explore what circuit breakers are, why they're essential for LLM integrations, and how to implement them in practice.

What Is a Circuit Breaker?

A circuit breaker wraps an external call and monitors its success and failure rates. Based on recent behavior, it switches between three states:

• Closed — Requests flow through normally. Failures and successes are recorded.
• Open — When failures exceed a threshold, the breaker "trips." All requests fail fast without hitting the remote service.
• Half-Open — After a cooldown period, a limited number of trial requests are allowed. If they succeed, the breaker returns to Closed; if they fail, it returns to Open.

This is conceptually similar to an electrical circuit breaker: when too much current flows, the breaker trips to prevent damage. Once conditions stabilize, it can be reset.

Why Circuit Breakers Matter for LLM APIs

LLM APIs have unique characteristics that make circuit breakers especially valuable:

• High latency and cost: A single request can take 10–60 seconds and cost real money. Retrying blindly wastes both time and budget.
• Rate limiting: Providers like OpenAI and Anthropic return 429 errors when limits are exceeded. Hammering the API only prolongs the problem.
• Token-based quotas: Some failures (e.g., TPM — tokens per minute) require waiting before retrying, not immediate reattempts.
• Partial outages: A provider may serve some models but fail others. A circuit breaker per model prevents one failing model from blocking your entire app.
• Downstream impact: Without a breaker, a slow LLM API can exhaust your connection pool, thread pool, or serverless concurrency, taking down unrelated features.

How to Use Circuit Breakers with LLM APIs

Let's walk through a practical implementation in Python. We'll build a circuit breaker from scratch first to understand the mechanics, then show how to use a production-grade library.

1. A Minimal Circuit Breaker from Scratch

The core idea is a state machine that tracks failures and decides whether to allow a call. Here's a simple implementation:

import time
import random
from enum import Enum
from functools import wraps

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

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30, half_open_max_calls=3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        self.state = State.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    def _before_call(self):
        if self.state == State.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                print("[CircuitBreaker] Transitioning OPEN -> HALF_OPEN")
                self.state = State.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitBreakerOpenError("Circuit is OPEN; failing fast")

        if self.state == State.HALF_OPEN and self.half_open_calls >= self.half_open_max_calls:
            raise CircuitBreakerOpenError("Circuit is HALF_OPEN and at trial limit")

        if self.state == State.HALF_OPEN:
            self.half_open_calls += 1

    def _on_success(self):
        if self.state == State.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.half_open_max_calls:
                print("[CircuitBreaker] Transitioning HALF_OPEN -> CLOSED")
                self._reset()
        else:
            self.failure_count = 0

    def _on_failure(self):
        self.last_failure_time = time.time()
        if self.state == State.HALF_OPEN:
            print("[CircuitBreaker] Transitioning HALF_OPEN -> OPEN")
            self.state = State.OPEN
            self.success_count = 0
        else:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                print("[CircuitBreaker] Transitioning CLOSED -> OPEN")
                self.state = State.OPEN

    def _reset(self):
        self.state = State.CLOSED
        self.failure_count = 0
        self.success_count = 0

    def call(self, func, *args, **kwargs):
        self._before_call()
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

class CircuitBreakerOpenError(Exception):
    pass

Now let's wrap an LLM call with this breaker:

import openai

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    try:
        return breaker.call(
            openai.chat.completions.create,
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
    except CircuitBreakerOpenError:
        # Fallback: use a cached response, a cheaper model, or a queue
        return "Service temporarily unavailable. Please try again later."
    except Exception as e:
        print(f"LLM call failed: {e}")
        raise

# Simulate usage
for i in range(20):
    try:
        response = call_llm("Summarize the benefits of circuit breakers.")
        print(f"Call {i}: OK")
    except Exception as e:
        print(f"Call {i}: FAILED - {e}")

2. Handling LLM-Specific Errors

Not every exception should count as a failure. A 400 Bad Request caused by a malformed prompt is a client error, not a service problem. You should only trip the breaker on transient or server-side failures:

from openai import RateLimitError, APIConnectionError, APITimeoutError, InternalServerError

def is_tripable_error(exc: Exception) -> bool:
    return isinstance(exc, (
        RateLimitError,
        APIConnectionError,
        APITimeoutError,
        InternalServerError,
    ))

class LLMSpecificCircuitBreaker(CircuitBreaker):
    def call(self, func, *args, **kwargs):
        self._before_call()
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            if is_tripable_error(e):
                self._on_failure()
            # Non-tripable errors (e.g., 400) don't affect circuit state
            raise

3. Using a Production-Grade Library: pybreaker

For production systems, prefer a battle-tested library. pybreaker is a popular Python circuit breaker that supports listeners, async usage, and custom storage backends:

pip install pybreaker openai

import pybreaker
import openai
from openai import RateLimitError, APIConnectionError, APITimeoutError, InternalServerError

# Define which exceptions trip the breaker
tripable_exceptions = (
    RateLimitError,
    APIConnectionError,
    APITimeoutError,
    InternalServerError,
)

# Create a breaker: trip after 5 consecutive failures, recover after 60s
llm_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    exclude=lambda exc: not isinstance(exc, tripable_exceptions),
)

@llm_breaker
def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    return openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )

# Register a listener for observability
class LoggingListener(pybreaker.CircuitBreakerListener):
    def state_change(self, cb, old, new):
        print(f"[Breaker] {old.name} -> {new.name}")

llm_breaker.add_listener(LoggingListener())

# Usage with fallback
def safe_call_llm(prompt: str) -> str:
    try:
        return call_llm(prompt)
    except pybreaker.CircuitBreakerError:
        return fallback_response(prompt)

def fallback_response(prompt: str) -> str:
    return "I'm unable to process your request right now. Please try again shortly."

4. Async Circuit Breakers for Streaming and Concurrent Calls

Modern LLM apps often use async clients and streaming. Here's how to combine pybreaker with the async OpenAI client:

import asyncio
import pybreaker
from openai import AsyncOpenAI
from openai import RateLimitError, APIConnectionError, APITimeoutError, InternalServerError

client = AsyncOpenAI()

async_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    exclude=lambda exc: not isinstance(exc, (
        RateLimitError, APIConnectionError, APITimeoutError, InternalServerError
    )),
)

@async_breaker
async def call_llm_async(prompt: str, model: str = "gpt-4o") -> str:
    response = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

async def stream_llm_async(prompt: str, model: str = "gpt-4o"):
    """Stream tokens with circuit breaker protection on connection."""
    if async_breaker.current_state == pybreaker.CircuitState.OPEN:
        raise pybreaker.CircuitBreakerError("Circuit is open")
    try:
        stream = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content
        async_breaker._inc_counter()
    except (RateLimitError, APIConnectionError, APITimeoutError, InternalServerError) as e:
        async_breaker._handle_error(e)
        raise

async def main():
    try:
        result = await call_llm_async("Explain quantum computing in one sentence.")
        print(result)
    except pybreaker.CircuitBreakerError:
        print("Circuit is open; using fallback.")

asyncio.run(main())

5. Per-Model and Per-Tenant Breakers

A single global breaker is rarely sufficient. Different models have different rate limits and reliability profiles. Use a registry keyed by model (and optionally tenant):

from dataclasses import dataclass, field
from typing import Dict
import pybreaker

@dataclass
class BreakerRegistry:
    breakers: Dict[str, pybreaker.CircuitBreaker] = field(default_factory=dict)

    def get(self, model: str) -> pybreaker.CircuitBreaker:
        if model not in self.breakers:
            self.breakers[model] = pybreaker.CircuitBreaker(
                fail_max=5,
                reset_timeout=60,
                name=f"llm-{model}",
            )
        return self.breakers[model]

registry = BreakerRegistry()

def call_llm_with_model(prompt: str, model: str) -> str:
    breaker = registry.get(model)
    try:
        return breaker.call(lambda: openai.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        ))
    except pybreaker.CircuitBreakerError:
        # Try a fallback model if the primary is down
        if model != "gpt-4o-mini":
            return call_llm_with_model(prompt, "gpt-4o-mini")
        raise

Best Practices

• Only trip on transient errors. Client errors (400, 401, 403, 404) should not affect circuit state. Trip on 429, 5xx, timeouts, and connection errors.
• Respect Retry-After headers. When a provider returns 429 with a retry hint, use that value as your reset timeout instead of a fixed value.
• Always provide a fallback. A circuit breaker without a fallback just moves the failure. Use cached responses, a cheaper model, a queue-and-retry-later pattern, or a graceful degradation message.
• Use per-model breakers. One failing model should not block requests to healthy models. Key your breakers by model, endpoint, or tenant.
• Combine with retries and timeouts. Circuit breakers complement retries (with exponential backoff) and timeouts. Retry transient failures a few times, then let the breaker trip if the problem persists.
• Instrument with metrics. Emit metrics on state transitions, failure counts, and time spent in each state. Integrate with Prometheus, Datadog, or OpenTelemetry.
• Tune thresholds based on traffic. A fail_max of 5 is fine for low traffic, but high-throughput systems may need higher thresholds or sliding-window failure-rate logic (e.g., trip when >50% of the last 100 calls fail).
• Test the half-open path. The most error-prone part of a circuit breaker is the half-open transition. Write integration tests that simulate failures and verify recovery behavior.
• Consider concurrency limits in half-open. Limit the number of simultaneous trial requests to avoid a thundering herd when the breaker transitions from open to half-open.
• Don't forget streaming. Streaming responses can fail mid-stream. Decide whether a mid-stream failure counts as a circuit failure, and handle partial output gracefully.

Conclusion

Circuit breakers are a critical resilience pattern for any application that depends on external LLM APIs. By failing fast when a provider is struggling, you protect your application's resources, reduce wasted spend on doomed retries, and create opportunities for graceful degradation. Start by identifying which LLM calls are on your critical path, wrap them with a breaker keyed by model or tenant, classify errors carefully so only transient failures trip the circuit, and always pair the breaker with a meaningful fallback. Combined with retries, timeouts, and good observability, circuit breakers turn unpredictable third-party dependencies into manageable, self-healing components of your system.