Smart RAG Caching: Cut Your LLM Costs by 80% (Without Losing Quality)

TL;DR

Caching is the #1 optimization to reduce RAG costs. By combining semantic cache, prompt caching (Anthropic/OpenAI), and embedding cache, you can cut your LLM costs by 60-80% while improving latency. This guide compares caching approaches (exact match, semantic, prompt caching), solutions (Redis, GPTCache, custom), and includes concrete ROI calculations.

Why Caching Is Critical for RAG

The Hidden Cost of a RAG Without Cache

Take a typical RAG chatbot serving 10,000 queries/day:

With a cache at 40% hit rate:

With an optimized cache at 70% hit rate:

Savings by Cache Level

No cache:        $4,836/mo  ████████████████████████████████
Basic cache:     $2,951/mo  ████████████████████
Optimized cache: $1,501/mo  ██████████
Advanced cache:  $967/mo    ██████
                              ↑ -80% cost reduction

The 5 Caching Strategies for RAG

Strategy 1: Exact Match Cache

The simplest. Same question = same answer.

DEVELOPERpython
import hashlib
import json
import redis

class ExactMatchCache:
    """Cache by exact question match."""

    def __init__(self, redis_url: str, ttl_seconds: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl_seconds

    def _make_key(self, query: str, context_hash: str = "") -> str:
        """Generate a deterministic cache key."""
        normalized = query.strip().lower()
        raw = f"{normalized}:{context_hash}"
        return f"rag:exact:{hashlib.sha256(raw.encode()).hexdigest()}"

    def get(self, query: str, context_hash: str = "") -> dict | None:
        key = self._make_key(query, context_hash)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None

    def set(
        self, query: str, response: dict, context_hash: str = ""
    ):
        key = self._make_key(query, context_hash)
        self.redis.setex(key, self.ttl, json.dumps(response))

    def invalidate(self, query: str, context_hash: str = ""):
        key = self._make_key(query, context_hash)
        self.redis.delete(key)


# Usage in the RAG pipeline
cache = ExactMatchCache("redis://localhost:6379", ttl_seconds=7200)

async def rag_with_cache(query: str, user_id: str) -> str:
    # 1. Check cache
    cached = cache.get(query)
    if cached:
        return cached["response"]  # Hit! No LLM call

    # 2. Normal RAG pipeline
    response = await full_rag_pipeline(query)

    # 3. Store in cache
    cache.set(query, {"response": response, "timestamp": time.time()})

    return response

Pros: Simple, fast, zero false positives. Cons: Low hit rate (10-20%), does not catch rephrased queries.

Strategy 2: Semantic Cache

Cache based on semantic similarity. "How to configure X?" and "I want to set up X" return the same result.

DEVELOPERpython
import numpy as np
from typing import Optional

class SemanticCache:
    """Semantic cache based on embeddings."""

    def __init__(
        self,
        embedding_model,
        redis_client,
        similarity_threshold: float = 0.92,
        ttl_seconds: int = 7200
    ):
        self.embedder = embedding_model
        self.redis = redis_client
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
        self._cache_embeddings = []
        self._cache_keys = []

    async def get(self, query: str) -> Optional[dict]:
        """Search for a semantically similar response."""
        query_embedding = await self.embedder.embed(query)

        if not self._cache_embeddings:
            return None

        # Calculate similarities
        similarities = np.dot(
            np.array(self._cache_embeddings),
            query_embedding
        )
        max_idx = np.argmax(similarities)
        max_sim = similarities[max_idx]

        if max_sim >= self.threshold:
            key = self._cache_keys[max_idx]
            cached = self.redis.get(key)
            if cached:
                return json.loads(cached)

        return None

    async def set(self, query: str, response: dict):
        """Store a response with its embedding."""
        query_embedding = await self.embedder.embed(query)
        key = f"rag:semantic:{hashlib.sha256(query.encode()).hexdigest()}"

        self.redis.setex(key, self.ttl, json.dumps(response))
        self._cache_embeddings.append(query_embedding)
        self._cache_keys.append(key)

    def set_threshold(self, threshold: float):
        """Adjust the similarity threshold.

        Higher = fewer hits but more precise.
        Lower = more hits but risk of incorrect responses.
        """
        self.threshold = threshold


# Recommended thresholds by use case
THRESHOLDS = {
    "technical_support": 0.95,    # Precision critical
    "general_faq": 0.90,          # Good balance
    "product_search": 0.88,       # More hits acceptable
    "informal_chat": 0.85,        # High tolerance
}

Pros: High hit rate (40-60%), catches rephrased queries. Cons: Embedding cost per query, risk of false positives.

Strategy 3: Prompt Caching (Anthropic / OpenAI)

Providers offer native caching of prompt prefixes.

DEVELOPERpython
# Anthropic Prompt Caching
import anthropic

client = anthropic.Anthropic()

# System prompt and documents are cached Anthropic-side
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a specialized RAG assistant...",
            "cache_control": {"type": "ephemeral"}
        },
        {
            "type": "text",
            "text": f"Reference documents:\n{all_documents}",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": query}]
)

