Circuit breaking LLM providers in LangChain apps

LangChain applications that depend on a single LLM provider inherit that provider’s uptime, latency, and rate limits. A langchain circuit breaker llm provider tutorial should show you how to detect degraded providers, trip a circuit, and route traffic to a healthy alternative without rewriting your chain logic. This post walks through a production-ready pattern you can drop into any LangChain app.

Step 1: Define the failure modes you care about

Not every error deserves a circuit trip. Network timeouts, 5xx responses, and explicit rate-limit headers (429 with Retry-After) are signals that a provider is unhealthy. Client errors (400, 401, 404) usually indicate a bug in your request — don’t circuit-break on those.

Create a small classification module so the rest of your code stays clean:

# circuit_classifier.py
from enum import Enum
from typing import TypeVar

import httpx
from langchain_core.runnables import RunnableConfig

class ProviderState(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    OPEN = "open"

T = TypeVar("T")

def classify_exception(exc: BaseException) -> ProviderState:
    """Map an exception to a provider state."""
    if isinstance(exc, httpx.TimeoutException):
        return ProviderState.DEGRADED
    if isinstance(exc, httpx.HTTPStatusError):
        status = exc.response.status_code
        if status >= 500:
            return ProviderState.DEGRADED
        if status == 429:
            return ProviderState.DEGRADED
    return ProviderState.HEALTHY  # treat everything else as non-tripping

This keeps your circuit breaker logic focused on infrastructure failures, not prompt engineering mistakes.

Step 2: Build a lightweight circuit breaker

LangChain’s Runnable interface makes it easy to wrap any model call with cross-cutting logic. We’ll implement a state machine with three states: closed (normal), open (tripped), and half-open (testing recovery).

# circuit_breaker.py
import time
import threading
from dataclasses import dataclass, field
from typing import Callable, Any

from circuit_classifier import ProviderState, classify_exception

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0  # seconds
    half_open_max_calls: int = 3

    _state: ProviderState = field(default=ProviderState.HEALTHY, init=False)
    _failure_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0.0, init=False)
    _half_open_calls: int = field(default=0, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    def call(self, func: Callable[[], Any]) -> Any:
        with self._lock:
            if self._state == ProviderState.OPEN:
                if time.time() - self._last_failure_time >= self.recovery_timeout:
                    self._state = ProviderState.DEGRADED  # half-open
                    self._half_open_calls = 0
                else:
                    raise CircuitOpenError("Circuit breaker is open")

            if self._state == ProviderState.DEGRADED:
                if self._half_open_calls >= self.half_open_max_calls:
                    raise CircuitOpenError("Half-open call limit reached")
                self._half_open_calls += 1

        try:
            result = func()
        except Exception as exc:
            self._record_failure(exc)
            raise
        else:
            self._record_success()
            return result

    def _record_failure(self, exc: BaseException) -> None:
        with self._lock:
            state = classify_exception(exc)
            if state == ProviderState.DEGRADED:
                self._failure_count += 1
                self._last_failure_time = time.time()
                if self._failure_count >= self.failure_threshold:
                    self._state = ProviderState.OPEN
            # HEALTHY exceptions don't increment the counter

    def _record_success(self) -> None:
        with self._lock:
            if self._state == ProviderState.DEGRADED:
                self._state = ProviderState.HEALTHY
                self._failure_count = 0
                self._half_open_calls = 0
            elif self._state == ProviderState.HEALTHY:
                self._failure_count = 0

class CircuitOpenError(Exception):
    """Raised when the circuit is open and the call is rejected."""
    pass

This implementation is deliberately minimal — no external dependencies, thread-safe, and easy to test in isolation.

Step 3: Wrap each provider with its own breaker

Each LLM provider gets an independent circuit breaker. This prevents a single degraded provider from contaminating the others.

# provider_registry.py
from typing import Dict
from langchain_core.language_models import BaseChatModel
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

from circuit_breaker import CircuitBreaker, CircuitOpenError

class ProviderRegistry:
    def __init__(self):
        self._models: Dict[str, BaseChatModel] = {}
        self._breakers: Dict[str, CircuitBreaker] = {}

    def register(self, name: str, model: BaseChatModel, breaker: CircuitBreaker | None = None) -> None:
        self._models[name] = model
        self._breakers[name] = breaker or CircuitBreaker()

    def get_model(self, name: str) -> BaseChatModel:
        return self._models[name]

    def get_breaker(self, name: str) -> CircuitBreaker:
        return self._breakers[name]

    def invoke_with_fallback(self, messages: list, preferred: list[str]) -> Any:
        """Try providers in order until one succeeds."""
        last_error = None
        for name in preferred:
            breaker = self._breakers[name]
            model = self._models[name]
            try:
                return breaker.call(lambda: model.invoke(messages))
            except CircuitOpenError:
                continue  # try next provider
            except Exception as exc:
                last_error = exc
                continue
        raise RuntimeError(f"All providers failed. Last error: {last_error}")

# Initialize once at app startup
registry = ProviderRegistry()
registry.register("openai-gpt4", ChatOpenAI(model="gpt-4o", temperature=0))
registry.register("anthropic-claude", ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0))
registry.register("google-gemini", ChatGoogleGenerativeAI(model="gemini-1.5-pro", temperature=0))

