Caching Strategies

The Query That Ran 50,000 Times

The analytics team at a B2B SaaS company built an LLM-powered dashboard feature: users could ask natural language questions about their business metrics. "What was our churn rate last quarter?" "Which customer segment grew the fastest?" The feature was popular. Too popular.

The engineering team noticed the LLM API bill was scaling faster than user growth. They pulled a sample of 10,000 queries from the past week and ran an analysis. The finding was eye-opening: the 10,000 queries were actually only 340 unique semantic questions. "What was our churn rate last quarter?" appeared 847 times with slight variations - "show me churn last quarter," "what's the Q3 churn?", "churn rate for Q3?" - all asking the exact same thing.

They had been calling GPT-4 50,000 times per week to answer questions that had already been answered. Worse, many of these queries were asking about metrics that hadn't changed - the Q3 churn rate is a historical fact. It will be the same answer tomorrow as it was today.

The team implemented a two-layer cache: an exact match cache for identical prompts and a semantic cache for similar ones. Within two weeks, cache hit rate reached 68%. LLM API costs dropped by 65%. Response latency for cached queries dropped from 3-5 seconds to under 50ms. The business metrics dashboard became the fastest feature in the product.

This is the value of caching for LLM applications. It is not a minor optimization - in the right product, it is a structural transformation.

Why This Exists

In traditional web applications, caching is well-understood: cache database queries, cache rendered HTML, cache API responses. The pattern is mature.

LLM applications share a property with traditional query-heavy applications: many user queries are repetitive. Search engines discovered this decades ago - the top 1,000 search queries account for a disproportionate fraction of all search traffic. The same applies to LLM applications in constrained domains.

But LLM caching has challenges that traditional caching doesn't:

• LLM outputs are verbose - you cache kilobytes per entry, not bytes
• "Same question" is fuzzy - "what's the weather?" and "how's the weather today?" are identical in intent but different in text
• LLM outputs can become stale - a cached answer about current events may be wrong tomorrow
• Context matters - the same question means different things in different conversation contexts

These challenges make LLM caching interesting engineering, not just redis.set(key, value).

The Four Caching Layers

Layer 1: Exact Cache

An exact cache stores the complete response for a specific prompt. On the next identical request, the cache returns the stored response without calling the LLM.

Key design decisions:

