aishastr.live

Agentic Architecture Pattern: Circuit Breaker

Written by

admin

in


    """
Pattern 182: Circuit Breaker Pattern

Prevents cascading failures by monitoring service health and "opening" the circuit
when failure rate exceeds threshold, allowing the service time to recover.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import time
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class CircuitState(Enum):
    """Circuit breaker states."""
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Blocking requests
    HALF_OPEN = "half_open"  # Testing if service recovered


@dataclass
class CircuitCall:
    """Track circuit calls."""
    timestamp: datetime
    success: bool
    error: Optional[str] = None
    response_time: float = 0.0


class CircuitBreaker:
    """Circuit breaker for a service."""
    
    def __init__(
        self,
        name: str,
        failure_threshold: int = 5,
        success_threshold: int = 2,
        timeout: int = 60,
        window_size: int = 10
    ):
        self.name = name
        self.failure_threshold = failure_threshold
        self.success_threshold = success_threshold
        self.timeout = timeout  # seconds to wait before trying again
        self.window_size = window_size  # number of recent calls to consider
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[datetime] = None
        self.call_history: List[CircuitCall] = []
    
    def can_execute(self) -> bool:
        """Check if execution is allowed."""
        if self.state == CircuitState.CLOSED:
            return True
        
        elif self.state == CircuitState.OPEN:
            # Check if timeout expired
            if self.last_failure_time:
                elapsed = (datetime.now() - self.last_failure_time).total_seconds()
                if elapsed >= self.timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                    print(f"Circuit {self.name}: OPEN -> HALF_OPEN (timeout expired)")
                    return True
            return False
        
        else:  # HALF_OPEN
            return True
    
    def record_success(self, response_time: float = 0.0):
        """Record successful call."""
        self.call_history.append(CircuitCall(
            timestamp=datetime.now(),
            success=True,
            response_time=response_time
        ))
        
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print(f"Circuit {self.name}: HALF_OPEN -> CLOSED (service recovered)")
        
        elif self.state == CircuitState.CLOSED:
            # Reset failure count on success
            self.failure_count = 0
    
    def record_failure(self, error: str):
        """Record failed call."""
        self.call_history.append(CircuitCall(
            timestamp=datetime.now(),
            success=False,
            error=error
        ))
        
        self.last_failure_time = datetime.now()
        
        if self.state == CircuitState.HALF_OPEN:
            # Failed during testing - go back to OPEN
            self.state = CircuitState.OPEN
            self.failure_count = 0
            print(f"Circuit {self.name}: HALF_OPEN -> OPEN (test failed)")
        
        elif self.state == CircuitState.CLOSED:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(f"Circuit {self.name}: CLOSED -> OPEN (threshold reached)")
    
    def get_status(self) -> Dict[str, Any]:
        """Get circuit status."""
        recent_calls = self.call_history[-self.window_size:] if self.call_history else []
        success_count = sum(1 for c in recent_calls if c.success)
        
        return {
            "name": self.name,
            "state": self.state.value,
            "failure_count": self.failure_count,
            "recent_success_rate": f"{(success_count/len(recent_calls)*100):.1f}%" if recent_calls else "N/A",
            "total_calls": len(self.call_history)
        }


class CircuitBreakerPattern:
    """
    Circuit Breaker Pattern - Manage circuit breakers for services.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.circuits: Dict[str, CircuitBreaker] = {}
    
    def register_circuit(self, circuit: CircuitBreaker):
        """Register a circuit breaker."""
        self.circuits[circuit.name] = circuit
        print(f"Circuit breaker registered: {circuit.name}")
    
    def execute_with_circuit_breaker(
        self,
        circuit_name: str,
        operation: Callable,
        *args,
        **kwargs
    ) -> Dict[str, Any]:
        """Execute operation with circuit breaker protection."""
        if circuit_name not in self.circuits:
            return {"success": False, "error": f"Circuit {circuit_name} not found"}
        
        circuit = self.circuits[circuit_name]
        
        if not circuit.can_execute():
            return {
                "success": False,
                "error": "Circuit is OPEN - service unavailable",
                "circuit_state": circuit.state.value
            }
        
        try:
            start = time.time()
            result = operation(*args, **kwargs)
            response_time = time.time() - start
            
            circuit.record_success(response_time)
            
            return {
                "success": True,
                "result": result,
                "circuit_state": circuit.state.value,
                "response_time": f"{response_time:.3f}s"
            }
        
        except Exception as e:
            circuit.record_failure(str(e))
            
            return {
                "success": False,
                "error": str(e),
                "circuit_state": circuit.state.value
            }
    
    def run(self, circuit_name: str, task: str, fail_probability: float = 0.0) -> Dict[str, Any]:
        """Execute task through circuit breaker."""
        
        def simulated_service():
            """Simulate a service that might fail."""
            import random
            if random.random()  List[Dict[str, Any]]:
        """Get status of all circuits."""
        return [circuit.get_status() for circuit in self.circuits.values()]


# Example usage
if __name__ == "__main__":
    print("=== Circuit Breaker Pattern Demo ===\n")
    
    # Create pattern
    pattern = CircuitBreakerPattern()
    
    # Register circuit breaker
    payment_circuit = CircuitBreaker(
        name="payment_service",
        failure_threshold=3,
        success_threshold=2,
        timeout=5,
        window_size=10
    )
    pattern.register_circuit(payment_circuit)
    
    print("\n--- Test 1: Normal Operation (low failure rate) ---")
    for i in range(5):
        result = pattern.run("payment_service", f"Process payment #{i+1}", fail_probability=0.1)
        print(f"Attempt {i+1}: Success={result['result']['success']}, State={result['circuit_status']['state']}")
    
    print("\n--- Test 2: High Failure Rate (should open circuit) ---")
    for i in range(6):
        result = pattern.run("payment_service", f"Process payment #{i+6}", fail_probability=0.9)
        print(f"Attempt {i+1}: Success={result['result']['success']}, State={result['circuit_status']['state']}")
        if result['circuit_status']['state'] == 'open':
            print("   ⚠ Circuit OPENED - blocking further requests")
            break
    
    print("\n--- Test 3: Circuit Open (requests blocked) ---")
    result = pattern.run("payment_service", "Process payment #blocked", fail_probability=0.1)
    print(f"Blocked request: {result['result']['error']}")
    
    print("\n--- Test 4: Wait for timeout (5 seconds) ---")
    print("Waiting for circuit to enter HALF_OPEN state...")
    time.sleep(5)
    
    print("\n--- Test 5: Recovery (testing in HALF_OPEN) ---")
    for i in range(3):
        result = pattern.run("payment_service", f"Process payment #{i+20}", fail_probability=0.0)
        print(f"Recovery attempt {i+1}: Success={result['result']['success']}, State={result['circuit_status']['state']}")
        if result['circuit_status']['state'] == 'closed':
            print("   ✓ Circuit CLOSED - service recovered")
            break
    
    # Show final status
    print("\n=== Final Circuit Status ===")
    for status in pattern.get_all_circuits_status():
        print(f"{status['name']}: {status['state'].upper()} - {status['total_calls']} total calls, {status['recent_success_rate']} recent success rate")

Agentic Architecture Pattern: Circuit Breaker

Pattern 182: 182 Circuit Breaker – Line-by-Line Explanation

Overview

Pattern 182: Circuit Breaker Pattern

Prevents cascading failures by monitoring service health and “opening” the circuit when failure rate exceeds threshold, allowing the service time to recover.

Code Structure Analysis

File Information

  • Pattern Number: 182
  • Pattern Name: 182 Circuit Breaker
  • File Path: patterns\error_handling\182_circuit_breaker.py
  • Total Lines: 271

Detailed Code Explanation

Module Docstring

"""
Pattern 182: Circuit Breaker Pattern

Prevents cascading failures by monitoring service health and "opening" the circuit
when failure rate exceeds threshold, allowing the service time to recover.
"""

Explanation: This module docstring describes the purpose and key features of the 182 Circuit Breaker pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import time
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
            import random  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

CircuitState (Line 17)

class CircuitState(Enum):
    CLOSED
    OPEN
    HALF_OPEN

Purpose: Circuit breaker states.

Values: – CLOSED: Represents closed – OPEN: Represents open – HALF_OPEN: Represents half open

Classes

CircuitState (Line 17)

Description: Circuit breaker states.

CircuitCall (Line 25)

Description: Track circuit calls.

CircuitBreaker (Line 33)

Description: Circuit breaker for a service.

Methods:

__init__ (Line 36)
def __init__(self, name, failure_threshold, success_threshold, timeout, window_size):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – failure_threshold: Parameter for failure_threshold – success_threshold: Parameter for success_threshold – timeout: Parameter for timeout – window_size: Parameter for window_size

can_execute (Line 56)
def can_execute(self):
    """Check if execution is allowed."""

Purpose: Check if execution is allowed.

Parameters: – self: Instance reference

record_success (Line 75)
def record_success(self, response_time):
    """Record successful call."""

Purpose: Record successful call.

Parameters: – self: Instance reference – response_time: Parameter for response_time

record_failure (Line 94)
def record_failure(self, error):
    """Record failed call."""

Purpose: Record failed call.

Parameters: – self: Instance reference – error: Parameter for error

get_status (Line 116)
def get_status(self):
    """Get circuit status."""

Purpose: Get circuit status.

Parameters: – self: Instance reference

CircuitBreakerPattern (Line 130)

Description: Circuit Breaker Pattern – Manage circuit breakers for services.

Methods:

__init__ (Line 135)
def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

register_circuit (Line 139)
def register_circuit(self, circuit):
    """Register a circuit breaker."""

Purpose: Register a circuit breaker.

Parameters: – self: Instance reference – circuit: Parameter for circuit

execute_with_circuit_breaker (Line 144)
def execute_with_circuit_breaker(self, circuit_name, operation):
    """Execute operation with circuit breaker protection."""

Purpose: Execute operation with circuit breaker protection.

Parameters: – self: Instance reference – circuit_name: Parameter for circuit_name – operation: Parameter for operation

run (Line 187)
def run(self, circuit_name, task, fail_probability):
    """Execute task through circuit breaker."""

Purpose: Execute task through circuit breaker.

Parameters: – self: Instance reference – circuit_name: Parameter for circuit_name – task: Parameter for task – fail_probability: Parameter for fail_probability

get_all_circuits_status (Line 216)
def get_all_circuits_status(self):
    """Get status of all circuits."""

Purpose: Get status of all circuits.

Parameters: – self: Instance reference

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = CircuitState(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts