aishastr.live

Agentic Design Pattern: Caching Patterns

Written by

admin

in


    """
Pattern 089: Caching Patterns

Description:
    The Caching Patterns involve storing and reusing previously computed results to
    improve performance, reduce latency, and minimize costs. In the context of LLM
    applications, caching is critical for optimizing response times and reducing API
    calls for repeated or similar queries. Effective caching strategies can
    dramatically improve user experience while significantly cutting operational costs.

    Caching in LLM applications targets:
    - Complete responses (exact match caching)
    - Embeddings (vector representations)
    - Prompt templates (reusable structures)
    - Intermediate results (multi-step workflows)
    - API responses (rate limit optimization)

Components:
    1. Cache Types
       - In-memory cache (fast, volatile)
       - Disk cache (persistent, slower)
       - Distributed cache (Redis, Memcached)
       - Database cache (structured, queryable)
       - CDN cache (geographic distribution)

    2. Cache Keys
       - Exact match (full prompt hash)
       - Semantic match (embedding similarity)
       - Parametric keys (template + parameters)
       - Hierarchical keys (nested cache structures)
       - Time-based keys (temporal validity)

    3. Cache Policies
       - TTL (Time To Live) - expire after time
       - LRU (Least Recently Used) - evict old entries
       - LFU (Least Frequently Used) - evict unpopular
       - Size-based - limit total cache size
       - Custom policies - application-specific

    4. Invalidation Strategies
       - Time-based expiration
       - Event-triggered invalidation
       - Manual invalidation
       - Cascade invalidation (dependencies)
       - Version-based invalidation

Use Cases:
    1. Repeated Queries
       - FAQ responses
       - Common questions
       - Popular content
       - Standard procedures
       - Frequently accessed data

    2. Expensive Operations
       - Complex reasoning chains
       - Multi-step workflows
       - Large context processing
       - Batch operations
       - Resource-intensive computations

    3. Rate Limit Optimization
       - Reduce API calls
       - Stay within quotas
       - Handle traffic spikes
       - Cost optimization
       - Performance consistency

    4. Embeddings and Vectors
       - Document embeddings
       - Query embeddings
       - Semantic search optimization
       - Similarity computations
       - Vector database queries

    5. Multi-User Applications
       - Shared responses
       - Common workflows
       - Popular queries
       - Template responses
       - Collaborative features

LangChain Implementation:
    LangChain supports caching through:
    - InMemoryCache for simple caching
    - SQLiteCache for persistent caching
    - Custom cache implementations
    - Semantic similarity caching
    - Embedding caching

Key Features:
    1. Multiple Cache Layers
       - L1: In-memory (fastest)
       - L2: Local disk (persistent)
       - L3: Distributed cache (shared)
       - L4: Database (long-term)

    2. Intelligent Matching
       - Exact match for identical queries
       - Semantic match for similar queries
       - Fuzzy match with thresholds
       - Parametric templates

    3. Performance Optimization
       - Fast lookup ( 70%)
       - Measure latency improvements
       - Calculate cost savings
       - Identify cache misses patterns

    4. Handle Cache Misses
       - Graceful fallback to computation
       - Transparent to users
       - Background cache warming
       - Predictive pre-caching

Trade-offs:
    Advantages:
    - Dramatically improved response times
    - Significant cost reduction
    - Better user experience
    - Reduced API rate limiting
    - Lower resource consumption
    - Improved scalability

    Disadvantages:
    - Stale data risk
    - Memory/storage overhead
    - Cache consistency challenges
    - Complexity in invalidation
    - Initial cache warming time
    - Debugging complexity

Production Considerations:
    1. Cache Size Management
       - Monitor memory usage
       - Implement eviction policies
       - Set size limits
       - Regular cleanup
       - Compression strategies

    2. Cache Consistency
       - Version tracking
       - Update propagation
       - Invalidation timing
       - Distributed consistency
       - Race condition handling

    3. Security
       - Access control for cached data
       - Encryption for sensitive content
       - Cache poisoning prevention
       - User isolation in multi-tenant
       - Audit trails

    4. Monitoring
       - Hit/miss rates
       - Cache size trends
       - Eviction frequency
       - Cost savings
       - Performance impact

    5. Disaster Recovery
       - Cache rebuild procedures
       - Backup strategies
       - Failover mechanisms
       - Data loss handling
       - Recovery time objectives
"""

import os
import time
import hashlib
import json
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
from collections import OrderedDict
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.caches import InMemoryCache
from langchain.globals import set_llm_cache

load_dotenv()


class CachePolicy(Enum):
    """Cache eviction policies"""
    LRU = "lru"  # Least Recently Used
    LFU = "lfu"  # Least Frequently Used
    FIFO = "fifo"  # First In First Out
    TTL = "ttl"  # Time To Live


