Caching LLM Responses with Redis

LLM caching with Redis means storing a model's response against a key derived from the prompt, then checking that key before you spend tokens on a new call. Done right, it cuts latency from seconds to milliseconds for repeat questions and can eliminate a large chunk of your API bill, since duplicate or near-duplicate prompts are common in chatbots, RAG pipelines, and any endpoint that serves the same handful of questions over and over. This guide builds two working caches: an exact-match cache for identical prompts, and a semantic cache that catches prompts that mean the same thing but aren't worded the same way.

We'll use redis-py for the client, Anthropic's Claude API for the actual generation calls being cached, and Redis Stack's vector search for the semantic layer. Everything here is runnable as-is; swap in your own prompts and model choice where it makes sense.

Why LLM Caching With Redis Beats Rolling Your Own

You could cache in a Python dict, but that dies with the process and doesn't scale past one worker. You could cache in Postgres, but you're paying disk I/O for something that's fundamentally a fast key lookup. Redis sits in the sweet spot: sub-millisecond reads, built-in TTL for expiry, atomic operations so concurrent requests don't race each other, and (with Redis Stack) native vector search so you don't need a separate vector database just for cache lookups.

The other reason Redis wins here is that it's almost certainly already in your stack for session storage, rate limiting, or queues. Adding an LLM cache to an existing Redis instance is a few hundred lines of code, not a new piece of infrastructure.

Two failure modes to keep in mind before you start:

• Caching too aggressively on prompts that include user-specific or time-sensitive data (today's date, account balance, live prices) will serve stale, wrong answers.
• Caching too narrowly (exact string match only) misses the bulk of real-world duplication, because users rarely phrase the same question identically.

The rest of this article addresses both.

Setting Up the Client

Install the dependencies:

pip install redis anthropic

Redis Stack (which bundles RediSearch for vector search) is the easiest way to get both exact-match and semantic caching from one server. If you're running plain Redis (no modules), the exact-match cache below still works fine; you'll just skip the semantic section or run a separate Redis Stack instance for it.

docker run -d --name redis-stack -p 6379:6379 redis/redis-stack-server:latest

Basic client setup:

import os
import redis
from anthropic import Anthropic

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
claude = Anthropic()  # reads ANTHROPIC_API_KEY from the environment

MODEL = "claude-opus-4-8"

If you're running high request volume and want a cheaper model for the underlying calls while you get the cache working, swap MODEL to claude-haiku-4-5. The caching logic below doesn't care which model produced the response.

Exact-Match Caching: The Simple Case

The core idea: hash the prompt (plus anything that changes the output, like the model name and temperature) into a cache key, check Redis for that key, and only call the model on a miss.

import hashlib
import json

def make_cache_key(prompt: str, model: str, system: str = "") -> str:
    """Deterministic key from everything that affects the output."""
    payload = json.dumps(
        {"prompt": prompt, "model": model, "system": system},
        sort_keys=True,
    )
    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    return f"llmcache:exact:{digest}"


def get_cached_response(prompt: str, model: str = MODEL, system: str = "") -> str | None:
    key = make_cache_key(prompt, model, system)
    cached = r.get(key)
    return cached


def set_cached_response(prompt: str, response_text: str, model: str = MODEL,
                         system: str = "", ttl_seconds: int = 3600) -> None:
    key = make_cache_key(prompt, model, system)
    r.set(key, response_text, ex=ttl_seconds)

Wire it into a call:

def ask_claude(prompt: str, system: str = "") -> str:
    cached = get_cached_response(prompt, system=system)
    if cached is not None:
        return cached

    message = claude.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )

    text = next((b.text for b in message.content if b.type == "text"), "")
    set_cached_response(prompt, text, system=system)
    return text

A few details worth calling out:

• sort_keys=True on the JSON dump matters. Without it, dict ordering can vary between processes and you'll silently generate different keys for the identical logical input, which defeats the cache.
• The key includes model and system. If you change either, you want a cache miss, not a wrong-model answer served from an old key.
• We're only caching the text output here. If your response includes tool calls or structured output, serialize the whole thing (JSON) instead of just .text.

This alone handles a surprising amount of traffic in systems with a fixed FAQ, a support bot answering the same handful of questions, or a batch job that reprocesses overlapping inputs. It does nothing, though, for "What's your refund policy?" versus "How do refunds work?" (same intent, different bytes, different hash). That's what semantic caching is for.

Semantic Caching: Catching Near-Duplicate Prompts

Semantic caching embeds the incoming prompt into a vector, searches for the nearest previously-cached prompt vector, and serves that cached response if the similarity is above a threshold. This is where Redis Stack's vector search earns its keep: you get the embedding index and the response store in the same database.

For the embedding model, use a small local model so you're not adding a second paid API call before every cache lookup. sentence-transformers with all-MiniLM-L6-v2 is a common, fast choice (384-dimensional vectors, runs on CPU):

