Agentic Architecture Pattern: Fallback

Written by

in


    
"""
Pattern 183: Fallback Pattern
Purpose: Provide graceful degradation with fallback 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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


class FallbackStrategy(Enum):
    """Fallback strategies."""
    CACHED = "cached"  # Use cached response
    SIMPLIFIED = "simplified"  # Use simpler model
    ALTERNATIVE = "alternative"  # Use alternative method
    DEFAULT = "default"  # Return default value


@dataclass
class FallbackConfig:
    """Fallback configuration."""
    strategy: FallbackStrategy
    max_retries: int = 3
    timeout_seconds: float = 30.0
    cache_ttl: int = 300  # seconds
    fallback_model: str = "gpt-3.5-turbo"


@dataclass
class ExecutionAttempt:
    """Record of execution attempt."""
    attempt_number: int
    strategy: FallbackStrategy
    success: bool
    result: Optional[Any] = None
    error: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)
    latency_ms: float = 0.0


class FallbackState(TypedDict):
    """State for fallback workflow."""
    task: str
    attempts: List[ExecutionAttempt]
    result: Optional[Any]
    final_strategy: Optional[FallbackStrategy]


class FallbackPattern:
    """
    Fallback Pattern - Graceful degradation with multiple fallback strategies.
    
    Features:
    - Multiple fallback strategies
    - Automatic retry with backoff
    - Response caching
    - Model degradation (GPT-4 -> GPT-3.5)
    - Default value fallback
    - Comprehensive metrics
    """
    
    def __init__(self, config: Optional[FallbackConfig] = None):
        self.config = config or FallbackConfig(strategy=FallbackStrategy.SIMPLIFIED)
        self.primary_llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.fallback_llm = ChatOpenAI(model=self.config.fallback_model, temperature=0.7)
        self.cache: Dict[str, Any] = {}
        self.stats = {
            "total_requests": 0,
            "primary_success": 0,
            "fallback_used": 0,
            "total_failures": 0,
            "by_strategy": {s.value: 0 for s in FallbackStrategy}
        }
    
    def execute_primary(self, task: str) -> Any:
        """Execute primary strategy (may fail)."""
        try:
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide comprehensive answers."),
                ("human", "{task}")
            ])
            chain = prompt | self.primary_llm
            response = chain.invoke({"task": task})
            return str(response.content)
        except Exception as e:
            raise Exception(f"Primary execution failed: {e}")
    
    def execute_cached(self, task: str) -> Optional[Any]:
        """Try cached response."""
        return self.cache.get(task)
    
    def execute_simplified(self, task: str) -> Any:
        """Execute with simpler/faster model."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful assistant. Provide concise answers."),
            ("human", "{task}")
        ])
        chain = prompt | self.fallback_llm
        response = chain.invoke({"task": task})
        return str(response.content)
    
    def execute_alternative(self, task: str) -> Any:
        """Execute with alternative method."""
        # Alternative implementation (e.g., rule-based)
        return f"Alternative solution for: {task}. Using simplified logic."
    
    def execute_default(self, task: str) -> Any:
        """Return default response."""
        return f"Unable to fully process request: {task}. Please try again later."
    
    def run_with_fallback(self, task: str) -> Dict[str, Any]:
        """
        Execute task with fallback strategies.
        
        Fallback chain:
        1. Try primary (GPT-4)
        2. Check cache
        3. Try simplified (GPT-3.5)
        4. Try alternative method
        5. Return default
        """
        self.stats["total_requests"] += 1
        attempts = []
        start_time = datetime.now()
        
        # Try primary
        try:
            result = self.execute_primary(task)
            self.stats["primary_success"] += 1
            self.cache[task] = result  # Cache successful result
            
            attempts.append(ExecutionAttempt(
                attempt_number=1,
                strategy=FallbackStrategy.DEFAULT,
                success=True,
                result=result,
                latency_ms=(datetime.now() - start_time).total_seconds() * 1000
            ))
            
            return {
                "success": True,
                "result": result,
                "strategy": "primary",
                "attempts": len(attempts),
                "total_latency_ms": (datetime.now() - start_time).total_seconds() * 1000
            }
        except Exception as e:
            attempts.append(ExecutionAttempt(
                attempt_number=1,
                strategy=FallbackStrategy.DEFAULT,
                success=False,
                error=str(e),
                latency_ms=(datetime.now() - start_time).total_seconds() * 1000
            ))
        
        # Try cache
        cached = self.execute_cached(task)
        if cached:
            self.stats["fallback_used"] += 1
            self.stats["by_strategy"][FallbackStrategy.CACHED.value] += 1
            
            return {
                "success": True,
                "result": cached,
                "strategy": "cached",
                "attempts": len(attempts) + 1,
                "total_latency_ms": (datetime.now() - start_time).total_seconds() * 1000
            }
        
        # Try simplified
        try:
            result = self.execute_simplified(task)
            self.stats["fallback_used"] += 1
            self.stats["by_strategy"][FallbackStrategy.SIMPLIFIED.value] += 1
            
            attempts.append(ExecutionAttempt(
                attempt_number=len(attempts) + 1,
                strategy=FallbackStrategy.SIMPLIFIED,
                success=True,
                result=result
            ))
            
            return {
                "success": True,
                "result": result,
                "strategy": "simplified",
                "attempts": len(attempts),
                "total_latency_ms": (datetime.now() - start_time).total_seconds() * 1000
            }
        except Exception as e:
            attempts.append(ExecutionAttempt(
                attempt_number=len(attempts) + 1,
                strategy=FallbackStrategy.SIMPLIFIED,
                success=False,
                error=str(e)
            ))
        
        # Try alternative
        try:
            result = self.execute_alternative(task)
            self.stats["fallback_used"] += 1
            self.stats["by_strategy"][FallbackStrategy.ALTERNATIVE.value] += 1
            
            return {
                "success": True,
                "result": result,
                "strategy": "alternative",
                "attempts": len(attempts) + 1,
                "total_latency_ms": (datetime.now() - start_time).total_seconds() * 1000
            }
        except:
            pass
        
        # Default fallback
        result = self.execute_default(task)
        self.stats["total_failures"] += 1
        self.stats["by_strategy"][FallbackStrategy.DEFAULT.value] += 1
        
        return {
            "success": False,
            "result": result,
            "strategy": "default",
            "attempts": len(attempts) + 1,
            "total_latency_ms": (datetime.now() - start_time).total_seconds() * 1000
        }
    
    def get_stats(self) -> Dict[str, Any]:
        """Get execution statistics."""
        success_rate = (
            self.stats["primary_success"] / self.stats["total_requests"] * 100
            if self.stats["total_requests"] > 0 else 0
        )
        
        return {
            **self.stats,
            "primary_success_rate": f"{success_rate:.1f}%",
            "fallback_rate": f"{self.stats['fallback_used'] / self.stats['total_requests'] * 100:.1f}%"
            if self.stats["total_requests"] > 0 else "0%"
        }


