Decorator Pattern

The Decorator pattern wraps objects to add new functionality without changing their core implementation. In LLM applications, decorators excel at adding cross-cutting concerns like caching, retry logic, monitoring, and cost tracking.

Why Decorator Pattern for LLM?

LLM applications often need:

• Response caching: Store expensive API calls to reduce costs and latency

• Retry mechanisms: Handle API failures gracefully with exponential backoff

• Cost tracking: Monitor spending across different providers and models

• Performance monitoring: Track response times and success rates

• Rate limiting: Control API call frequency to avoid quota issues

Key LLM Use Cases

1. Response Caching Strategy

Cache LLM responses to improve performance and reduce costs:

import functools
import hashlib
import json
from typing import Dict, Any, Optional

class LLMCache:
    def __init__(self, max_size: int = 1000, ttl: int = 3600):
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.max_size = max_size
        self.ttl = ttl  # Time to live in seconds
    
    def _generate_key(self, *args, **kwargs) -> str:
        """Generate cache key from function arguments"""
        key_data = {'args': str(args), 'kwargs': kwargs}
        return hashlib.md5(json.dumps(key_data, sort_keys=True).encode()).hexdigest()
    
    def get(self, key: str) -> Optional[Any]:
        """Get cached value if exists and not expired"""
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry['timestamp'] < self.ttl:
                return entry['value']
            else:
                del self.cache[key]  # Remove expired entry
        return None
    
    def set(self, key: str, value: Any):
        """Set cache value with timestamp"""
        if len(self.cache) >= self.max_size:
            # Remove oldest entry
            oldest_key = min(self.cache.keys(), key=lambda k: self.cache[k]['timestamp'])
            del self.cache[oldest_key]
        
        self.cache[key] = {
            'value': value,
            'timestamp': time.time()
        }

def cache_llm_response(cache_instance: Optional[LLMCache] = None):
    """Decorator to cache LLM responses"""
    if cache_instance is None:
        cache_instance = LLMCache()
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Generate cache key
            cache_key = cache_instance._generate_key(*args, **kwargs)
            
            # Check cache first
            cached_result = cache_instance.get(cache_key)
            if cached_result is not None:
                print(f"🎯 Cache hit! Saved API call")
                return cached_result
            
            # Call original function
            print(f"🌐 Making API call...")
            result = func(*args, **kwargs)
            
            # Store in cache
            cache_instance.set(cache_key, result)
            print(f"💾 Response cached")
            
            return result
        
        wrapper.cache = cache_instance  # Expose cache for debugging
        return wrapper
    return decorator

# Usage example
@cache_llm_response()
def call_llm(prompt: str, model: str = "gpt-4") -> str:
    # Simulate expensive LLM API call
    time.sleep(2)  # Simulate network delay
    return f"Response to: {prompt[:50]}..."

Benefits:

• Significant cost reduction for repeated queries

• Improved response times for cached content

• Reduced API quota usage

• Better user experience with faster responses

2. Retry and Resilience Decorator

Handle API failures with intelligent retry logic:

import time
import random
from functools import wraps

def retry_llm_call(max_attempts: int = 3, base_delay: float = 1.0, 
                   max_delay: float = 60.0, backoff_factor: float = 2.0):
    """Decorator to retry LLM calls with exponential backoff"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                
                except Exception as e:
                    last_exception = e
                    
                    if attempt == max_attempts - 1:
                        # Last attempt failed
                        print(f"❌ All {max_attempts} attempts failed")
                        raise last_exception
                    
                    # Calculate delay with exponential backoff
                    delay = min(max_delay, base_delay * (backoff_factor ** attempt))
                    jitter = random.uniform(0, 0.1) * delay  # Add jitter
                    total_delay = delay + jitter
                    
                    print(f"⚠️ Attempt {attempt + 1} failed: {str(e)[:50]}...")
                    print(f"🔄 Retrying in {total_delay:.2f} seconds...")
                    time.sleep(total_delay)
            
            raise last_exception
        
        return wrapper
    return decorator

# Usage example
@retry_llm_call(max_attempts=3, base_delay=1.0)
@cache_llm_response()
def reliable_llm_call(prompt: str) -> str:
    # Simulate potentially failing API call
    if random.random() < 0.3:  # 30% failure rate
        raise Exception("API temporarily unavailable")
    return f"Successful response to: {prompt}"

Benefits:

• Automatic recovery from transient failures

• Exponential backoff prevents API overwhelming

• Configurable retry policies per use case

• Maintains service reliability despite API issues

3. Cost Tracking Decorator

Monitor and control LLM spending:

from datetime import datetime
from typing import Dict, List

class CostTracker:
    def __init__(self):
        self.costs: List[Dict[str, Any]] = []
        self.total_cost = 0.0
        
        # Pricing per 1K tokens (example prices)
        self.model_prices = {
            'gpt-4': {'input': 0.03, 'output': 0.06},
            'gpt-3.5-turbo': {'input': 0.001, 'output': 0.002},
            'claude-3-opus': {'input': 0.015, 'output': 0.075},
            'gemini-pro': {'input': 0.0005, 'output': 0.0015}
        }
    
    def estimate_tokens(self, text: str) -> int:
        """Rough token estimation (4 characters ≈ 1 token)"""
        return max(1, len(text) // 4)
    
    def calculate_cost(self, model: str, input_text: str, output_text: str) -> float:
        """Calculate cost for LLM call"""
        if model not in self.model_prices:
            return 0.0  # Unknown model
        
        prices = self.model_prices[model]
        input_tokens = self.estimate_tokens(input_text)
        output_tokens = self.estimate_tokens(output_text)
        
        cost = (input_tokens / 1000 * prices['input'] + 
                output_tokens / 1000 * prices['output'])
        return round(cost, 6)
    
    def add_cost(self, model: str, input_text: str, output_text: str):
        """Add cost entry"""
        cost = self.calculate_cost(model, input_text, output_text)
        self.total_cost += cost
        
        self.costs.append({
            'timestamp': datetime.now(),
            'model': model,
            'input_tokens': self.estimate_tokens(input_text),
            'output_tokens': self.estimate_tokens(output_text),
            'cost': cost
        })
    
    def get_summary(self) -> Dict[str, Any]:
        """Get cost summary"""
        if not self.costs:
            return {'total_cost': 0, 'total_calls': 0}
        
        return {
            'total_cost': round(self.total_cost, 4),
            'total_calls': len(self.costs),
            'average_cost': round(self.total_cost / len(self.costs), 4),
            'most_expensive_model': max(set(entry['model'] for entry in self.costs),
                                      key=lambda m: sum(e['cost'] for e in self.costs if e['model'] == m))
        }

def track_llm_cost(cost_tracker: CostTracker, model: str = "gpt-4"):
    """Decorator to track LLM costs"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Extract input from arguments
            input_text = str(args[0]) if args else str(kwargs.get('prompt', ''))
            
            # Call original function
            result = func(*args, **kwargs)
            
            # Track cost
            output_text = str(result)
            cost_tracker.add_cost(model, input_text, output_text)
            
            # Show cost info
            cost = cost_tracker.calculate_cost(model, input_text, output_text)
            print(f"💰 Cost: ${cost:.4f} | Total: ${cost_tracker.total_cost:.4f}")
            
            return result
        
        return wrapper
    return decorator

