Semantic Caching and Vector Search for LLM Applications: Reducing Latency and Cost in High-Volume Enterprise AI Pipelines

Introduction

Your customer support AI is fielding 50,000 queries a day. Your internal knowledge assistant is answering questions from 3,000 employees. Your code generation tool is processing requests from hundreds of developers across multiple time zones. Each one of those requests hits a commercial LLM API — at anywhere from $0.002 to $0.06 per 1,000 tokens — and returns a response in 2 to 8 seconds. Now do the math. A pipeline that generates 50,000 responses a day at an average of 500 output tokens per response is spending tens of thousands of dollars monthly, and users are staring at loading spinners for the better part of a workday.

The instinctive solution is exact-match caching: if someone asks the same question twice, serve the cached response. In practice, this barely helps. Real users rephrase constantly. "What's the refund policy?" and "How do I get my money back?" and "Can I return this?" are semantically identical questions that will never share a cache key. Traditional caching misses all three matches. What you need is a cache that understands meaning — one that can recognize when two different strings are asking the same thing, and serve the cached answer for all of them. That's semantic caching, and it's one of the highest-leverage optimizations available to teams running LLM applications at scale.

By the end of this lesson, you'll understand how semantic caching works at the architectural level, how to implement it using vector databases and embedding models, and how to tune it for production — including the hard trade-offs around similarity thresholds, embedding quality, cache invalidation, and multi-tenancy. This isn't a toy implementation. We're building something you could actually deploy.

What you'll learn:

• How vector embeddings enable semantic similarity matching and why this matters for caching
• How to design a semantic cache layer that integrates cleanly with an LLM pipeline
• How to choose similarity thresholds that balance cache hit rate against answer quality
• How to handle cache invalidation, multi-tenancy, and namespace isolation in enterprise environments
• How to measure, monitor, and continuously improve semantic cache performance in production

---

Prerequisites

You should be comfortable with:

• Python at an intermediate level (async/await, class design, decorators)
• Basic familiarity with LLM APIs (OpenAI, Anthropic, or similar)
• Conceptual understanding of embeddings and vector similarity (dot product, cosine similarity)
• Some exposure to Redis, a relational database, or a vector database like Pinecone, Weaviate, or pgvector

If you haven't worked with embeddings before, spend 30 minutes with the OpenAI embeddings quickstart first. This lesson will explain the concepts, but will move quickly through the math.

---

What Semantic Caching Actually Is (and Why Exact Caching Fails You)

Let's be precise about what we're building before we write a single line of code.

An exact-match cache is a hash map. You compute a hash of the input string, check if that hash exists in the cache, and return the value if it does. This works beautifully for deterministic computations where identical inputs produce identical outputs. SQL query results, rendered templates, computed statistics — all great candidates. Natural language queries are not. Two strings that mean the same thing will produce completely different hashes.

Semantic caching works differently. Instead of hashing the raw text, you convert it into a dense vector embedding — a list of floating-point numbers that encodes the meaning of the text in a high-dimensional space. The core insight of embedding models is that semantically similar texts produce vectors that are close together in this space, regardless of their surface-level wording.

When a user asks "What's the refund policy?", you embed that query into a vector and search the cache for any stored vectors that are sufficiently similar. If you find one — say, a previous response to "How do I return a product?" — you return the cached answer without ever calling the LLM. The threshold for "sufficiently similar" is a cosine similarity score, typically somewhere between 0.85 and 0.95 depending on your tolerance for semantic drift.

This framing immediately reveals the fundamental trade-off: cache hit rate versus answer precision. A lower threshold catches more similar queries, improving your hit rate and reducing costs, but risks serving a cached response that doesn't quite fit the new question. A higher threshold is more conservative but misses more opportunities. Getting this balance right is the core engineering challenge of semantic caching.

---

The Architecture of a Semantic Cache Layer

Before diving into code, let's understand the system architecture. A well-designed semantic cache sits as a middleware layer between your application and your LLM provider. Every query flows through it, and the cache makes a decision: return a cached response, or forward to the LLM and store the result.

Here's the request flow:

User Query
    │
    ▼
Embed Query ──────────────────────────────────────────────┐
    │                                                      │
    ▼                                                      │
