Semantic Caching for LLM Applications

Why Traditional Caching Falls Short

Traditional caching uses exact key matching. If a user asks “What is Docker?” and you cache the response, a second user asking “What’s Docker?” gets a cache miss. “Explain Docker to me” is another miss. “Can you tell me about Docker?” is yet another. The semantic meaning is identical, but the strings are different.

In LLM applications, this pattern is common. Users phrase the same question in many ways. Without semantic caching, you pay for a new API call every time, even though you already have a perfectly good answer.

Semantic caching solves this by comparing the meaning of queries rather than their text. If a new query is semantically similar enough to a cached query, return the cached response.

How Semantic Caching Works

The flow is straightforward:

1. Convert the incoming query into an embedding vector.
2. Search the cache for vectors within a similarity threshold.
3. If a match is found, return the cached response.
4. If no match, call the LLM, cache the query embedding and response, and return the response.

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.92):
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self.threshold = similarity_threshold
        self.cache: list[dict] = []  # Simple list-based cache for illustration

    def _embed(self, text: str) -> np.ndarray:
        return self.encoder.encode(text, normalize_embeddings=True)

    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b))

    def get(self, query: str) -> str | None:
        query_embedding = self._embed(query)

        best_score = 0.0
        best_response = None

        for entry in self.cache:
            score = self._cosine_similarity(query_embedding, entry["embedding"])
            if score > best_score:
                best_score = score
                best_response = entry["response"]

        if best_score >= self.threshold:
            return best_response
        return None

    def put(self, query: str, response: str):
        embedding = self._embed(query)
        self.cache.append({
            "query": query,
            "embedding": embedding,
            "response": response,
        })

# Usage
cache = SemanticCache(similarity_threshold=0.92)
cache.put("What is Docker?", "Docker is a containerization platform that...")

# These will all hit the cache
print(cache.get("What's Docker?"))          # Hit
print(cache.get("Explain Docker to me"))     # Hit
print(cache.get("How does Kubernetes work?")) # Miss (different topic)

Scaling with FAISS

The list-based approach above scans every cached entry linearly. With thousands of cached queries, this becomes slow. FAISS (Facebook AI Similarity Search) provides fast approximate nearest neighbor search.

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class FAISSSemanticCache:
    def __init__(self, similarity_threshold: float = 0.92, dimension: int = 384):
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self.threshold = similarity_threshold
        self.dimension = dimension

        # FAISS index for cosine similarity (using inner product on normalized vectors)
        self.index = faiss.IndexFlatIP(dimension)
        self.responses: list[str] = []
        self.queries: list[str] = []

    def _embed(self, text: str) -> np.ndarray:
        vec = self.encoder.encode(text, normalize_embeddings=True)
        return vec.reshape(1, -1).astype("float32")

    def get(self, query: str) -> str | None:
        if self.index.ntotal == 0:
            return None

        query_vec = self._embed(query)
        scores, indices = self.index.search(query_vec, k=1)

        if scores[0][0] >= self.threshold:
            return self.responses[indices[0][0]]
        return None

    def put(self, query: str, response: str):
        vec = self._embed(query)
        self.index.add(vec)
        self.queries.append(query)
        self.responses.append(response)

    @property
    def size(self) -> int:
        return self.index.ntotal

FAISS handles millions of vectors efficiently. For even larger scales, use IndexIVFFlat with clustering for sublinear search time.

Production Cache with Redis

For production deployments, you need persistence, TTL, and multi-process access. Redis with its vector search module handles all of this.

import redis
import json
import hashlib
import numpy as np
from sentence_transformers import SentenceTransformer

class RedisSemanticCache:
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        similarity_threshold: float = 0.92,
        ttl_seconds: int = 3600,
        prefix: str = "sem_cache",
    ):
        self.redis = redis.from_url(redis_url)
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
        self.prefix = prefix
        self._ensure_index()

    def _ensure_index(self):
        """Create the RediSearch index for vector similarity."""
        try:
            self.redis.execute_command(
                "FT.CREATE", f"{self.prefix}:idx",
                "ON", "HASH",
                "PREFIX", "1", f"{self.prefix}:entry:",
                "SCHEMA",
                "query", "TEXT",
                "response", "TEXT",
                "embedding", "VECTOR", "FLAT", "6",
                    "TYPE", "FLOAT32",
                    "DIM", "384",
                    "DISTANCE_METRIC", "COSINE",
            )
        except redis.ResponseError:
            pass  # Index already exists

    def _embed(self, text: str) -> bytes:
        vec = self.encoder.encode(text, normalize_embeddings=True)
        return vec.astype(np.float32).tobytes()

    def get(self, query: str) -> str | None:
        query_vec = self._embed(query)

        results = self.redis.execute_command(
            "FT.SEARCH", f"{self.prefix}:idx",
            f"*=>[KNN 1 @embedding $vec AS score]",
            "PARAMS", "2", "vec", query_vec,
            "RETURN", "2", "response", "score",
            "SORTBY", "score",
            "DIALECT", "2",
        )

        if results[0] > 0:
            # Parse results - score is cosine distance, lower is better
            score = float(results[2][3])
            if score <= (1 - self.threshold):  # Convert distance to similarity
                response = results[2][1].decode() if isinstance(results[2][1], bytes) else results[2][1]
                return response

        return None

    def put(self, query: str, response: str):
        key = f"{self.prefix}:entry:{hashlib.md5(query.encode()).hexdigest()}"
        embedding = self._embed(query)

        self.redis.hset(key, mapping={
            "query": query,
            "response": response,
            "embedding": embedding,
        })
        self.redis.expire(key, self.ttl)