@dataclass
class CacheEntry:
    """Represents a cached entry"""
    key: str
    value: Any
    created_at: datetime
    last_accessed: datetime
    access_count: int = 0
    ttl: Optional[int] = None  # TTL in seconds
    
    def is_expired(self) -> bool:
        """Check if entry is expired"""
        if self.ttl is None:
            return False
        age = (datetime.now() - self.created_at).total_seconds()
        return age > self.ttl
    
    def touch(self):
        """Update access information"""
        self.last_accessed = datetime.now()
        self.access_count += 1


@dataclass
class CacheStats:
    """Cache performance statistics"""
    hits: int = 0
    misses: int = 0
    evictions: int = 0
    total_requests: int = 0
    
    @property
    def hit_rate(self) -> float:
        """Calculate cache hit rate"""
        if self.total_requests == 0:
            return 0.0
        return (self.hits / self.total_requests) * 100
    
    @property
    def miss_rate(self) -> float:
        """Calculate cache miss rate"""
        return 100.0 - self.hit_rate


class SimpleCache:
    """
    Simple in-memory cache with multiple eviction policies.
    
    This cache supports:
    1. Multiple eviction policies (LRU, LFU, FIFO, TTL)
    2. Size limits
    3. TTL per entry
    4. Performance statistics
    """
    
    def __init__(
        self,
        max_size: int = 1000,
        policy: CachePolicy = CachePolicy.LRU,
        default_ttl: Optional[int] = None
    ):
        """
        Initialize cache.
        
        Args:
            max_size: Maximum number of entries
            policy: Eviction policy
            default_ttl: Default TTL in seconds
        """
        self.max_size = max_size
        self.policy = policy
        self.default_ttl = default_ttl
        self.cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self.stats = CacheStats()
    
    def _make_key(self, key: Any) -> str:
        """Create cache key from input"""
        if isinstance(key, str):
            return hashlib.md5(key.encode()).hexdigest()
        return hashlib.md5(json.dumps(key, sort_keys=True).encode()).hexdigest()
    
    def _evict(self):
        """Evict entries based on policy"""
        if len(self.cache)  Optional[Any]:
        """
        Get value from cache.
        
        Args:
            key: Cache key
            
        Returns:
            Cached value or None
        """
        self.stats.total_requests += 1
        cache_key = self._make_key(key)
        
        # Remove expired entries
        self._remove_expired()
        
        if cache_key in self.cache:
            entry = self.cache[cache_key]
            
            if entry.is_expired():
                del self.cache[cache_key]
                self.stats.misses += 1
                return None
            
            entry.touch()
            
            # Move to end for LRU
            if self.policy == CachePolicy.LRU:
                self.cache.move_to_end(cache_key)
            
            self.stats.hits += 1
            return entry.value
        
        self.stats.misses += 1
        return None
    
    def set(self, key: Any, value: Any, ttl: Optional[int] = None):
        """
        Set value in cache.
        
        Args:
            key: Cache key
            value: Value to cache
            ttl: Optional TTL in seconds
        """
        cache_key = self._make_key(key)
        
        # Evict if necessary
        if cache_key not in self.cache:
            self._evict()
        
        entry = CacheEntry(
            key=cache_key,
            value=value,
            created_at=datetime.now(),
            last_accessed=datetime.now(),
            ttl=ttl or self.default_ttl
        )
        
        self.cache[cache_key] = entry
        
        # Move to end for LRU
        if self.policy == CachePolicy.LRU:
            self.cache.move_to_end(cache_key)
    
    def clear(self):
        """Clear all cache entries"""
        self.cache.clear()
        self.stats = CacheStats()
    
    def size(self) -> int:
        """Get current cache size"""
        return len(self.cache)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics"""
        return {
            "hits": self.stats.hits,
            "misses": self.stats.misses,
            "evictions": self.stats.evictions,
            "total_requests": self.stats.total_requests,
            "hit_rate": f"{self.stats.hit_rate:.2f}%",
            "miss_rate": f"{self.stats.miss_rate:.2f}%",
            "current_size": len(self.cache),
            "max_size": self.max_size
        }


class SemanticCache:
    """
    Semantic cache using embeddings for similarity matching.
    
    This cache finds similar queries using vector similarity,
    allowing cache hits for semantically similar but not identical queries.
    """
    
    def __init__(
        self,
        similarity_threshold: float = 0.95,
        max_size: int = 500
    ):
        """
        Initialize semantic cache.
        
        Args:
            similarity_threshold: Minimum similarity for cache hit
            max_size: Maximum cache entries
        """
        self.similarity_threshold = similarity_threshold
        self.max_size = max_size
        self.embeddings = OpenAIEmbeddings()
        self.cache: List[Dict[str, Any]] = []
        self.stats = CacheStats()
    
    def _compute_similarity(self, query1: str, query2: str) -> float:
        """
        Compute semantic similarity between queries.
        
        Args:
            query1: First query
            query2: Second query
            
        Returns:
            Similarity score (0-1)
        """
        # Simple implementation - in production use precomputed embeddings
        # This is for demonstration only
        query1_lower = query1.lower()
        query2_lower = query2.lower()
        
        # Simple word overlap similarity
        words1 = set(query1_lower.split())
        words2 = set(query2_lower.split())
        
        if not words1 or not words2:
            return 0.0
        
        intersection = words1.intersection(words2)
        union = words1.union(words2)
        
        return len(intersection) / len(union)
    
    def get(self, query: str) -> Optional[Any]:
        """
        Get value for semantically similar query.
        
        Args:
            query: Query string
            
        Returns:
            Cached value or None
        """
        self.stats.total_requests += 1
        
        # Find most similar cached query
        best_match = None
        best_similarity = 0.0
        
        for entry in self.cache:
            similarity = self._compute_similarity(query, entry["query"])
            if similarity > best_similarity:
                best_similarity = similarity
                best_match = entry
        
        if best_match and best_similarity >= self.similarity_threshold:
            self.stats.hits += 1
            # Update access time
            best_match["last_accessed"] = datetime.now()
            return best_match["value"]
        
        self.stats.misses += 1
        return None
    
    def set(self, query: str, value: Any):
        """
        Cache value for query.
        
        Args:
            query: Query string
            value: Value to cache
        """
        # Evict oldest if at capacity
        if len(self.cache) >= self.max_size:
            self.cache.pop(0)
            self.stats.evictions += 1
        
        self.cache.append({
            "query": query,
            "value": value,
            "created_at": datetime.now(),
            "last_accessed": datetime.now()
        })
    
    def clear(self):
        """Clear cache"""
        self.cache.clear()
        self.stats = CacheStats()
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics"""
        return {
            "hits": self.stats.hits,
            "misses": self.stats.misses,
            "evictions": self.stats.evictions,
            "total_requests": self.stats.total_requests,
            "hit_rate": f"{self.stats.hit_rate:.2f}%",
            "current_size": len(self.cache),
            "max_size": self.max_size,
            "similarity_threshold": self.similarity_threshold
        }