# Usage example
cost_tracker = CostTracker()

@track_llm_cost(cost_tracker, model="gpt-4")
@cache_llm_response()
def cost_aware_llm_call(prompt: str) -> str:
    return f"Processed: {prompt[:30]}... [Response length: 150 tokens]"

Benefits:

• Real-time cost monitoring

• Budget control and alerts

• Cost optimization insights

• Spending transparency across models

4. Performance Monitoring Decorator

Track response times and success rates:

import time
from collections import defaultdict
from statistics import mean

class PerformanceMonitor:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.success_count = 0
        self.failure_count = 0
    
    def record_call(self, duration: float, success: bool, model: str):
        """Record performance metrics"""
        self.metrics[model].append(duration)
        if success:
            self.success_count += 1
        else:
            self.failure_count += 1
    
    def get_stats(self) -> Dict[str, Any]:
        """Get performance statistics"""
        total_calls = self.success_count + self.failure_count
        
        model_stats = {}
        for model, durations in self.metrics.items():
            if durations:
                model_stats[model] = {
                    'calls': len(durations),
                    'avg_duration': round(mean(durations), 3),
                    'min_duration': round(min(durations), 3),
                    'max_duration': round(max(durations), 3)
                }
        
        return {
            'total_calls': total_calls,
            'success_rate': round((self.success_count / total_calls) * 100, 2) if total_calls > 0 else 0,
            'model_performance': model_stats
        }

def monitor_performance(monitor: PerformanceMonitor, model: str = "default"):
    """Decorator to monitor LLM performance"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            success = False
            
            try:
                result = func(*args, **kwargs)
                success = True
                return result
            
            except Exception as e:
                success = False
                raise
            
            finally:
                duration = time.time() - start_time
                monitor.record_call(duration, success, model)
                
                if success:
                    print(f"⚡ Call completed in {duration:.3f}s")
                else:
                    print(f"💥 Call failed after {duration:.3f}s")
        
        return wrapper
    return decorator

# Usage example
performance_monitor = PerformanceMonitor()

@monitor_performance(performance_monitor, model="gpt-4")
@track_llm_cost(cost_tracker, model="gpt-4")
@retry_llm_call(max_attempts=2)
@cache_llm_response()
def full_featured_llm_call(prompt: str) -> str:
    """LLM call with all decorators applied"""
    time.sleep(random.uniform(0.5, 2.0))  # Simulate variable response time
    return f"Enhanced response to: {prompt}"

Benefits:

• Performance bottleneck identification

• SLA compliance monitoring

• Model comparison insights

• System health tracking

Implementation Advantages

1. Separation of Concerns

• Core LLM logic remains clean and focused

• Cross-cutting concerns handled separately

• Easy to enable/disable features per environment

• Independent testing of each concern

2. Composability

• Stack multiple decorators for comprehensive functionality

• Order-dependent behavior (cache before retry, monitor everything)

• Mix and match based on requirements

• Reusable across different LLM functions

3. Transparency

• Original function interface unchanged

• Callers unaware of decorator enhancements

• Easy migration from basic to enhanced functions

• Backward compatibility maintained

4. Configuration Flexibility

• Runtime decorator configuration

• Environment-specific behavior

• A/B testing different decorator combinations

• Dynamic feature toggling

Real-World Impact

The Decorator pattern in LLM applications provides:

• Cost Savings: Caching reduces API calls by 30-70% in typical applications

• Reliability: Retry mechanisms improve success rates from 95% to 99.9%

• Performance: Cached responses serve in milliseconds vs seconds

• Observability: Complete visibility into LLM usage patterns and costs

---

🔗 Interactive Implementation

📓 Decorator Pattern Notebook - LLM response caching, performance monitoring, and cost tracking with real OpenAI API integration.

This pattern is essential for production LLM systems where reliability, performance, and cost control are critical requirements.