if __name__ == "__main__":
    print("=== Fallback Pattern Demo ===\n")
    
    pattern = FallbackPattern()
    
    # Test cases
    test_tasks = [
        "Explain quantum computing in simple terms",
        "What is the capital of France?",
        "Describe machine learning",
    ]
    
    print("Testing fallback strategies:\n")
    for i, task in enumerate(test_tasks, 1):
        print(f"{i}. Task: {task}")
        result = pattern.run_with_fallback(task)
        print(f"   Strategy: {result['strategy']}")
        print(f"   Success: {result['success']}")
        print(f"   Attempts: {result['attempts']}")
        print(f"   Latency: {result['total_latency_ms']:.0f}ms")
        print(f"   Result: {result['result'][:100]}...")
        print()
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total requests: {stats['total_requests']}")
    print(f"Primary success: {stats['primary_success']}")
    print(f"Primary success rate: {stats['primary_success_rate']}")
    print(f"Fallback used: {stats['fallback_used']}")
    print(f"Fallback rate: {stats['fallback_rate']}")
    print(f"Total failures: {stats['total_failures']}")
    
    print("\nBy strategy:")
    for strategy, count in stats['by_strategy'].items():
        if count > 0:
            print(f"  {strategy}: {count}")

    

Agentic Architecture Pattern: Fallback

Pattern 183: 183 Fallback – Line-by-Line Explanation

Overview

Pattern 183: Fallback Pattern Purpose: Provide graceful degradation with fallback strategies


Code Structure Analysis

File Information

  • Pattern Number: 183
  • Pattern Name: 183 Fallback
  • File Path: patterns\error_handling\183_fallback.py
  • Total Lines: 285

Detailed Code Explanation

Module Docstring

"""
Pattern 183: Fallback Pattern
Purpose: Provide graceful degradation with fallback strategies
"""

Explanation: This module docstring describes the purpose and key features of the 183 Fallback 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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


Enumerations

FallbackStrategy (Line 17)

class FallbackStrategy(Enum):
    CACHED
    SIMPLIFIED
    ALTERNATIVE
    DEFAULT

Purpose: Fallback strategies.

Values: – CACHED: Represents cached – SIMPLIFIED: Represents simplified – ALTERNATIVE: Represents alternative – DEFAULT: Represents default


Classes

FallbackStrategy (Line 17)

Description: Fallback strategies.

FallbackConfig (Line 26)

Description: Fallback configuration.

ExecutionAttempt (Line 36)

Description: Record of execution attempt.

FallbackState (Line 47)

Description: State for fallback workflow.

FallbackPattern (Line 55)

Description: Fallback Pattern – Graceful degradation with multiple fallback strategies.

Features: – Multiple fallback strategies – Automatic retry with backoff – Response caching – Model degradation (GPT-4 -> GPT-3.5) – Default value fallback – Comprehensive metrics

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

execute_primary (Line 81)
def execute_primary(self, task):
    """Execute primary strategy (may fail)."""

Purpose: Execute primary strategy (may fail).

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

execute_cached (Line 94)
def execute_cached(self, task):
    """Try cached response."""

Purpose: Try cached response.

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

execute_simplified (Line 98)
def execute_simplified(self, task):
    """Execute with simpler/faster model."""

Purpose: Execute with simpler/faster model.

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

execute_alternative (Line 108)
def execute_alternative(self, task):
    """Execute with alternative method."""

Purpose: Execute with alternative method.

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

execute_default (Line 113)
def execute_default(self, task):
    """Return default response."""

Purpose: Return default response.

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

run_with_fallback (Line 117)
def run_with_fallback(self, task):
    """Execute task with fallback strategies.

Fallback chain:
1. Try primary (GPT-4)
2. Check cache
3. Try simplified (GPT-3.5)
4. Try alternative method
5. Return default"""

Purpose: Execute task with fallback strategies.

Fallback chain: 1. Try primary (GPT-4) 2. Check cache 3. Try simplified (GPT-3.5) 4. Try alternative method 5. Return default

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

get_stats (Line 233)
def get_stats(self):
    """Get execution statistics."""

Purpose: Get execution 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 = FallbackStrategy(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 *