pip install sentence-transformers numpy

from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer("all-MiniLM-L6-v2")
VECTOR_DIM = 384

def embed(text: str) -> np.ndarray:
    return embedder.encode(text, normalize_embeddings=True).astype(np.float32)

Create a vector index in Redis once, at startup:

from redis.commands.search.field import TextField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType

INDEX_NAME = "llmcache_semantic_idx"
PREFIX = "llmcache:semantic:"

def ensure_semantic_index() -> None:
    try:
        r.ft(INDEX_NAME).info()
        return  # already exists
    except redis.exceptions.ResponseError:
        pass

    schema = (
        TextField("prompt"),
        TextField("response"),
        VectorField(
            "embedding",
            "HNSW",
            {
                "TYPE": "FLOAT32",
                "DIM": VECTOR_DIM,
                "DISTANCE_METRIC": "COSINE",
            },
        ),
    )
    r.ft(INDEX_NAME).create_index(
        schema,
        definition=IndexDefinition(prefix=[PREFIX], index_type=IndexType.HASH),
    )

Store a prompt-response pair with its embedding:

def store_semantic_entry(prompt: str, response_text: str, ttl_seconds: int = 3600) -> None:
    vector = embed(prompt).tobytes()
    key = f"{PREFIX}{hashlib.sha256(prompt.encode()).hexdigest()}"
    r.hset(key, mapping={
        "prompt": prompt,
        "response": response_text,
        "embedding": vector,
    })
    r.expire(key, ttl_seconds)

Query for a near match before calling the model:

from redis.commands.search.query import Query

SIMILARITY_THRESHOLD = 0.92  # cosine similarity; tune per use case

def find_semantic_match(prompt: str) -> str | None:
    vector = embed(prompt).tobytes()
    q = (
        Query("*=>[KNN 1 @embedding $vec AS score]")
        .sort_by("score")
        .return_fields("response", "score", "prompt")
        .dialect(2)
    )
    results = r.ft(INDEX_NAME).search(q, query_params={"vec": vector})

    if not results.docs:
        return None

    top = results.docs[0]
    # Redis returns cosine DISTANCE, so lower is more similar; convert to similarity
    similarity = 1 - float(top.score)
    if similarity >= SIMILARITY_THRESHOLD:
        return top.response
    return None

Put it all together:

def ask_claude_semantic(prompt: str, system: str = "") -> str:
    exact = get_cached_response(prompt, system=system)
    if exact is not None:
        return exact

    semantic = find_semantic_match(prompt)
    if semantic is not None:
        return semantic

    message = claude.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    text = next((b.text for b in message.content if b.type == "text"), "")

    set_cached_response(prompt, text, system=system)
    store_semantic_entry(prompt, text)
    return text

Checking exact match first is a deliberate ordering choice: it's cheaper (no embedding call, no vector search) and it's guaranteed correct, so there's no reason to skip it in favor of the fuzzier semantic path.

Threshold tuning matters more than the model you pick. At 0.95+ you'll only match prompts that are nearly word-for-word identical, which barely improves on exact match. At 0.80 or below you'll start serving answers for questions that are related but meaningfully different, which is worse than no cache at all because it's silently wrong. Start around 0.90 to 0.93, log every semantic hit with its similarity score for a week, and manually spot-check the low-confidence ones before tightening or loosening the threshold.

TTL and Cache Invalidation Strategy

There's no single right TTL. Pick it based on how fast the underlying facts change:

• Static reference content (documentation Q&A, policy text that rarely updates): hours to days.
• Product or pricing information: minutes to a few hours, and invalidate explicitly on any update to the source data rather than relying on TTL alone.
• Anything involving a live value (stock price, inventory count, "what time is it"): don't cache, or cache for seconds only.

For explicit invalidation, tag your cache keys so you can wipe a category on demand instead of waiting for TTL:

def invalidate_by_prefix(prefix: str) -> int:
    """Delete all cache entries under a given key prefix. Use SCAN, not KEYS,
    in production so you don't block Redis on a large keyspace."""
    deleted = 0
    for key in r.scan_iter(match=f"{prefix}*", count=500):
        r.delete(key)
        deleted += 1
    return deleted

# Example: wipe every cached answer after updating the refund policy doc
invalidate_by_prefix("llmcache:exact:")

SCAN iterates without holding a lock on the whole keyspace, which is why it's the right choice over KEYS * once you have more than a trivial number of cached entries. KEYS blocks the Redis event loop until it finishes scanning everything, and on a large cache that pause is visible to every other client hitting that Redis instance.

If different parts of your app cache different kinds of content, namespace the keys accordingly (llmcache:exact:faq:, llmcache:exact:support:) so you can invalidate one category without touching the others.

Handling Streaming Responses

If you're streaming tokens to the client, you can still cache, but you need to buffer the full response before writing to Redis and decide whether to serve cache hits as a stream or all at once:

