Agentic Architecture Pattern: Exception Handling

Written by

in


    """
Pattern 186: Exception Handling Pattern
Purpose: Comprehensive exception handling with classification and logging
"""

from typing import Dict, Any, List, Optional, Callable, Type
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI


class ExceptionSeverity(Enum):
    """Exception severity levels."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class ExceptionInfo:
    """Information about an exception."""
    exception_type: str
    message: str
    severity: ExceptionSeverity
    timestamp: datetime = field(default_factory=datetime.now)
    stack_trace: Optional[str] = None
    context: Dict[str, Any] = field(default_factory=dict)
    handled: bool = False


class ExceptionHandlingPattern:
    """
    Exception Handling - Comprehensive exception management.
    
    Features:
    - Exception classification by severity
    - Custom exception handlers
    - Exception logging and tracking
    - Context preservation
    - Recovery strategies
    - Statistics and reporting
    """
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.handlers: Dict[Type[Exception], Callable] = {}
        self.exception_log: List[ExceptionInfo] = []
        self.stats = {
            "total_exceptions": 0,
            "handled": 0,
            "unhandled": 0,
            "by_severity": {s.value: 0 for s in ExceptionSeverity}
        }
    
    def register_handler(
        self,
        exception_type: Type[Exception],
        handler: Callable,
        severity: ExceptionSeverity = ExceptionSeverity.MEDIUM
    ):
        """Register custom exception handler."""
        self.handlers[exception_type] = (handler, severity)
    
    def classify_exception(self, exception: Exception) -> ExceptionSeverity:
        """Classify exception severity."""
        if isinstance(exception, (SystemExit, KeyboardInterrupt)):
            return ExceptionSeverity.CRITICAL
        elif isinstance(exception, (ValueError, TypeError, KeyError)):
            return ExceptionSeverity.MEDIUM
        elif isinstance(exception, (FileNotFoundError, ConnectionError)):
            return ExceptionSeverity.HIGH
        else:
            return ExceptionSeverity.LOW
    
    def handle_exception(
        self,
        exception: Exception,
        context: Optional[Dict[str, Any]] = None
    ) -> Any:
        """Handle exception with registered handler or default."""
        self.stats["total_exceptions"] += 1
        
        exc_type = type(exception)
        severity = self.classify_exception(exception)
        
        # Log exception
        import traceback
        exc_info = ExceptionInfo(
            exception_type=exc_type.__name__,
            message=str(exception),
            severity=severity,
            stack_trace=traceback.format_exc(),
            context=context or {}
        )
        self.exception_log.append(exc_info)
        self.stats["by_severity"][severity.value] += 1
        
        # Try registered handler
        if exc_type in self.handlers:
            handler, _ = self.handlers[exc_type]
            try:
                result = handler(exception, context)
                exc_info.handled = True
                self.stats["handled"] += 1
                return result
            except Exception as handler_error:
                print(f"Handler failed: {handler_error}")
        
        # Default handling
        self.stats["unhandled"] += 1
        
        if severity == ExceptionSeverity.CRITICAL:
            raise exception
        elif severity == ExceptionSeverity.HIGH:
            print(f"HIGH severity exception: {exception}")
            return None
        else:
            print(f"{severity.value.upper()} severity exception: {exception}")
            return None
    
    def execute_with_handling(
        self,
        func: Callable,
        *args,
        context: Optional[Dict[str, Any]] = None,
        **kwargs
    ) -> Any:
        """Execute function with exception handling."""
        try:
            return func(*args, **kwargs)
        except Exception as e:
            return self.handle_exception(e, context)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get exception statistics."""
        handling_rate = (
            self.stats["handled"] / self.stats["total_exceptions"] * 100
            if self.stats["total_exceptions"] > 0 else 0
        )
        
        return {
            **self.stats,
            "handling_rate": f"{handling_rate:.1f}%",
            "total_logged": len(self.exception_log)
        }
    
    def get_recent_exceptions(self, count: int = 10) -> List[ExceptionInfo]:
        """Get recent exceptions."""
        return self.exception_log[-count:]


