4. Caching Strategies

Overview

Cutting redundant API calls with caching strategies: exact match (prompt hash), semantic caching (embedding similarity), TTL strategies by volatility, and cache invalidation. Implementation with Redis, in-memory, and GPTCache.

Why Cache LLM Responses

LLM responses have traits that make them good caching candidates:

Traits of LLM requests:
1. High repetition: "What's the price?" gets asked 1000 times/day
2. Determinism (temperature=0): Same prompt → same answer
3. High cost per request: The cache complexity pays for itself
4. Variable latency: A cache makes p50 consistent

When NOT to cache:
- Queries that need real-time information (stock prices, weather)
- Conversations where each message depends on the specific history
- Creative outputs where variety is the point

The 3 Types of Caching

Type 1: EXACT MATCH CACHE
  "What are your business hours?" (exactly the same)
  → Prompt hash → answer in Redis
  → Hit rate: 30-40% typical

Type 2: SEMANTIC CACHE
  "What time do you open?" ~ "What are your business hours?"
  → Embeddings + cosine similarity → answer from a similar query
  → Extra hit rate: 10-20% more than exact match

Type 3: PROMPT CACHING (API-level)
  Reused system prompt
  → Token discount (50-90%) at the API
  → Doesn't avoid the request, but cuts the cost

Type 1: Exact Match Cache with Redis

import redis
import hashlib
import json
import time
from openai import OpenAI
from typing import Optional

client = OpenAI()

