Building Production-Grade Semantic Cache for LLMs with Redis and OpenAI Embeddings (2026 Guide)

Why Exact-Match Caching Fails for Generative AI

In high-scale enterprise applications powered by Large Language Models (LLMs), API inference costs and latency present massive operational hurdles. Traditional HTTP or in-memory caches rely on exact string hashing (MD5 or SHA-256). If a user asks “How do I reset my password?” and another asks “Steps to change forgotten password?”, exact-match caches miss 100% of the time, incurring redundant API calls and 2–5 second latency delays.

By implementing a Semantic Cache using Redis Vector Search and dense text embeddings, applications can identify conceptually identical questions and return cached responses in under 15 milliseconds, slashing LLM API expenditures by up to 68%.

Documented by Saizul Amin for AI systems engineering and production ML infrastructure.

Semantic Caching Architecture & Workflow

1. Query Vectorization: User prompt is converted into a high-dimensional vector embedding using text-embedding-3-small.
2. K-Nearest Neighbors (KNN) Search: Redis searches existing cached vectors using cosine distance.
3. Threshold Evaluation: If the cosine similarity exceeds a strict threshold (e.g., > 0.92), return the cached LLM completion instantly (Cache Hit).
4. LLM Fallback & Cache Write: If below threshold (Cache Miss), forward prompt to Claude/GPT-4o, return completion to user, and asynchronously persist vector + answer into Redis.

Python Implementation: Fast Semantic Cache with Redis-Py

Below is a production-ready Python service implementing semantic caching:

import numpy as np
import redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from openai import OpenAI

# Initialize clients
client = OpenAI()
r = redis.Redis(host='\''localhost'\'', port=6379, decode_responses=False)

INDEX_NAME = "idx:semantic_cache"
VECTOR_DIM = 1536  # Dimension for text-embedding-3-small
SIMILARITY_THRESHOLD = 0.92

# Ensure Vector Search Index Exists
def init_redis_index():
    try:
        r.ft(INDEX_NAME).info()
    except Exception:
        schema = (
            TextField("prompt"),
            TextField("response"),
            VectorField(
                "prompt_vector",
                "HNSW",
                {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": "COSINE"}
            )
        )
        r.ft(INDEX_NAME).create_index(
            schema,
            definition=IndexDefinition(prefix=["cache:"], index_type=IndexType.HASH)
        )
        print("Initialized Redis Semantic Cache Vector Index.")

def get_embedding(text: str) -> list[float]:
    res = client.embeddings.create(input=[text], model="text-embedding-3-small")
    return res.data[0].embedding

def query_semantic_cache(prompt: str):
    prompt_vector = np.array(get_embedding(prompt), dtype=np.float32).tobytes()
    
    # KNN Vector Query
    q = Query(f"*=>[KNN 1 @prompt_vector $vec AS score]")        .sort_by("score")        .return_fields("prompt", "response", "score")        .dialect(2)
        
    results = r.ft(INDEX_NAME).search(q, query_params={"vec": prompt_vector})
    
    if results.docs:
        top_match = results.docs[0]
        # Cosine distance = 1 - similarity. Score is distance in Redis.
        distance = float(top_match.score)
        similarity = 1.0 - distance
        
        if similarity >= SIMILARITY_THRESHOLD:
            return {
                "source": "CACHE_HIT",
                "similarity": round(similarity, 4),
                "response": top_match.response.decode('\''utf-8'\'') if isinstance(top_match.response, bytes) else top_match.response
            }

    # Cache Miss: Call LLM
    completion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    answer = completion.choices[0].message.content

    # Store in Redis Cache
    cache_id = f"cache:{abs(hash(prompt))}"
    r.hset(cache_id, mapping={
        "prompt": prompt,
        "response": answer,
        "prompt_vector": prompt_vector
    })
    r.expire(cache_id, 86400 * 7) # 7 days TTL

    return {
        "source": "LLM_GENERATED",
        "similarity": 0.0,
        "response": answer
    }

Operational Caveats: When NOT to Use Semantic Caching

• Time-Sensitive Queries: Queries requesting live financial data, stock quotes, or real-time sports results must bypass the semantic cache.
• Personalized / PII Content: Prompts containing user account credentials or user-specific private records should never be stored in shared semantic vector indexes.