Integrating with Your LLM Client

Wrap your LLM client with the cache layer.

from openai import OpenAI

class CachedLLMClient:
    def __init__(self, cache: FAISSSemanticCache, model: str = "gpt-4o"):
        self.cache = cache
        self.client = OpenAI()
        self.model = model
        self.stats = {"hits": 0, "misses": 0}

    def chat(self, query: str, system_prompt: str = "") -> str:
        # Check cache
        cache_key = f"{system_prompt}|||{query}" if system_prompt else query
        cached = self.cache.get(cache_key)

        if cached is not None:
            self.stats["hits"] += 1
            return cached

        # Cache miss - call LLM
        self.stats["misses"] += 1
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": query})

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
        )
        answer = response.choices[0].message.content

        # Store in cache
        self.cache.put(cache_key, answer)
        return answer

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

# Usage
cache = FAISSSemanticCache(similarity_threshold=0.92)
llm = CachedLLMClient(cache)

# First call - cache miss
answer1 = llm.chat("What is Docker?")

# Second call - cache hit (semantically similar)
answer2 = llm.chat("Explain what Docker is")

print(f"Hit rate: {llm.hit_rate:.1%}")
print(f"Cache size: {cache.size}")

Choosing the Right Similarity Threshold

The threshold controls the tradeoff between cache hits and answer quality.

• 0.98+: Very conservative. Only near-identical queries match. Low hit rate but virtually no wrong answers served.
• 0.92-0.97: Good default range. Catches common paraphrases while avoiding semantic drift.
• 0.85-0.91: Aggressive. Higher hit rate but risk of serving answers to meaningfully different questions.

Test with your actual query distribution. Collect query pairs that should and should not match, then find the threshold that correctly classifies most of them.

# Evaluate threshold with labeled pairs
test_pairs = [
    ("What is Docker?", "Explain Docker", True),
    ("What is Docker?", "How to install Docker", False),
    ("Python async await", "How does async/await work in Python", True),
    ("Python async await", "JavaScript async await", False),
]

for threshold in [0.85, 0.90, 0.92, 0.95, 0.98]:
    correct = 0
    for q1, q2, should_match in test_pairs:
        cache = SemanticCache(similarity_threshold=threshold)
        cache.put(q1, "cached response")
        matched = cache.get(q2) is not None
        if matched == should_match:
            correct += 1
    print(f"Threshold {threshold}: {correct}/{len(test_pairs)} correct")

Cache Invalidation

Cached LLM responses can become stale. A cached answer about “the latest Python version” ages quickly. Strategies for invalidation:

TTL-based: Set an expiration time on each entry. Good for general knowledge queries where freshness matters.

Version-based: Include a version key in the cache. When you update your system prompt or model, bump the version to invalidate all entries.

Topic-based: Tag cached entries by topic. When you know a topic’s information has changed, invalidate entries with that tag.

class VersionedCache(FAISSSemanticCache):
    def __init__(self, version: str = "v1", **kwargs):
        super().__init__(**kwargs)
        self.version = version

    def get(self, query: str) -> str | None:
        return super().get(f"{self.version}:{query}")

    def put(self, query: str, response: str):
        super().put(f"{self.version}:{query}", response)

Cost Savings

The economics of semantic caching are compelling. A GPT-4o call costs roughly $0.01-0.05 depending on input and output length. An embedding computation with a local model costs effectively nothing. With a 40% cache hit rate on 100,000 daily queries, you save 40,000 API calls per day.

The embedding model itself runs locally and processes thousands of queries per second on a single CPU. The bottleneck is never the cache lookup.

Summary

Semantic caching matches queries by meaning rather than exact text, dramatically improving cache hit rates for LLM applications. Use embedding models to convert queries to vectors, FAISS for fast similarity search at scale, and Redis for production persistence with TTL. Set your similarity threshold between 0.92 and 0.97 as a starting point and tune based on your query distribution. Include the system prompt in your cache key to avoid cross-context contamination. Implement TTL or version-based invalidation for time-sensitive content.