Aller au contenu principal

Chapitre 2 : Mise en cache et performance

Types de mise en cache​

Type de cacheCe qui est mis en cacheAvantage
EmbeddingEmbeddings de requêtesRéduire les appels API
SémantiqueRésultats de requêtes similairesÉviter la récupération
RéponseRéponses RAG complètesLe plus rapide

Cache d'Embedding​

from functools import lru_cache
import hashlib

class CachedEmbeddings:
def __init__(self, embeddings, cache_size=10000):
self.embeddings = embeddings
self._cache = {}

def embed_query(self, text):
key = hashlib.md5(text.encode()).hexdigest()
if key not in self._cache:
self._cache[key] = self.embeddings.embed_query(text)
return self._cache[key]

Cache Sémantique​

Cache basé sur la similarité des requêtes :

from langchain.cache import GPTCache
import gptcache

def get_similar_query(query):
"""Trouver le résultat mis en cache pour une requête similaire"""
query_embedding = embed(query)

# Recherche dans le magasin de vecteurs de cache
results = cache_store.similarity_search(
query_embedding,
threshold=0.95 # Requêtes très similaires
)

if results:
return results[0].metadata["response"]
return None

Cache de Réponse avec Redis​

import redis
import json

r = redis.Redis()
CACHE_TTL = 3600 # 1 heure

def cached_rag(query, rag_chain):
# Vérifier le cache
cached = r.get(f"rag:{query}")
if cached:
return json.loads(cached)

# Générer et mettre en cache
response = rag_chain.invoke(query)
r.setex(f"rag:{query}", CACHE_TTL, json.dumps(response))

return response

Invalidation du Cache​

def invalidate_on_update(document_id):
"""Invalider les caches lorsque les documents changent"""
# Effacer les caches de réponse associés
r.delete_pattern(f"rag:*doc:{document_id}*")

# Supprimer du cache sémantique
semantic_cache.delete(document_id)

Comparaison de Performance​

ScénarioLatenceCoût
Pas de cache2-5s$0.01/requête
Cache d'embedding1-3s$0.005/requête
Succès du cache sémantique<100ms~$0
Succès du cache de réponse<50ms~$0