Agentic Architecture Pattern: Graceful Degradation

Written by

in


    """
Pattern 187: Graceful Degradation Pattern
Purpose: Maintain partial functionality when components fail
"""

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 ServiceLevel(Enum):
    """Service level tiers."""
    FULL = "full"  # All features available
    DEGRADED = "degraded"  # Reduced features
    MINIMAL = "minimal"  # Core features only
    UNAVAILABLE = "unavailable"  # Service down


@dataclass
class Feature:
    """Feature definition."""
    name: str
    required_level: ServiceLevel
    enabled: bool = True
    fallback: Optional[Callable] = None


@dataclass
class DegradationEvent:
    """Record of degradation event."""
    from_level: ServiceLevel
    to_level: ServiceLevel
    reason: str
    affected_features: List[str]
    timestamp: datetime = field(default_factory=datetime.now)


class GracefulDegradationPattern:
    """
    Graceful Degradation - Maintain partial functionality under failure.
    
    Features:
    - Multiple service levels
    - Feature-level control
    - Automatic degradation
    - Fallback mechanisms
    - Recovery detection
    - Performance monitoring
    """
    
    def __init__(self):
        self.primary_llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.fallback_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        self.current_level = ServiceLevel.FULL
        self.features: Dict[str, Feature] = {}
        self.degradation_history: List[DegradationEvent] = []
        self.stats = {
            "total_requests": 0,
            "full_service": 0,
            "degraded_service": 0,
            "minimal_service": 0,
            "failures": 0
        }
    
    def register_feature(self, feature: Feature):
        """Register a feature with required service level."""
        self.features[feature.name] = feature
    
    def degrade_service(self, reason: str, target_level: ServiceLevel):
        """Degrade service to target level."""
        if target_level.value == self.current_level.value:
            return
        
        affected_features = [
            name for name, feat in self.features.items()
            if feat.required_level.value  bool:
        """Check if feature is available at current service level."""
        if feature_name not in self.features:
            return False
        
        feature = self.features[feature_name]
        return feature.enabled and feature.required_level.value >= self.current_level.value
    
    def execute_query(self, query: str, feature_name: str = "full_query") -> str:
        """Execute query with graceful degradation."""
        self.stats["total_requests"] += 1
        
        try:
            # Try full service
            if self.current_level == ServiceLevel.FULL:
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are a helpful AI assistant providing comprehensive answers."),
                    ("human", "{query}")
                ])
                chain = prompt | self.primary_llm
                response = chain.invoke({"query": query})
                self.stats["full_service"] += 1
                return str(response.content)
            
        except Exception as e:
            self.degrade_service(f"Primary service failed: {e}", ServiceLevel.DEGRADED)
        
        try:
            # Try degraded service
            if self.current_level in [ServiceLevel.FULL, ServiceLevel.DEGRADED]:
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are a helpful assistant. Provide concise answers."),
                    ("human", "{query}")
                ])
                chain = prompt | self.fallback_llm
                response = chain.invoke({"query": query})
                self.stats["degraded_service"] += 1
                return str(response.content)
            
        except Exception as e:
            self.degrade_service(f"Fallback service failed: {e}", ServiceLevel.MINIMAL)
        
        try:
            # Minimal service - simple responses
            if self.current_level == ServiceLevel.MINIMAL:
                self.stats["minimal_service"] += 1
                return f"Service operating at minimal capacity. Query: {query}"
            
        except:
            self.degrade_service("All services failed", ServiceLevel.UNAVAILABLE)
        
        self.stats["failures"] += 1
        return "Service temporarily unavailable. Please try again later."
    
    def get_stats(self) -> Dict[str, Any]:
        """Get service statistics."""
        return {
            **self.stats,
            "current_level": self.current_level.value,
            "degradation_events": len(self.degradation_history),
            "available_features": sum(1 for f in self.features.values() if f.enabled)
        }


