AI Engineering Academy · Lesson

Semantic Caching with Embeddings

Build a semantic cache that retrieves stored responses for semantically similar but not identical queries by comparing query embeddings to a cache of previous request embeddings.

Semantic Caching with Embeddings is a free AI Engineering Academy lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

The Limitation of Exact Caching

Exact caching only helps when users send byte-for-byte identical requests. In reality, users phrase the same question differently: 'How do I cancel my subscription?', 'What is the process to unsubscribe?', and 'Can I stop my plan?' all intend the same question but produce different cache keys. Exact caching misses all these variants. Semantic caching solves this by matching similar queries instead of identical ones, dramatically increasing cache hit rates.

How Semantic Caching Works

A semantic cache stores the embedding of each cached query alongside the cached response. When a new query arrives, embed it and search the cache for a previously seen query with high cosine similarity. If the closest cached query is above a similarity threshold (typically 0.95+), return its cached response. If no match is found, call the LLM, store the new response, and add the new query's embedding to the cache index for future lookups.

# Semantic cache flow
# 1. New query arrives: 'How do I cancel my subscription?'
# 2. Embed it: embed_query = embed('How do I cancel my subscription?')
# 3. Search cache index for nearest cached query embedding
# 4. Find cached: 'What is the process to unsubscribe?' (similarity=0.97)
# 5. 0.97 >= threshold (0.95) → cache HIT, return cached response
# 6. If 0.82 < threshold → cache MISS, call LLM, cache result, add embedding to index

In-Memory Semantic Cache with NumPy

For small applications or prototypes, implement semantic caching in memory using NumPy for cosine similarity computation. Store cached query embeddings in a 2D array and responses in a parallel list. On each new query, compute cosine similarity between the new embedding and all cached embeddings, and return the closest match if it exceeds the threshold.

import numpy as np
from openai import OpenAI

client = OpenAI()

class InMemorySemanticCache:
    def __init__(self, threshold: float = 0.95):
        self.threshold = threshold
        self.embeddings = []    # list of np.ndarray
        self.responses = []     # list of str
        self.queries = []       # list of str (for inspection)

    def _embed(self, text: str) -> np.ndarray:
        resp = client.embeddings.create(model='text-embedding-3-small', input=text)
        return np.array(resp.data[0].embedding)

    def get(self, query: str) -> str | None:
        if not self.embeddings:
            return None
        q_emb = self._embed(query)
        cache_matrix = np.array(self.embeddings)
        # Cosine similarity: dot product of normalized vectors
        norms = np.linalg.norm(cache_matrix, axis=1)
        q_norm = np.linalg.norm(q_emb)
        sims = (cache_matrix @ q_emb) / (norms * q_norm + 1e-8)
        best_idx = int(np.argmax(sims))
        if sims[best_idx] >= self.threshold:
            print(f'[SEMANTIC HIT] sim={sims[best_idx]:.3f} matched: {self.queries[best_idx]!r}')
            return self.responses[best_idx]
        return None

    def set(self, query: str, response: str):
        emb = self._embed(query)
        self.embeddings.append(emb)
        self.responses.append(response)
        self.queries.append(query)

Semantic Cache with Redis and Pinecone

For production semantic caching, store query embeddings in a vector database for fast approximate nearest neighbor search, and store responses in Redis keyed by a unique ID. When a new query arrives, search the vector database for the closest cached query, retrieve the response from Redis using the ID in the vector metadata, and return it — all without calling the LLM.

import redis
from pinecone import Pinecone
import hashlib

r = redis.Redis(decode_responses=True)
pc = Pinecone(api_key='YOUR_KEY')
index = pc.Index('semantic-cache')

SIMILARITY_THRESHOLD = 0.95

def semantic_cache_get(query: str) -> str | None:
    q_emb = embed(query)  # from earlier lesson
    results = index.query(vector=q_emb, top_k=1, include_metadata=True)
    if not results.matches:
        return None
    best = results.matches[0]
    if best.score >= SIMILARITY_THRESHOLD:
        response_key = best.metadata.get('response_key')
        return r.get(response_key)
    return None

def semantic_cache_set(query: str, response: str):
    q_emb = embed(query)
    entry_id = hashlib.sha256(query.encode()).hexdigest()[:16]
    response_key = f'sem_cache_resp:{entry_id}'
    r.setex(response_key, 86400, response)  # 24h TTL
    index.upsert(vectors=[{
        'id': entry_id,
        'values': q_emb,
        'metadata': {'query': query[:200], 'response_key': response_key},
    }])

