"""
Pattern 181: Retry Pattern

Automatically retry failed operations with configurable strategies (exponential backoff,
fixed delay, etc.). Essential for handling transient failures.
"""

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


class RetryStrategy(Enum):
    """Retry strategies."""
    FIXED_DELAY = "fixed_delay"
    EXPONENTIAL_BACKOFF = "exponential_backoff"
    LINEAR_BACKOFF = "linear_backoff"
    RANDOM_JITTER = "random_jitter"


@dataclass
class RetryAttempt:
    """Track retry attempts."""
    attempt_number: int
    timestamp: datetime
    error: Optional[str] = None
    success: bool = False
    delay: float = 0.0


@dataclass
class RetryConfig:
    """Retry configuration."""
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    multiplier: float = 2.0
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF


class RetryPattern:
    """
    Retry Pattern - Automatic retry with configurable strategies.
    """
    
    def __init__(self, config: Optional[RetryConfig] = None, model_name: str = "gpt-4"):
        self.config = config or RetryConfig()
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.retry_history: List[RetryAttempt] = []
    
    def calculate_delay(self, attempt: int) -> float:
        """Calculate delay based on retry strategy."""
        if self.config.strategy == RetryStrategy.FIXED_DELAY:
            return self.config.initial_delay
        
        elif self.config.strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
            delay = self.config.initial_delay * (self.config.multiplier ** (attempt - 1))
            return min(delay, self.config.max_delay)
        
        elif self.config.strategy == RetryStrategy.LINEAR_BACKOFF:
            delay = self.config.initial_delay * attempt
            return min(delay, self.config.max_delay)
        
        elif self.config.strategy == RetryStrategy.RANDOM_JITTER:
            base_delay = self.config.initial_delay * (self.config.multiplier ** (attempt - 1))
            jitter = random.uniform(0, base_delay * 0.1)
            return min(base_delay + jitter, self.config.max_delay)
        
        return self.config.initial_delay
    
    def execute_with_retry(self, operation: Callable, *args, **kwargs) -> Dict[str, Any]:
        """Execute operation with retry logic."""
        last_error = None
        
        for attempt in range(1, self.config.max_attempts + 1):
            try:
                # Execute operation
                result = operation(*args, **kwargs)
                
                # Record successful attempt
                self.retry_history.append(RetryAttempt(
                    attempt_number=attempt,
                    timestamp=datetime.now(),
                    success=True
                ))
                
                return {
                    "success": True,
                    "result": result,
                    "attempts": attempt,
                    "total_attempts": self.config.max_attempts
                }
            
            except Exception as e:
                last_error = str(e)
                
                # Record failed attempt
                self.retry_history.append(RetryAttempt(
                    attempt_number=attempt,
                    timestamp=datetime.now(),
                    error=last_error,
                    success=False
                ))
                
                # Don't sleep on last attempt
                if attempt  Dict[str, Any]:
        """Execute with retry - simulates operation with potential failures."""
        
        def simulated_operation():
            """Simulate an operation that might fail."""
            if random.random()  Dict[str, Any]:
        """Get retry statistics."""
        total_attempts = len(self.retry_history)
        successful = sum(1 for a in self.retry_history if a.success)
        failed = total_attempts - successful
        
        return {
            "total_attempts": total_attempts,
            "successful": successful,
            "failed": failed,
            "success_rate": f"{(successful/total_attempts*100):.1f}%" if total_attempts > 0 else "0%"
        }