if __name__ == "__main__":
    print("=== Exception Handling Pattern Demo ===\n")
    
    pattern = ExceptionHandlingPattern()
    
    # Register custom handlers
    def handle_value_error(exc, context):
        print(f"  Custom handler for ValueError: {exc}")
        return f"Recovered from ValueError: {exc}"
    
    pattern.register_handler(ValueError, handle_value_error, ExceptionSeverity.MEDIUM)
    
    # Test cases
    print("Testing exception handling:\n")
    
    # Test 1: ValueError (handled)
    print("1. Testing ValueError (custom handler):")
    def raise_value_error():
        raise ValueError("Invalid input value")
    
    result = pattern.execute_with_handling(
        raise_value_error,
        context={"test": "value_error"}
    )
    print(f"   Result: {result}\n")
    
    # Test 2: KeyError (default handling)
    print("2. Testing KeyError (default handling):")
    def raise_key_error():
        d = {}
        return d["missing_key"]
    
    result = pattern.execute_with_handling(
        raise_key_error,
        context={"test": "key_error"}
    )
    print(f"   Result: {result}\n")
    
    # Test 3: Successful execution
    print("3. Testing successful execution:")
    def successful_func():
        return "Success!"
    
    result = pattern.execute_with_handling(successful_func)
    print(f"   Result: {result}\n")
    
    # Show statistics
    print("=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total exceptions: {stats['total_exceptions']}")
    print(f"Handled: {stats['handled']}")
    print(f"Unhandled: {stats['unhandled']}")
    print(f"Handling rate: {stats['handling_rate']}")
    
    print("\nBy severity:")
    for severity, count in stats['by_severity'].items():
        if count > 0:
            print(f"  {severity}: {count}")
    
    # Show recent exceptions
    print("\n=== Recent Exceptions ===")
    for exc in pattern.get_recent_exceptions(5):
        print(f"{exc.timestamp.strftime('%H:%M:%S')} - {exc.exception_type}: {exc.message}")
        print(f"  Severity: {exc.severity.value}, Handled: {exc.handled}")

    

Agentic Architecture Pattern: Exception Handling

Pattern 186: 186 Exception Handling – Line-by-Line Explanation

Overview

Pattern 186: Exception Handling Pattern Purpose: Comprehensive exception handling with classification and logging


Code Structure Analysis

File Information

  • Pattern Number: 186
  • Pattern Name: 186 Exception Handling
  • File Path: patterns\error_handling\186_exception_handling.py
  • Total Lines: 218

Detailed Code Explanation

Module Docstring

"""
Pattern 186: Exception Handling Pattern
Purpose: Comprehensive exception handling with classification and logging
"""

Explanation: This module docstring describes the purpose and key features of the 186 Exception Handling pattern.


Imports Section

from typing import Dict, Any, List, Optional, Callable, Type
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
        import traceback  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable, Type: 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


Enumerations

ExceptionSeverity (Line 13)

class ExceptionSeverity(Enum):
    LOW
    MEDIUM
    HIGH
    CRITICAL

Purpose: Exception severity levels.

Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical


Classes

ExceptionSeverity (Line 13)

Description: Exception severity levels.

ExceptionInfo (Line 22)

Description: Information about an exception.

ExceptionHandlingPattern (Line 33)

Description: Exception Handling – Comprehensive exception management.

Features: – Exception classification by severity – Custom exception handlers – Exception logging and tracking – Context preservation – Recovery strategies – Statistics and reporting

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_handler (Line 57)
def register_handler(self, exception_type, handler, severity):
    """Register custom exception handler."""

Purpose: Register custom exception handler.

Parameters: – self: Instance reference – exception_type: Parameter for exception_type – handler: Parameter for handler – severity: Parameter for severity

classify_exception (Line 66)
def classify_exception(self, exception):
    """Classify exception severity."""

Purpose: Classify exception severity.

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

handle_exception (Line 77)
def handle_exception(self, exception, context):
    """Handle exception with registered handler or default."""

Purpose: Handle exception with registered handler or default.

Parameters: – self: Instance reference – exception: Parameter for exception – context: Parameter for context

execute_with_handling (Line 123)
def execute_with_handling(self, func):
    """Execute function with exception handling."""

Purpose: Execute function with exception handling.

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

get_stats (Line 136)
def get_stats(self):
    """Get exception statistics."""

Purpose: Get exception statistics.

Parameters: – self: Instance reference

get_recent_exceptions (Line 149)
def get_recent_exceptions(self, count):
    """Get recent exceptions."""

Purpose: Get recent exceptions.

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


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 = ExceptionSeverity(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 *