"""
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
Leave a Reply