"""
Pattern 184: Timeout Pattern
Purpose: Prevent indefinite waiting with timeouts
"""

from typing import Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


class TimeoutAction(Enum):
    """Action to take on timeout."""
    RAISE = "raise"  # Raise exception
    RETURN_NONE = "return_none"  # Return None
    RETURN_DEFAULT = "return_default"  # Return default value
    RETRY = "retry"  # Retry operation


@dataclass
class TimeoutConfig:
    """Timeout configuration."""
    timeout_seconds: float = 30.0
    action: TimeoutAction = TimeoutAction.RAISE
    default_value: Any = None
    max_retries: int = 0


@dataclass
class TimeoutEvent:
    """Record of timeout event."""
    operation: str
    timeout_seconds: float
    actual_duration: float
    action_taken: TimeoutAction
    timestamp: datetime = field(default_factory=datetime.now)
    success: bool = False


class TimeoutPattern:
    """
    Timeout Pattern - Prevent indefinite waiting.
    
    Features:
    - Configurable timeout duration
    - Multiple timeout actions
    - Async and sync support
    - Timeout statistics
    - Retry on timeout
    """
    
    def __init__(self, config: Optional[TimeoutConfig] = None):
        self.config = config or TimeoutConfig()
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7, request_timeout=self.config.timeout_seconds)
        self.events: list[TimeoutEvent] = []
        self.stats = {
            "total_operations": 0,
            "timeouts": 0,
            "successful": 0,
            "retried": 0
        }
    
    def execute_with_timeout(
        self, 
        func: Callable, 
        timeout_seconds: Optional[float] = None,
        *args, 
        **kwargs
    ) -> Any:
        """Execute function with timeout."""
        timeout = timeout_seconds or self.config.timeout_seconds
        self.stats["total_operations"] += 1
        start_time = datetime.now()
        
        try:
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                result = future.result(timeout=timeout)
                
            duration = (datetime.now() - start_time).total_seconds()
            self.stats["successful"] += 1
            
            self.events.append(TimeoutEvent(
                operation=func.__name__,
                timeout_seconds=timeout,
                actual_duration=duration,
                action_taken=TimeoutAction.RAISE,
                success=True
            ))
            
            return result
            
        except FuturesTimeoutError:
            duration = (datetime.now() - start_time).total_seconds()
            self.stats["timeouts"] += 1
            
            event = TimeoutEvent(
                operation=func.__name__,
                timeout_seconds=timeout,
                actual_duration=duration,
                action_taken=self.config.action,
                success=False
            )
            self.events.append(event)
            
            # Handle based on action
            if self.config.action == TimeoutAction.RAISE:
                raise TimeoutError(f"Operation {func.__name__} timed out after {timeout}s")
            elif self.config.action == TimeoutAction.RETURN_NONE:
                return None
            elif self.config.action == TimeoutAction.RETURN_DEFAULT:
                return self.config.default_value
            elif self.config.action == TimeoutAction.RETRY and self.config.max_retries > 0:
                self.stats["retried"] += 1
                return self.execute_with_timeout(func, timeout_seconds, *args, **kwargs)
            
            return None
    
    def llm_query_with_timeout(self, query: str, timeout_seconds: Optional[float] = None) -> str:
        """Execute LLM query with timeout."""
        def _query():
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("human", "{query}")
            ])
            chain = prompt | self.llm
            response = chain.invoke({"query": query})
            return str(response.content)
        
        return self.execute_with_timeout(_query, timeout_seconds)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get timeout statistics."""
        timeout_rate = (
            self.stats["timeouts"] / self.stats["total_operations"] * 100
            if self.stats["total_operations"] > 0 else 0
        )
        
        avg_duration = (
            sum(e.actual_duration for e in self.events) / len(self.events)
            if self.events else 0
        )
        
        return {
            **self.stats,
            "timeout_rate": f"{timeout_rate:.1f}%",
            "avg_duration_seconds": f"{avg_duration:.2f}",
            "config": {
                "timeout_seconds": self.config.timeout_seconds,
                "action": self.config.action.value
            }
        }


if __name__ == "__main__":
    print("=== Timeout Pattern Demo ===\n")
    
    # Test with short timeout
    config = TimeoutConfig(
        timeout_seconds=5.0,
        action=TimeoutAction.RETURN_DEFAULT,
        default_value="Timeout occurred"
    )
    pattern = TimeoutPattern(config)
    
    print("Testing timeout pattern:\n")
    
    # Quick query (should succeed)
    print("1. Quick query:")
    result = pattern.llm_query_with_timeout("What is 2+2?", timeout_seconds=30.0)
    print(f"   Result: {result}\n")
    
    # Potentially slow query
    print("2. Complex query:")
    result = pattern.llm_query_with_timeout(
        "Explain quantum computing in detail",
        timeout_seconds=30.0
    )
    print(f"   Result: {result[:100]}...\n")
    
    # Test timeout with custom function
    print("3. Custom function with timeout:")
    import time
    
    def slow_function():
        time.sleep(2)
        return "Completed"
    
    result = pattern.execute_with_timeout(slow_function, timeout_seconds=5.0)
    print(f"   Result: {result}\n")
    
    # Show statistics
    print("=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total operations: {stats['total_operations']}")
    print(f"Successful: {stats['successful']}")
    print(f"Timeouts: {stats['timeouts']}")
    print(f"Timeout rate: {stats['timeout_rate']}")
    print(f"Average duration: {stats['avg_duration_seconds']}s")
    print(f"\nConfiguration:")
    print(f"  Timeout: {stats['config']['timeout_seconds']}s")
    print(f"  Action: {stats['config']['action']}")

    

Agentic Architecture Pattern: Timeout

Pattern 184: 184 Timeout – Line-by-Line Explanation

Overview

Pattern 184: Timeout Pattern Purpose: Prevent indefinite waiting with timeouts

Code Structure Analysis

File Information

• Pattern Number: 184
• Pattern Name: 184 Timeout
• File Path: patterns\error_handling\184_timeout.py
• Total Lines: 207

Detailed Code Explanation

Module Docstring

"""
Pattern 184: Timeout Pattern
Purpose: Prevent indefinite waiting with timeouts
"""

Explanation: This module docstring describes the purpose and key features of the 184 Timeout pattern.

Imports Section

from typing import Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
    import time  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: 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

TimeoutAction (Line 16)

class TimeoutAction(Enum):
    RAISE
    RETURN_NONE
    RETURN_DEFAULT
    RETRY

Purpose: Action to take on timeout.

Values: – RAISE: Represents raise – RETURN_NONE: Represents return none – RETURN_DEFAULT: Represents return default – RETRY: Represents retry

Classes

TimeoutAction (Line 16)

Description: Action to take on timeout.

TimeoutConfig (Line 25)

Description: Timeout configuration.

TimeoutEvent (Line 34)

Description: Record of timeout event.

TimeoutPattern (Line 44)

Description: Timeout Pattern – Prevent indefinite waiting.

Features: – Configurable timeout duration – Multiple timeout actions – Async and sync support – Timeout statistics – Retry on timeout

Methods:

__init__ (Line 56)

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

Purpose: No method docstring.

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

execute_with_timeout (Line 67)

def execute_with_timeout(self, func, timeout_seconds):
    """Execute function with timeout."""

Purpose: Execute function with timeout.

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

llm_query_with_timeout (Line 123)

def llm_query_with_timeout(self, query, timeout_seconds):
    """Execute LLM query with timeout."""

Purpose: Execute LLM query with timeout.

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

get_stats (Line 136)

def get_stats(self):
    """Get timeout statistics."""

Purpose: Get timeout 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 = TimeoutAction(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