class ExactMatchCache:
    """
    Exact match cache for LLM responses.
    Uses Redis as the backend with a configurable TTL.
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        ttl_default: int = 3600,  # 1 hour default
        prefix: str = "llm_cache:"
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.ttl_default = ttl_default
        self.prefix = prefix
        
        # Cache stats
        self._hits = 0
        self._misses = 0
    
    def _cache_key(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> str:
        """Generates a deterministic cache key."""
        content = f"{model}|{temperature}|{prompt}"
        hash_hex = hashlib.sha256(content.encode()).hexdigest()
        return f"{self.prefix}{hash_hex}"
    
    def get(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> Optional[str]:
        """Looks in the cache. Returns the answer or None."""
        key = self._cache_key(prompt, model, temperature)
        cached = self.redis.get(key)
        
        if cached:
            self._hits += 1
            return json.loads(cached)
        
        self._misses += 1
        return None
    
    def set(
        self,
        prompt: str,
        response: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0,
        ttl: Optional[int] = None
    ) -> None:
        """Stores the answer in the cache."""
        key = self._cache_key(prompt, model, temperature)
        ttl_value = ttl if ttl is not None else self.ttl_default
        
        self.redis.setex(
            key,
            ttl_value,
            json.dumps(response)
        )
    
    def delete(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> bool:
        """Invalidates one cache entry."""
        key = self._cache_key(prompt, model, temperature)
        return bool(self.redis.delete(key))
    
    def invalidate_by_pattern(self, pattern: str) -> int:
        """
        Invalidates every entry matching a pattern.
        Useful for wiping a prompt's whole cache when it changes.
        """
        keys = list(self.redis.scan_iter(f"{self.prefix}*"))
        deleted = 0
        for key in keys:
            value = self.redis.get(key)
            if value and pattern in str(value):
                self.redis.delete(key)
                deleted += 1
        return deleted
    
    def stats(self) -> dict:
        """Cache usage statistics."""
        total = self._hits + self._misses
        hit_rate = self._hits / total if total > 0 else 0.0
        
        return {
            "hits": self._hits,
            "misses": self._misses,
            "total_requests": total,
            "hit_rate": f"{hit_rate:.1%}",
            "redis_keys": self.redis.dbsize()
        }
    
    def call_with_cache(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        ttl: Optional[int] = None,
        **kwargs
    ) -> dict:
        """
        Runs the LLM with the cache built in.
        On a cache hit: returns immediately.
        If not: calls the API and caches the result.
        """
        # Try the cache first
        cached = self.get(prompt, model)
        if cached:
            return {
                "output": cached,
                "source": "cache",
                "latency_ms": 1  # < 1ms from the cache
            }
        
        # Cache miss: call the API
        start = time.time()
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,  # Important: temperature=0 for reproducibility
            **kwargs
        )
        
        output = response.choices[0].message.content
        latency_ms = (time.time() - start) * 1000
        
        # Store in the cache
        self.set(prompt, output, model, ttl=ttl)
        
        return {
            "output": output,
            "source": "api",
            "latency_ms": latency_ms,
            "tokens": response.usage.total_tokens
        }


# Usage demo:
cache = ExactMatchCache(ttl_default=3600)

# First call (cache miss)
r1 = cache.call_with_cache("What is the capital of Mexico?")
print(f"1st call: {r1['source']} - {r1['latency_ms']:.0f}ms")
# 1st call: api - 743ms

# Second call (cache hit)
r2 = cache.call_with_cache("What is the capital of Mexico?")
print(f"2nd call: {r2['source']} - {r2['latency_ms']:.0f}ms")
# 2nd call: cache - 1ms

print(cache.stats())
# {'hits': 1, 'misses': 1, 'hit_rate': '50.0%', ...}

Type 2: Semantic Cache

The semantic cache catches queries that are similar but not identical:

import numpy as np
from openai import OpenAI
import redis
import json

client = OpenAI()

class SemanticCache:
    """
    Semantic cache that also finds answers for similar queries.
    
    Uses embeddings to compare similarity between queries.
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        threshold: float = 0.92,  # Minimum similarity to count as a match
        embedding_model: str = "text-embedding-3-small",
        ttl: int = 86400,          # 24 hours
        prefix: str = "sem_cache:"
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.threshold = threshold
        self.embedding_model = embedding_model
        self.ttl = ttl
        self.prefix = prefix
    
    def _get_embedding(self, text: str) -> list[float]:
        """Gets the embedding of a text."""
        response = client.embeddings.create(
            input=text,
            model=self.embedding_model
        )
        return response.data[0].embedding
    
    def _cosine_similarity(self, v1: list[float], v2: list[float]) -> float:
        """Computes the cosine similarity between two vectors."""
        a = np.array(v1)
        b = np.array(v2)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    def get(self, query: str) -> Optional[dict]:
        """
        Looks for an answer to the query using semantic similarity.
        
        Returns: dict with {response, similarity, cached_query} or None
        """
        query_embedding = self._get_embedding(query)
        
        # Search every embedding in Redis
        keys = list(self.redis.scan_iter(f"{self.prefix}emb:*"))
        
        best_match = None
        best_similarity = 0.0
        
        for key in keys:
            cached_emb_json = self.redis.get(key)
            if not cached_emb_json:
                continue
            
            cached_emb = json.loads(cached_emb_json)
            similarity = self._cosine_similarity(query_embedding, cached_emb["embedding"])
            
            if similarity > best_similarity:
                best_similarity = similarity
                best_match = {
                    "key_id": key.split(":")[-1],
                    "similarity": similarity,
                    "cached_query": cached_emb["query"]
                }
        
        if best_match and best_similarity >= self.threshold:
            # Get the associated answer
            resp_key = f"{self.prefix}resp:{best_match['key_id']}"
            answer = self.redis.get(resp_key)
            
            if answer:
                return {
                    "response": json.loads(answer),
                    "similarity": best_similarity,
                    "cached_query": best_match["cached_query"],
                    "cache_hit": True
                }
        
        return None
    
    def set(self, query: str, response: str) -> str:
        """
        Stores the query and its answer with an embedding.
        Returns the generated ID.
        """
        import time
        
        cache_id = hashlib.sha256(f"{query}:{time.time()}".encode()).hexdigest()[:16]
        embedding = self._get_embedding(query)
        
        # Store the embedding
        emb_key = f"{self.prefix}emb:{cache_id}"
        self.redis.setex(
            emb_key,
            self.ttl,
            json.dumps({"embedding": embedding, "query": query})
        )
        
        # Store the answer
        resp_key = f"{self.prefix}resp:{cache_id}"
        self.redis.setex(resp_key, self.ttl, json.dumps(response))
        
        return cache_id
    
    def call_with_semantic_cache(
        self,
        prompt_template: str,
        query: str,
        model: str = "gpt-4o-mini"
    ) -> dict:
        """
        Runs with the semantic cache.
        First looks for an answer to a similar query, if there is none it calls the API.
        """
        # Look in the semantic cache
        cached = self.get(query)
        if cached:
            return {
                "output": cached["response"],
                "source": "semantic_cache",
                "similarity": f"{cached['similarity']:.3f}",
                "cached_query": cached["cached_query"]
            }
        
        # Cache miss: call the API
        full_prompt = prompt_template.format(input=query)
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": full_prompt}],
            temperature=0
        )
        
        output = response.choices[0].message.content
        
        # Store in the semantic cache
        self.set(query, output)
        
        return {
            "output": output,
            "source": "api",
            "similarity": None,
            "tokens": response.usage.total_tokens
        }


