aishastr.live

Agentic Design Pattern: Result Memoization

Written by

admin

in


    """
Pattern 143: Result Memoization

Description:
    The Result Memoization pattern caches computation results to avoid redundant
    processing, significantly improving performance for repeated operations. When
    an agent encounters the same input or query it has processed before, it returns
    the cached result immediately instead of recomputing. This pattern is essential
    for production AI systems where identical or similar queries occur frequently.

    Memoization provides dramatic performance improvements for deterministic operations,
    expensive computations, and frequently accessed data. It trades memory for speed,
    storing results with appropriate cache keys, expiration policies, and invalidation
    strategies. The pattern supports various caching strategies including LRU, TTL,
    and semantic similarity-based caching.

    This implementation provides comprehensive memoization capabilities including
    function-level caching, LLM response caching, semantic similarity caching,
    cache warming, invalidation strategies, and performance metrics. It handles
    both simple key-value caching and advanced semantic caching for AI operations.

Components:
    - Function Memoization: Cache function results by arguments
    - LLM Response Caching: Cache model outputs for queries
    - Semantic Caching: Cache based on semantic similarity
    - Cache Warming: Preload frequently accessed results
    - Invalidation Strategy: Remove stale cache entries
    - Performance Metrics: Track cache effectiveness

Use Cases:
    - Repeated queries in chatbots
    - Expensive computations
    - API call reduction
    - Database query caching
    - Embedding generation
    - Translation caching
    - Recommendation systems
    - Data processing pipelines

LangChain Implementation:
    This implementation uses:
    - Python functools.lru_cache for function memoization
    - Custom cache implementations with TTL
    - Semantic similarity-based caching
    - LangChain cache integration
    - Hash-based cache keys
    - Cache metrics and monitoring

Benefits:
    - Dramatically reduces latency
    - Lowers computational costs
    - Reduces API calls
    - Improves scalability
    - Provides consistent responses
    - Enables offline operation
    - Reduces load on services

Trade-offs:
    - Memory overhead for cache storage
    - Stale data if not invalidated properly
    - Cache consistency challenges
    - Cold start performance
    - Complex invalidation logic
    - Potential memory leaks

Production Considerations:
    - Set appropriate cache sizes
    - Implement TTL for freshness
    - Monitor cache hit rates
    - Use distributed caching for scale
    - Handle cache misses gracefully
    - Implement cache warming
    - Monitor memory usage
    - Use appropriate invalidation
    - Consider cache coherence
    - Profile cache effectiveness
    - Implement cache preloading
    - Handle cache failures
"""

import os
import time
import hashlib
import json
from typing import List, Dict, Any, Optional, Callable, Tuple
from datetime import datetime, timedelta
from functools import wraps, lru_cache
from collections import OrderedDict
from dataclasses import dataclass, field
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


@dataclass
class CacheEntry:
    """Cache entry with metadata."""
    key: str
    value: Any
    created_at: datetime
    last_accessed: datetime
    access_count: int
    ttl_seconds: Optional[float]
    
    def is_expired(self) -> bool:
        """Check if entry is expired."""
        if self.ttl_seconds is None:
            return False
        age = (datetime.now() - self.created_at).total_seconds()
        return age > self.ttl_seconds
    
    def access(self):
        """Record access."""
        self.last_accessed = datetime.now()
        self.access_count += 1


class LRUCache:
    """LRU (Least Recently Used) Cache implementation."""
    
    def __init__(self, max_size: int = 100, ttl_seconds: Optional[float] = None):
        """Initialize LRU cache."""
        self.max_size = max_size
        self.ttl_seconds = ttl_seconds
        self.cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self.hits = 0
        self.misses = 0
    
    def get(self, key: str) -> Optional[Any]:
        """Get value from cache."""
        if key not in self.cache:
            self.misses += 1
            return None
        
        entry = self.cache[key]
        
        # Check if expired
        if entry.is_expired():
            del self.cache[key]
            self.misses += 1
            return None
        
        # Move to end (most recently used)
        self.cache.move_to_end(key)
        entry.access()
        self.hits += 1
        return entry.value
    
    def set(self, key: str, value: Any):
        """Set value in cache."""
        # Remove if exists
        if key in self.cache:
            del self.cache[key]
        
        # Create entry
        entry = CacheEntry(
            key=key,
            value=value,
            created_at=datetime.now(),
            last_accessed=datetime.now(),
            access_count=0,
            ttl_seconds=self.ttl_seconds
        )
        
        self.cache[key] = entry
        
        # Evict oldest if over size
        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)
    
    def clear(self):
        """Clear cache."""
        self.cache.clear()
        self.hits = 0
        self.misses = 0
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics."""
        total = self.hits + self.misses
        hit_rate = self.hits / total if total > 0 else 0
        return {
            "size": len(self.cache),
            "max_size": self.max_size,
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": f"{hit_rate:.2%}",
            "total_requests": total
        }


class SemanticCache:
    """Cache based on semantic similarity of inputs."""
    
    def __init__(self, similarity_threshold: float = 0.9, max_size: int = 100):
        """Initialize semantic cache."""
        self.similarity_threshold = similarity_threshold
        self.max_size = max_size
        self.entries: List[Tuple[str, Any, str]] = []  # (key, value, original_text)
        self.hits = 0
        self.misses = 0
    
    def _compute_similarity(self, text1: str, text2: str) -> float:
        """Compute simple similarity (in production, use embeddings)."""
        # Simple word overlap similarity
        words1 = set(text1.lower().split())
        words2 = set(text2.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 from cache based on semantic similarity."""
        for key, value, original_text in self.entries:
            similarity = self._compute_similarity(query, original_text)
            if similarity >= self.similarity_threshold:
                self.hits += 1
                return value
        
        self.misses += 1
        return None
    
    def set(self, query: str, value: Any):
        """Set value in cache."""
        key = hashlib.md5(query.encode()).hexdigest()
        self.entries.append((key, value, query))
        
        # Evict oldest if over size
        if len(self.entries) > self.max_size:
            self.entries.pop(0)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics."""
        total = self.hits + self.misses
        hit_rate = self.hits / total if total > 0 else 0
        return {
            "size": len(self.entries),
            "max_size": self.max_size,
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": f"{hit_rate:.2%}",
            "similarity_threshold": self.similarity_threshold
        }


def memoize(cache: Optional[LRUCache] = None, ttl_seconds: Optional[float] = None):
    """Decorator for function memoization."""
    if cache is None:
        cache = LRUCache(ttl_seconds=ttl_seconds)
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Create cache key from arguments
            key_data = {
                "func": func.__name__,
                "args": args,
                "kwargs": kwargs
            }
            key = hashlib.md5(
                json.dumps(key_data, sort_keys=True, default=str).encode()
            ).hexdigest()
            
            # Check cache
            cached_result = cache.get(key)
            if cached_result is not None:
                return cached_result
            
            # Compute and cache
            result = func(*args, **kwargs)
            cache.set(key, result)
            return result
        
        wrapper.cache = cache
        return wrapper
    
    return decorator


class MemoizationAgent:
    """
    Agent that uses memoization for performance optimization.
    
    This agent caches results to avoid redundant computation,
    dramatically improving performance for repeated operations.
    """
    
    def __init__(self, temperature: float = 0.3):
        """Initialize the memoization agent."""
        self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        
        # Different caches for different purposes
        self.llm_cache = LRUCache(max_size=100, ttl_seconds=3600)
        self.computation_cache = LRUCache(max_size=50, ttl_seconds=None)
        self.semantic_cache = SemanticCache(similarity_threshold=0.85)
        
        # Create LLM chain (we'll cache its results)
        self.chain = (
            ChatPromptTemplate.from_template("{query}")
            | self.llm
            | StrOutputParser()
        )
    
    def query_llm_cached(self, query: str, use_semantic: bool = False) -> str:
        """Query LLM with caching."""
        # Create cache key
        cache_key = hashlib.md5(query.encode()).hexdigest()
        
        # Try semantic cache first if enabled
        if use_semantic:
            result = self.semantic_cache.get(query)
            if result:
                return f"[SEMANTIC CACHE HIT] {result}"
        
        # Try exact cache
        result = self.llm_cache.get(cache_key)
        if result:
            return f"[CACHE HIT] {result}"
        
        # Cache miss - query LLM
        result = self.chain.invoke({"query": query})
        
        # Store in caches
        self.llm_cache.set(cache_key, result)
        if use_semantic:
            self.semantic_cache.set(query, result)
        
        return f"[CACHE MISS] {result}"
    
    @memoize(ttl_seconds=300)
    def expensive_computation(self, value: int) -> int:
        """Simulate expensive computation with memoization."""
        time.sleep(0.5)  # Simulate expensive work
        return value ** 2
    
    def cache_warm(self, common_queries: List[str]):
        """Warm up cache with common queries."""
        print(f"Warming cache with {len(common_queries)} common queries...")
        for query in common_queries:
            result = self.chain.invoke({"query": query})
            cache_key = hashlib.md5(query.encode()).hexdigest()
            self.llm_cache.set(cache_key, result)
            self.semantic_cache.set(query, result)
        print("Cache warming complete")
    
    def invalidate_cache(self, pattern: Optional[str] = None):
        """Invalidate cache entries."""
        if pattern is None:
            # Clear all caches
            self.llm_cache.clear()
            self.computation_cache.clear()
            print("All caches cleared")
        else:
            print(f"Pattern-based invalidation not implemented for: {pattern}")
    
    def get_all_stats(self) -> Dict[str, Any]:
        """Get statistics for all caches."""
        return {
            "llm_cache": self.llm_cache.get_stats(),
            "computation_cache": self.computation_cache.get_stats(),
            "semantic_cache": self.semantic_cache.get_stats()
        }


def demonstrate_memoization():
    """Demonstrate the result memoization pattern."""
    print("=" * 80)
    print("Result Memoization Pattern Demonstration")
    print("=" * 80)
    
    agent = MemoizationAgent()
    
    # Demonstration 1: Basic Function Memoization
    print("\n" + "=" * 80)
    print("Demonstration 1: Basic Function Memoization")
    print("=" * 80)
    
    print("\nFirst call (cache miss, will take 0.5s):")
    start = time.time()
    result1 = agent.expensive_computation(10)
    elapsed1 = time.time() - start
    print(f"Result: {result1}, Time: {elapsed1:.3f}s")
    
    print("\nSecond call with same argument (cache hit, instant):")
    start = time.time()
    result2 = agent.expensive_computation(10)
    elapsed2 = time.time() - start
    print(f"Result: {result2}, Time: {elapsed2:.3f}s")
    
    print(f"\nSpeedup: {elapsed1/elapsed2:.1f}x faster")
    
    # Demonstration 2: LLM Response Caching
    print("\n" + "=" * 80)
    print("Demonstration 2: LLM Response Caching")
    print("=" * 80)
    
    query = "What is 2+2?"
    
    print(f"\nQuery: {query}")
    print("First call (cache miss):")
    start = time.time()
    response1 = agent.query_llm_cached(query)
    elapsed1 = time.time() - start
    print(f"Response: {response1[:100]}...")
    print(f"Time: {elapsed1:.3f}s")
    
    print("\nSecond call (cache hit):")
    start = time.time()
    response2 = agent.query_llm_cached(query)
    elapsed2 = time.time() - start
    print(f"Response: {response2[:100]}...")
    print(f"Time: {elapsed2:.3f}s")
    
    # Demonstration 3: Semantic Caching
    print("\n" + "=" * 80)
    print("Demonstration 3: Semantic Caching")
    print("=" * 80)
    
    query1 = "What is the capital of France?"
    query2 = "Tell me the capital city of France"  # Similar semantic meaning
    
    print(f"\nFirst query: {query1}")
    response1 = agent.query_llm_cached(query1, use_semantic=True)
    print(f"Response: {response1[:80]}...")
    
    print(f"\nSimilar query: {query2}")
    response2 = agent.query_llm_cached(query2, use_semantic=True)
    print(f"Response: {response2[:80]}...")
    
    if "[SEMANTIC CACHE HIT]" in response2:
        print("\n✓ Semantic cache successfully matched similar query!")
    
    # Demonstration 4: Cache Warming
    print("\n" + "=" * 80)
    print("Demonstration 4: Cache Warming")
    print("=" * 80)
    
    common_queries = [
        "What is the weather?",
        "Tell me a joke",
        "Hello, how are you?"
    ]
    
    agent.cache_warm(common_queries)
    
    print("\nQuerying warmed cache:")
    for query in common_queries[:2]:
        response = agent.query_llm_cached(query)
        print(f"  {query}: {response[:60]}...")
    
    # Demonstration 5: TTL Expiration
    print("\n" + "=" * 80)
    print("Demonstration 5: TTL (Time-To-Live) Expiration")
    print("=" * 80)
    
    # Create cache with short TTL
    short_ttl_cache = LRUCache(max_size=10, ttl_seconds=2)
    
    @memoize(cache=short_ttl_cache)
    def short_lived_computation(x):
        time.sleep(0.1)
        return x * 2
    
    print("\nComputation with 2-second TTL:")
    print("First call:")
    result1 = short_lived_computation(5)
    print(f"Result: {result1}")
    
    print("\nImmediate second call (within TTL):")
    result2 = short_lived_computation(5)
    print(f"Result: {result2} (cached)")
    
    print("\nWaiting 2.5 seconds for TTL expiration...")
    time.sleep(2.5)
    
    print("Third call (after TTL expiration):")
    result3 = short_lived_computation(5)
    print(f"Result: {result3} (recomputed)")
    
    # Demonstration 6: LRU Eviction
    print("\n" + "=" * 80)
    print("Demonstration 6: LRU Cache Eviction")
    print("=" * 80)
    
    small_cache = LRUCache(max_size=3)
    
    print("\nCache with max size 3:")
    print("Adding 5 items...")
    for i in range(5):
        small_cache.set(f"key_{i}", f"value_{i}")
        print(f"  Added key_{i}, cache size: {len(small_cache.cache)}")
    
    print(f"\nFinal cache size: {len(small_cache.cache)} (oldest evicted)")
    print(f"Remaining keys: {list(small_cache.cache.keys())}")
    
    # Demonstration 7: Cache Invalidation
    print("\n" + "=" * 80)
    print("Demonstration 7: Cache Invalidation")
    print("=" * 80)
    
    print("\nQuerying LLM to populate cache...")
    agent.query_llm_cached("Test query 1")
    agent.query_llm_cached("Test query 2")
    
    stats_before = agent.get_all_stats()
    print(f"LLM cache size before: {stats_before['llm_cache']['size']}")
    
    print("\nInvalidating all caches...")
    agent.invalidate_cache()
    
    stats_after = agent.get_all_stats()
    print(f"LLM cache size after: {stats_after['llm_cache']['size']}")
    
    # Demonstration 8: Cache Statistics
    print("\n" + "=" * 80)
    print("Demonstration 8: Cache Performance Statistics")
    print("=" * 80)
    
    # Generate some cache activity
    for i in range(10):
        agent.query_llm_cached(f"Query {i % 3}")  # Repeat some queries
    
    stats = agent.get_all_stats()
    
    print("\nLLM Cache Statistics:")
    for key, value in stats['llm_cache'].items():
        print(f"  {key}: {value}")
    
    print("\nSemantic Cache Statistics:")
    for key, value in stats['semantic_cache'].items():
        print(f"  {key}: {value}")
    
    # Summary
    print("\n" + "=" * 80)
    print("Summary: Result Memoization Pattern")
    print("=" * 80)
    print("""
The Result Memoization pattern dramatically improves performance through caching:

Key Features Demonstrated:
1. Function Memoization - Cache expensive computations
2. LLM Response Caching - Avoid redundant API calls
3. Semantic Caching - Match similar queries
4. Cache Warming - Preload common queries
5. TTL Expiration - Auto-invalidate stale data
6. LRU Eviction - Manage cache size
7. Cache Invalidation - Clear stale entries
8. Performance Metrics - Track cache effectiveness

Benefits:
- Dramatically reduces latency
- Lowers computational costs
- Reduces API calls
- Improves scalability
- Provides consistent responses
- Reduces load on services

Best Practices:
- Set appropriate cache sizes
- Implement TTL for freshness
- Monitor cache hit rates
- Use distributed caching for scale
- Handle cache misses gracefully
- Implement cache warming
- Monitor memory usage
- Use appropriate invalidation
- Consider cache coherence
- Profile cache effectiveness

Common Use Cases:
- Repeated queries in chatbots
- Expensive computations
- API call reduction
- Database query caching
- Embedding generation
- Translation caching
- Recommendation systems
- Data processing pipelines

Performance Considerations:
- Memory overhead for cache storage
- Balance cache size vs. hit rate
- Implement appropriate eviction policies
- Monitor for stale data
- Consider distributed caching
- Profile actual performance gains

This pattern is essential for production AI systems requiring high performance
and cost optimization through intelligent caching.
""")


if __name__ == "__main__":
    demonstrate_memoization()

Agentic Design Pattern: Result Memoization

Pattern 143: Result Memoization – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Result Memoization pattern implementation.

Pattern Description

Pattern 143: Result Memoization

Description: The Result Memoization pattern caches computation results to avoid redundant processing, significantly improving performance for repeated operations. When an agent encounters the same input or query it has processed before, it returns the cached result immediately instead of recomputing. This pattern is essential for production AI systems where identical or similar queries occur frequently.

Memoization provides dramatic performance improvements for deterministic operations,
expensive computations, and frequently accessed data. It trades memory for speed,
storing results with appropriate cache keys, expiration policies, and invalidation
strategies. The pattern supports various caching strategies including LRU, TTL,
and semantic similarity-based caching.

This implementation provides comprehensive memoization capabilities including
function-level caching, LLM response caching, semantic similarity caching,
cache warming, invalidation strategies, and performance metrics. It handles
both simple key-value caching and advanced semantic caching for AI operations.

Components: – Function Memoization: Cache function results by arguments – LLM Response Caching: Cache model outputs for queries – Semantic Caching: Cache based on semantic similarity – Cache Warming: Preload frequently accessed results – Invalidation Strategy: Remove stale cache entries – Performance Metrics: Track cache effectiveness

Use Cases: – Repeated queries in chatbots – Expensive computations – API call reduction – Database query caching – Embedding generation – Translation caching – Recommendation systems – Data processing pipelines

LangChain Implementation: This implementation uses: – Python functools.lru_cache for function memoization – Custom cache implementations with TTL – Semantic similarity-based caching – LangChain cache integration – Hash-based cache keys – Cache metrics and monitoring

Benefits: – Dramatically reduces latency – Lowers computational costs – Reduces API calls – Improves scalability – Provides consistent responses – Enables offline operation – Reduces load on services

Trade-offs: – Memory overhead for cache storage – Stale data if not invalidated properly – Cache consistency challenges – Cold start performance – Complex invalidation logic – Potential memory leaks

Production Considerations: – Set appropriate cache sizes – Implement TTL for freshness – Monitor cache hit rates – Use distributed caching for scale – Handle cache misses gracefully – Implement cache warming – Monitor memory usage – Use appropriate invalidation – Consider cache coherence – Profile cache effectiveness – Implement cache preloading – Handle cache failures

File Statistics

  • Total Lines: 603
  • Classes: 4
  • Functions: 2
  • Constants: 0

Module Documentation

"""
Pattern 143: Result Memoization

Description:
    The Result Memoization pattern caches computation results to avoid redundant
    processing, significantly improving performance for repeated operations. When
    an agent encounters the same input or query it has processed before, it returns
    the cached result immediately instead of recomputing. This pattern is essential
    for production AI systems where identical or similar queries occur frequently.

    Memoization provides dramatic performance improvements for deterministic operations,
    expensive computations, and frequently accessed data. It trades memory for speed,
    storing results with appropriate cache keys, expiration policies, and invalidation
    strategies. The pattern supports various caching strategies including LRU, TTL,
    and semantic similarity-based caching.

    This implementation provides comprehensive memoization capabilities including
    function-level caching, LLM response caching, semantic similarity caching,
    cache warming, invalidation strategies, and performance metrics. It handles
    both simple key-value caching and advanced semantic caching for AI operations.

Components:
    - Function Memoization: Cache function results by arguments
    - LLM Response Caching: Cache model outputs for queries
    - Semantic Caching: Cache based on semantic similarity
    - Cache Warming: Preload frequently accessed results
    - Invalidation Strategy: Remove stale cache entries
    - Performance Metrics: Track cache effectiveness

Use Cases:
    - Repeated queries in chatbots
    - Expensive computations
    - API call reduction
    - Database query caching
    - Embedding generation
    - Translation caching
    - Recommendation systems
    - Data processing pipelines

LangChain Implementation:
    This implementation uses:
    - Python functools.lru_cache for function memoization
    - Custom cache implementations with TTL
    - Semantic similarity-based caching
    - LangChain cache integration
    - Hash-based cache keys
    - Cache metrics and monitoring

Benefits:
    - Dramatically reduces latency
    - Lowers computational costs
    - Reduces API calls
    - Improves scalability
    - Provides consistent responses
    - Enables offline operation
    - Reduces load on services