# Example usage
if __name__ == "__main__":
    print("=== Retry Pattern Demo ===\n")
    
    # Test with exponential backoff
    print("1. Exponential Backoff Strategy")
    config1 = RetryConfig(
        max_attempts=4,
        initial_delay=0.5,
        multiplier=2.0,
        strategy=RetryStrategy.EXPONENTIAL_BACKOFF
    )
    pattern1 = RetryPattern(config1)
    
    result1 = pattern1.run("Process customer payment", simulate_failure_rate=0.6)
    print(f"   Operation: {result1['operation']}")
    print(f"   Success: {result1['result']['success']}")
    print(f"   Attempts: {result1['result']['attempts']}/{result1['result']['total_attempts']}")
    if result1['result']['success']:
        print(f"   Result: {result1['result']['result'][:100]}...")
    else:
        print(f"   Error: {result1['result']['error']}")
    
    # Test with fixed delay
    print("\n\n2. Fixed Delay Strategy")
    config2 = RetryConfig(
        max_attempts=3,
        initial_delay=1.0,
        strategy=RetryStrategy.FIXED_DELAY
    )
    pattern2 = RetryPattern(config2)
    
    result2 = pattern2.run("Fetch external API data", simulate_failure_rate=0.4)
    print(f"   Operation: {result2['operation']}")
    print(f"   Success: {result2['result']['success']}")
    print(f"   Attempts: {result2['result']['attempts']}/{result2['result']['total_attempts']}")
    
    # Show statistics
    print("\n\n=== Statistics ===")
    stats1 = pattern1.get_statistics()
    print(f"Pattern 1: {stats1['successful']}/{stats1['total_attempts']} successful ({stats1['success_rate']})")
    stats2 = pattern2.get_statistics()
    print(f"Pattern 2: {stats2['successful']}/{stats2['total_attempts']} successful ({stats2['success_rate']})")

    

Agentic Architecture Pattern: Retry

Pattern 181: 181 Retry – Line-by-Line Explanation

Overview

Pattern 181: Retry Pattern

Automatically retry failed operations with configurable strategies (exponential backoff, fixed delay, etc.). Essential for handling transient failures.

Code Structure Analysis

File Information

• Pattern Number: 181
• Pattern Name: 181 Retry
• File Path: patterns\error_handling\181_retry.py
• Total Lines: 211

Detailed Code Explanation

Module Docstring

"""
Pattern 181: Retry Pattern

Automatically retry failed operations with configurable strategies (exponential backoff,
fixed delay, etc.). Essential for handling transient failures.
"""

Explanation: This module docstring describes the purpose and key features of the 181 Retry pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import time
import random
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # 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_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

RetryStrategy (Line 18)

class RetryStrategy(Enum):
    FIXED_DELAY
    EXPONENTIAL_BACKOFF
    LINEAR_BACKOFF
    RANDOM_JITTER

Purpose: Retry strategies.

Values: – FIXED_DELAY: Represents fixed delay – EXPONENTIAL_BACKOFF: Represents exponential backoff – LINEAR_BACKOFF: Represents linear backoff – RANDOM_JITTER: Represents random jitter

Classes

RetryStrategy (Line 18)

Description: Retry strategies.

RetryAttempt (Line 27)

Description: Track retry attempts.

RetryConfig (Line 37)

Description: Retry configuration.

RetryPattern (Line 46)

Description: Retry Pattern – Automatic retry with configurable strategies.

Methods:

__init__ (Line 51)

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

Purpose: No method docstring.

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

calculate_delay (Line 56)

def calculate_delay(self, attempt):
    """Calculate delay based on retry strategy."""

Purpose: Calculate delay based on retry strategy.

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

execute_with_retry (Line 76)

def execute_with_retry(self, operation):
    """Execute operation with retry logic."""

Purpose: Execute operation with retry logic.

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

run (Line 124)

def run(self, operation_description, simulate_failure_rate):
    """Execute with retry - simulates operation with potential failures."""

Purpose: Execute with retry – simulates operation with potential failures.

Parameters: – self: Instance reference – operation_description: Parameter for operation_description – simulate_failure_rate: Parameter for simulate_failure_rate

get_statistics (Line 154)

def get_statistics(self):
    """Get retry statistics."""

Purpose: Get retry 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 = RetryStrategy(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