# Check cache usage
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Cache read: {response.usage.cache_read_input_tokens}")
print(f"Cache creation: {response.usage.cache_creation_input_tokens}")
# Cache read = 90% cheaper than normal input tokens

DEVELOPERpython
# OpenAI Prompt Caching (automatic since 2024)
from openai import OpenAI

client = OpenAI()

# Identical prefix is automatically cached
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": long_system_prompt  # Auto-cached
        },
        {
            "role": "user",
            "content": f"Documents:\n{documents}\n\nQuestion: {query}"
        }
    ]
)

# OpenAI: 50% reduction on cached tokens
print(f"Cached tokens: {response.usage.prompt_tokens_details.cached_tokens}")

Prompt caching savings:

Strategy 4: Embedding Cache

Cache embeddings to avoid recomputation.

DEVELOPERpython
class EmbeddingCache:
    """Cache embeddings to avoid API calls."""

    def __init__(self, redis_client, ttl_seconds: int = 86400):
        self.redis = redis_client
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    async def get_or_compute(
        self,
        text: str,
        embedding_fn
    ) -> list[float]:
        """Return embedding from cache or compute it."""
        key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"

        # Try cache
        cached = self.redis.get(key)
        if cached:
            self.hits += 1
            return json.loads(cached)

        # Compute and store
        self.misses += 1
        embedding = await embedding_fn(text)
        self.redis.setex(key, self.ttl, json.dumps(embedding))

        return embedding

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


# Usage
emb_cache = EmbeddingCache(redis_client)

async def embed_query(query: str) -> list[float]:
    return await emb_cache.get_or_compute(
        query,
        lambda q: openai_embed(q)
    )

Strategy 5: Retrieval Cache

Cache vector search results.

DEVELOPERpython
class RetrievalCache:
    """Cache vector search results."""

    def __init__(
        self,
        redis_client,
        ttl_seconds: int = 1800,
        max_results: int = 10
    ):
        self.redis = redis_client
        self.ttl = ttl_seconds
        self.max_results = max_results

    def _cache_key(self, query_embedding: list[float]) -> str:
        """Key based on quantized embedding."""
        quantized = [round(x, 3) for x in query_embedding[:32]]
        return f"ret:{hashlib.md5(str(quantized).encode()).hexdigest()}"

    async def get_or_search(
        self,
        query_embedding: list[float],
        search_fn
    ) -> list[dict]:
        key = self._cache_key(query_embedding)

        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        results = await search_fn(query_embedding)
        self.redis.setex(
            key,
            self.ttl,
            json.dumps(results[:self.max_results])
        )

        return results

Caching Approaches Comparison

Redis vs GPTCache vs Custom

Comparison Table

GPTCache - Implementation

DEVELOPERpython
from gptcache import cache
from gptcache.adapter import openai as gptcache_openai
from gptcache.embedding import Onnx
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

# GPTCache configuration
onnx = Onnx()
data_manager = get_data_manager(
    CacheBase("sqlite"),
    VectorBase("faiss", dimension=onnx.dimension)
)

cache.init(
    embedding_func=onnx.to_embeddings,
    data_manager=data_manager,
    similarity_evaluation=SearchDistanceEvaluation(),
)
cache.set_openai_key()

# Transparent usage (replaces OpenAI API)
response = gptcache_openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ],
)
# GPTCache automatically checks semantic cache

Redis - Advanced Implementation

DEVELOPERpython
import redis
from redis.commands.search.field import (
    VectorField, TextField, NumericField
)
from redis.commands.search.indexDefinition import IndexDefinition
from redis.commands.search.query import Query

class RedisSemanticCache:
    """Semantic cache with Redis Stack (RediSearch)."""

    def __init__(self, redis_url: str, embedding_dim: int = 1536):
        self.redis = redis.from_url(redis_url)
        self.dim = embedding_dim
        self._create_index()

    def _create_index(self):
        try:
            self.redis.ft("rag_cache").info()
        except Exception:
            schema = (
                TextField("query"),
                TextField("response"),
                NumericField("timestamp"),
                NumericField("ttl"),
                VectorField(
                    "embedding",
                    "FLAT",
                    {
                        "TYPE": "FLOAT32",
                        "DIM": self.dim,
                        "DISTANCE_METRIC": "COSINE",
                    }
                ),
            )
            self.redis.ft("rag_cache").create_index(
                schema,
                definition=IndexDefinition(prefix=["cache:"])
            )

    async def search(
        self,
        query_embedding: list[float],
        threshold: float = 0.92,
    ) -> dict | None:
        query_bytes = np.array(
            query_embedding, dtype=np.float32
        ).tobytes()

        q = (
            Query(f"*=>[KNN 1 @embedding $vec AS score]")
            .sort_by("score")
            .return_fields("query", "response", "score")
            .dialect(2)
        )

        results = self.redis.ft("rag_cache").search(
            q, query_params={"vec": query_bytes}
        )

        if results.docs:
            doc = results.docs[0]
            if float(doc.score) <= (1 - threshold):
                return {
                    "query": doc.query,
                    "response": doc.response,
                    "similarity": 1 - float(doc.score)
                }

        return None