GPTCache: A Ready-Made Semantic Cache

GPTCache is an open-source library that implements semantic caching for LLM applications. It wraps the OpenAI client, embeds queries automatically, checks a vector similarity cache, and falls back to the real API on misses. It supports multiple vector stores (FAISS, Milvus, Redis) and multiple embedding models out of the box, making it a fast way to add semantic caching to an existing application.

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

# Configure GPTCache with FAISS vector store
cache.init(
    embedding_func=EmbeddingOpenAI().to_embeddings,
    data_manager=get_data_manager(
        CacheBase('sqlite'),
        VectorBase('faiss', dimension=1536),
    ),
    similarity_evaluation=SearchDistanceEvaluation(),
)

# Now use the wrapped openai client — caching is transparent
response = openai.ChatCompletion.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'What is RAG?'}],
)

Choosing the Similarity Threshold

The similarity threshold is the most important hyperparameter for semantic caching. Too high (0.99+) and you miss most paraphrase variants. Too low (0.85-) and you return wrong cached answers for different but superficially similar queries. Validate your threshold empirically by sampling query pairs and checking whether queries above the threshold truly have the same intended answer. Typical values: 0.92-0.97 for factual Q&A, 0.98+ for code generation where small differences matter a lot.

def validate_threshold(cache, query_pairs_with_labels):
    '''
    query_pairs_with_labels: list of (query1, query2, should_match: bool)
    '''
    true_pos = true_neg = false_pos = false_neg = 0
    for q1, q2, should_match in query_pairs_with_labels:
        e1, e2 = embed(q1), embed(q2)
        sim = cosine_similarity(e1, e2)
        matched = sim >= cache.threshold
        if should_match and matched: true_pos += 1
        elif not should_match and not matched: true_neg += 1
        elif not should_match and matched: false_pos += 1
        else: false_neg += 1
    precision = true_pos / (true_pos + false_pos) if (true_pos + false_pos) else 0
    recall = true_pos / (true_pos + false_neg) if (true_pos + false_neg) else 0
    print(f'Precision: {precision:.3f}, Recall: {recall:.3f}')

Semantic Cache Scope: System Prompt Matters

A critical detail: semantic caching must account for the system prompt. Two identical user queries produce different answers if the system prompt differs (different personas, different knowledge bases, different response formats). Always include the system prompt in the embedding input or create separate cache namespaces per system prompt. A clean pattern is to hash the system prompt and use it as a cache namespace prefix.

import hashlib

def make_semantic_cache_namespace(system_prompt: str) -> str:
    return 'sc:' + hashlib.md5(system_prompt.encode()).hexdigest()[:8]

def semantic_cache_get_namespaced(system_prompt: str, user_query: str) -> str | None:
    namespace = make_semantic_cache_namespace(system_prompt)
    q_emb = embed(user_query)
    # Search only within this namespace
    results = index.query(
        vector=q_emb,
        top_k=1,
        filter={'namespace': namespace},
        include_metadata=True,
    )
    if results.matches and results.matches[0].score >= SIMILARITY_THRESHOLD:
        return r.get(results.matches[0].metadata['response_key'])
    return None

Semantic Cache Hit Rate Analysis

After deploying semantic caching, analyze hit rates segmented by query cluster. Use the cached query embeddings themselves — cluster them with K-means and compute hit rate per cluster. High-hit clusters represent common question themes where caching pays off most. Low-hit clusters of diverse unique queries may not benefit from caching at all and could be excluded from the cache to reduce index size and embedding costs.

from sklearn.cluster import KMeans
import numpy as np

def analyze_cache_clusters(cache, n_clusters=10):
    if len(cache.embeddings) < n_clusters:
        print('Not enough cache entries to cluster')
        return

    matrix = np.array(cache.embeddings)
    kmeans = KMeans(n_clusters=n_clusters, n_init=10, random_state=42)
    labels = kmeans.fit_predict(matrix)

    from collections import Counter
    cluster_sizes = Counter(labels)
    print('Query clusters by size:')
    for cluster_id, count in cluster_sizes.most_common():
        representative = cache.queries[labels.tolist().index(cluster_id)]
        print(f'  Cluster {cluster_id}: {count} queries, e.g. {representative!r}')

