Stratégies de Mise en Cache pour Réduire la Latence et le Coût RAG

Pourquoi utiliser le cache ?

Sans cache :

• Chaque requête → Appel API ($$$)
• Latence de 500ms+
• Limitations de débit

Avec cache :

• Réduction des coûts de 80%
• Réponses en 10ms depuis le cache
• Pas de limitations de débit

1. Cache sémantique de requêtes

Ne cachez pas les correspondances exactes - cachez les requêtes similaires :

DEVELOPERpython
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
cache = {}  # {embedding: response}

def semantic_cache_lookup(query, threshold=0.95):
    query_emb = model.encode(query)

    # Vérifier si une requête similaire est dans le cache
    for cached_emb, response in cache.items():
        similarity = np.dot(query_emb, cached_emb)
        if similarity > threshold:
            return response  # Cache hit!

    return None  # Cache miss

def rag_with_cache(query):
    # Vérifier d'abord le cache
    cached = semantic_cache_lookup(query)
    if cached:
        return cached

    # Cache miss - exécuter le pipeline RAG complet
    response = full_rag_pipeline(query)

    # Stocker dans le cache
    cache[model.encode(query)] = response

    return response

2. Cache d'embeddings

Mettez en cache les embeddings pour éviter de les recalculer :

DEVELOPERpython
import hashlib
import redis

redis_client = redis.Redis(host='localhost', port=6379)

def get_embedding_cached(text):
    # Créer une clé de cache
    cache_key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"

    # Vérifier le cache
    cached = redis_client.get(cache_key)
    if cached:
        return np.frombuffer(cached, dtype=np.float32)

    # Calculer l'embedding
    embedding = openai_embed(text)

    # Stocker dans le cache (expire après 7 jours)
    redis_client.setex(
        cache_key,
        604800,  # 7 jours
        embedding.tobytes()
    )

    return embedding

3. Intégration GPTCache

DEVELOPERpython
from gptcache import Cache
from gptcache.embedding import OpenAI
from gptcache.similarity_evaluation import SearchDistanceEvaluation

cache = Cache()
cache.init(
    embedding_func=OpenAI().to_embeddings,
    similarity_evaluation=SearchDistanceEvaluation(),
)

def cached_llm_call(prompt):
    # Vérifier le cache
    cached_response = cache.get(prompt)
    if cached_response:
        return cached_response

    # Appeler le LLM
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    # Mettre en cache la réponse
    cache.set(prompt, response)

    return response

4. Cache à deux niveaux

Mémoire rapide + Redis persistant :

DEVELOPERpython
from functools import lru_cache
import redis

redis_client = redis.Redis()

@lru_cache(maxsize=1000)
def l1_cache(query):
    # Cache L2 (Redis)
    cached = redis_client.get(f"rag:{query}")
    if cached:
        return cached.decode()

    # Cache miss - calculer
    result = rag_pipeline(query)

    # Stocker dans L2
    redis_client.setex(f"rag:{query}", 3600, result)

    return result

5. Invalidation du cache

DEVELOPERpython
import time

cache_with_ttl = {}

def get_with_ttl(key, ttl=3600):
    if key in cache_with_ttl:
        value, timestamp = cache_with_ttl[key]
        if time.time() - timestamp < ttl:
            return value
        else:
            del cache_with_ttl[key]  # Expiré

    return None

def set_with_ttl(key, value):
    cache_with_ttl[key] = (value, time.time())

Analyse des coûts

Sans cache (1M requêtes/mois) :

• Embeddings : 100$
• LLM : 3000$
• Total : 3100$

Avec cache (taux de succès de 80%) :

• Embeddings : 20$
• LLM : 600$
• Redis : 50$
• Total : 670$ (économie de 78%)

Le cache est le fruit le plus accessible pour l'optimisation RAG. Implémentez-le dès le début.