When NOT to Use Cache

Cases Where Caching Is Counterproductive

Invalidation Strategies

DEVELOPERpython
class CacheInvalidator:
    """Intelligent RAG cache invalidation."""

    def __init__(self, cache: SemanticCache):
        self.cache = cache

    async def on_document_updated(self, doc_id: str):
        """Invalidate cache when a document is updated."""
        affected_keys = await self.find_keys_by_doc(doc_id)
        for key in affected_keys:
            self.cache.invalidate(key)

    async def on_knowledge_base_refresh(self):
        """Invalidate all cache on full refresh."""
        self.cache.flush_all()

    def schedule_ttl_cleanup(self, max_age_hours: int = 24):
        """Clean entries that are too old."""
        cutoff = time.time() - (max_age_hours * 3600)
        self.cache.delete_older_than(cutoff)

Detailed ROI Calculation

Scenario: E-commerce Support Chatbot

Volume: 15,000 queries/day
Model: GPT-4o ($2.50/1M input, $10/1M output)
Average tokens: 2,000 input, 500 output per query

12-Month ROI

Investment:
  - Cache development: 40h x $100/h = $4,000
  - Redis infrastructure: $100/mo x 12 = $1,200
  - Total investment: $5,200

Annual savings (combined cache 75%):
  - $3,276/mo x 12 = $39,312

ROI = ($39,312 - $5,200) / $5,200 = 656%
Payback period: < 2 months

Complete Cache Pipeline

DEVELOPERpython
class RAGCachePipeline:
    """Multi-layer cache pipeline for RAG."""

    def __init__(self):
        self.exact_cache = ExactMatchCache(redis_url, ttl=7200)
        self.semantic_cache = SemanticCache(
            embedder, redis_client, threshold=0.92
        )
        self.embedding_cache = EmbeddingCache(redis_client)
        self.retrieval_cache = RetrievalCache(redis_client)
        self.metrics = CacheMetrics()

    async def process(self, query: str) -> str:
        # Layer 1: Exact match (fastest)
        exact = self.exact_cache.get(query)
        if exact:
            self.metrics.record("exact_hit")
            return exact["response"]

        # Layer 2: Semantic cache
        semantic = await self.semantic_cache.get(query)
        if semantic:
            self.metrics.record("semantic_hit")
            return semantic["response"]

        # Layer 3: Embedding cache (for embedding)
        embedding = await self.embedding_cache.get_or_compute(
            query, self.embed_fn
        )

        # Layer 4: Retrieval cache
        docs = await self.retrieval_cache.get_or_search(
            embedding, self.search_fn
        )

        # Layer 5: Generation (prompt caching provider-side)
        response = await self.generate_with_prompt_cache(
            query, docs
        )

        # Store in caches
        result = {"response": response, "timestamp": time.time()}
        self.exact_cache.set(query, result)
        await self.semantic_cache.set(query, result)

        self.metrics.record("miss")
        return response

Going Further

• RAG Generation and LLM Optimization: the parent guide on generation
• Reduce RAG Latency: beyond caching
• RAG Cost Optimization: global strategies
• RAG Observability: monitoring hit rates
• Small Language Models: reduce costs at the source

FAQ

What similarity threshold should I use for semantic cache?

For technical support where precision is critical, use 0.95. For general FAQ, 0.90 offers a good balance between hit rate and precision. For informal chat, 0.85 is acceptable. Start high (0.95) and gradually lower while monitoring the quality of cached responses through your observability dashboard.

Is Anthropic and OpenAI prompt caching compatible with RAG?

Yes, and it is actually one of the best use cases. The system prompt (RAG instructions) and reference documents form a stable prefix that is automatically cached. Only the user question changes with each request. With Anthropic, you save 90% on prefix tokens. With OpenAI, 50%. It is transparent and requires no significant code changes.

How do I handle cache invalidation when my document base changes?

Three complementary approaches. Automatic TTL (2-24h depending on update frequency) handles common cases. Event-driven invalidation (trigger a flush when a document is updated) handles critical cases. Combining both is recommended: short TTL (2h) + event-driven invalidation for important updates.

Is GPTCache production-ready?

GPTCache is excellent for prototyping and small to medium projects. For large-scale production, we recommend Redis Stack with a custom semantic cache implementation. Redis offers better persistence, scalability, and a mature monitoring ecosystem. GPTCache remains an excellent starting point to validate the concept quickly.

Does Ailog use a caching system?

Yes. Ailog integrates an optimized multi-layer caching system: exact match, semantic cache, and provider-side prompt caching. Our clients' average hit rate is 55-65%, which significantly reduces costs and latency. The cache is automatically invalidated when data sources are updated, with no manual intervention required.