Semantic Cache Security Considerations

Semantic caching introduces a privacy risk: if user A asks a sensitive question, its response might be returned to user B who asks a similar question. This is acceptable for public knowledge bases but not for applications with user-specific data or sensitive content. Apply strict namespace isolation per user or organization, and consider excluding queries that match patterns like personal information from the cache entirely.

import re

PII_PATTERNS = [
    r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',   # phone numbers
    r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b',  # emails
    r'\b\d{9}\b',                           # SSN-like
]

def should_cache(query: str) -> bool:
    for pattern in PII_PATTERNS:
        if re.search(pattern, query, re.IGNORECASE):
            return False  # do not cache queries with PII
    return True

def secure_semantic_completion(user_id: str, query: str) -> str:
    if should_cache(query):
        cached = semantic_cache_get_namespaced(f'user:{user_id}', query)
        if cached:
            return cached
    result = call_llm_api(query)  # actual API call
    if should_cache(query):
        semantic_cache_set_namespaced(f'user:{user_id}', query, result)
    return result

Combining Exact and Semantic Caching

The most efficient caching strategy uses both exact and semantic caching in a two-tier hierarchy. Check exact cache first (fastest, zero embedding cost) and return immediately on hit. If exact cache misses, check semantic cache (requires one embedding API call). If semantic cache misses, call the LLM. This ordering minimizes both latency and cost per request.

async def two_tier_cached_completion(messages: list[dict], model: str = 'gpt-4o-mini') -> str:
    user_query = messages[-1].get('content', '')
    system_prompt = messages[0].get('content', '') if messages and messages[0]['role'] == 'system' else ''

    # Tier 1: exact cache (instant, free)
    exact_key = make_cache_key(messages, model, temperature=0.0)
    exact_cached = await async_r.get(exact_key)
    if exact_cached:
        return json.loads(exact_cached)

    # Tier 2: semantic cache (one embedding call ~5ms)
    sem_result = semantic_cache_get_namespaced(system_prompt, user_query)
    if sem_result:
        # Backfill exact cache to avoid embedding next time
        await async_r.setex(exact_key, 3600, json.dumps(sem_result))
        return sem_result

    # Tier 3: actual LLM call
    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=0.0
    )
    result = response.choices[0].message.content
    await async_r.setex(exact_key, 3600, json.dumps(result))
    semantic_cache_set_namespaced(system_prompt, user_query, result)
    return result

Cache Warming for Cold Start

A fresh semantic cache provides zero benefit until it is populated. For applications with predictable traffic patterns, pre-warm the cache at startup by embedding and caching responses for the most frequently asked questions from your historical query logs. This eliminates the cold-start period where every user during the first hours of operation gets a cache miss and incurs full API cost.

async def warm_semantic_cache(faq_list: list[dict], system_prompt: str):
    print(f'Warming cache with {len(faq_list)} FAQ entries...')
    for entry in faq_list:
        cached = semantic_cache_get_namespaced(system_prompt, entry['question'])
        if cached:
            print(f'  Already cached: {entry["question"][:50]}')
            continue
        # Generate and cache the response
        response = await async_client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': entry['question']},
            ],
            temperature=0.0,
        )
        answer = response.choices[0].message.content
        semantic_cache_set_namespaced(system_prompt, entry['question'], answer)
        print(f'  Cached: {entry["question"][:50]}')
    print('Cache warming complete')

Quick Check

Test your understanding of semantic caching from this lesson.

Lesson Recap

In this lesson you learned: semantic caching matches similar but non-identical queries by comparing query embeddings against a vector store of cached query embeddings, the similarity threshold controls the trade-off between hit rate and answer correctness, and system prompt namespacing prevents incorrect cross-context cache hits. A two-tier architecture checking exact cache before semantic cache minimizes both latency and embedding costs. Next up we leverage OpenAI's built-in prompt prefix caching.

Frequently asked questions

Is the “Semantic Caching with Embeddings” lesson free?

Yes — the full text of “Semantic Caching with Embeddings” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Semantic Caching with Embeddings”?

Build a semantic cache that retrieves stored responses for semantically similar but not identical queries by comparing query embeddings to a cache of previous request embeddings. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Semantic Caching with Embeddings” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

Exact Caching with Redis
Semantic Caching with Embeddings
OpenAI Prompt Prefix Caching
Batching, Model Routing, and Cost Dashboards

← Back to AI Engineering Academy