Vector Similarity Search                                   │
    │                                                      │
    ├─── Cache Hit (similarity ≥ threshold) ──► Return     │
    │                                         Cached       │
    └─── Cache Miss ──────────────────────────► Call LLM   │
                                                  │        │
                                                  ▼        │
                                             Embed Query   │
                                             Store in DB ◄─┘
                                                  │
                                                  ▼
                                             Return LLM
                                             Response

Notice that on a cache miss, we embed the query again before storing. In practice you can reuse the embedding from the initial lookup — I've drawn it separately to clarify the logical steps. We store the embedding alongside the original query text and the LLM response.

The key components you need:

1. An embedding model — converts text to vectors. OpenAI's text-embedding-3-small is a cost-effective choice; for on-premise or privacy-sensitive deployments, consider a local model like sentence-transformers/all-MiniLM-L6-v2.

2. A vector store — indexes the embeddings and supports approximate nearest-neighbor (ANN) search. Options include pgvector (if you're already on Postgres), Redis Stack, Pinecone, Weaviate, or Qdrant.

3. A cache store — holds the actual query/response pairs. This can be the same database as your vector store, or a fast key-value store like Redis if you want sub-millisecond retrieval after finding a match.

4. A similarity threshold — the cosine similarity score above which you consider a cached entry a valid match.

---

Building the Core Cache Implementation

Let's build this in Python. We'll use OpenAI embeddings, Qdrant as our vector store (it runs locally in memory for development, which makes testing easy), and a simple in-memory dict as the response store. We'll then evolve this toward a production-ready design.

Start with your dependencies:

pip install openai qdrant-client numpy python-dotenv

Here's the foundational class:

import hashlib
import time
import numpy as np
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    VectorParams,
    PointStruct,
    SearchParams,
    Filter,
    FieldCondition,
    MatchValue,
)
from typing import Optional
import uuid

class SemanticCache:
    """
    A semantic cache layer for LLM applications.
    Uses vector similarity to match semantically equivalent queries.
    """

    EMBEDDING_MODEL = "text-embedding-3-small"
    EMBEDDING_DIM = 1536
    COLLECTION_NAME = "semantic_cache"

    def __init__(
        self,
        similarity_threshold: float = 0.92,
        ttl_seconds: int = 86400,  # 24 hours
        namespace: str = "default",
    ):
        self.client = OpenAI()
        self.qdrant = QdrantClient(":memory:")  # use URL for production
        self.similarity_threshold = similarity_threshold
        self.ttl_seconds = ttl_seconds
        self.namespace = namespace
        self._response_store: dict[str, dict] = {}
        self._ensure_collection()

    def _ensure_collection(self):
        """Create the vector collection if it doesn't exist."""
        collections = [c.name for c in self.qdrant.get_collections().collections]
        if self.COLLECTION_NAME not in collections:
            self.qdrant.create_collection(
                collection_name=self.COLLECTION_NAME,
                vectors_config=VectorParams(
                    size=self.EMBEDDING_DIM,
                    distance=Distance.COSINE,
                ),
            )

    def _embed(self, text: str) -> list[float]:
        """Embed a query string into a dense vector."""
        response = self.client.embeddings.create(
            model=self.EMBEDDING_MODEL,
            input=text,
        )
        return response.data[0].embedding

    def _cache_key(self, point_id: str) -> str:
        return f"{self.namespace}:{point_id}"

    def lookup(self, query: str) -> Optional[str]:
        """
        Search the cache for a semantically similar query.
        Returns the cached response if found, None otherwise.
        """
        query_vector = self._embed(query)

        results = self.qdrant.search(
            collection_name=self.COLLECTION_NAME,
            query_vector=query_vector,
            limit=1,
            search_params=SearchParams(hnsw_ef=128, exact=False),
            query_filter=Filter(
                must=[
                    FieldCondition(
                        key="namespace",
                        match=MatchValue(value=self.namespace),
                    )
                ]
            ),
            with_payload=True,
        )

        if not results:
            return None

        top_result = results[0]
        score = top_result.score
        point_id = top_result.payload.get("point_id")
        stored_at = top_result.payload.get("stored_at", 0)

        # Check TTL
        if time.time() - stored_at > self.ttl_seconds:
            return None

        # Check similarity threshold
        if score < self.similarity_threshold:
            return None

        # Retrieve the cached response
        cache_entry = self._response_store.get(self._cache_key(point_id))
        if not cache_entry:
            return None

        return cache_entry["response"]

    def store(self, query: str, response: str) -> None:
        """
        Store a query/response pair in the semantic cache.
        """
        query_vector = self._embed(query)
        point_id = str(uuid.uuid4())
        stored_at = time.time()

        # Upsert the vector into Qdrant
        self.qdrant.upsert(
            collection_name=self.COLLECTION_NAME,
            points=[
                PointStruct(
                    id=point_id,
                    vector=query_vector,
                    payload={
                        "query": query,
                        "namespace": self.namespace,
                        "stored_at": stored_at,
                        "point_id": point_id,
                    },
                )
            ],
        )

        # Store the response in the fast lookup store
        self._response_store[self._cache_key(point_id)] = {
            "query": query,
            "response": response,
            "stored_at": stored_at,
        }