# Usage example:
sem_cache = SemanticCache(threshold=0.92)

# First query
r1 = sem_cache.call_with_semantic_cache(
    "Answer the user's question: {input}",
    "What are the customer service business hours?"
)
print(f"1st: {r1['source']}")

# Similar query (different wording)
r2 = sem_cache.call_with_semantic_cache(
    "Answer the user's question: {input}",
    "What time is support available?"
)
print(f"2nd: {r2['source']} (similarity: {r2.get('similarity', 'N/A')})")
# 2nd: semantic_cache (similarity: 0.941) — we saved an API call

In-Memory Cache (No Redis)

For simple projects or testing, an in-memory cache:

from functools import lru_cache
import time
from typing import Optional


class InMemoryCache:
    """In-memory cache with TTL. Simpler than Redis, doesn't survive restarts."""
    
    def __init__(self, ttl_default: int = 3600, max_size: int = 1000):
        self._cache: dict = {}  # {key: {"value": ..., "expires_at": ...}}
        self.ttl_default = ttl_default
        self.max_size = max_size
        self._hits = 0
        self._misses = 0
    
    def _evict_expired(self) -> None:
        """Removes expired entries."""
        now = time.time()
        expired_keys = [k for k, v in self._cache.items() if v["expires_at"] < now]
        for k in expired_keys:
            del self._cache[k]
    
    def _evict_lru(self) -> None:
        """If it's full, removes the least recent entry."""
        if len(self._cache) >= self.max_size:
            # Remove the oldest one (with the lowest expires_at)
            oldest = min(self._cache, key=lambda k: self._cache[k]["expires_at"])
            del self._cache[oldest]
    
    def get(self, key: str) -> Optional[str]:
        self._evict_expired()
        entry = self._cache.get(key)
        
        if entry and entry["expires_at"] > time.time():
            self._hits += 1
            return entry["value"]
        
        self._misses += 1
        return None
    
    def set(self, key: str, value: str, ttl: Optional[int] = None) -> None:
        self._evict_lru()
        self._cache[key] = {
            "value": value,
            "expires_at": time.time() + (ttl or self.ttl_default)
        }
    
    def stats(self) -> dict:
        total = self._hits + self._misses
        return {
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": f"{self._hits/total*100:.1f}%" if total > 0 else "0%",
            "entries": len(self._cache)
        }


# Simple functional wrapper with a decorator:
_local_cache = InMemoryCache(ttl_default=1800, max_size=500)

def cached_llm_call(
    prompt: str,
    model: str = "gpt-4o-mini",
    ttl: int = 1800
) -> str:
    """Cached wrapper for LLM calls."""
    cache_key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()
    
    # Try the cache
    cached = _local_cache.get(cache_key)
    if cached:
        return cached
    
    # API call
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    output = response.choices[0].message.content
    
    # Cache it
    _local_cache.set(cache_key, output, ttl=ttl)
    
    return output

TTL Strategies: How Long to Cache

The cache duration depends on how volatile the data is:

