Chapter 2: Caching and Performance

Types of Caching

Cache Type	What's Cached	Benefit
Embedding	Query embeddings	Reduce API calls
Semantic	Similar query results	Skip retrieval
Response	Full RAG responses	Fastest

Embedding Cache

from functools import lru_cache
import hashlib

class CachedEmbeddings:
    def __init__(self, embeddings, cache_size=10000):
        self.embeddings = embeddings
        self._cache = {}
    
    def embed_query(self, text):
        key = hashlib.md5(text.encode()).hexdigest()
        if key not in self._cache:
            self._cache[key] = self.embeddings.embed_query(text)
        return self._cache[key]

Semantic Cache

Cache based on query similarity:

from langchain.cache import GPTCache
import gptcache

def get_similar_query(query):
    """Find cached result for similar query"""
    query_embedding = embed(query)
    
    # Search cache vector store
    results = cache_store.similarity_search(
        query_embedding, 
        threshold=0.95  # Very similar queries
    )
    
    if results:
        return results[0].metadata["response"]
    return None

Response Cache with Redis

import redis
import json

r = redis.Redis()
CACHE_TTL = 3600  # 1 hour

def cached_rag(query, rag_chain):
    # Check cache
    cached = r.get(f"rag:{query}")
    if cached:
        return json.loads(cached)
    
    # Generate and cache
    response = rag_chain.invoke(query)
    r.setex(f"rag:{query}", CACHE_TTL, json.dumps(response))
    
    return response

Cache Invalidation

def invalidate_on_update(document_id):
    """Invalidate caches when documents change"""
    # Clear related response caches
    r.delete_pattern(f"rag:*doc:{document_id}*")
    
    # Remove from semantic cache
    semantic_cache.delete(document_id)

Performance Comparison

Scenario	Latency	Cost
No cache	2-5s	$0.01/query
Embedding cache	1-3s	$0.005/query
Semantic cache hit	<100ms	~$0
Response cache hit	<50ms	~$0