def ask_claude_streaming(prompt: str, system: str = ""):
    cached = get_cached_response(prompt, system=system)
    if cached is not None:
        yield cached
        return

    chunks = []
    with claude.messages.stream(
        model=MODEL,
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            chunks.append(text)
            yield text

    full_text = "".join(chunks)
    set_cached_response(prompt, full_text, system=system)

Cache hits here return the whole response in one yield rather than a token-by-token stream. That's usually fine, since a cache hit means you already have the full text in hand and there's no reason to artificially throttle its delivery. If your frontend expects a consistent streaming cadence regardless of cache status, chunk the cached string into small pieces and yield those with a short delay.

Measuring Whether the Cache Is Actually Helping

Don't just deploy this and assume it's working. Track hit rate and cost savings explicitly:

def record_cache_event(hit_type: str) -> None:
    """hit_type is one of: exact_hit, semantic_hit, miss"""
    day_key = f"llmcache:stats:{hit_type}"
    r.incr(day_key)

def cache_stats() -> dict:
    return {
        "exact_hits": int(r.get("llmcache:stats:exact_hit") or 0),
        "semantic_hits": int(r.get("llmcache:stats:semantic_hit") or 0),
        "misses": int(r.get("llmcache:stats:miss") or 0),
    }

Call record_cache_event at each branch in ask_claude_semantic. A healthy production cache for a support or FAQ-style workload usually settles somewhere between 30% and 60% combined hit rate once you have a reasonable volume of repeat traffic; if you're seeing single digits, your TTL is probably too short, your similarity threshold is too strict, or your traffic genuinely doesn't have much repetition and caching isn't the right lever for that endpoint.

Common Pitfalls

• Caching prompts that embed a timestamp or session ID. If your prompt template includes "Today is {date}" or a user ID, every request generates a unique hash and the exact-match cache never hits. Strip volatile fields out of the cache key, or move them into a separate parameter that doesn't participate in hashing, while keeping them in the actual prompt sent to the model.
• Sharing a cache across users with different permissions. If two users can ask the same literal question but should get different answers based on their access level, include a permission or tenant identifier in the cache key. Otherwise you'll leak one user's data to another.
• Trusting semantic similarity blindly. "Cancel my subscription" and "How do I pause my subscription" can score high on cosine similarity while requiring different answers. Log every semantic hit for a while before trusting the threshold in production, especially for anything with financial or legal consequences.
• Forgetting to version the cache when you change the system prompt. If you tweak your system instructions, old cached responses reflect the old behavior. Either fold the system prompt into the cache key (as shown above) or add a version string to the key and bump it on every prompt change.
• Skipping error responses correctly, but also skipping partial ones. Don't cache a response if the model call errored out, and don't cache a response that was cut off by hitting max_tokens. Check stop_reason before writing to the cache, since a truncated answer served from cache repeatedly is worse than a fresh, complete one.

FAQ

Does Redis caching reduce LLM API costs? Yes, directly. Every cache hit is a request you don't send to the model provider, so your token spend drops in proportion to your hit rate. A 40% hit rate on a workload with meaningful repetition translates roughly to a 40% reduction in generation costs for that endpoint, minus the small overhead of running Redis itself.

Is exact-match caching or semantic caching better? They solve different problems and work best combined. Exact-match is nearly free (a hash lookup) and 100% correct when it hits, but it only catches identical prompts. Semantic caching catches paraphrases and near-duplicates but costs an embedding computation per request and carries a small risk of false positives if the similarity threshold is too loose. Check exact match first, fall through to semantic, and only call the model on a full miss.

What TTL should I use for an LLM cache? Base it on how quickly the answer becomes stale, not on a fixed default. Static or reference content can live for hours or days; anything tied to changing data (pricing, inventory, current events) should have a short TTL or skip caching entirely. When in doubt, start conservative (minutes) and lengthen it once you've confirmed the underlying content doesn't change on a shorter cycle.

Can I cache tool-use or structured output responses, not just plain text? Yes. Serialize the full response content (including tool_use blocks) as JSON before writing to Redis, and deserialize it on a cache hit rather than just storing the text field. The cache key logic doesn't change; only what you store and retrieve does.

What happens if Redis goes down? Treat the cache as an optimization, not a dependency. Wrap your Redis calls in a try/except that falls through to calling the model directly on any connection error, so a Redis outage degrades your latency and cost profile but doesn't take down the feature. Don't let a cache failure become an application failure.

Does semantic caching need a separate vector database? Not if you're already running Redis Stack, which bundles RediSearch and supports vector fields with KNN search directly on hash or JSON documents. That's enough for most LLM caching use cases: you don't need a dedicated vector database unless you're also doing large-scale retrieval-augmented generation with millions of documents, which is a different workload from caching a moderate number of prompt-response pairs.