if __name__ == "__main__":
    print("=== Graceful Degradation Pattern Demo ===\n")
    
    pattern = GracefulDegradationPattern()
    
    # Register features
    pattern.register_feature(Feature("full_query", ServiceLevel.FULL))
    pattern.register_feature(Feature("detailed_analysis", ServiceLevel.FULL))
    pattern.register_feature(Feature("basic_query", ServiceLevel.DEGRADED))
    pattern.register_feature(Feature("health_check", ServiceLevel.MINIMAL))
    
    print("Testing graceful degradation:\n")
    
    # Test 1: Full service
    print("1. Full service query:")
    result = pattern.execute_query("What is machine learning?")
    print(f"   Level: {pattern.current_level.value}")
    print(f"   Result: {result[:100]}...\n")
    
    # Test 2: Another query
    print("2. Second query:")
    result = pattern.execute_query("Explain neural networks")
    print(f"   Level: {pattern.current_level.value}")
    print(f"   Result: {result[:100]}...\n")
    
    # Show statistics
    print("=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total requests: {stats['total_requests']}")
    print(f"Full service: {stats['full_service']}")
    print(f"Degraded service: {stats['degraded_service']}")
    print(f"Minimal service: {stats['minimal_service']}")
    print(f"Failures: {stats['failures']}")
    print(f"\nCurrent level: {stats['current_level']}")
    print(f"Available features: {stats['available_features']}")
    print(f"Degradation events: {stats['degradation_events']}")

    

Agentic Architecture Pattern: Graceful Degradation

Pattern 187: 187 Graceful Degradation – Line-by-Line Explanation

Overview

Pattern 187: Graceful Degradation Pattern Purpose: Maintain partial functionality when components fail


Code Structure Analysis

File Information

  • Pattern Number: 187
  • Pattern Name: 187 Graceful Degradation
  • File Path: patterns\error_handling\187_graceful_degradation.py
  • Total Lines: 211

Detailed Code Explanation

Module Docstring

"""
Pattern 187: Graceful Degradation Pattern
Purpose: Maintain partial functionality when components fail
"""

Explanation: This module docstring describes the purpose and key features of the 187 Graceful Degradation 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  # 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

ServiceLevel (Line 14)

class ServiceLevel(Enum):
    FULL
    DEGRADED
    MINIMAL
    UNAVAILABLE

Purpose: Service level tiers.

Values: – FULL: Represents full – DEGRADED: Represents degraded – MINIMAL: Represents minimal – UNAVAILABLE: Represents unavailable


Classes

ServiceLevel (Line 14)

Description: Service level tiers.

Feature (Line 23)

Description: Feature definition.

DegradationEvent (Line 32)

Description: Record of degradation event.

GracefulDegradationPattern (Line 41)

Description: Graceful Degradation – Maintain partial functionality under failure.

Features: – Multiple service levels – Feature-level control – Automatic degradation – Fallback mechanisms – Recovery detection – Performance monitoring

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_feature (Line 68)
def register_feature(self, feature):
    """Register a feature with required service level."""

Purpose: Register a feature with required service level.

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

degrade_service (Line 72)
def degrade_service(self, reason, target_level):
    """Degrade service to target level."""

Purpose: Degrade service to target level.

Parameters: – self: Instance reference – reason: Parameter for reason – target_level: Parameter for target_level

restore_service (Line 99)
def restore_service(self):
    """Attempt to restore full service."""

Purpose: Attempt to restore full service.

Parameters: – self: Instance reference

is_feature_available (Line 111)
def is_feature_available(self, feature_name):
    """Check if feature is available at current service level."""

Purpose: Check if feature is available at current service level.

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

execute_query (Line 119)
def execute_query(self, query, feature_name):
    """Execute query with graceful degradation."""

Purpose: Execute query with graceful degradation.

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

get_stats (Line 165)
def get_stats(self):
    """Get service statistics."""

Purpose: Get service 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 = ServiceLevel(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 *