class CachedLLMAgent:
    """
    LLM agent with caching support.
    
    This agent automatically caches responses and retrieves them
    for repeated or similar queries.
    """
    
    def __init__(
        self,
        cache: SimpleCache,
        model: str = "gpt-3.5-turbo"
    ):
        """
        Initialize cached agent.
        
        Args:
            cache: Cache instance to use
            model: LLM model name
        """
        self.cache = cache
        self.llm = ChatOpenAI(model=model, temperature=0.7)
    
    def generate(self, prompt: str) -> Dict[str, Any]:
        """
        Generate response with caching.
        
        Args:
            prompt: Input prompt
            
        Returns:
            Dictionary with response and cache info
        """
        # Check cache first
        cached_response = self.cache.get(prompt)
        
        if cached_response is not None:
            return {
                "response": cached_response,
                "from_cache": True,
                "cache_stats": self.cache.get_stats()
            }
        
        # Generate new response
        start_time = time.time()
        response = self.llm.invoke(prompt)
        generation_time = time.time() - start_time
        
        # Cache the response
        self.cache.set(prompt, response.content)
        
        return {
            "response": response.content,
            "from_cache": False,
            "generation_time": generation_time,
            "cache_stats": self.cache.get_stats()
        }


def demonstrate_caching_patterns():
    """Demonstrate caching patterns"""
    print("=" * 80)
    print("CACHING PATTERNS DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Basic LRU Cache
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Basic LRU Cache")
    print("=" * 80)
    
    cache = SimpleCache(max_size=5, policy=CachePolicy.LRU)
    agent = CachedLLMAgent(cache)
    
    print("\nMaking requests (first time - cache misses):\n")
    
    queries = [
        "What is Python?",
        "What is JavaScript?",
        "What is Python?",  # Repeat
        "What is Java?",
        "What is Python?",  # Repeat again
    ]
    
    for i, query in enumerate(queries, 1):
        print(f"Request {i}: {query}")
        result = agent.generate(query)
        
        cache_status = "HIT ✓" if result["from_cache"] else "MISS ✗"
        print(f"  Status: {cache_status}")
        
        if not result["from_cache"]:
            print(f"  Generation time: {result['generation_time']:.2f}s")
        print(f"  Response: {result['response'][:80]}...")
        print()
    
    stats = cache.get_stats()
    print("Cache Statistics:")
    print(f"  Hit Rate: {stats['hit_rate']}")
    print(f"  Total Requests: {stats['total_requests']}")
    print(f"  Hits: {stats['hits']}")
    print(f"  Misses: {stats['misses']}")
    
    # Example 2: TTL-Based Caching
    print("\n" + "=" * 80)
    print("EXAMPLE 2: TTL-Based Caching (Time Expiration)")
    print("=" * 80)
    
    ttl_cache = SimpleCache(max_size=100, default_ttl=5)  # 5 second TTL
    
    print("\nCaching value with 5-second TTL...")
    ttl_cache.set("test_key", "test_value", ttl=5)
    
    print("Immediate retrieval:")
    value = ttl_cache.get("test_key")
    print(f"  Result: {'HIT' if value else 'MISS'} - {value}")
    
    print("\nWaiting 6 seconds for expiration...")
    time.sleep(6)
    
    print("Retrieval after TTL:")
    value = ttl_cache.get("test_key")
    print(f"  Result: {'HIT' if value else 'MISS'} - {value}")
    
    # Example 3: Cache Eviction Policies
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Cache Eviction Policies Comparison")
    print("=" * 80)
    
    policies = [CachePolicy.LRU, CachePolicy.LFU, CachePolicy.FIFO]
    
    for policy in policies:
        print(f"\n{policy.value.upper()} Policy:")
        test_cache = SimpleCache(max_size=3, policy=policy)
        
        # Add 5 items to cache of size 3
        for i in range(5):
            key = f"item_{i}"
            test_cache.set(key, f"value_{i}")
            print(f"  Added {key}, cache size: {test_cache.size()}")
        
        stats = test_cache.get_stats()
        print(f"  Evictions: {stats['evictions']}")
    
    # Example 4: Semantic Cache
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Semantic Cache (Similar Query Matching)")
    print("=" * 80)
    
    semantic_cache = SemanticCache(similarity_threshold=0.7, max_size=100)
    
    # Cache a response
    original_query = "What is machine learning?"
    original_response = "Machine learning is a subset of AI that enables systems to learn from data."
    semantic_cache.set(original_query, original_response)
    
    print(f"\nCached query: '{original_query}'")
    print(f"Response: {original_response}\n")
    
    # Try similar queries
    similar_queries = [
        "What is machine learning?",  # Exact match
        "What's machine learning?",   # Slight variation
        "Tell me about machine learning",  # Similar
        "Explain artificial intelligence",  # Different topic
    ]
    
    for query in similar_queries:
        result = semantic_cache.get(query)
        status = "HIT ✓" if result else "MISS ✗"
        print(f"Query: '{query}'")
        print(f"  Status: {status}")
        if result:
            print(f"  Retrieved: {result}")
        print()
    
    # Example 5: Multi-Layer Cache
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Multi-Layer Cache Strategy")
    print("=" * 80)
    
    # L1: Small, fast, in-memory cache
    l1_cache = SimpleCache(max_size=10, policy=CachePolicy.LRU)
    
    # L2: Larger, still in-memory
    l2_cache = SimpleCache(max_size=100, policy=CachePolicy.LRU)
    
    print("\nMulti-layer cache simulation:")
    print("  L1: Small (10 items), fast")
    print("  L2: Large (100 items), moderate\n")
    
    def multi_layer_get(key: str) -> Optional[str]:
        """Get from multi-layer cache"""
        # Try L1 first
        value = l1_cache.get(key)
        if value:
            print(f"  L1 HIT for '{key}'")
            return value
        
        # Try L2
        value = l2_cache.get(key)
        if value:
            print(f"  L2 HIT for '{key}'")
            # Promote to L1
            l1_cache.set(key, value)
            return value
        
        print(f"  MISS for '{key}'")
        return None
    
    # Simulate access pattern
    l1_cache.set("frequently_accessed", "hot_data")
    l2_cache.set("occasionally_accessed", "warm_data")
    
    print("Access patterns:")
    multi_layer_get("frequently_accessed")
    multi_layer_get("occasionally_accessed")
    multi_layer_get("never_cached")
    
    # Example 6: Cache Performance Metrics
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Cache Performance Analysis")
    print("=" * 80)
    
    perf_cache = SimpleCache(max_size=50, policy=CachePolicy.LRU)
    perf_agent = CachedLLMAgent(perf_cache)
    
    print("\nSimulating workload with repeated queries...\n")
    
    workload = [
        "What is AI?",
        "What is ML?",
        "What is AI?",  # Repeat
        "What is DL?",
        "What is AI?",  # Repeat
        "What is ML?",  # Repeat
        "What is NLP?",
        "What is AI?",  # Repeat
    ]
    
    total_time_with_cache = 0
    total_time_without_cache = 0
    
    for query in workload:
        result = perf_agent.generate(query)
        
        if result["from_cache"]:
            total_time_with_cache += 0.001  # Assume 1ms for cache hit
        else:
            gen_time = result.get("generation_time", 1.0)
            total_time_with_cache += gen_time
            total_time_without_cache += gen_time
    
    # Estimate time without cache (all queries would need generation)
    total_time_without_cache = len(workload) * 1.0  # Assume 1s per generation
    
    print(f"Performance Analysis:")
    print(f"  Total Queries: {len(workload)}")
    print(f"  Time WITH cache: {total_time_with_cache:.2f}s")
    print(f"  Time WITHOUT cache (estimated): {total_time_without_cache:.2f}s")
    print(f"  Speedup: {total_time_without_cache/total_time_with_cache:.1f}x")
    print(f"  Time Saved: {total_time_without_cache - total_time_with_cache:.2f}s")
    
    final_stats = perf_cache.get_stats()
    print(f"\nFinal Cache Statistics:")
    print(f"  Hit Rate: {final_stats['hit_rate']}")
    print(f"  Hits: {final_stats['hits']}")
    print(f"  Misses: {final_stats['misses']}")
    
    # Summary
    print("\n" + "=" * 80)
    print("CACHING PATTERNS SUMMARY")
    print("=" * 80)
    print("""
Caching Pattern Benefits:
1. Performance: Dramatically faster response times (10-1000x)
2. Cost Reduction: Fewer API calls = lower costs
3. Scalability: Handle more requests with same resources
4. Reliability: Reduce dependency on external services
5. User Experience: Instant responses for cached queries

Cache Types Demonstrated:
1. Simple Cache: Basic in-memory caching with eviction policies
2. TTL Cache: Time-based expiration for freshness
3. Semantic Cache: Similarity-based matching for related queries
4. Multi-Layer Cache: Hierarchical caching strategy
5. Performance-Optimized: Focus on hit rates and speedup

Eviction Policies:
1. LRU (Least Recently Used): Remove oldest accessed items
2. LFU (Least Frequently Used): Remove least popular items
3. FIFO (First In First Out): Remove oldest items
4. TTL (Time To Live): Expire after time period

Best Practices:
1. Cache Strategy Selection:
   - Exact match: Simple, fast, for identical queries
   - Semantic: Better coverage, for similar queries
   - TTL: For time-sensitive data
   - Multi-layer: For diverse access patterns

2. TTL Configuration:
   - Short (minutes): Dynamic, frequently changing data
   - Medium (hours): Semi-static content
   - Long (days): Stable, rarely changing data
   - No TTL: Static reference data

3. Size Management:
   - Monitor memory usage
   - Set appropriate limits
   - Implement eviction policies
   - Regular cleanup

4. Performance Monitoring:
   - Target >70% hit rate for good performance
   - Track cache size trends
   - Monitor eviction frequency
   - Measure latency improvements

When to Use Caching:
✓ Repeated identical queries
✓ Similar semantic queries
✓ Expensive computations
✓ Rate-limited APIs
✓ Stable, unchanging data
✗ Real-time data requirements
✗ Highly personalized responses
✗ Security-sensitive information
✗ Rapidly changing data

Production Considerations:
- Distributed caching (Redis) for multi-instance deployments
- Cache warming strategies for cold starts
- Monitoring and alerting on hit rates
- Regular cache analysis and optimization
- Security: encrypt sensitive cached data
- Compliance: respect data retention policies
- Backup and recovery procedures
- Cache versioning for updates

Cost-Benefit Analysis:
- Typical API call: $0.001-0.01 per request
- Cache hit cost: ~$0.000001 per request
- Potential savings: 90-99% cost reduction
- Performance gain: 10-1000x faster responses
- ROI: Immediate for high-volume applications
""")
    
    print("\n" + "=" * 80)
    print("Pattern 089 (Caching Patterns) demonstration complete!")
    print("=" * 80)


if __name__ == "__main__":
    demonstrate_caching_patterns()

Agentic Design Pattern: Caching Patterns

Pattern 089: Caching Patterns – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Caching Patterns pattern implementation.

Pattern Description

Pattern 089: Caching Patterns

Description: The Caching Patterns involve storing and reusing previously computed results to improve performance, reduce latency, and minimize costs. In the context of LLM applications, caching is critical for optimizing response times and reducing API calls for repeated or similar queries. Effective caching strategies can dramatically improve user experience while significantly cutting operational costs.

Caching in LLM applications targets:
- Complete responses (exact match caching)
- Embeddings (vector representations)
- Prompt templates (reusable structures)
- Intermediate results (multi-step workflows)
- API responses (rate limit optimization)

Components: 1. Cache Types – In-memory cache (fast, volatile) – Disk cache (persistent, slower) – Distributed cache (Redis, Memcached) – Database cache (structured, queryable) – CDN cache (geographic distribution)

2. Cache Keys
   - Exact match (full prompt hash)
   - Semantic match (embedding similarity)
   - Parametric keys (template + parameters)
   - Hierarchical keys (nested cache structures)
   - Time-based keys (temporal validity)

3. Cache Policies
   - TTL (Time To Live) - expire after time
   - LRU (Least Recently Used) - evict old entries
   - LFU (Least Frequently Used) - evict unpopular
   - Size-based - limit total cache size
   - Custom policies - application-specific

4. Invalidation Strategies
   - Time-based expiration
   - Event-triggered invalidation
   - Manual invalidation
   - Cascade invalidation (dependencies)
   - Version-based invalidation

Use Cases: 1. Repeated Queries – FAQ responses – Common questions – Popular content – Standard procedures – Frequently accessed data

2. Expensive Operations
   - Complex reasoning chains
   - Multi-step workflows
   - Large context processing
   - Batch operations
   - Resource-intensive computations

3. Rate Limit Optimization
   - Reduce API calls
   - Stay within quotas
   - Handle traffic spikes
   - Cost optimization
   - Performance consistency

4. Embeddings and Vectors
   - Document embeddings
   - Query embeddings
   - Semantic search optimization
   - Similarity computations
   - Vector database queries

5. Multi-User Applications
   - Shared responses
   - Common workflows
   - Popular queries
   - Template responses
   - Collaborative features

LangChain Implementation: LangChain supports caching through: – InMemoryCache for simple caching – SQLiteCache for persistent caching – Custom cache implementations – Semantic similarity caching – Embedding caching

Key Features: 1. Multiple Cache Layers – L1: In-memory (fastest) – L2: Local disk (persistent) – L3: Distributed cache (shared) – L4: Database (long-term)

2. Intelligent Matching
   - Exact match for identical queries
   - Semantic match for similar queries
   - Fuzzy match with thresholds
   - Parametric templates

3. Performance Optimization
   - Fast lookup (< 1ms for in-memory)
   - Efficient serialization
   - Compression for large entries
   - Lazy loading

4. Observability
   - Hit rate monitoring
   - Miss analysis
   - Performance metrics
   - Cost savings tracking

Best Practices: 1. Cache Appropriately – Cache stable, reusable results – Don’t cache rapidly changing data – Consider data sensitivity – Balance freshness vs performance

2. Set Proper TTLs
   - Shorter for dynamic content
   - Longer for stable content
   - Consider update frequency
   - Account for user expectations

3. Monitor Performance
   - Track hit rates (target > 70%)
   - Measure latency improvements
   - Calculate cost savings
   - Identify cache misses patterns

4. Handle Cache Misses
   - Graceful fallback to computation
   - Transparent to users
   - Background cache warming
   - Predictive pre-caching

Trade-offs: Advantages: – Dramatically improved response times – Significant cost reduction – Better user experience – Reduced API rate limiting – Lower resource consumption – Improved scalability

Disadvantages:
- Stale data risk
- Memory/storage overhead
- Cache consistency challenges
- Complexity in invalidation
- Initial cache warming time
- Debugging complexity

Production Considerations: 1. Cache Size Management – Monitor memory usage – Implement eviction policies – Set size limits – Regular cleanup – Compression strategies

2. Cache Consistency
   - Version tracking
   - Update propagation
   - Invalidation timing
   - Distributed consistency
   - Race condition handling

3. Security
   - Access control for cached data
   - Encryption for sensitive content
   - Cache poisoning prevention
   - User isolation in multi-tenant
   - Audit trails

4. Monitoring
   - Hit/miss rates
   - Cache size trends
   - Eviction frequency
   - Cost savings
   - Performance impact

5. Disaster Recovery
   - Cache rebuild procedures
   - Backup strategies
   - Failover mechanisms
   - Data loss handling
   - Recovery time objectives

File Statistics

  • Total Lines: 897
  • Classes: 6
  • Functions: 1
  • Constants: 0

Module Documentation

"""
Pattern 089: Caching Patterns

Description:
    The Caching Patterns involve storing and reusing previously computed results to
    improve performance, reduce latency, and minimize costs. In the context of LLM
    applications, caching is critical for optimizing response times and reducing API
    calls for repeated or similar queries. Effective caching strategies can
    dramatically improve user experience while significantly cutting operational costs.

    Caching in LLM applications targets:
    - Complete responses (exact match caching)
    - Embeddings (vector representations)
    - Prompt templates (reusable structures)
    - Intermediate results (multi-step workflows)
    - API responses (rate limit optimization)

Components:
    1. Cache Types
       - In-memory cache (fast, volatile)
       - Disk cache (persistent, slower)
       - Distributed cache (Redis, Memcached)
       - Database cache (structured, queryable)
       - CDN cache (geographic distribution)

    2. Cache Keys
       - Exact match (full prompt hash)
       - Semantic match (embedding similarity)
       - Parametric keys (template + parameters)
       - Hierarchical keys (nested cache structures)
       - Time-based keys (temporal validity)

    3. Cache Policies
       - TTL (Time To Live) - expire after time
       - LRU (Least Recently Used) - evict old entries
       - LFU (Least Frequently Used) - evict unpopular
       - Size-based - limit total cache size
       - Custom policies - application-specific

    4. Invalidation Strategies
       - Time-based expiration
       - Event-triggered invalidation
       - Manual invalidation
       - Cascade invalidation (dependencies)
       - Version-based invalidation

Use Cases:
    1. Repeated Queries
       - FAQ responses
       - Common questions
       - Popular content
       - Standard procedures
       - Frequently accessed data

    2. Expensive Operations
       - Complex reasoning chains
       - Multi-step workflows
       - Large context processing
       - Batch operations
       - Resource-intensive computations

    3. Rate Limit Optimization
       - Reduce API calls
       - Stay within quotas
       - Handle traffic spikes
       - Cost optimization
       - Performance consistency

    4. Embeddings and Vectors
       - Document embeddings
       - Query embeddings
       - Semantic search optimization
       - Similarity computations
       - Vector database queries

    5. Multi-User Applications
       - Shared responses
       - Common workflows
       - Popular queries
       - Template responses
       - Collaborative features

LangChain Implementation:
    LangChain supports caching through:
    - InMemoryCache for simple caching
    - SQLiteCache for persistent caching
    - Custom cache implementations
    - Semantic similarity caching
    - Embedding caching

Key Features:
    1. Multiple Cache Layers
       - L1: In-memory (fastest)
       - L2: Local disk (persistent)
       - L3: Distributed cache (shared)
       - L4: Database (long-term)

    2. Intelligent Matching
       - Exact match for identical queries
       - Semantic match for similar queries
       - Fuzzy match with thresholds
       - Parametric templates

    3. Performance Optimization
       - Fast lookup (< 1ms for in-memory)
       - Efficient serialization
       - Compression for large entries
       - Lazy loading

    4. Observability
       - Hit rate monitoring
       - Miss analysis
       - Performance metrics
       - Cost savings tracking

Best Practices:
    1. Cache Appropriately
       - Cache stable, reusable results
       - Don't cache rapidly changing data
       - Consider data sensitivity
       - Balance freshness vs performance

    2. Set Proper TTLs
       - Shorter for dynamic content
       - Longer for stable content
       - Consider update frequency
       - Account for user expectations

    3. Monitor Performance
       - Track hit rates (target > 70%)
       - Measure latency improvements
       - Calculate cost savings
       - Identify cache misses patterns

    4. Handle Cache Misses
       - Graceful fallback to computation
       - Transparent to users
       - Background cache warming
       - Predictive pre-caching

Trade-offs:
    Advantages:
    - Dramatically improved response times
    - Significant cost reduction
    - Better user experience
    - Reduced API rate limiting
    - Lower resource consumption
    - Improved scalability

    Disadvantages:
    - Stale data risk
    - Memory/storage overhead
    - Cache consistency challenges
    - Complexity in invalidation
    - Initial cache warming time
    - Debugging complexity

Production Considerations:
    1. Cache Size Management
       - Monitor memory usage
       - Implement eviction policies
       - Set size limits
       - Regular cleanup
       - Compression strategies

    2. Cache Consistency
       - Version tracking
       - Update propagation
       - Invalidation timing
       - Distributed consistency
       - Race condition handling

    3. Security
       - Access control for cached data
       - Encryption for sensitive content
       - Cache poisoning prevention
       - User isolation in multi-tenant
       - Audit trails

    4. Monitoring
       - Hit/miss rates
       - Cache size trends
       - Eviction frequency
       - Cost savings
       - Performance impact

    5. Disaster Recovery
       - Cache rebuild procedures
       - Backup strategies
       - Failover mechanisms
       - Data loss handling
       - Recovery time objectives
"""

Purpose: This module implements the Caching Patterns pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import collections.OrderedDict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import enum.Enum
import json
import langchain_core.caches.InMemoryCache
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – collections.OrderedDict: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.caches.InMemoryCache: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain.globals.set_llm_cache
from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from hashlib

Purpose: Additional dependencies for specialized functionality.

Classes

Class: CachePolicy

Lines 215-220

Inherits from: Enum

Purpose:

Cache eviction policies

Class: CacheEntry

Lines 224-243

Purpose:

Represents a cached entry

Methods:

is_expired(self)

Lines 233-238

Purpose:

Check if entry is expired

Parameters:

Returns: bool

touch(self)

Lines 240-243

Purpose:

Update access information

Parameters:

Class: CacheStats

Lines 247-264

Purpose:

Cache performance statistics

Methods:

hit_rate(self)

Lines 255-259

Decorators: property

Purpose:

Calculate cache hit rate

Parameters:

Returns: float

miss_rate(self)

Lines 262-264

Decorators: property

Purpose:

Calculate cache miss rate

Parameters:

Returns: float

Class: SimpleCache

Lines 267-416

Purpose:

Simple in-memory cache with multiple eviction policies.

This cache supports:
1. Multiple eviction policies (LRU, LFU, FIFO, TTL)
2. Size limits
3. TTL per entry
4. Performance statistics

Methods:

__init__(self, max_size, policy, default_ttl)

Lines 278-296

Purpose:

Initialize cache.

Args:
    max_size: Maximum number of entries
    policy: Eviction policy
    default_ttl: Default TTL in seconds

Parameters: – max_size: Max size – policy: Policy – default_ttl: Default ttl

_make_key(self, key)

Lines 298-302

Purpose:

Create cache key from input

Parameters: – key: Key

Returns: str

_evict(self)

Lines 304-320

Purpose:

Evict entries based on policy

Parameters:

_remove_expired(self)

Lines 322-329

Purpose:

Remove expired entries

Parameters:

get(self, key)

Lines 331-365

Purpose:

Get value from cache.

Args:
    key: Cache key

Returns:
    Cached value or None

Parameters: – key: Key

Returns: <ast.Subscript object at 0x000001A06D62AED0>

set(self, key, value, ttl)

Lines 367-394

Purpose:

Set value in cache.

Args:
    key: Cache key
    value: Value to cache
    ttl: Optional TTL in seconds

Parameters: – key: Key – value: Value – ttl: Ttl

clear(self)

Lines 396-399

Purpose:

Clear all cache entries

Parameters:

size(self)

Lines 401-403

Purpose:

Get current cache size

Parameters:

Returns: int

get_stats(self)

Lines 405-416

Purpose:

Get cache statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D74A3D0>

Class: SemanticCache

Lines 419-540

Purpose:

Semantic cache using embeddings for similarity matching.

This cache finds similar queries using vector similarity,
allowing cache hits for semantically similar but not identical queries.

Methods:

__init__(self, similarity_threshold, max_size)

Lines 427-443

Purpose:

Initialize semantic cache.

Args:
    similarity_threshold: Minimum similarity for cache hit
    max_size: Maximum cache entries

Parameters: – similarity_threshold: Similarity threshold – max_size: Max size

_compute_similarity(self, query1, query2)

Lines 445-471

Purpose:

Compute semantic similarity between queries.

Args:
    query1: First query
    query2: Second query

Returns:
    Similarity score (0-1)

Parameters: – query1: Input query or question – query2: Input query or question

Returns: float

get(self, query)

Lines 473-502

Purpose:

Get value for semantically similar query.

Args:
    query: Query string

Returns:
    Cached value or None

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D6A1F50>

set(self, query, value)

Lines 504-522

Purpose:

Cache value for query.

Args:
    query: Query string
    value: Value to cache

Parameters: – query: Input query or question – value: Value

clear(self)

Lines 524-527

Purpose:

Clear cache

Parameters:

get_stats(self)

Lines 529-540

Purpose:

Get cache statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7FC110>

Class: CachedLLMAgent

Lines 543-599

Purpose:

LLM agent with caching support.

This agent automatically caches responses and retrieves them
for repeated or similar queries.

Methods:

__init__(self, cache, model)

Lines 551-564

Purpose:

Initialize cached agent.

Args:
    cache: Cache instance to use
    model: LLM model name

Parameters: – cache: Cache – model: Language model instance

generate(self, prompt)

Lines 566-599

Purpose:

Generate response with caching.

Args:
    prompt: Input prompt

Returns:
    Dictionary with response and cache info

Parameters: – prompt: Prompt template or text

Returns: <ast.Subscript object at 0x000001A06D7C1A90>

Functions

demonstrate_caching_patterns()

Lines 602-892

Purpose:

Demonstrate caching patterns

Key Concepts Summary

Pattern Overview

The Caching Patterns pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

  1. Agent Class: Main implementation of the pattern logic
  2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
  3. Tool Integration: Optional tools for extended functionality
  4. Memory Management: State handling across interactions
  5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

Best Practices Demonstrated

1. Environment Configuration

  • ✅ Use .env files for API keys
  • ✅ Never hardcode credentials
  • ✅ Validate environment variables on startup

2. Error Handling

  • ✅ Graceful degradation on failures
  • ✅ Informative error messages
  • ✅ Logging for debugging

3. Prompt Engineering

  • ✅ Clear, specific instructions
  • ✅ Structured output formats
  • ✅ Few-shot examples when beneficial

4. Code Organization

  • ✅ Separation of concerns (class-based design)
  • ✅ Type hints for clarity
  • ✅ Docstrings for documentation

5. Performance Optimization

  • ✅ Appropriate temperature settings
  • ✅ Token budget management
  • ✅ Caching where applicable

LangChain Components Used

  • ChatOpenAI: OpenAI model interface
  • Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

Performance Considerations

Computational Cost

  • LLM Calls: Primary cost factor (based on tokens)
  • Latency: Depends on model and input size
  • Memory: Minimal for most patterns

Optimization Strategies

  1. Prompt Optimization: Minimize tokens while maintaining clarity
  2. Caching: Cache repeated queries/responses
  3. Model Selection: Use smaller models when appropriate
  4. Batch Processing: Group similar requests
  5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

  • GPT-3.5-turbo: ~$0.002 per request
  • GPT-4: ~$0.03 per request
  • GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

  1. Monitoring: Add logging, metrics, tracing
  2. Validation: Input/output schema validation
  3. Rate Limiting: Implement request throttling
  4. Caching: Redis or in-memory cache for responses
  5. Async Support: Non-blocking operations for scalability

Feature Extensions

  1. Multi-Model Support: Switch between providers
  2. Fallback Models: Graceful degradation strategy
  3. Custom Tools: Domain-specific function calling
  4. Advanced Memory: Vector stores for long-term memory
  5. Evaluation: Automated testing and benchmarking

Code Quality

  1. Testing: Unit tests, integration tests
  2. Documentation: API docs, usage examples
  3. Type Safety: Stricter type hints
  4. Configuration: Externalized config files

Related Patterns

  • Pattern 088
  • Pattern 090

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

Summary

The Caching Patterns pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

File: 089_caching_patterns.py
Total Lines: 897
Pattern Category: AI Agent Design Pattern #089

Generated from actual implementation code
For educational and reference purposes

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts