"""
Pattern 166: Prompt Caching & Reuse
Description:
The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
similarity to match queries with cached results and manages cache invalidation
and updates.
Components:
1. Cache Storage: Stores prompts and results
2. Similarity Matcher: Finds semantically similar cached prompts
3. Cache Manager: Handles cache lifecycle
4. Invalidation Strategy: Manages cache freshness
5. Hit/Miss Tracker: Monitors cache performance
6. Compression: Optimizes cache storage
Use Cases:
- Repeated query handling
- Cost optimization
- Latency reduction
- Common question answering
- Template-based generation
- Batch processing optimization
Benefits:
- Reduced API costs
- Lower latency
- Better resource utilization
- Consistent responses
- Offline capability
Trade-offs:
- Storage overhead
- Cache invalidation complexity
- Stale data risk
- Memory usage
- Similarity matching overhead
LangChain Implementation:
Combines semantic similarity search with LangChain caching mechanisms.
Uses embeddings for matching similar queries efficiently.
"""
import os
import json
import hashlib
import time
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from collections import defaultdict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
load_dotenv()
@dataclass
class CacheEntry:
"""Represents a cached prompt and response"""
key: str
prompt: str
response: str
timestamp: datetime
hit_count: int = 0
embedding: Optional[List[float]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
ttl: Optional[int] = None # Time-to-live in seconds
def is_expired(self) -> bool:
"""Check if cache entry has expired"""
if self.ttl is None:
return False
age = (datetime.now() - self.timestamp).total_seconds()
return age > self.ttl
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for serialization"""
data = asdict(self)
data['timestamp'] = self.timestamp.isoformat()
return data
@staticmethod
def from_dict(data: Dict[str, Any]) -> 'CacheEntry':
"""Create from dictionary"""
data['timestamp'] = datetime.fromisoformat(data['timestamp'])
return CacheEntry(**data)
class PromptCacheManager:
"""Manages prompt caching and reuse with semantic similarity matching"""
def __init__(self,
similarity_threshold: float = 0.85,
max_cache_size: int = 1000,
default_ttl: Optional[int] = None):
"""
Initialize prompt cache manager
Args:
similarity_threshold: Minimum similarity score for cache hit (0-1)
max_cache_size: Maximum number of entries to cache
default_ttl: Default time-to-live in seconds
"""
self.similarity_threshold = similarity_threshold
self.max_cache_size = max_cache_size
self.default_ttl = default_ttl
self.cache: Dict[str, CacheEntry] = {}
self.embeddings = OpenAIEmbeddings()
self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# Statistics
self.stats = {
"total_requests": 0,
"cache_hits": 0,
"cache_misses": 0,
"total_cost_saved": 0.0,
"total_time_saved": 0.0
}
def get_or_generate(self, prompt: str,
force_refresh: bool = False,
ttl: Optional[int] = None) -> Tuple[str, bool]:
"""
Get cached response or generate new one
Args:
prompt: The prompt to process
force_refresh: Skip cache and generate new response
ttl: Time-to-live for this entry
Returns:
Tuple of (response, was_cached)
"""
self.stats["total_requests"] += 1
if not force_refresh:
# Try to find in cache
cached_response = self._find_similar_cached(prompt)
if cached_response:
self.stats["cache_hits"] += 1
self.stats["total_cost_saved"] += 0.002 # Estimated savings
self.stats["total_time_saved"] += 0.5 # Estimated time saved
return cached_response, True
# Cache miss - generate new response
self.stats["cache_misses"] += 1
response = self._generate_response(prompt)
# Cache the result
self._add_to_cache(prompt, response, ttl or self.default_ttl)
return response, False
def batch_get_or_generate(self, prompts: List[str]) -> List[Tuple[str, bool]]:
"""Process multiple prompts efficiently"""
results = []
for prompt in prompts:
result, cached = self.get_or_generate(prompt)
results.append((result, cached))
return results
def invalidate(self, pattern: Optional[str] = None):
"""Invalidate cache entries matching pattern"""
if pattern is None:
# Clear all
self.cache.clear()
else:
# Remove matching entries
keys_to_remove = [
key for key, entry in self.cache.items()
if pattern.lower() in entry.prompt.lower()
]
for key in keys_to_remove:
del self.cache[key]
def _find_similar_cached(self, prompt: str) -> Optional[str]:
"""Find semantically similar cached prompt"""
if not self.cache:
return None
# Clean expired entries
self._clean_expired()
# Get embedding for query prompt
query_embedding = self.embeddings.embed_query(prompt)
# Find most similar cached entry
best_match = None
best_similarity = 0.0
for entry in self.cache.values():
if entry.embedding is None:
continue
# Calculate cosine similarity
similarity = self._cosine_similarity(query_embedding, entry.embedding)
if similarity > best_similarity and similarity >= self.similarity_threshold:
best_similarity = similarity
best_match = entry
if best_match:
best_match.hit_count += 1
return best_match.response
return None
def _generate_response(self, prompt: str) -> str:
"""Generate new response using LLM"""
chain = ChatPromptTemplate.from_messages([
("user", "{prompt}")
]) | self.llm | StrOutputParser()
response = chain.invoke({"prompt": prompt})
return response
def _add_to_cache(self, prompt: str, response: str, ttl: Optional[int] = None):
"""Add entry to cache"""
# Generate cache key
key = self._generate_key(prompt)
# Get embedding
embedding = self.embeddings.embed_query(prompt)
# Create entry
entry = CacheEntry(
key=key,
prompt=prompt,
response=response,
timestamp=datetime.now(),
embedding=embedding,
ttl=ttl
)
# Check cache size
if len(self.cache) >= self.max_cache_size:
self._evict_entry()
# Add to cache
self.cache[key] = entry
def _generate_key(self, prompt: str) -> str:
"""Generate cache key from prompt"""
return hashlib.sha256(prompt.encode()).hexdigest()
def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
"""Calculate cosine similarity between two vectors"""
dot_product = sum(a * b for a, b in zip(vec1, vec2))
magnitude1 = sum(a * a for a in vec1) ** 0.5
magnitude2 = sum(b * b for b in vec2) ** 0.5
if magnitude1 == 0 or magnitude2 == 0:
return 0.0
return dot_product / (magnitude1 * magnitude2)
def _clean_expired(self):
"""Remove expired entries"""
keys_to_remove = [
key for key, entry in self.cache.items()
if entry.is_expired()
]
for key in keys_to_remove:
del self.cache[key]
def _evict_entry(self):
"""Evict least recently used entry"""
if not self.cache:
return
# Find entry with lowest hit count and oldest timestamp
lru_key = min(
self.cache.keys(),
key=lambda k: (self.cache[k].hit_count, self.cache[k].timestamp)
)
del self.cache[lru_key]
def get_stats(self) -> Dict[str, Any]:
"""Get cache statistics"""
total = self.stats["total_requests"]
hit_rate = self.stats["cache_hits"] / total if total > 0 else 0
return {
**self.stats,
"cache_hit_rate": hit_rate,
"cache_size": len(self.cache),
"avg_hit_count": sum(e.hit_count for e in self.cache.values()) / len(self.cache)
if self.cache else 0
}
def get_popular_queries(self, top_n: int = 10) -> List[Dict[str, Any]]:
"""Get most frequently accessed cached queries"""
sorted_entries = sorted(
self.cache.values(),
key=lambda e: e.hit_count,
reverse=True
)[:top_n]
return [
{
"prompt": entry.prompt,
"hit_count": entry.hit_count,
"age_seconds": (datetime.now() - entry.timestamp).total_seconds()
}
for entry in sorted_entries
]
def save_cache(self, filepath: str):
"""Save cache to file"""
cache_data = {
key: entry.to_dict()
for key, entry in self.cache.items()
}
with open(filepath, 'w') as f:
json.dump(cache_data, f, indent=2)
def load_cache(self, filepath: str):
"""Load cache from file"""
try:
with open(filepath, 'r') as f:
cache_data = json.load(f)
self.cache = {
key: CacheEntry.from_dict(data)
for key, data in cache_data.items()
}
except FileNotFoundError:
print(f"Cache file {filepath} not found")
except Exception as e:
print(f"Error loading cache: {e}")
def demonstrate_prompt_caching():
"""Demonstrate prompt caching and reuse"""
print("=" * 80)
print("PROMPT CACHING & REUSE PATTERN DEMONSTRATION")
print("=" * 80)
cache_manager = PromptCacheManager(
similarity_threshold=0.85,
max_cache_size=100,
default_ttl=3600 # 1 hour
)
# Example 1: Basic caching
print("\n" + "=" * 80)
print("Example 1: Basic Caching")
print("=" * 80)
prompts = [
"What is the capital of France?",
"Tell me about the capital of France", # Similar query
"What is the capital of Germany?",
]
print("\nProcessing prompts with caching:")
for i, prompt in enumerate(prompts, 1):
print(f"\n{i}. Query: {prompt}")
response, was_cached = cache_manager.get_or_generate(prompt)
status = "CACHE HIT" if was_cached else "CACHE MISS"
print(f" Status: {status}")
print(f" Response: {response[:100]}...")
# Example 2: Similar query matching
print("\n" + "=" * 80)
print("Example 2: Semantic Similarity Matching")
print("=" * 80)
original = "Explain quantum computing"
similar = "Can you describe quantum computing?"
print(f"\nOriginal query: {original}")
response1, cached1 = cache_manager.get_or_generate(original)
print(f"Status: {'CACHED' if cached1 else 'GENERATED'}")
print(f"Response: {response1[:100]}...")
print(f"\nSimilar query: {similar}")
response2, cached2 = cache_manager.get_or_generate(similar)
print(f"Status: {'CACHED' if cached2 else 'GENERATED'}")
print(f"Match found: {cached2}")
if cached2:
print("✓ Similar query matched with cached result!")
# Example 3: Batch processing
print("\n" + "=" * 80)
print("Example 3: Batch Processing with Cache")
print("=" * 80)
batch_prompts = [
"What is machine learning?",
"Define machine learning", # Similar
"What is deep learning?",
"Explain deep learning", # Similar
"What is machine learning?", # Exact duplicate
]
print(f"\nProcessing batch of {len(batch_prompts)} prompts:")
start_time = time.time()
results = cache_manager.batch_get_or_generate(batch_prompts)
elapsed = time.time() - start_time
hits = sum(1 for _, cached in results if cached)
misses = len(results) - hits
print(f"\nResults:")
print(f" Total queries: {len(results)}")
print(f" Cache hits: {hits}")
print(f" Cache misses: {misses}")
print(f" Hit rate: {hits/len(results):.1%}")
print(f" Processing time: {elapsed:.2f}s")
# Example 4: Cache statistics
print("\n" + "=" * 80)
print("Example 4: Cache Statistics")
print("=" * 80)
stats = cache_manager.get_stats()
print("\nCache Performance:")
print(f" Total requests: {stats['total_requests']}")
print(f" Cache hits: {stats['cache_hits']}")
print(f" Cache misses: {stats['cache_misses']}")
print(f" Hit rate: {stats['cache_hit_rate']:.1%}")
print(f" Cache size: {stats['cache_size']} entries")
print(f" Estimated cost saved: ${stats['total_cost_saved']:.4f}")
print(f" Estimated time saved: {stats['total_time_saved']:.2f}s")
# Example 5: Popular queries
print("\n" + "=" * 80)
print("Example 5: Popular Queries Analysis")
print("=" * 80)
popular = cache_manager.get_popular_queries(top_n=5)
print("\nMost frequently accessed queries:")
for i, query_info in enumerate(popular, 1):
age_min = query_info['age_seconds'] / 60
print(f"\n{i}. Hit count: {query_info['hit_count']}")
print(f" Age: {age_min:.1f} minutes")
print(f" Query: {query_info['prompt'][:60]}...")
# Example 6: TTL and expiration
print("\n" + "=" * 80)
print("Example 6: Time-to-Live (TTL) Management")
print("=" * 80)
# Add entry with short TTL
short_ttl_prompt = "What's the current time?"
response, cached = cache_manager.get_or_generate(short_ttl_prompt, ttl=2)
print(f"\nAdded query with 2 second TTL: {short_ttl_prompt}")
print(f"Response: {response[:80]}...")
print("\nImmediate retry (should hit cache):")
response, cached = cache_manager.get_or_generate(short_ttl_prompt)
print(f"Cached: {cached}")
print("\nWaiting 3 seconds for expiration...")
time.sleep(3)
print("Retry after expiration (should miss cache):")
response, cached = cache_manager.get_or_generate(short_ttl_prompt)
print(f"Cached: {cached}")
# Example 7: Cache invalidation
print("\n" + "=" * 80)
print("Example 7: Cache Invalidation")
print("=" * 80)
print(f"\nCache size before invalidation: {len(cache_manager.cache)}")
# Invalidate specific pattern
cache_manager.invalidate(pattern="capital")
print(f"Cache size after invalidating 'capital' queries: {len(cache_manager.cache)}")
# Test that invalidated query is regenerated
test_prompt = "What is the capital of France?"
response, cached = cache_manager.get_or_generate(test_prompt)
print(f"\nRetrying invalidated query: {test_prompt}")
print(f"Cached: {cached} (should be False)")
# Final statistics
print("\n" + "=" * 80)
print("Final Statistics")
print("=" * 80)
final_stats = cache_manager.get_stats()
print(f"\nOverall Performance:")
print(f" Total requests processed: {final_stats['total_requests']}")
print(f" Overall hit rate: {final_stats['cache_hit_rate']:.1%}")
print(f" Final cache size: {final_stats['cache_size']} entries")
print(f" Total cost saved: ${final_stats['total_cost_saved']:.4f}")
print(f" Total time saved: {final_stats['total_time_saved']:.2f}s")
# Summary
print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)
print("""
The Prompt Caching & Reuse pattern enables:
✓ Semantic similarity matching for cache hits
✓ Significant cost reduction through reuse
✓ Lower latency for repeated queries
✓ TTL-based cache expiration
✓ Cache invalidation strategies
✓ Performance monitoring and analytics
✓ Batch processing optimization
This pattern is valuable for:
- High-traffic applications with repeated queries
- Cost-sensitive deployments
- Latency-critical systems
- Common question answering systems
- Template-based generation
- Offline or low-connectivity scenarios
""")
if __name__ == "__main__":
demonstrate_prompt_caching()
Agentic Design Pattern: Prompt Caching And Reuse
Pattern 166: Prompt Caching & Reuse – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Prompt Caching & Reuse pattern implementation.
Pattern Description
Pattern 166: Prompt Caching & Reuse
Description: The Prompt Caching & Reuse pattern stores and retrieves prompts and their results to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic similarity to match queries with cached results and manages cache invalidation and updates.
Components: 1. Cache Storage: Stores prompts and results 2. Similarity Matcher: Finds semantically similar cached prompts 3. Cache Manager: Handles cache lifecycle 4. Invalidation Strategy: Manages cache freshness 5. Hit/Miss Tracker: Monitors cache performance 6. Compression: Optimizes cache storage
Use Cases: – Repeated query handling – Cost optimization – Latency reduction – Common question answering – Template-based generation – Batch processing optimization
Benefits: – Reduced API costs – Lower latency – Better resource utilization – Consistent responses – Offline capability
Trade-offs: – Storage overhead – Cache invalidation complexity – Stale data risk – Memory usage – Similarity matching overhead
LangChain Implementation: Combines semantic similarity search with LangChain caching mechanisms. Uses embeddings for matching similar queries efficiently.
File Statistics
- Total Lines: 527
- Classes: 2
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 166: Prompt Caching & Reuse
Description:
The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
similarity to match queries with cached results and manages cache invalidation
and updates.
Components:
1. Cache Storage: Stores prompts and results
2. Similarity Matcher: Finds semantically similar cached prompts
3. Cache Manager: Handles cache lifecycle
4. Invalidation Strategy: Manages cache freshness
5. Hit/Miss Tracker: Monitors cache performance
6. Compression: Optimizes cache storage
Use Cases:
- Repeated query handling
- Cost optimization
- Latency reduction
- Common question answering
- Template-based generation
- Batch processing optimization
Benefits:
- Reduced API costs
- Lower latency
- Better resource utilization
- Consistent responses
- Offline capability
Trade-offs:
- Storage overhead
- Cache invalidation complexity
- Stale data risk
- Memory usage
- Similarity matching overhead
LangChain Implementation:
Combines semantic similarity search with LangChain caching mechanisms.
Uses embeddings for matching similar queries efficiently.
"""
Purpose: This module implements the Prompt Caching & Reuse pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import collections.defaultdict
import dataclasses.asdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple
Purpose: Core Python functionality:
– collections.defaultdict: Specialized container datatypes
– dataclasses.asdict: Simplified class definitions with @dataclass
– 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
– 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.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
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: CacheEntry
Lines 63-91
Purpose:
Represents a cached prompt and response
Methods:
is_expired(self)
Lines 74-79
Purpose:
Check if cache entry has expired
Parameters:
Returns: bool
to_dict(self)
Lines 81-85
Purpose:
Convert to dictionary for serialization
Parameters:
Returns: <ast.Subscript object at 0x000001A06D6AA950>
from_dict(data)
Lines 88-91
Decorators: staticmethod
Purpose:
Create from dictionary
Parameters:
– data: Data
Returns: CacheEntry
Class: PromptCacheManager
Lines 94-337
Purpose:
Manages prompt caching and reuse with semantic similarity matching
Methods:
__init__(self, similarity_threshold, max_cache_size, default_ttl)
Lines 97-124
Purpose:
Initialize prompt cache manager
Args:
similarity_threshold: Minimum similarity score for cache hit (0-1)
max_cache_size: Maximum number of entries to cache
default_ttl: Default time-to-live in seconds
Parameters:
– similarity_threshold: Similarity threshold
– max_cache_size: Max cache size
– default_ttl: Default ttl
get_or_generate(self, prompt, force_refresh, ttl)
Lines 126-158
Purpose:
Get cached response or generate new one
Args:
prompt: The prompt to process
force_refresh: Skip cache and generate new response
ttl: Time-to-live for this entry
Returns:
Tuple of (response, was_cached)
Parameters:
– prompt: Prompt template or text
– force_refresh: Force refresh
– ttl: Ttl
Returns: <ast.Subscript object at 0x000001A06D824D50>
batch_get_or_generate(self, prompts)
Lines 160-166
Purpose:
Process multiple prompts efficiently
Parameters:
– prompts: Prompt template or text
Returns: <ast.Subscript object at 0x000001A06D827310>
invalidate(self, pattern)
Lines 168-180
Purpose:
Invalidate cache entries matching pattern
Parameters:
– pattern: Pattern
_find_similar_cached(self, prompt)
Lines 182-212
Purpose:
Find semantically similar cached prompt
Parameters:
– prompt: Prompt template or text
Returns: <ast.Subscript object at 0x000001A06D6147D0>
_generate_response(self, prompt)
Lines 214-221
Purpose:
Generate new response using LLM
Parameters:
– prompt: Prompt template or text
Returns: str
_add_to_cache(self, prompt, response, ttl)
Lines 223-246
Purpose:
Add entry to cache
Parameters:
– prompt: Prompt template or text
– response: Response
– ttl: Ttl
_generate_key(self, prompt)
Lines 248-250
Purpose:
Generate cache key from prompt
Parameters:
– prompt: Prompt template or text
Returns: str
_cosine_similarity(self, vec1, vec2)
Lines 252-261
Purpose:
Calculate cosine similarity between two vectors
Parameters:
– vec1: Vec1
– vec2: Vec2
Returns: float
_clean_expired(self)
Lines 263-270
Purpose:
Remove expired entries
Parameters:
_evict_entry(self)
Lines 272-282
Purpose:
Evict least recently used entry
Parameters:
get_stats(self)
Lines 284-295
Purpose:
Get cache statistics
Parameters:
Returns: <ast.Subscript object at 0x000001A06D8656D0>
get_popular_queries(self, top_n)
Lines 297-312
Purpose:
Get most frequently accessed cached queries
Parameters:
– top_n: Top n
Returns: <ast.Subscript object at 0x000001A06D7B8510>
save_cache(self, filepath)
Lines 314-322
Purpose:
Save cache to file
Parameters:
– filepath: Filepath
load_cache(self, filepath)
Lines 324-337
Purpose:
Load cache from file
Parameters:
– filepath: Filepath
Functions
demonstrate_prompt_caching()
Lines 340-522
Purpose:
Demonstrate prompt caching and reuse
Key Concepts Summary
Pattern Overview
The Prompt Caching & Reuse 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
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- 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
.envfiles 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
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- 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
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 165
- Pattern 167
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 Prompt Caching & Reuse 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: 166_prompt_caching_&_reuse.py
Total Lines: 527
Pattern Category: AI Agent Design Pattern #166
Generated from actual implementation code
For educational and reference purposes
Leave a Reply