LLM APIs have outages. Rate limits hit. Timeouts happen. A circuit breaker stops your system from hammering a failing API — and returns a degraded-but-functional response instead of cascading failures that take your entire service down.
All requests pass through to the LLM. The breaker tracks failures. If failures exceed threshold within a time window, it opens.
import time
from enum import Enum
from dataclasses import dataclass, field
import anthropic
client = anthropic.Anthropic()
class BreakerState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreaker:
failure_threshold: int = 5 # Open after N failures
recovery_timeout: float = 60.0 # Seconds before trying again
half_open_max_calls: int = 1 # Trial calls in half-open state
_state: BreakerState = field(default=BreakerState.CLOSED, init=False)
_failure_count: int = field(default=0, init=False)
_last_failure_time: float = field(default=0.0, init=False)
def call(self, func, *args, fallback=None, **kwargs):
"""Wrap a function call with circuit breaker protection."""
if self._state == BreakerState.OPEN:
if time.monotonic() - self._last_failure_time >= self.recovery_timeout:
self._state = BreakerState.HALF_OPEN
else:
return fallback() if callable(fallback) else fallback
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
if callable(fallback):
return fallback()
raise
def _on_success(self):
self._failure_count = 0
self._state = BreakerState.CLOSED
def _on_failure(self):
self._failure_count += 1
self._last_failure_time = time.monotonic()
if self._failure_count >= self.failure_threshold:
self._state = BreakerState.OPEN
print(f"Circuit OPENED after {self._failure_count} failures")
# Usage
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
def llm_call(prompt: str) -> str:
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
def cached_fallback() -> str:
return "I'm temporarily unavailable. Please try again shortly."
answer = breaker.call(llm_call, "What is Python?", fallback=cached_fallback)import httpx
import anthropic
# Always set explicit timeouts — no timeout = potentially blocked forever
client = anthropic.Anthropic(
http_client=httpx.Client(
timeout=httpx.Timeout(
connect=5.0, # Max time to establish connection
read=30.0, # Max time between bytes (stream timeout)
write=5.0, # Max time to send request
pool=5.0, # Max time to get a connection from pool
)
)
)
# Async version
async_client = anthropic.AsyncAnthropic(
http_client=httpx.AsyncClient(
timeout=httpx.Timeout(connect=5.0, read=30.0, write=5.0, pool=5.0)
)
)Return the most recent successful response for a similar query. Best for FAQ-style content that rarely changes.
"I'm temporarily unavailable" + retry-after header. Honest and non-confusing for users.
Try claude-haiku if claude-sonnet fails. Or try a local model (Ollama) if the API is down.
For narrow tasks (classification, routing), keep a simple if/else logic as the fallback that runs without any LLM.
Test your fallback before you need it.Run a chaos test monthly: inject failures into your LLM client and verify that the fallback activates and users see a graceful response — not a 500.
Lesson 6 moves from resilience to performance: load testing your LLM application to find the breaking points before they hit production.