The invoke_with_fallback method encodes your routing policy: try OpenAI first, then Anthropic, then Google. Adjust the order based on your cost, latency, or quality preferences.

Step 4: Expose a LangChain-compatible runnable

Your chains shouldn’t know about circuit breakers. Wrap the registry in a Runnable that implements the standard interface.

# resilient_chat.py
from typing import Any, Dict, List, Optional
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_core.messages import BaseMessage
from langchain_core.outputs import ChatResult

from provider_registry import registry

class ResilientChatModel(Runnable[List[BaseMessage], ChatResult]):
    """A chat model that automatically fails over across providers."""

    def __init__(self, provider_order: Optional[List[str]] = None):
        self.provider_order = provider_order or ["openai-gpt4", "anthropic-claude", "google-gemini"]

    def invoke(self, input: List[BaseMessage], config: Optional[RunnableConfig] = None) -> ChatResult:
        # Allow per-request override via config
        order = config.get("configurable", {}).get("provider_order", self.provider_order) if config else self.provider_order
        return registry.invoke_with_fallback(input, order)

    async def ainvoke(self, input: List[BaseMessage], config: Optional[RunnableConfig] = None) -> ChatResult:
        # Async version delegates to sync for simplicity; replace with true async if needed
        return self.invoke(input, config)

    def batch(self, inputs: List[List[BaseMessage]], config: Optional[RunnableConfig] = None) -> List[ChatResult]:
        return [self.invoke(inp, config) for inp in inputs]

Now you can drop ResilientChatModel() anywhere you’d use a regular ChatOpenAI or ChatAnthropic:

# main_chain.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

from resilient_chat import ResilientChatModel

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical assistant."),
    ("human", "{question}"),
])

model = ResilientChatModel()
chain = prompt | model | StrOutputParser()

# Usage
response = chain.invoke({"question": "Explain circuit breakers in three sentences."})
print(response)

Step 5: Add observability so you know when circuits trip

A circuit breaker that trips silently is a debugging nightmare. Emit structured logs and metrics at each state transition.

# observability.py
import logging
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Generator

import structlog

logger = structlog.get_logger(__name__)

@dataclass
class CircuitEvent:
    provider: str
    from_state: str
    to_state: str
    failure_count: int
    timestamp: float = field(default_factory=time.time)

def log_circuit_event(event: CircuitEvent) -> None:
    logger.warning(
        "circuit_breaker_state_change",
        provider=event.provider,
        from_state=event.from_state,
        to_state=event.to_state,
        failure_count=event.failure_count,
    )