Now let's wire this into an LLM call:

def get_llm_response(
    query: str,
    cache: SemanticCache,
    system_prompt: str = "You are a helpful assistant.",
) -> dict:
    """
    Fetch a response, using the semantic cache when possible.
    Returns metadata about whether the response was cached.
    """
    client = OpenAI()
    start_time = time.time()

    # Try the cache first
    cached_response = cache.lookup(query)
    if cached_response:
        return {
            "response": cached_response,
            "source": "cache",
            "latency_ms": round((time.time() - start_time) * 1000, 2),
        }

    # Cache miss — call the LLM
    llm_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query},
        ],
        temperature=0,  # deterministic outputs make caching more reliable
    )
    response_text = llm_response.choices[0].message.content

    # Store in cache
    cache.store(query, response_text)

    return {
        "response": response_text,
        "source": "llm",
        "latency_ms": round((time.time() - start_time) * 1000, 2),
        "tokens_used": llm_response.usage.total_tokens,
    }

Let's test this with some semantically similar queries:

cache = SemanticCache(similarity_threshold=0.92, namespace="customer_support")

queries = [
    "What is your return policy?",              # original
    "How do I return a product I bought?",       # paraphrase
    "Can I get a refund on my purchase?",        # paraphrase
    "What are the terms for returning items?",   # paraphrase
    "How do I track my order?",                  # different topic
]

for q in queries:
    result = get_llm_response(q, cache)
    print(f"[{result['source'].upper()}] ({result['latency_ms']}ms) {q[:50]}")

On a typical run, you'll see the first query go to the LLM (~1500ms), and subsequent semantically similar queries come from cache (~50-100ms). The "track my order" query will correctly miss the cache and go to the LLM.

---

Choosing and Tuning Your Similarity Threshold

The similarity threshold is the most important tuning parameter in your entire system, and it's deeply domain-dependent. There is no universal correct value.

Here's how to think about it empirically. Collect a sample of real user queries from your application — at least a few hundred. Manually group them into clusters of semantically equivalent questions. Then compute the cosine similarity distribution within clusters (true matches) and across clusters (false matches). Your threshold should sit somewhere in the gap between these two distributions.

import itertools
from scipy.spatial.distance import cosine

def analyze_threshold_distribution(query_clusters: dict[str, list[str]]):
    """
    Analyze within-cluster vs across-cluster similarity scores.
    query_clusters: {"intent_name": ["query1", "query2", ...], ...}
    """
    client = OpenAI()

    # Embed all queries
    all_queries = []
    all_labels = []
    for intent, queries in query_clusters.items():
        all_queries.extend(queries)
        all_labels.extend([intent] * len(queries))

    embeddings = []
    for query in all_queries:
        resp = client.embeddings.create(
            model="text-embedding-3-small", input=query
        )
        embeddings.append(np.array(resp.data[0].embedding))

    within_cluster_scores = []
    across_cluster_scores = []

    for i, j in itertools.combinations(range(len(all_queries)), 2):
        sim = 1 - cosine(embeddings[i], embeddings[j])
        if all_labels[i] == all_labels[j]:
            within_cluster_scores.append(sim)
        else:
            across_cluster_scores.append(sim)

    print(f"Within-cluster: mean={np.mean(within_cluster_scores):.3f}, "
          f"min={np.min(within_cluster_scores):.3f}, "
          f"p10={np.percentile(within_cluster_scores, 10):.3f}")

    print(f"Across-cluster: mean={np.mean(across_cluster_scores):.3f}, "
          f"max={np.max(across_cluster_scores):.3f}, "
          f"p90={np.percentile(across_cluster_scores, 90):.3f}")

