"""
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