class TTLStrategy:
    """TTL strategies by content type."""
    
    # Predefined configurations
    CONFIGS = {
        "static": {
            "ttl": 7 * 24 * 3600,    # 7 days
            "description": "Data that almost never changes (definitions, static FAQs)",
            "examples": ["What is machine learning?", "Definition of terms"]
        },
        "semi_static": {
            "ttl": 24 * 3600,         # 24 hours
            "description": "Changes occasionally (policies, product catalogs)",
            "examples": ["Return policy", "List of available services"]
        },
        "dynamic": {
            "ttl": 3600,              # 1 hour
            "description": "Changes regularly (inventory, availability)",
            "examples": ["Stock available", "Opening hours this week"]
        },
        "real_time": {
            "ttl": 300,               # 5 minutes
            "description": "Changes frequently (prices, news)",
            "examples": ["Current stock price", "Recent news"]
        },
        "no_cache": {
            "ttl": 0,
            "description": "Don't cache — always different",
            "examples": ["Personalized conversations", "Creative outputs"]
        }
    }
    
    @classmethod
    def recommend(cls, query_type: str) -> dict:
        """Recommends a TTL based on the query type."""
        real_time_words = ["price", "current price", "now", "today", "latest"]
        dynamic_words = ["available", "stock", "hours", "this week"]
        static_words = ["what is", "define", "explain", "concept"]
        
        query_lower = query_type.lower()
        
        for word in real_time_words:
            if word in query_lower:
                return cls.CONFIGS["real_time"]
        
        for word in dynamic_words:
            if word in query_lower:
                return cls.CONFIGS["dynamic"]
        
        for word in static_words:
            if word in query_lower:
                return cls.CONFIGS["static"]
        
        return cls.CONFIGS["semi_static"]  # Default


# Usage in the cache:
def call_with_smart_ttl(prompt: str, query: str) -> dict:
    """Determines the TTL automatically based on the nature of the query."""
    ttl_config = TTLStrategy.recommend(query)
    
    if ttl_config["ttl"] == 0:
        # Don't cache
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt.format(input=query)}],
            temperature=0
        )
        return {"output": response.choices[0].message.content, "cached": False}
    
    # Cache with the right TTL
    result = cache.call_with_cache(
        prompt.format(input=query),
        ttl=ttl_config["ttl"]
    )
    return {**result, "ttl_used": ttl_config["ttl"]}

Cache Invalidation

Correct invalidation keeps you from serving stale data:

class CacheInvalidator:
    """Manages cache invalidation in an orderly way."""
    
    def __init__(self, redis_client: redis.Redis, prefix: str = "llm_cache:"):
        self.redis = redis_client
        self.prefix = prefix
    
    def invalidate_prompt_version(self, prompt_name: str, old_version: str) -> int:
        """
        Wipes the whole cache when a prompt's version changes.
        
        Strategy: Include the prompt version in the cache key,
        so when the version changes, the old keys are automatically invalid.
        """
        # If the keys include the prompt version, simply switching
        # the active version orphans the old cache and it expires by TTL
        pattern = f"{self.prefix}*{prompt_name}:{old_version}*"
        
        keys = list(self.redis.scan_iter(pattern))
        if keys:
            self.redis.delete(*keys)
        
        print(f"Invalidated {len(keys)} cache entries for {prompt_name}:{old_version}")
        return len(keys)
    
    def invalidate_by_query_pattern(self, pattern: str) -> int:
        """Invalidates entries related to a query pattern."""
        # Scan and inspect content (only works with decode_responses=True)
        keys_to_delete = []
        
        for key in self.redis.scan_iter(f"{self.prefix}*"):
            value = self.redis.get(key)
            if value and pattern.lower() in value.lower():
                keys_to_delete.append(key)
        
        if keys_to_delete:
            self.redis.delete(*keys_to_delete)
        
        return len(keys_to_delete)
    
    def invalidate_all(self) -> int:
        """Clears the whole cache (useful on major deploys)."""
        keys = list(self.redis.scan_iter(f"{self.prefix}*"))
        
        if keys:
            self.redis.delete(*keys)
        
        print(f"Cache cleared: {len(keys)} entries removed")
        return len(keys)
    
    def invalidate_on_deploy(self, prompt_name: str, new_version: str) -> None:
        """
        Hook to run when deploying a new prompt.
        Invalidates the cache specific to the prompt that changed.
        """
        print(f"Deploy detected: {prompt_name} → {new_version}")
        
        # Strategy 1: Wipe the prompt's whole cache
        # invalidated = self.invalidate_prompt_version(prompt_name, "*")
        
        # Strategy 2: Let it expire by TTL (less aggressive)
        # Only invalidate if the change is breaking (MAJOR version change)
        is_breaking = new_version.startswith("v") and new_version[1] != "1"
        
        if is_breaking:
            invalidated = self.invalidate_by_query_pattern(prompt_name)
            print(f"Breaking change detected: {invalidated} entries invalidated")
        else:
            print("Non-breaking change: letting the cache expire by TTL")

