Cache RAG Intelligent : Comment Réduire vos Coûts LLM de 80% (Sans Sacrifier la Qualité)

TL;DR

Le cache est l'optimisation n1 pour réduire les coûts RAG. En combinant semantic cache, prompt caching (Anthropic/OpenAI) et embedding cache, vous pouvez réduire vos coûts LLM de 60-80% tout en améliorant la latence. Ce guide compare les approches de cache (exact match, sémantique, prompt caching), les solutions (Redis, GPTCache, custom) et inclut des calculs de ROI concrets.

Pourquoi le cache est critique pour le RAG

Le coût caché d'un RAG sans cache

Prenons un chatbot RAG typique servant 10 000 requêtes/jour :

Avec un cache à 40% de hit rate :

Avec un cache optimisé à 70% de hit rate :

Économies par niveau de cache

Sans cache:     $4 836/mois  ████████████████████████████████
Cache basique:  $2 951/mois  ████████████████████
Cache optimisé: $1 501/mois  ██████████
Cache avancé:   $967/mois    ██████
                              ↑ -80% de réduction

Les 5 stratégies de cache pour le RAG

Stratégie 1 : Exact Match Cache

Le plus simple. Même question = même réponse.

DEVELOPERpython
import hashlib
import json
import redis

class ExactMatchCache:
    """Cache par correspondance exacte de la question."""

    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:
        """Génère une clé de cache déterministe."""
        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)


# Utilisation dans le pipeline RAG
cache = ExactMatchCache("redis://localhost:6379", ttl_seconds=7200)

async def rag_with_cache(query: str, user_id: str) -> str:
    # 1. Vérifier le cache
    cached = cache.get(query)
    if cached:
        return cached["response"]  # Hit! Pas d'appel LLM

    # 2. Pipeline RAG normal
    response = await full_rag_pipeline(query)

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

    return response

Avantages : Simple, rapide, 0 faux positif. Inconvénients : Hit rate faible (10-20%), ne capte pas les reformulations.

Stratégie 2 : Semantic Cache

Cache basé sur la similarité sémantique. "Comment configurer X ?" et "Je veux paramétrer X" retournent le même résultat.

DEVELOPERpython
import numpy as np
from typing import Optional

class SemanticCache:
    """Cache sémantique basé sur les 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]:
        """Cherche une réponse sémantiquement similaire."""
        query_embedding = await self.embedder.embed(query)

        if not self._cache_embeddings:
            return None

        # Calculer les similarités
        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):
        """Stocke une réponse avec son 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):
        """Ajuste le seuil de similarité.

        Plus haut = moins de hits mais plus précis.
        Plus bas = plus de hits mais risque de réponses incorrectes.
        """
        self.threshold = threshold


# Seuils recommandés par cas d'usage
THRESHOLDS = {
    "support_technique": 0.95,  # Précision critique
    "faq_generale": 0.90,      # Bon équilibre
    "recherche_produit": 0.88,  # Plus de hits acceptables
    "chat_informel": 0.85,     # Tolérance élevée
}

Avantages : Hit rate élevé (40-60%), capte les reformulations. Inconvénients : Coût d'embedding par requête, risque de faux positifs.

Stratégie 3 : Prompt Caching (Anthropic / OpenAI)

Les providers offrent un cache natif des préfixes de prompt.

DEVELOPERpython
# Anthropic Prompt Caching
import anthropic

client = anthropic.Anthropic()

# Le system prompt et les documents sont mis en cache côté Anthropic
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "Vous êtes un assistant RAG spécialisé...",
            "cache_control": {"type": "ephemeral"}
        },
        {
            "type": "text",
            "text": f"Documents de référence:\n{all_documents}",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": query}]
)

# Vérifier l'utilisation du cache
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% moins cher que les input tokens normaux

DEVELOPERpython
# OpenAI Prompt Caching (automatique depuis 2024)
from openai import OpenAI

client = OpenAI()

# Le préfixe identique est automatiquement mis en cache
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": long_system_prompt  # Mis en cache auto
        },
        {
            "role": "user",
            "content": f"Documents:\n{documents}\n\nQuestion: {query}"
        }
    ]
)

# OpenAI : 50% de réduction sur les tokens cached
print(f"Cached tokens: {response.usage.prompt_tokens_details.cached_tokens}")

Économies du prompt caching :

Stratégie 4 : Embedding Cache

Mettre en cache les embeddings pour éviter de recalculer.

DEVELOPERpython
class EmbeddingCache:
    """Cache des embeddings pour éviter les appels API."""

    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]:
        """Retourne l'embedding depuis le cache ou le calcule."""
        key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"

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

        # Calculer et stocker
        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


# Utilisation
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)
    )

Stratégie 5 : Retrieval Cache

Cache des résultats de recherche vectorielle.

DEVELOPERpython
class RetrievalCache:
    """Cache des résultats de recherche vectorielle."""

    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:
        """Clé basée sur l'embedding quantifié."""
        # Quantifier l'embedding pour augmenter les hits
        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

Comparaison des approches de cache

Redis vs GPTCache vs Custom

Tableau comparatif

GPTCache - Implémentation

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

# Configuration GPTCache
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()

# Utilisation transparente (remplace l'API OpenAI)
response = gptcache_openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ],
)
# GPTCache vérifie automatiquement le cache sémantique

Redis - Implémentation avancée

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:
    """Cache sémantique avec 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

Quand NE PAS utiliser le cache

Cas où le cache est contre-productif

Stratégies d'invalidation

DEVELOPERpython
class CacheInvalidator:
    """Invalidation intelligente du cache RAG."""

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

    async def on_document_updated(self, doc_id: str):
        """Invalide le cache quand un document est mis à jour."""
        # Trouver toutes les entrées de cache
        # qui référencent ce document
        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):
        """Invalide tout le cache lors d'un refresh complet."""
        self.cache.flush_all()

    def schedule_ttl_cleanup(self, max_age_hours: int = 24):
        """Nettoie les entrées trop anciennes."""
        cutoff = time.time() - (max_age_hours * 3600)
        self.cache.delete_older_than(cutoff)

Calcul de ROI détaillé

Scénario : Chatbot support e-commerce

Volume : 15 000 requêtes/jour
Modèle : GPT-4o ($2.50/1M input, $10/1M output)
Tokens moyens : 2000 input, 500 output par requête

ROI sur 12 mois

Investissement :
  - Développement cache : 40h × $100/h = $4 000
  - Infrastructure Redis : $100/mois × 12 = $1 200
  - Total investissement : $5 200

Économies annuelles (cache combiné 75%) :
  - $3 276/mois × 12 = $39 312

ROI = ($39 312 - $5 200) / $5 200 = 656%
Temps de retour : < 2 mois

Pipeline de cache complet

DEVELOPERpython
class RAGCachePipeline:
    """Pipeline de cache multicouche pour 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:
        # Couche 1 : Exact match (le plus rapide)
        exact = self.exact_cache.get(query)
        if exact:
            self.metrics.record("exact_hit")
            return exact["response"]

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

        # Couche 3 : Embedding cache (pour l'embedding)
        embedding = await self.embedding_cache.get_or_compute(
            query, self.embed_fn
        )

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

        # Couche 5 : Génération (prompt caching côté provider)
        response = await self.generate_with_prompt_cache(
            query, docs
        )

        # Stocker dans les 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

Pour aller plus loin

• Génération RAG et optimisation LLM : le guide parent sur la génération
• Réduire la latence RAG : au-delà du cache
• Optimisation des coûts RAG : stratégies globales
• Observabilité RAG : monitoring du hit rate
• Small Language Models : réduire les coûts à la source

FAQ

Quel seuil de similarité choisir pour le semantic cache ?

Pour un support technique où la précision est critique, utilisez 0.95. Pour une FAQ générale, 0.90 offre un bon équilibre entre hit rate et précision. Pour un chat informel, 0.85 est acceptable. Commencez haut (0.95) et baissez progressivement en surveillant la qualité des réponses cachées via votre dashboard d'observabilité.

Le prompt caching d'Anthropic et d'OpenAI est-il compatible avec le RAG ?

Oui, et c'est même l'un des meilleurs cas d'usage. Le system prompt (instructions RAG) et les documents de référence forment un préfixe stable qui est automatiquement mis en cache. Seule la question utilisateur change à chaque requête. Avec Anthropic, vous économisez 90% sur les tokens du préfixe. Avec OpenAI, 50%. C'est transparent et ne nécessite aucune modification de code significative.

Comment gérer l'invalidation du cache quand ma base documentaire change ?

Trois approches complémentaires. Le TTL automatique (2-24h selon la fréquence de mise à jour) gère les cas courants. L'invalidation événementielle (déclencher un flush quand un document est mis à jour) gère les cas critiques. La combinaison des deux est recommandée : TTL court (2h) + invalidation sur événement pour les mises à jour importantes.

GPTCache est-il prêt pour la production ?

GPTCache est excellent pour le prototypage et les projets de petite à moyenne taille. Pour la production à grande échelle, nous recommandons Redis Stack avec une implémentation custom du semantic cache. Redis offre une meilleure persistance, scalabilité et un écosystème de monitoring mature. GPTCache reste un excellent point de départ pour valider le concept rapidement.

Ailog utilise-t-il un système de cache ?

Oui. Ailog intègre un système de cache multicouche optimisé : exact match, semantic cache et prompt caching côté provider. Le hit rate moyen de nos clients est de 55-65%, ce qui réduit significativement les coûts et la latence. Le cache est automatiquement invalidé lors de la mise à jour des sources de données, sans intervention manuelle.