Circuit Breaker Pattern

What is a Circuit Breaker? A pattern that prevents cascading failures by stopping requests to a failing service, allowing it to recover.

Circuit States

CLOSED (normal) --[failures > threshold]--> OPEN (blocking)
                                              |
                            [timeout] --------+
                                              v
                                      HALF-OPEN (testing)
                                              |
                    [success] ----------------+--> CLOSED
                    [failure] ----------------+--> OPEN

Implementation

import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = "CLOSED"
        self.last_failure_time = None

    def call(self, func):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitOpenError()

        try:
            result = func()
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = "CLOSED"

    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "OPEN"

PyBreaker Library

import pybreaker

breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=30
)

@breaker
def call_llm_api(prompt):
    return openai.chat.completions.create(...)

Fallback Strategies

def call_with_fallback(prompt):
    try:
        return call_primary_llm(prompt)
    except CircuitOpenError:
        return call_fallback_llm(prompt)
    except Exception:
        return cached_response(prompt)

Graceful Degradation

def degrade_gracefully(prompt):
    try:
        return primary_model(prompt)  # GPT-4
    except ServiceUnavailable:
        try:
            return fallback_model(prompt)  # GPT-3.5
        except ServiceUnavailable:
            cached = get_similar_cached(prompt)
            if cached:
                return cached
            return "Service temporarily unavailable"

Best Practices

• Set appropriate thresholds based on error rates
• Use half-open state to test recovery
• Implement fallback chains
• Monitor circuit state
• Log state transitions for debugging