Cache Metrics

class CacheMetrics:
    """Monitors how effective the cache is."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def full_report(self, cache: ExactMatchCache) -> str:
        """Generates a cache effectiveness report."""
        stats = cache.stats()
        hits = cache._hits
        misses = cache._misses
        total = hits + misses
        hit_rate = hits / total if total > 0 else 0
        
        # Compute estimated savings
        # Assuming $0.001 per uncached request (prompt + completion)
        cost_without_cache = total * 0.001
        cost_with_cache = misses * 0.001
        savings = cost_without_cache - cost_with_cache
        
        return f"""
Cache Performance Report
━━━━━━━━━━━━━━━━━━━━━━━
Total requests:  {total:,}
Cache hits:      {hits:,} ({hit_rate:.1%})
Cache misses:    {misses:,}
Keys in Redis:   {stats['redis_keys']:,}

Savings estimate:
Without cache:   ${cost_without_cache:.4f}
With cache:      ${cost_with_cache:.4f}
Savings:         ${savings:.4f} ({hit_rate:.1%})

Recommendation:
{'✅ Excellent hit rate' if hit_rate > 0.5 else 
 '⚠️  Low hit rate — consider tuning the TTL or adding semantic caching' if hit_rate > 0.2 else
 '❌ Very low hit rate — check whether caching fits this use case at all'}