In my experience across customer support, HR knowledge base, and internal documentation use cases, within-cluster similarities for well-phrased paraphrases typically land between 0.88 and 0.97. Across-cluster similarities for genuinely different questions typically stay below 0.85. This suggests a threshold in the 0.90–0.93 range for most applications — but domain-specific jargon, short queries, and multilingual inputs can all compress this gap.

---

Embedding Model Selection and Its Impact on Cache Quality

Your choice of embedding model has a direct, measurable impact on cache hit rate and answer quality. This deserves serious attention.

The key dimensions to evaluate are:

Semantic precision: How well does the model distinguish between queries that are similar-sounding but mean different things? This matters for your false positive rate. "How do I cancel my subscription?" and "How do I pause my subscription?" sound similar but require different answers.

Dimensionality: Higher-dimensional embeddings (3072 for text-embedding-3-large) capture more nuance but require more storage and slower ANN search. Lower-dimensional embeddings (text-embedding-3-small at 1536 dimensions, or local models at 384 dimensions) are faster and cheaper but may miss subtle distinctions.

Multilingual support: If your application receives queries in multiple languages, you need a multilingual embedding model. text-embedding-3-small handles this reasonably well. For on-premise, multilingual-e5-large is a strong performer.

Latency: Embedding latency directly adds to your response time on cache misses (and on cache hits, since you must embed the query before searching). Local embedding models can reduce this to single-digit milliseconds, while API calls typically add 50–200ms.

Here's a practical comparison setup:

from sentence_transformers import SentenceTransformer
import time

LOCAL_MODEL = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def embed_local(text: str) -> list[float]:
    return LOCAL_MODEL.encode(text).tolist()

def benchmark_embedding(text: str, n: int = 100):
    # Local model
    start = time.perf_counter()
    for _ in range(n):
        embed_local(text)
    local_ms = (time.perf_counter() - start) / n * 1000

    # OpenAI API
    client = OpenAI()
    start = time.perf_counter()
    for _ in range(n):
        client.embeddings.create(model="text-embedding-3-small", input=text)
    api_ms = (time.perf_counter() - start) / n * 1000

    print(f"Local model: {local_ms:.1f}ms avg")
    print(f"OpenAI API:  {api_ms:.1f}ms avg")

For high-volume production workloads, the math often favors deploying a local embedding model on GPU: you eliminate API latency and per-call embedding costs entirely, at the cost of infrastructure overhead. At 50,000 queries/day, the embedding API calls alone can add $1–5/day depending on query length — meaningful at scale.

---

Multi-Tenancy and Namespace Isolation

Enterprise applications almost always serve multiple tenants — different business units, customer accounts, or product lines — and the cache must isolate their data appropriately. Naively sharing a single cache across tenants creates correctness problems: an answer appropriate for Tenant A's context might be completely wrong for Tenant B.

There are two architectural approaches: namespace filtering and collection-per-tenant.

Namespace filtering (what we implemented above) stores all vectors in a single collection but tags each entry with a namespace identifier and filters on that field during search. This is operationally simple but has a scaling ceiling — as the collection grows, the filtered search has more work to do, and Qdrant (like most vector DBs) filters after the ANN search, not before. At hundreds of thousands of entries per tenant with many tenants, this degrades.

Collection-per-tenant creates a separate Qdrant collection for each tenant. ANN search is fully isolated and scales linearly with tenant size. The downside is operational overhead: you now manage N collections instead of one, and collection creation has overhead that makes it unsuitable for ad-hoc tenant creation.

For most enterprise deployments with a bounded set of known tenants (departments, product lines), collection-per-tenant is the right choice. For SaaS platforms with thousands of customer accounts, namespace filtering is more practical, and you can supplement it with payload indexing to improve filter performance.

class MultiTenantSemanticCache:
    """
    Semantic cache with collection-per-tenant isolation.
    Suitable for enterprise deployments with a known set of tenants.
    """

    EMBEDDING_MODEL = "text-embedding-3-small"
    EMBEDDING_DIM = 1536

    def __init__(self, qdrant_url: str, similarity_threshold: float = 0.92):
        self.client = OpenAI()
        self.qdrant = QdrantClient(url=qdrant_url)
        self.similarity_threshold = similarity_threshold
        self._collections: set[str] = set()

    def _collection_name(self, tenant_id: str) -> str:
        # Sanitize to avoid injection or naming issues
        safe_tenant = "".join(c for c in tenant_id if c.isalnum() or c == "_")
        return f"cache_{safe_tenant}"

    def _ensure_tenant_collection(self, tenant_id: str):
        name = self._collection_name(tenant_id)
        if name not in self._collections:
            existing = {c.name for c in self.qdrant.get_collections().collections}
            if name not in existing:
                self.qdrant.create_collection(
                    collection_name=name,
                    vectors_config=VectorParams(
                        size=self.EMBEDDING_DIM,
                        distance=Distance.COSINE,
                    ),
                )
            self._collections.add(name)

    def lookup(self, query: str, tenant_id: str) -> Optional[str]:
        self._ensure_tenant_collection(tenant_id)
        collection = self._collection_name(tenant_id)
        query_vector = self._embed(query)

        results = self.qdrant.search(
            collection_name=collection,
            query_vector=query_vector,
            limit=1,
            with_payload=True,
        )

        if results and results[0].score >= self.similarity_threshold:
            return results[0].payload.get("response")
        return None

    def store(self, query: str, response: str, tenant_id: str) -> None:
        self._ensure_tenant_collection(tenant_id)
        collection = self._collection_name(tenant_id)
        query_vector = self._embed(query)
        point_id = str(uuid.uuid4())

        self.qdrant.upsert(
            collection_name=collection,
            points=[
                PointStruct(
                    id=point_id,
                    vector=query_vector,
                    payload={
                        "query": query,
                        "response": response,
                        "stored_at": time.time(),
                    },
                )
            ],
        )

    def _embed(self, text: str) -> list[float]:
        return self.client.embeddings.create(
            model=self.EMBEDDING_MODEL, input=text
        ).data[0].embedding

---

Cache Invalidation: The Hard Problem

Phil Karlton famously said there are only two hard things in computer science: cache invalidation and naming things. Semantic caches add a third dimension to the invalidation problem: entries don't just need to expire by time, they need to expire when the underlying knowledge they reflect becomes incorrect.

Consider a customer support cache for a software product. You cache the response to "How do I reset my password?" The app ships a new authentication system three weeks later. Now every cache hit for that question serves wrong information. TTL alone doesn't help unless you've set it short enough to catch the update — but a short TTL defeats the cost savings of caching.

There are four practical strategies:

Strategy 1: TTL with topic-aware duration. Assign different TTLs based on the likely volatility of the information. Policy questions might be stable for 30 days. Pricing questions might be unstable and warrant a 1-day TTL. Feature how-to questions might warrant 7 days. You can implement this by tagging cache entries with a topic category at storage time.

Strategy 2: Event-driven invalidation. When your knowledge base changes, proactively evict cache entries that might be affected. This requires a link between your content management system and your cache — when a knowledge base article is updated, you tag cache entries associated with it and expire them. This is the most precise approach but requires significant integration work.

Strategy 3: Invalidation by metadata tag. Store cache entries with tags corresponding to the source documents or topics they depend on. When a source document changes, batch-delete all cache entries with that tag.

def invalidate_by_tag(
    qdrant: QdrantClient,
    collection_name: str,
    tag: str,
) -> int:
    """
    Delete all cache entries associated with a given topic tag.
    Returns the number of entries deleted.
    """
    # Scroll through matching points
    points_to_delete = []
    offset = None

    while True:
        results, offset = qdrant.scroll(
            collection_name=collection_name,
            scroll_filter=Filter(
                must=[
                    FieldCondition(
                        key="tags",
                        match=MatchValue(value=tag),
                    )
                ]
            ),
            limit=100,
            offset=offset,
            with_payload=False,
        )
        points_to_delete.extend([p.id for p in results])
        if offset is None:
            break

    if points_to_delete:
        qdrant.delete(
            collection_name=collection_name,
            points_selector=points_to_delete,
        )

    return len(points_to_delete)

Strategy 4: Human-in-the-loop review. For high-stakes applications (legal, medical, financial), supplement automated invalidation with a review queue. Flag cache entries that are frequently hit but occasionally followed by user correction signals (negative feedback, follow-up questions). These candidates go to a human reviewer for quality verification before being re-served.

