AI Engineering Academy · Lección

Caché semántica con embeddings

Caché semántica con embeddings es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

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.

Preguntas frecuentes

¿La lección «Caché semántica con embeddings» es gratis?

Sí — el texto completo de «Caché semántica con embeddings» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Caché semántica con embeddings»?

Cree una caché semántica que recupere respuestas almacenadas para consultas semánticamente similares, aunque no idénticas, comparando los embeddings de las consultas con una caché de embeddings de so… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.

¿Cuánto tiempo toma la lección «Caché semántica con embeddings»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Caché exacta con Redis
Caché semántica con embeddings
Caché de prefijos de prompts de OpenAI
Procesamiento por lotes, enrutamiento de modelos y paneles de costes

← Volver a AI Engineering Academy