"""

Comparison: Caching Options

Troubleshooting

Problem 1: Cache hit rate very low

Symptom: Hit rate < 15% after several days.

Diagnosis:

def diagnose_low_hit_rate(cache: ExactMatchCache) -> list[str]:
    """Identifies why the hit rate is low."""
    suggestions = []
    
    # 1. Check whether prompts have dynamic variables that break the cache
    suggestions.append(
        "Check: do the prompts include timestamps, user IDs, "
        "or other dynamic variables? Those break exact match."
    )
    
    # 2. TTL too short
    if cache.ttl_default < 300:
        suggestions.append(
            f"TTL too short ({cache.ttl_default}s). "
            "Consider 3600s (1h) for semi-static queries."
        )
    
    # 3. If queries are all unique, exact match doesn't apply
    suggestions.append(
        "Are the queries mostly unique (personalized conversations)? "
        "If so, exact match doesn't apply. Consider semantic caching, or no caching."
    )
    
    return suggestions

Problem 2: Stale data

Symptom: The cache serves outdated answers.

Solution:

# Include the prompt version in the cache key
def cache_key_with_version(prompt: str, prompt_version: str, model: str) -> str:
    content = f"{model}:{prompt_version}:{prompt}"
    return hashlib.sha256(content.encode()).hexdigest()

# When you move the prompt to v1.2, the v1.1 keys no longer match
# You don't need to invalidate explicitly, they simply expire by TTL

Problem 3: Redis memory growing without bound

Symptom: Redis uses more and more memory.

Solution:

# In redis.conf:
maxmemory 256mb
maxmemory-policy allkeys-lru  # Evict the least used when it's full

# Configure it at initialization:
r = redis.Redis(host="localhost", port=6379)
r.config_set("maxmemory", "256mb")
r.config_set("maxmemory-policy", "allkeys-lru")

Exercises

Exercise 1: Implement an exact match cache without Redis

Implement an in-memory cache with TTL that works without Redis (Python stdlib only):

import hashlib
import time
from typing import Optional

class SimpleCache:
    """In-memory cache with no external dependencies."""
    
    def __init__(self, ttl_seconds: int = 3600, max_items: int = 500):
        self._store: dict = {}
        self.ttl = ttl_seconds
        self.max_items = max_items
    
    def _key(self, prompt: str) -> str:
        return hashlib.md5(prompt.encode()).hexdigest()
    
    def get(self, prompt: str) -> Optional[str]:
        key = self._key(prompt)
        if key in self._store:
            value, expires = self._store[key]
            if time.time() < expires:
                return value
            del self._store[key]
        return None
    
    def set(self, prompt: str, answer: str) -> None:
        if len(self._store) >= self.max_items:
            # Remove the oldest one
            oldest_key = min(self._store, key=lambda k: self._store[k][1])
            del self._store[oldest_key]
        
        self._key(prompt)
        self._store[self._key(prompt)] = (answer, time.time() + self.ttl)
    
    def hit_rate(self, n_queries: int = 0) -> float:
        return 0.0  # Simplified — in production, track hits/misses

# Test:
from openai import OpenAI
client = OpenAI()
cache = SimpleCache(ttl_seconds=300)

def ask(question: str) -> tuple[str, str]:
    """Returns (answer, source)."""
    cached = cache.get(question)
    if cached:
        return cached, "cache"
    
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        temperature=0
    )
    answer = r.choices[0].message.content
    cache.set(question, answer)
    return answer, "api"

r1, s1 = ask("What is Python?")
r2, s2 = ask("What is Python?")  # Should be a cache hit
print(f"1st: {s1}, 2nd: {s2}")

Exercise 2: Measure the savings from caching

Simulate 100 requests where 40% are repeats and compute the estimated savings:

import random
import time

def simulate_caching(n_requests: int = 100, cache_hit_rate: float = 0.40) -> dict:
    """Simulates the impact of caching on cost and latency."""
    
    api_cost_per_request = 0.001    # $0.001 per API request
    api_latency_ms = 800            # Typical API latency
    cache_latency_ms = 2            # Typical cache hit latency
    
    n_hits = int(n_requests * cache_hit_rate)
    n_misses = n_requests - n_hits
    
    # Cost
    cost_without_cache = n_requests * api_cost_per_request
    cost_with_cache = n_misses * api_cost_per_request
    
    # Total latency
    latency_without_cache_ms = n_requests * api_latency_ms
    latency_with_cache_ms = n_misses * api_latency_ms + n_hits * cache_latency_ms
    
    return {
        "n_requests": n_requests,
        "cache_hits": n_hits,
        "cache_misses": n_misses,
        "hit_rate": f"{cache_hit_rate:.0%}",
        "cost_without_cache": f"${cost_without_cache:.4f}",
        "cost_with_cache": f"${cost_with_cache:.4f}",
        "cost_savings": f"${cost_without_cache - cost_with_cache:.4f} ({cache_hit_rate:.0%})",
        "total_latency_without_cache": f"{latency_without_cache_ms/1000:.1f}s",
        "total_latency_with_cache": f"{latency_with_cache_ms/1000:.1f}s",
        "latency_reduction": f"{(1-latency_with_cache_ms/latency_without_cache_ms):.0%}"
    }

result = simulate_caching(100, 0.40)
for k, v in result.items():
    print(f"{k}: {v}")

Summary

• Exact match cache: Prompt hash as the key in Redis — simple and effective for repeated queries
• Semantic cache: Embeddings + cosine similarity — catches similar queries, 10-20% more hits
• In-memory cache: No external dependencies — only for development or a single process
• TTL strategy: Tune by volatility: 7d for static, 1h for dynamic, 5min for real-time
• Cache invalidation: With the prompt version in the key, or explicit invalidation on deploy
• Target hit rate: >30% to justify the complexity; >50% for a semantic cache

Additional resources

1. Redis Documentation — Full Redis documentation
2. GPTCache — Caching library built for LLMs
3. redis-py — Python client for Redis
4. OpenAI Embeddings — For semantic caching
5. Caching Best Practices — AWS guide to caching