---

Measuring and Monitoring Cache Performance

You can't improve what you don't measure. A semantic cache needs its own observability layer.

The metrics that matter:

Cache hit rate: The percentage of queries resolved from cache. A well-tuned semantic cache for a stable customer support domain should hit 40–70%. Below 20% suggests your threshold is too conservative or your query distribution is too diverse. Above 80% is impressive but warrants scrutiny — are you accidentally serving wrong answers with high confidence?

P50/P95 latency by source: Cache hits and LLM calls have very different latency profiles. Track these separately. If your cache hit latency is approaching your LLM latency, you have a vector search performance problem.

Semantic drift rate: The percentage of cache hits that, upon human or automated review, served an answer that didn't adequately address the actual query. This is your false positive rate, and it's the hardest metric to collect at scale.

Cost per query by source: Track (llm_calls × avg_token_cost) vs. (cache_calls × embedding_cost + vector_search_cost). At sufficient volume, the difference is dramatic.

import collections
from dataclasses import dataclass, field
from threading import Lock

@dataclass
class CacheMetrics:
    hits: int = 0
    misses: int = 0
    total_latency_ms_cache: float = 0.0
    total_latency_ms_llm: float = 0.0
    tokens_saved: int = 0
    _lock: Lock = field(default_factory=Lock)

    def record_hit(self, latency_ms: float):
        with self._lock:
            self.hits += 1
            self.total_latency_ms_cache += latency_ms

    def record_miss(self, latency_ms: float, tokens_used: int):
        with self._lock:
            self.misses += 1
            self.total_latency_ms_llm += latency_ms
            # Approximate: each future cache hit for this response saves tokens
            self.tokens_saved += tokens_used  # optimistic estimate

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

    @property
    def avg_cache_latency_ms(self) -> float:
        return self.total_latency_ms_cache / self.hits if self.hits > 0 else 0.0

    @property
    def avg_llm_latency_ms(self) -> float:
        return self.total_latency_ms_llm / self.misses if self.misses > 0 else 0.0

    def report(self) -> dict:
        return {
            "hit_rate": f"{self.hit_rate:.1%}",
            "total_queries": self.hits + self.misses,
            "cache_hits": self.hits,
            "llm_calls": self.misses,
            "avg_cache_latency_ms": round(self.avg_cache_latency_ms, 1),
            "avg_llm_latency_ms": round(self.avg_llm_latency_ms, 1),
            "estimated_tokens_saved": self.tokens_saved,
        }

Feed these metrics into whatever observability stack you're using — Prometheus, Datadog, CloudWatch. Set alerts for sudden drops in hit rate (which might indicate a schema change in user queries) and spikes in cache latency (which might indicate vector search index degradation).

---

Advanced Pattern: Hierarchical Semantic Caching

For very high-volume applications, a single-layer semantic cache may not be sufficient. Consider a hierarchical architecture with two levels:

Level 1: Exact-match cache (Redis, sub-millisecond) — catches identical queries with zero embedding overhead. User interface patterns, templated queries, and bot traffic often produce exact duplicates.

Level 2: Semantic cache (vector DB, 20–100ms) — catches paraphrases and rewording.

Level 3: LLM (1000–5000ms) — only when the first two miss.

import redis
import hashlib

