Caching

Caching Strategies for LLM Applications

Why Cache?
LLM calls are expensive and slow. Caching reduces latency, costs, and API load.

What to Cache

Semantic Caching
Cache by query meaning, not exact match:

class SemanticCache:
    def __init__(self, vector_store, threshold=0.95):
        self.vector_store = vector_store
        self.threshold = threshold

    def get(self, query):
        query_embedding = embed(query)
        results = self.vector_store.search(query_embedding, k=1)

        if results and results[0].score > self.threshold:
            return results[0].cached_response
        return None

    def set(self, query, response):
        query_embedding = embed(query)
        self.vector_store.add(query_embedding, {"response": response})

Embedding Caching

@lru_cache(maxsize=10000)
def cached_embed(text_hash):
    return embedding_model.embed(text)

def embed_with_cache(text):
    text_hash = hash(text)
    return cached_embed(text_hash)

Response Caching

import redis

cache = redis.Redis()

def cached_llm_call(prompt, model, ttl=3600):
    cache_key = f"llm:{model}:{hash(prompt)}"

    cached = cache.get(cache_key)
    if cached:
        return json.loads(cached)

    response = llm.generate(prompt, model=model)
    cache.setex(cache_key, ttl, json.dumps(response))
    return response

Cache Strategies
| Strategy | Description | Use Case |
|----------|-------------|----------|
| Cache-aside | App manages cache | General purpose |
| Write-through | Write cache + DB | Consistency critical |
| Write-behind | Write cache, async DB | High write volume |
| TTL-based | Expire after time | Time-sensitive data |

Cache Invalidation

def invalidate_on_update(document_id):
    # Invalidate all cached queries mentioning this doc
    pattern = f"rag:{document_id}:*"
    keys = cache.keys(pattern)
    cache.delete(*keys)

Redis Setup

import redis

# Connection pool
pool = redis.ConnectionPool(
    host="localhost",
    port=6379,
    max_connections=20
)
cache = redis.Redis(connection_pool=pool)

# With TTL and tags
def cache_with_metadata(key, value, ttl=3600, tags=None):
    cache.setex(key, ttl, value)
    for tag in tags or []:
        cache.sadd(f"tag:{tag}", key)

Best Practices
- Use semantic caching for similar queries
- Set appropriate TTLs for freshness
- Monitor cache hit rates
- Consider cache warming for common queries