Trade-offs:
    - Memory overhead for cache storage
    - Stale data if not invalidated properly
    - Cache consistency challenges
    - Cold start performance
    - Complex invalidation logic
    - Potential memory leaks

Production Considerations:
    - Set appropriate cache sizes
    - Implement TTL for freshness
    - Monitor cache hit rates
    - Use distributed caching for scale
    - Handle cache misses gracefully
    - Implement cache warming
    - Monitor memory usage
    - Use appropriate invalidation
    - Consider cache coherence
    - Profile cache effectiveness
    - Implement cache preloading
    - Handle cache failures
"""

Purpose: This module implements the Result Memoization 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 functools.lru_cache
import functools.wraps
import json
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
import typing.Tuple

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 – functools.lru_cache: Higher-order functions and operations on callable objects – functools.wraps: Higher-order functions and operations on callable objects – json: JSON serialization/deserialization – 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) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

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: CacheEntry

Lines 100-119

Purpose:

Cache entry with metadata.

Methods:

is_expired(self)

Lines 109-114

Purpose:

Check if entry is expired.

Parameters:

Returns: bool

access(self)

Lines 116-119

Purpose:

Record access.

Parameters:

Class: LRUCache

Lines 122-192

Purpose:

LRU (Least Recently Used) Cache implementation.

Methods:

__init__(self, max_size, ttl_seconds)

Lines 125-131

Purpose:

Initialize LRU cache.

Parameters: – max_size: Max size – ttl_seconds: Ttl seconds

get(self, key)

Lines 133-151

Purpose:

Get value from cache.

Parameters: – key: Key

Returns: <ast.Subscript object at 0x000001A06D811350>

set(self, key, value)

Lines 153-173

Purpose:

Set value in cache.

Parameters: – key: Key – value: Value

clear(self)

Lines 175-179

Purpose:

Clear cache.

Parameters:

get_stats(self)

Lines 181-192

Purpose:

Get cache statistics.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D164C10>

Class: SemanticCache

Lines 195-251

Purpose:

Cache based on semantic similarity of inputs.

Methods:

__init__(self, similarity_threshold, max_size)

Lines 198-204

Purpose:

Initialize semantic cache.

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

_compute_similarity(self, text1, text2)

Lines 206-218

Purpose:

Compute simple similarity (in production, use embeddings).

Parameters: – text1: Text1 – text2: Text2

Returns: float

get(self, query)

Lines 220-229

Purpose:

Get value from cache based on semantic similarity.

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D669D10>

set(self, query, value)

Lines 231-238

Purpose:

Set value in cache.

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

get_stats(self)

Lines 240-251

Purpose:

Get cache statistics.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D859510>

Class: MemoizationAgent

Lines 288-370

Purpose:

Agent that uses memoization for performance optimization.

This agent caches results to avoid redundant computation,
dramatically improving performance for repeated operations.

Methods:

__init__(self, temperature)

Lines 296-310

Purpose:

Initialize the memoization agent.

Parameters: – temperature: Sampling temperature (0-1)

query_llm_cached(self, query, use_semantic)

Lines 312-336

Purpose:

Query LLM with caching.

Parameters: – query: Input query or question – use_semantic: Use semantic

Returns: str

expensive_computation(self, value)

Lines 339-342

Decorators:

Purpose:

Simulate expensive computation with memoization.

Parameters: – value: Value

Returns: int

cache_warm(self, common_queries)

Lines 344-352

Purpose:

Warm up cache with common queries.

Parameters: – common_queries: Common queries

invalidate_cache(self, pattern)

Lines 354-362

Purpose:

Invalidate cache entries.

Parameters: – pattern: Pattern

get_all_stats(self)

Lines 364-370

Purpose:

Get statistics for all caches.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D605C10>

Functions

memoize(cache, ttl_seconds)

Lines 254-285

Purpose:

Decorator for function memoization.

Parameters: – cache: Cache – ttl_seconds: Ttl seconds

demonstrate_memoization()

Lines 373-598

Purpose:

Demonstrate the result memoization pattern.

Key Concepts Summary

Pattern Overview

The Result Memoization 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 142
  • Pattern 144

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 Result Memoization 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: 143_result_memoization.py
Total Lines: 603
Pattern Category: AI Agent Design Pattern #143

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