class HierarchicalCache:
    """
    Two-tier cache: exact match (Redis) + semantic match (Qdrant).
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        qdrant_url: str = "http://localhost:6333",
        similarity_threshold: float = 0.92,
        exact_ttl: int = 3600,       # 1 hour for exact matches
        semantic_ttl: int = 86400,   # 24 hours for semantic matches
    ):
        self.redis = redis.from_url(redis_url)
        self.semantic = SemanticCache(
            similarity_threshold=similarity_threshold,
            ttl_seconds=semantic_ttl,
        )
        self.exact_ttl = exact_ttl

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

    def lookup(self, query: str, namespace: str = "default") -> Optional[dict]:
        # L1: exact match
        key = self._exact_key(query, namespace)
        cached = self.redis.get(key)
        if cached:
            return {"response": cached.decode(), "source": "exact_cache"}

        # L2: semantic match
        response = self.semantic.lookup(query)
        if response:
            # Promote to L1 for future exact matches
            self.redis.setex(key, self.exact_ttl, response)
            return {"response": response, "source": "semantic_cache"}

        return None

    def store(self, query: str, response: str, namespace: str = "default") -> None:
        # Store in both tiers
        key = self._exact_key(query, namespace)
        self.redis.setex(key, self.exact_ttl, response)
        self.semantic.store(query, response)

The "promote to L1" step on semantic cache hits is a subtle optimization worth highlighting: if we serve a semantic match, we store the exact query text as an L1 key too, so the next time this exact phrasing appears, it resolves in under a millisecond.

---

Vector Search Performance Tuning

As your cache grows to hundreds of thousands of entries, ANN search performance becomes a real concern. The HNSW (Hierarchical Navigable Small World) index that most vector databases use by default involves two tuning parameters that matter:

m (number of connections per node): Higher values improve recall (finding the true nearest neighbor) but use more memory. Default is typically 16. For caching, where you care deeply about not returning a false positive, consider increasing to 32.

ef_construction (search depth during indexing): Higher values build a better index but slow down inserts. For a cache where writes are infrequent relative to reads, you can afford a higher value.

ef (search depth at query time): Higher values improve recall at the cost of search latency. This is the hnsw_ef parameter we passed in our search call. For caching, where a false positive is worse than a miss, keep this relatively high (128–256).

# Creating a collection with tuned HNSW parameters
from qdrant_client.models import HnswConfigDiff

qdrant.create_collection(
    collection_name="semantic_cache_tuned",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
    ),
    hnsw_config=HnswConfigDiff(
        m=32,                   # more connections = better recall
        ef_construct=200,       # better index quality at write time
        full_scan_threshold=10_000,  # use exact search below this count
    ),
)

---

Hands-On Exercise

In this exercise, you'll build a semantic cache for a customer support chatbot, measure its performance under a realistic query distribution, and tune the similarity threshold based on empirical data.

Setup:

pip install openai qdrant-client numpy pandas sentence-transformers

Step 1: Generate a realistic query dataset.

Create a file called support_queries.py:

# Simulated customer support queries with intent labels
QUERY_DATASET = [
    # Refund/Return intent
    ("What is the return policy?", "refund"),
    ("How do I get a refund?", "refund"),
    ("Can I return something I bought?", "refund"),
    ("I want to send back an item", "refund"),
    ("How long do I have to return a product?", "refund"),
    ("What's the process for getting my money back?", "refund"),

    # Shipping intent
    ("When will my order arrive?", "shipping"),
    ("How do I track my package?", "shipping"),
    ("Where is my order?", "shipping"),
    ("My shipment is late, what should I do?", "shipping"),
    ("How long does shipping take?", "shipping"),

    # Password reset intent
    ("I forgot my password", "password"),
    ("How do I reset my password?", "password"),
    ("Can't log in to my account", "password"),
    ("I need to change my password", "password"),
    ("My account is locked, how do I get back in?", "password"),

    # Billing intent
    ("I was charged twice", "billing"),
    ("There's an error on my bill", "billing"),
    ("Can you explain my invoice?", "billing"),
    ("I see an unexpected charge", "billing"),
    ("Why was my card charged?", "billing"),
]

Step 2: Run threshold analysis.

Using the analyze_threshold_distribution function from earlier in this lesson, compute within-intent and across-intent similarity scores across the full dataset. Group your queries by intent and find the threshold value that minimizes false positives while maximizing true matches.

Step 3: Build and warm the cache.

Take the first query from each intent group and use it to warm the cache (store it with a handcrafted response). Then run all remaining queries through the cache and record:

• Whether each query hit the cache
• The similarity score of the match
• Whether the cached response was semantically appropriate for the query

Step 4: Compute and report.

Calculate:

• Cache hit rate overall and per intent
• Average similarity score on hits vs. misses
• Any false positives (hits where the cached answer was wrong)

Step 5: Tune and re-run.

Based on your analysis, adjust the threshold up or down by 0.02 and re-run. Does the change improve or worsen your precision/recall trade-off? Document your finding.

This exercise is designed to take about 2 hours. You should come out of it with an empirically grounded threshold recommendation and a working cache implementation.

---

Common Mistakes & Troubleshooting

Mistake 1: Not normalizing query text before embedding.

Trailing whitespace, inconsistent capitalization, and punctuation differences don't meaningfully change semantics, but they can cause slight embedding drift in some models. Normalize your query text before embedding:

def normalize_query(text: str) -> str:
    return " ".join(text.strip().lower().split())

Mistake 2: Caching LLM outputs that depend on external state.

If your prompt template includes real-time data (current date, user account details, inventory levels), the cached response may be valid for the semantic question but wrong for the specific user context. Either exclude these queries from caching or include the dynamic context as part of the cache key. One approach: maintain a secondary "context fingerprint" alongside the semantic key, and only serve a cache hit if both the semantic similarity and the context fingerprint match.

Mistake 3: Treating cache hits as ground truth.

Your cache will accumulate errors over time — wrong answers stored in the early days, outdated responses that slipped past TTL, responses to edge-case queries that happen to match common ones. Implement a sampling strategy: for some percentage of cache hits (say, 1–5%), pass the query through to the LLM anyway and compare the responses. If they diverge significantly, flag the cache entry for review. This "shadow mode" approach catches quality degradation before it compounds.

Mistake 4: Forgetting to account for embedding model version changes.

If you upgrade your embedding model, all existing cached vectors become incompatible. The new model will produce different vector representations for the same text, and similarity scores against old vectors will be meaningless. You must either regenerate all cached embeddings with the new model or maintain separate collections per model version. Plan for this before you need it.

Mistake 5: Underestimating the latency of the embedding call on cache hits.

Cache hits are supposed to be fast, but if you're calling the OpenAI embeddings API to embed the incoming query before searching, you're adding 50–200ms of network latency to every cache hit. This can make your cache hits barely faster than LLM calls. Consider:

• Batching embedding requests
• Using a local embedding model
• Caching frequently-seen queries by exact match first (the hierarchical architecture)

Mistake 6: Ignoring conversation history.

If your application is a multi-turn conversation, individual messages are context-dependent. "Can you tell me more?" is a very different question depending on what's been said before. Naive semantic caching of individual turns without including conversation context will serve wildly wrong responses. For multi-turn applications, either cache only the first turn of conversations (where context is absent), or include a hash of the recent conversation history as part of your cache namespace.

---

Summary & Next Steps

Let's consolidate what we've covered. Semantic caching transforms the economics of high-volume LLM applications by extending traditional caching to work on meaning rather than exact string matches. The core mechanism — embedding queries into vector space and searching for near-neighbors — is conceptually simple but requires careful engineering to deploy correctly.

The key decisions that determine whether your semantic cache succeeds or fails in production are:

• Similarity threshold selection — empirically derived from your actual query distribution, not guessed
• Embedding model choice — balancing quality, latency, and cost, with awareness of versioning implications
• Multi-tenancy architecture — namespace filtering for simplicity, collection-per-tenant for isolation and scale
• Invalidation strategy — TTL alone is insufficient; you need event-driven or topic-tagged invalidation for production correctness
• Observability — hit rate, latency by source, and semantic drift rate are your three non-negotiable metrics

At scale, a well-tuned semantic cache can reduce LLM API costs by 40–70% and cut P50 response latency by 90%. These aren't marginal improvements — they can determine whether a product is economically viable or not.

Where to go next:

• Retrieval-Augmented Generation (RAG): Semantic caching and RAG share foundational infrastructure (embedding models, vector databases), and many production systems combine both. Understanding how a semantic cache interacts with a RAG retrieval pipeline is the natural next step.

• Embedding model fine-tuning: Off-the-shelf embedding models are general-purpose. For specialized domains (legal, medical, financial, code), fine-tuning an embedding model on domain-specific pairs can dramatically improve cache precision.

• Query routing: Semantic similarity isn't only useful for caching — it's also the foundation for routing queries to different LLM models based on topic, complexity, or required expertise. A query classifier built on the same vector infrastructure as your cache is a natural extension.

• LLM response quality evaluation: The shadow mode pattern mentioned in the troubleshooting section is the entry point to automated LLM evaluation. Understanding how to quantify response similarity and semantic correctness opens up a full discipline of LLM quality engineering.

The infrastructure you've built in this lesson — embedding pipelines, vector stores, similarity-based retrieval — is the foundation for almost every advanced LLM system pattern. Understand it deeply, and you'll find it appearing everywhere.