"""
Pattern 185: Error Recovery Pattern
Purpose: Automatic error recovery strategies
"""

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


class RecoveryStrategy(Enum):
    """Error recovery strategies."""
    RETRY = "retry"
    FALLBACK = "fallback"
    COMPENSATE = "compensate"
    IGNORE = "ignore"
    ESCALATE = "escalate"


@dataclass
class RecoveryAction:
    """Recovery action definition."""
    strategy: RecoveryStrategy
    max_attempts: int = 3
    backoff_seconds: float = 1.0
    fallback_func: Optional[Callable] = None


@dataclass
class ErrorRecord:
    """Record of error and recovery."""
    error_type: str
    error_message: str
    recovery_strategy: RecoveryStrategy
    recovery_successful: bool
    attempts: int
    timestamp: datetime = field(default_factory=datetime.now)


class ErrorRecoveryPattern:
    """
    Error Recovery - Automatic error recovery with multiple strategies.
    
    Features:
    - Multiple recovery strategies
    - Exponential backoff
    - Fallback functions
    - Compensation actions
    - Error history
    - Recovery metrics
    """
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.recovery_actions: Dict[type, RecoveryAction] = {}
        self.error_history: List[ErrorRecord] = []
        self.stats = {
            "total_errors": 0,
            "recovered": 0,
            "failed": 0,
            "by_strategy": {s.value: 0 for s in RecoveryStrategy}
        }
    
    def register_recovery(self, error_type: type, action: RecoveryAction):
        """Register recovery action for error type."""
        self.recovery_actions[error_type] = action
    
    def execute_with_recovery(
        self,
        func: Callable,
        *args,
        **kwargs
    ) -> Any:
        """Execute function with automatic error recovery."""
        last_error = None
        
        for attempt in range(1, 4):  # Default 3 attempts
            try:
                return func(*args, **kwargs)
                
            except Exception as e:
                self.stats["total_errors"] += 1
                last_error = e
                error_type = type(e)
                
                # Check if recovery action registered
                if error_type in self.recovery_actions:
                    action = self.recovery_actions[error_type]
                    
                    if action.strategy == RecoveryStrategy.RETRY and attempt  str:
        """Execute LLM query with error recovery."""
        def fallback_query(query):
            # Simplified fallback using gpt-3.5
            fallback_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful assistant."),
                ("human", "{query}")
            ])
            chain = prompt | fallback_llm
            response = chain.invoke({"query": query})
            return str(response.content)
        
        # Register recovery for common errors
        self.register_recovery(
            Exception,
            RecoveryAction(
                strategy=RecoveryStrategy.FALLBACK,
                fallback_func=fallback_query
            )
        )
        
        def _query(query):
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("human", "{query}")
            ])
            chain = prompt | self.llm
            response = chain.invoke({"query": query})
            return str(response.content)
        
        return self.execute_with_recovery(_query, query)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get recovery statistics."""
        recovery_rate = (
            self.stats["recovered"] / self.stats["total_errors"] * 100
            if self.stats["total_errors"] > 0 else 0
        )
        
        return {
            **self.stats,
            "recovery_rate": f"{recovery_rate:.1f}%",
            "recent_errors": len(self.error_history[-10:])
        }


if __name__ == "__main__":
    print("=== Error Recovery Pattern Demo ===\n")
    
    pattern = ErrorRecoveryPattern()
    
    # Test recovery strategies
    print("Testing error recovery:\n")
    
    # Test 1: Successful query
    print("1. Successful query:")
    result = pattern.llm_query_with_recovery("What is the capital of France?")
    print(f"   Result: {result}\n")
    
    # Test 2: Query that might fail
    print("2. Complex query:")
    result = pattern.llm_query_with_recovery("Explain machine learning")
    print(f"   Result: {result[:100]}...\n")
    
    # Show statistics
    print("=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total errors: {stats['total_errors']}")
    print(f"Recovered: {stats['recovered']}")
    print(f"Failed: {stats['failed']}")
    print(f"Recovery rate: {stats['recovery_rate']}")
    
    print("\nBy strategy:")
    for strategy, count in stats['by_strategy'].items():
        if count > 0:
            print(f"  {strategy}: {count}")

    

Agentic Architecture Pattern: Error Recovery

Pattern 185: 185 Error Recovery – Line-by-Line Explanation

Overview

Pattern 185: Error Recovery Pattern Purpose: Automatic error recovery strategies

Code Structure Analysis

File Information

• Pattern Number: 185
• Pattern Name: 185 Error Recovery
• File Path: patterns\error_handling\185_error_recovery.py
• Total Lines: 223

Detailed Code Explanation

Module Docstring

"""
Pattern 185: Error Recovery Pattern
Purpose: Automatic error recovery strategies
"""

Explanation: This module docstring describes the purpose and key features of the 185 Error Recovery pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
                        import time  # 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: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Enumerations

RecoveryStrategy (Line 14)

class RecoveryStrategy(Enum):
    RETRY
    FALLBACK
    COMPENSATE
    IGNORE
    ESCALATE

Purpose: Error recovery strategies.

Values: – RETRY: Represents retry – FALLBACK: Represents fallback – COMPENSATE: Represents compensate – IGNORE: Represents ignore – ESCALATE: Represents escalate

Classes

RecoveryStrategy (Line 14)

Description: Error recovery strategies.

RecoveryAction (Line 24)

Description: Recovery action definition.

ErrorRecord (Line 33)

Description: Record of error and recovery.

ErrorRecoveryPattern (Line 43)

Description: Error Recovery – Automatic error recovery with multiple strategies.

Features: – Multiple recovery strategies – Exponential backoff – Fallback functions – Compensation actions – Error history – Recovery metrics

Methods:

__init__ (Line 56)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_recovery (Line 67)

def register_recovery(self, error_type, action):
    """Register recovery action for error type."""

Purpose: Register recovery action for error type.

Parameters: – self: Instance reference – error_type: Parameter for error_type – action: Parameter for action

execute_with_recovery (Line 71)

def execute_with_recovery(self, func):
    """Execute function with automatic error recovery."""

Purpose: Execute function with automatic error recovery.

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

llm_query_with_recovery (Line 146)

def llm_query_with_recovery(self, query):
    """Execute LLM query with error recovery."""

Purpose: Execute LLM query with error recovery.

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

get_stats (Line 179)

def get_stats(self):
    """Get recovery statistics."""

Purpose: Get recovery statistics.

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 = RecoveryStrategy(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