@contextmanager
def track_latency(provider: str) -> Generator[None, None, None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        duration_ms = (time.perf_counter() - start) * 1000
        logger.info("provider_latency", provider=provider, duration_ms=duration_ms)

Wire this into CircuitBreaker._record_failure and _record_success:

# Inside CircuitBreaker._record_failure
if self._state == ProviderState.OPEN and previous_state != ProviderState.OPEN:
    log_circuit_event(CircuitEvent(
        provider=provider_name,  # pass this in or store on the breaker
        from_state=previous_state.value,
        to_state=ProviderState.OPEN.value,
        failure_count=self._failure_count,
    ))

If you’re already using n4n.ai as your inference gateway, its per-token usage metering and automatic fallback when a provider is rate-limited or degraded complement this application-level circuit breaker — the gateway handles infrastructure failover while your code handles semantic routing.

Step 6: Write integration tests that verify the behavior

Unit-test the state machine in isolation, then write an integration test that exercises the full fallback chain against a mock server.

# test_circuit_breaker.py
import pytest
import httpx
from unittest.mock import Mock, patch

from circuit_breaker import CircuitBreaker, CircuitOpenError, ProviderState
from circuit_classifier import classify_exception

def test_circuit_opens_after_threshold():
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
    mock_func = Mock(side_effect=httpx.HTTPStatusError("500", request=Mock(), response=Mock(status_code=500)))

    for _ in range(3):
        with pytest.raises(httpx.HTTPStatusError):
            breaker.call(mock_func)

    assert breaker._state == ProviderState.OPEN
    with pytest.raises(CircuitOpenError):
        breaker.call(lambda: "success")

def test_half_open_allows_recovery():
    breaker = CircuitBreaker(failure_threshold=2, recovery_timeout=0.01, half_open_max_calls=1)
    mock_func = Mock(side_effect=httpx.HTTPStatusError("500", request=Mock(), response=Mock(status_code=500)))

    # Trip the circuit
    for _ in range(2):
        with pytest.raises(httpx.HTTPStatusError):
            breaker.call(mock_func)

    # Wait for recovery timeout
    import time
    time.sleep(0.02)

    # Half-open call succeeds
    mock_func.side_effect = None
    mock_func.return_value = "ok"
    result = breaker.call(mock_func)
    assert result == "ok"
    assert breaker._state == ProviderState.HEALTHY

For integration testing, spin up a local mock server that returns configurable responses:

# test_integration.py
import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
import uvicorn
import threading
import time

from provider_registry import ProviderRegistry
from resilient_chat import ResilientChatModel
from langchain_core.messages import HumanMessage

app = FastAPI()

@app.post("/v1/chat/completions")
async def mock_chat(request: Request):
    body = await request.json()
    provider = request.headers.get("x-test-provider", "unknown")
    if provider == "fail":
        return JSONResponse(status_code=500, content={"error": "internal error"})
    return {"choices": [{"message": {"content": f"response from {provider}"}}]}

def start_mock_server():
    config = uvicorn.Config(app, host="127.0.0.1", port=18080, log_level="error")
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    time.sleep(0.5)  # wait for server to start
    return "http://127.0.0.1:18080"

@pytest.fixture(scope="session")
def mock_base_url():
    return start_mock_server()

def test_fallback_on_provider_failure(mock_base_url):
    from langchain_openai import ChatOpenAI

    registry = ProviderRegistry()
    registry.register("primary", ChatOpenAI(
        base_url=f"{mock_base_url}/v1",
        api_key="test",
        model="test",
        default_headers={"x-test-provider": "fail"}
    ))
    registry.register("fallback", ChatOpenAI(
        base_url=f"{mock_base_url}/v1",
        api_key="test",
        model="test",
        default_headers={"x-test-provider": "ok"}
    ))

    model = ResilientChatModel(provider_order=["primary", "fallback"])
    # Monkey-patch the registry used by ResilientChatModel
    import resilient_chat
    resilient_chat.registry = registry

    result = model.invoke([HumanMessage(content="hello")])
    assert "response from ok" in result.generations[0].message.content

Run with pytest -v test_circuit_breaker.py test_integration.py. All tests should pass.

Step 7: Configure timeouts and retries at the HTTP layer

Circuit breaking works best when the underlying HTTP client fails fast. Configure your LangChain model clients with aggressive timeouts so the breaker sees failures quickly.

# http_config.py
import httpx
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

def create_http_client() -> httpx.Client:
    return httpx.Client(
        timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    )

# Usage
http_client = create_http_client()

openai_model = ChatOpenAI(
    model="gpt-4o",
    http_client=http_client,
    max_retries=0,  # let the circuit breaker handle retries via fallback
)

anthropic_model = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    http_client=http_client,
    max_retries=0,
)

Setting max_retries=0 on the model prevents double-retry behavior — the circuit breaker and fallback logic own the retry policy.

Step 8: Verify success in production

Deploy with feature flags so you can enable the resilient model for a subset of traffic first.

# feature_flags.py
import os

def use_resilient_model() -> bool:
    return os.getenv("USE_RESILIENT_MODEL", "false").lower() == "true"

# In your application entry point
from feature_flags import use_resilient_model
from resilient_chat import ResilientChatModel
from langchain_openai import ChatOpenAI

if use_resilient_model():
    chat_model = ResilientChatModel()
else:
    chat_model = ChatOpenAI(model="gpt-4o")

Monitor these key metrics after rollout:

Metric	Target	Alert threshold
Circuit open rate (per provider)	< 0.1% of requests	> 1% for 5 minutes
Fallback latency p99	< 2x primary latency	> 3x primary latency
Fallback success rate	> 99.9%	< 99.5%
Provider error rate (5xx + 429)	< 0.5%	> 2%

Dashboards in Datadog, Grafana, or your observability stack should show circuit state transitions over time. A healthy system shows brief, rare trips followed by quick recovery. Persistent open circuits indicate a provider issue that needs escalation, not just fallback.

Common pitfalls to avoid

Don’t share a single circuit breaker across providers. Each provider fails independently. A shared breaker would trip all providers when one degrades.

Don’t treat 4xx errors as circuit-breakable. A 400 means your prompt is malformed. A 401 means your API key is invalid. Fix the bug instead of failing over.

Don’t forget to reset failure counts on success. The _record_success method must clear the counter when the provider recovers, otherwise a single success after many failures leaves the breaker primed to trip again immediately.

Don’t hardcode provider order in the chain. Pass it via RunnableConfig so you can A/B test routing policies or implement per-tenant preferences without code changes:

chain.invoke(
    {"question": "..."},
    config={"configurable": {"provider_order": ["anthropic-claude", "openai-gpt4"]}}
)

What this gives you

With ~150 lines of framework code, you now have:

Automatic failover when any provider degrades
Independent circuit state per provider
Observable state transitions for debugging
Zero changes to your existing chain logic
Configurable routing policy per request

The pattern scales to any number of providers and works with any LangChain-compatible model. When you add a new provider, register it in ProviderRegistry and add it to your preferred order — no other code changes required.