1. Prompt normalization. Before hashing, normalize the prompt to increase hit rate on trivially different inputs. Strip leading/trailing whitespace, normalize punctuation, lowercase (for queries where case doesn't affect the answer).

import hashlib
import re


def normalize_prompt(prompt: str) -> str:
    """Normalize prompt for exact cache key generation."""
    # Strip leading/trailing whitespace
    normalized = prompt.strip()
    # Collapse multiple whitespace to single space
    normalized = re.sub(r'\s+', ' ', normalized)
    # Lowercase (safe for most queries, disable for code prompts)
    normalized = normalized.lower()
    # Remove trailing punctuation that doesn't affect meaning
    normalized = re.sub(r'[.!?]+$', '', normalized)
    return normalized


def make_cache_key(
    system_prompt: str,
    user_message: str,
    model: str,
    temperature: float = 0,
) -> str:
    """Generate a deterministic cache key for an LLM request."""
    # Only cache deterministic requests (temperature=0 or very low)
    if temperature > 0.1:
        return None  # Don't cache non-deterministic outputs

    key_material = {
        "system": normalize_prompt(system_prompt),
        "user": normalize_prompt(user_message),
        "model": model,
    }
    key_string = str(sorted(key_material.items()))
    return hashlib.sha256(key_string.encode()).hexdigest()

2. TTL strategy. Not all cached responses have the same staleness profile:

Response Type	Appropriate TTL	Reason
Historical facts	30 days	Doesn't change
Documentation Q&A	7 days	Updates occasionally
Product pricing info	1 day	Changes weekly
Current events	1 hour	Changes rapidly
Real-time data	No caching	Always fresh needed

3. What NOT to cache. Personalized responses (the answer depends on the specific user's data), non-deterministic outputs (temperature > 0.1), safety-critical queries (medical advice, legal advice), and responses that include the current date/time.

import json
import redis.asyncio as redis
from datetime import timedelta


class ExactCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.client = redis.from_url(redis_url, decode_responses=True)

    async def get(self, cache_key: str) -> dict | None:
        """Retrieve cached response. Returns None on miss."""
        if not cache_key:
            return None

        data = await self.client.get(f"exact:{cache_key}")
        if data:
            return json.loads(data)
        return None

    async def set(
        self,
        cache_key: str,
        response: str,
        ttl_seconds: int = 86400,  # 24 hours default
        metadata: dict = None,
    ) -> bool:
        """Store response in cache."""
        if not cache_key:
            return False

        payload = {
            "response": response,
            "cached_at": time.time(),
            "metadata": metadata or {},
        }

        await self.client.setex(
            f"exact:{cache_key}",
            ttl_seconds,
            json.dumps(payload),
        )
        return True

    async def invalidate_pattern(self, pattern: str) -> int:
        """Invalidate all cache entries matching a pattern."""
        keys = await self.client.keys(f"exact:{pattern}*")
        if keys:
            await self.client.delete(*keys)
        return len(keys)

    async def get_stats(self) -> dict:
        """Return cache statistics."""
        keys = await self.client.keys("exact:*")
        return {
            "total_entries": len(keys),
            "estimated_memory_mb": len(keys) * 2 / 1024,  # rough estimate
        }

Layer 2: Semantic Cache

An exact cache misses queries that are semantically identical but textually different. Semantic caching uses embedding similarity to find cached responses for queries that mean the same thing.

How it works:

1. When a new query arrives, compute its embedding
2. Search the vector store for cached queries with high cosine similarity
3. If similarity exceeds threshold, return the cached response
4. If miss, call the LLM, then store the query embedding + response pair

The similarity threshold decision:

Threshold	Hit Rate	Risk
0.99	Low (~5%)	Very low - only near-identical queries
0.95	Medium (~25%)	Low - very similar phrasing
0.92	Higher (~40%)	Medium - may conflate distinct questions
0.85	High (~60%)	High - semantically similar but possibly different answers

The right threshold depends on your domain. For factual Q&A, 0.95+ is safe. For nuanced conversational queries, stay at 0.98+ or disable semantic caching.

import numpy as np
from openai import AsyncOpenAI


class SemanticCache:
    """
    Semantic cache using vector similarity.
    Uses an in-memory index for simplicity; use Pinecone/Weaviate for production.
    """

    def __init__(
        self,
        similarity_threshold: float = 0.95,
        max_entries: int = 10000,
        embedding_model: str = "text-embedding-3-small",
    ):
        self.threshold = similarity_threshold
        self.max_entries = max_entries
        self.embedding_model = embedding_model
        self.client = AsyncOpenAI()

        # In production, replace with vector store
        self.entries: list[dict] = []

    async def embed(self, text: str) -> np.ndarray:
        """Compute embedding for a query."""
        response = await self.client.embeddings.create(
            model=self.embedding_model,
            input=text,
        )
        return np.array(response.data[0].embedding)

    def cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

    async def get(self, query: str) -> dict | None:
        """Find a cached response for a semantically similar query."""
        if not self.entries:
            return None

        query_embedding = await self.embed(query)

        best_score = 0.0
        best_entry = None

        for entry in self.entries:
            score = self.cosine_similarity(query_embedding, entry["embedding"])
            if score > best_score:
                best_score = score
                best_entry = entry

        if best_score >= self.threshold:
            return {
                "response": best_entry["response"],
                "similarity": best_score,
                "original_query": best_entry["query"],
            }

        return None

    async def set(self, query: str, response: str, metadata: dict = None):
        """Store a query-response pair with its embedding."""
        embedding = await self.embed(query)

        entry = {
            "query": query,
            "response": response,
            "embedding": embedding,
            "cached_at": time.time(),
            "metadata": metadata or {},
        }

        self.entries.append(entry)

        # Evict oldest entries if at capacity
        if len(self.entries) > self.max_entries:
            self.entries.sort(key=lambda e: e["cached_at"])
            self.entries = self.entries[-self.max_entries:]

GPTCache: An open-source semantic cache library that handles embedding, storage, and similarity search:

# pip install gptcache
from gptcache import cache
from gptcache.adapter import openai
from gptcache.embedding import OpenAI as OpenAIEmbedding
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

def init_gptcache():
    embedding_function = OpenAIEmbedding()

    data_manager = get_data_manager(
        CacheBase("sqlite"),
        VectorBase("faiss", dimension=embedding_function.dimension),
    )

    cache.init(
        embedding_func=embedding_function.to_embeddings,
        data_manager=data_manager,
        similarity_evaluation=SearchDistanceEvaluation(),
    )
    cache.set_openai_key()

# After init, openai.ChatCompletion.create() automatically checks cache

Risks of semantic caching:

The fundamental risk is that semantically similar does NOT always mean same answer. Consider:

• "What is the capital of France?" → "Paris"
• "What was the capital of France during WWII?" → "Vichy"

These queries are semantically similar (both about France's capital) but have different answers. For domains where contextual nuance matters, semantic caching must be used conservatively or disabled.

:::warning Disable semantic caching for: factual queries where the answer depends on subtle phrasing differences, personalized responses, anything involving current state, and any safety-sensitive domains (medical, legal, financial advice). :::

Layer 3: Provider-Level Prompt Caching

Anthropic and OpenAI both offer prefix caching, which works at the KV (key-value) computation level within the Transformer. The provider stores the computed attention keys and values for repeated prompt prefixes, so the model can skip recomputing them on repeated requests.

Economics:

• Anthropic: 90% discount on cached tokens ($3.00/M →$0.30/M for Claude 3.5 Sonnet)
• OpenAI: 50% discount on cached tokens ($2.50/M →$1.25/M for GPT-4o)
• Minimum prefix length: 1024 tokens (both providers)
• Anthropic TTL: 5 minutes (ephemeral), renewed on each cache hit
• OpenAI: automatic, no TTL disclosed

Designing for provider cache hits:

# WRONG: Dynamic content before static content - breaks caching
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": f"Today is {today}. {static_context}"},  # Bad!
    *history,
    {"role": "user", "content": user_message},
]

# RIGHT: Static content first, dynamic last - maximizes cache hits
messages = [
    {
        "role": "system",
        "content": (
            f"{system_prompt}\n\n"         # Static ─┐
            f"{tool_definitions}\n\n"       # Static  ├─ cached prefix
            f"{few_shot_examples}\n\n"      # Static ─┘
            f"Context:\n{retrieved_docs}"  # Dynamic (RAG = per query)
        ),
    },
    *history,                              # Dynamic (per session)
    {"role": "user", "content": user_message},  # Dynamic (per request)
]

Measuring cache performance:

async def call_with_cache_metrics(messages: list[dict], model: str) -> dict:
    """Call Anthropic API and report cache performance."""
    import anthropic

    client = anthropic.Anthropic()

    response = client.messages.create(
        model=model,
        max_tokens=1000,
        messages=messages,
    )

    usage = response.usage
    cache_read = getattr(usage, "cache_read_input_tokens", 0)
    cache_create = getattr(usage, "cache_creation_input_tokens", 0)

    # Calculate cost savings
    regular_input_cost = usage.input_tokens * 3.00 / 1_000_000  # without caching
    actual_input_cost = (
        (usage.input_tokens - cache_read) * 3.00 / 1_000_000
        + cache_read * 0.30 / 1_000_000  # 90% discount
    )
    savings = regular_input_cost - actual_input_cost

    return {
        "content": response.content[0].text,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cache_read_tokens": cache_read,
        "cache_creation_tokens": cache_create,
        "cache_hit": cache_read > 0,
        "cost_savings_usd": savings,
    }

The Multi-Layer Cache Implementation

Combining all layers into a production-ready cache pipeline:

import hashlib
import json
import time
from dataclasses import dataclass

import redis.asyncio as redis
from openai import AsyncOpenAI


@dataclass
class CacheResult:
    hit: bool
    response: str | None
    layer: str | None  # "exact" | "semantic" | "miss"
    similarity: float | None = None
    latency_ms: float = 0


class MultiLayerLLMCache:
    """
    Production multi-layer cache:
    1. Exact cache (Redis)
    2. Semantic cache (embeddings + vector search)
    3. Falls through to LLM API (with provider-level caching active)
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        semantic_threshold: float = 0.95,
        exact_ttl: int = 86400,        # 24 hours
        semantic_ttl: int = 43200,     # 12 hours
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.openai = AsyncOpenAI()
        self.semantic_threshold = semantic_threshold
        self.exact_ttl = exact_ttl
        self.semantic_ttl = semantic_ttl

    async def get(self, query: str, context_key: str = "") -> CacheResult:
        """
        Check all cache layers.
        context_key: a hash of the conversation context (ensures context-aware caching).
        """
        start = time.perf_counter()

        # Layer 1: Exact cache
        exact_key = self._exact_key(query, context_key)
        exact_result = await self.redis.get(f"cache:exact:{exact_key}")
        if exact_result:
            data = json.loads(exact_result)
            return CacheResult(
                hit=True,
                response=data["response"],
                layer="exact",
                latency_ms=(time.perf_counter() - start) * 1000,
            )

        # Layer 2: Semantic cache
        query_embedding = await self._embed(query)
        semantic_result = await self._semantic_lookup(query_embedding, context_key)
        if semantic_result:
            return CacheResult(
                hit=True,
                response=semantic_result["response"],
                layer="semantic",
                similarity=semantic_result["score"],
                latency_ms=(time.perf_counter() - start) * 1000,
            )

        return CacheResult(
            hit=False,
            response=None,
            layer="miss",
            latency_ms=(time.perf_counter() - start) * 1000,
        )

    async def set(
        self,
        query: str,
        response: str,
        context_key: str = "",
        query_embedding: list[float] | None = None,
    ):
        """Store in both exact and semantic cache."""
        exact_key = self._exact_key(query, context_key)

        # Store in exact cache
        await self.redis.setex(
            f"cache:exact:{exact_key}",
            self.exact_ttl,
            json.dumps({"response": response, "query": query, "stored_at": time.time()}),
        )

        # Store in semantic cache (using Redis with vector search, or external vector DB)
        if query_embedding is None:
            query_embedding = await self._embed(query)

        await self._semantic_store(query, response, query_embedding, context_key)

    def _exact_key(self, query: str, context_key: str) -> str:
        normalized = query.strip().lower()
        return hashlib.sha256(f"{normalized}:{context_key}".encode()).hexdigest()

    async def _embed(self, text: str) -> list[float]:
        response = await self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        )
        return response.data[0].embedding

    async def _semantic_lookup(
        self, query_embedding: list[float], context_key: str
    ) -> dict | None:
        """
        In production: use Redis Vector Similarity Search or Pinecone.
        Simplified version below using sorted set simulation.
        """
        # Real implementation would use: redis.ft().search() with vector query
        # or pinecone.query(vector=query_embedding, top_k=1)
        # Placeholder for illustration:
        return None

    async def _semantic_store(
        self,
        query: str,
        response: str,
        embedding: list[float],
        context_key: str,
    ):
        """Store in vector search index for semantic retrieval."""
        # In production: upsert to Pinecone/Weaviate/Redis Vector
        # Placeholder for illustration:
        pass


# ─── LLM Service with Integrated Cache ───────────────────────────────────────

class CachedLLMService:
    def __init__(self, cache: MultiLayerLLMCache):
        self.cache = cache
        self.client = AsyncOpenAI()
        self.hits = 0
        self.misses = 0

    async def complete(
        self,
        system_prompt: str,
        user_message: str,
        conversation_context: str = "",
        model: str = "gpt-4o-mini",
        max_tokens: int = 500,
    ) -> dict:
        """Get completion with multi-layer caching."""
        context_key = hashlib.md5(
            (system_prompt + conversation_context).encode()
        ).hexdigest()[:16]

        # Check cache
        cache_result = await self.cache.get(user_message, context_key)

        if cache_result.hit:
            self.hits += 1
            return {
                "response": cache_result.response,
                "cached": True,
                "cache_layer": cache_result.layer,
                "latency_ms": cache_result.latency_ms,
                "cost_usd": 0.0,
            }

        # Cache miss - call LLM
        self.misses += 1
        start = time.perf_counter()

        completion = await self.client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message},
            ],
            max_tokens=max_tokens,
        )

        response_text = completion.choices[0].message.content
        latency = (time.perf_counter() - start) * 1000

        # Store in cache
        await self.cache.set(user_message, response_text, context_key)

        input_tokens = completion.usage.prompt_tokens
        output_tokens = completion.usage.completion_tokens
        cost = input_tokens * 0.15 / 1_000_000 + output_tokens * 0.60 / 1_000_000

        return {
            "response": response_text,
            "cached": False,
            "cache_layer": "miss",
            "latency_ms": latency,
            "cost_usd": cost,
            "tokens_used": input_tokens + output_tokens,
        }

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

Cache Invalidation

Cache invalidation is famously one of the two hard problems in computer science. For LLM caches, the challenge is knowing when a cached response is stale.

Time-based invalidation (TTL): Simplest approach. Every entry has a time-to-live. Stale entries expire automatically. The downside: entries may expire before they need to (wasteful) or remain valid after the underlying data changes (incorrect).

Event-based invalidation: When data changes, explicitly invalidate affected cache entries. More precise but requires knowing which cache entries depend on which data.

async def invalidate_on_data_change(
    topic: str,
    redis_client: redis.Redis,
) -> int:
    """Invalidate all cache entries related to a specific topic."""
    # In practice: maintain a reverse index of topic -> cache keys
    # Or use Redis keyspace tagging
    pattern = f"cache:exact:*"
    keys = await redis_client.keys(pattern)

    invalidated = 0
    for key in keys:
        data = json.loads(await redis_client.get(key) or "{}")
        # Check if the query mentions the changed topic
        if topic.lower() in data.get("query", "").lower():
            await redis_client.delete(key)
            invalidated += 1

    return invalidated

Hybrid (TTL + event-based): Use TTL as a safety net to prevent permanently stale entries, and event-based invalidation for known data changes.

Cold Start Problem

When your cache is empty (new deployment, cache wipe, cache miss storm), the cache provides no benefit and your LLM API is hit with full traffic. Solutions:

Warm-up from historical data: Before deployment, run your top 100-500 queries through the LLM and pre-populate the cache.

async def warm_cache(
    service: CachedLLMService,
    top_queries: list[str],
    system_prompt: str,
    model: str = "gpt-4o-mini",
):
    """Pre-warm cache with historical high-frequency queries."""
    print(f"Warming cache with {len(top_queries)} queries...")

    for i, query in enumerate(top_queries):
        result = await service.complete(
            system_prompt=system_prompt,
            user_message=query,
            model=model,
        )
        if not result["cached"]:
            print(f"  [{i+1}/{len(top_queries)}] Cached: {query[:60]}...")

    print(f"Cache warming complete. Hit rate: {service.hit_rate:.1%}")

Staggered rollout: When replacing cached responses with new model versions, roll out to a small fraction of traffic first. This allows the new cache to warm up before taking full traffic.

Production Notes

Cache hit rate benchmarks:

Application Type	Typical Hit Rate	Notes
FAQ / documentation	60-80%	Highly repetitive queries
Customer support	40-60%	Moderate repetition
General chat	5-15%	Low repetition, personalized
Business analytics	50-70%	Repetitive metric queries
Code generation	15-30%	Some repeated patterns

If your hit rate is below what you'd expect for your application type, investigate: are you normalizing queries before cache lookup? Is your semantic threshold too conservative? Are context keys over-partitioning the cache?

Cost savings calculation:

def calculate_cache_savings(
    total_requests: int,
    hit_rate: float,
    avg_input_tokens: int,
    avg_output_tokens: int,
    model: str = "gpt-4o",
) -> dict:
    pricing = {
        "gpt-4o": {"input": 2.50 / 1e6, "output": 10.00 / 1e6},
        "gpt-4o-mini": {"input": 0.15 / 1e6, "output": 0.60 / 1e6},
    }
    p = pricing[model]

    cost_per_request = (
        avg_input_tokens * p["input"] + avg_output_tokens * p["output"]
    )
    llm_requests = int(total_requests * (1 - hit_rate))
    actual_cost = llm_requests * cost_per_request
    full_cost = total_requests * cost_per_request

    return {
        "total_requests": total_requests,
        "llm_requests": llm_requests,
        "cache_hits": total_requests - llm_requests,
        "hit_rate": hit_rate,
        "full_cost_usd": full_cost,
        "actual_cost_usd": actual_cost,
        "savings_usd": full_cost - actual_cost,
        "savings_pct": (full_cost - actual_cost) / full_cost * 100,
    }

Cache observability: Always track the following metrics:

• Exact cache hit rate
• Semantic cache hit rate
• Cache latency (p50, p95, p99 for cache lookups)
• Cache eviction rate (indicates under-provisioned cache size)
• Cache error rate (failed lookups/writes)
• Cost savings per day/week

Common Mistakes

:::danger Mistake: Caching personalized responses without user isolation If user A asks "summarize my account history" and the cache stores this response, and user B asks the same question, they must NOT receive user A's account summary. Cache keys for personalized responses MUST include the user ID or a hash of the user-specific context. Failure to isolate cached responses across users is a serious security vulnerability. :::

:::danger Mistake: Setting semantic similarity threshold too low A 0.85 cosine similarity threshold sounds conservative, but in practice, semantically similar queries in the same domain can have meaningfully different answers. "What is the max upload size?" and "What is the max file size for attachments?" are 0.88 similar but might have different answers if uploads and attachments are different features. Start at 0.95 and lower only after testing on your specific domain. :::

:::warning Mistake: Not invalidating cache on model version changes When you upgrade from GPT-4o to GPT-4o-mini, or change your system prompt, the cached responses from the old configuration may produce inconsistent quality or behavior. Tag cache entries with the model version and system prompt hash, and invalidate when either changes. :::

:::warning Mistake: Applying semantic caching to conversational context Semantic caching works best for context-free queries - questions whose answer does not depend on the conversation history. For multi-turn conversations, the context_key must include a hash of the conversation state. Two queries that appear identical may need different answers based on what was said earlier in the conversation. :::

Interview Q&A

Q1: What is semantic caching and when would you use it vs. exact caching?

Exact caching stores responses keyed by a hash of the prompt. It works for truly identical queries - same text, same system prompt, same context. The hit rate in most applications is low (5-15%) because users rarely phrase questions identically.

Semantic caching embeds the query and finds cached responses for semantically similar queries. Hit rates can reach 40-70% in domain-specific applications where users ask variations of the same underlying questions.

I would use exact caching universally (low overhead, high precision). I would add semantic caching on top when: (1) the application has a constrained domain with predictable query patterns; (2) the queries are largely context-free (same answer regardless of conversation history); (3) the domain is not one where subtle phrasing differences change the answer. I would NOT use semantic caching for general-purpose chat, medical/legal/financial advice, or any domain where the answer depends on precise phrasing.

Q2: Explain how Anthropic and OpenAI prompt caching works and how to design prompts to maximize cache hits.

Both providers cache the KV (key-value) state of the Transformer attention layers for repeated prompt prefixes. When the same prefix is re-encountered, the model skips recomputing those layers, saving both compute time and the cost of processing those tokens.

Anthropic's cache requires explicit opt-in via cache_control: {"type": "ephemeral"} in the system prompt blocks. Minimum 1024 tokens to qualify. 90% discount on cached tokens. TTL is 5 minutes, renewed on each hit.

OpenAI's cache is automatic - any prefix of 1024+ tokens that repeats is automatically cached with a 50% discount. No explicit opt-in needed.

To maximize hit rate: put all stable content at the very beginning of the prompt - system instructions, persona, tool definitions, static knowledge base. Put all dynamic content at the end - retrieved context, conversation history, current message. Even a single token difference at position 1 invalidates the entire cache. Never put dates, user IDs, or session-specific information in the stable prefix.

Q3: How would you implement cache invalidation for an LLM product that answers questions about product documentation?

I would use a hybrid TTL + event-based approach.

TTL as baseline: all cache entries expire after 7 days. This prevents permanently stale information regardless of events.

Event-based invalidation for known changes: when a documentation page is updated, extract the topics covered by that page and invalidate all cache entries whose query is semantically related to those topics. In practice, this means: when a page update webhook fires, embed the page title and key concepts, then scan the semantic cache for entries with high similarity (above 0.80 threshold) to the updated content and delete them.

For the exact cache: maintain a reverse index mapping document IDs to cache keys. When a document is updated, look up all cache keys tagged with that document ID and delete them.

The cold start after invalidation: use the top 50 queries from the past week to re-warm the cache immediately after invalidation, so users hit the cache on their next visit rather than experiencing a latency spike.

Q4: What is the cold start problem for LLM caches and how do you solve it?

The cold start problem occurs when the cache is empty - on first deployment, after a cache wipe, or after invalidating a large fraction of entries. During cold start, all requests miss the cache and hit the LLM API, which means full latency and full cost for every user.

The primary solution is cache warm-up: before directing traffic to the new deployment, run the top N queries from historical data through the LLM and populate the cache. For an application with 100K daily queries, the top 500 queries typically cover 30-40% of traffic - warming these 500 entries can bring the hit rate from 0% to 30-40% before the first real user arrives.

The secondary solution is staged rollout: don't switch 100% of traffic to the new deployment immediately. Start at 1-5% of traffic, let the cache warm up naturally, then increase the percentage over hours. This ensures most users are served by a warm cache.

For semantic caches, warm-up also needs to populate the embedding index. Generate embeddings for historical queries and pre-populate the vector store alongside the responses.

Q5: How do you calculate the ROI of implementing a semantic cache for your LLM application?

ROI calculation requires four inputs: (1) current request volume and mix; (2) current cost per request; (3) expected hit rate; (4) implementation cost (engineering time + infrastructure).

Start by sampling 1,000-10,000 recent queries and running them through an offline semantic similarity analysis to estimate the hit rate at different similarity thresholds. This gives you the expected hit rate without building anything.

Cost savings formula: savings_per_day = requests_per_day * hit_rate * avg_cost_per_request. Subtract the cache infrastructure cost (Redis with vector search: ~$100-300/month for most scales) and the embedding cost for new misses (negligible for `text-embedding-3-small` at$0.02/million tokens).

Also quantify latency savings: cached responses return in under 50ms vs 3-10 seconds for LLM calls. If your application charges per successful interaction or has SLO requirements, the latency benefit has additional business value.

Typical result for FAQ/support applications: semantic caching pays back the implementation cost in 2-4 weeks at moderate traffic levels (10K+ requests/day).

:::tip 🎮 Interactive Playground

Visualize this concept: Try the LLM Caching Strategies demo on the EngineersOfAI Playground - no code required.

:::