Stratégies de Cache pour Embeddings : Précalcul et Réutilisation des Vecteurs pour Requêtes Fréquentes

Les embeddings constituent le fondement de nombreuses applications d'intelligence artificielle moderne, des moteurs de recherche sémantique aux systèmes de recommandation. Cependant, le coût computationnel du génération d'embedding peut rapidement devenir prohibitif lorsqu'il s'agit de traiter des millions de requêtes quotidiennes. Dans ce tutoriel, nous explorons les stratégies de mise en cache permettant d'optimiser les performances et de réduire les coûts.

Comparatif des Solutions d'Embedding

Critère	HolySheep AI	API Officielle	Services Relais
Latence moyenne	<50ms	80-150ms	60-120ms
Coût par million de tokens	À partir de ¥0.50	$0.13-0.20	$0.10-0.18
Méthodes de paiement	WeChat/Alipay/Carte	Carte internationale	Variable
Crédits gratuits	Oui (offerts à l'inscription)	Non	Variable

Principe Fondamental du Cache d'Embeddings

Le principe est simple : au lieu de recalculer l'embedding d'un texte déjà traité, nous le stockons et le récupérons lors des requêtes ultérieures. Cette approche répond particulièrement aux scénarios où un petit nombre de requêtes représente une grande partie du trafic total.

Implémentation en Python

Architecture de Base avec Redis

import hashlib
import redis
import json
from typing import List, Optional

class EmbeddingCache:
    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=0,
            decode_responses=True
        )
        self.cache_ttl = 86400  # 24 heures par défaut
    
    def _generate_cache_key(self, text: str, model: str) -> str:
        """Génère une clé de cache unique basée sur le hash du texte."""
        text_hash = hashlib.sha256(text.encode('utf-8')).hexdigest()
        return f"embedding:{model}:{text_hash}"
    
    def get_cached_embedding(self, text: str, model: str) -> Optional[List[float]]:
        """Récupère un embedding depuis le cache."""
        cache_key = self._generate_cache_key(text, model)
        cached = self.redis_client.get(cache_key)
        
        if cached:
            return json.loads(cached)
        return None
    
    def cache_embedding(self, text: str, model: str, embedding: List[float], ttl: int = None):
        """Stocke un embedding dans le cache."""
        cache_key = self._generate_cache_key(text, model)
        ttl = ttl or self.cache_ttl
        
        self.redis_client.setex(
            cache_key,
            ttl,
            json.dumps(embedding)
        )

Utilisation
cache = EmbeddingCache()
cached = cache.get_cached_embedding("Quelle est la capitale de la France?", "text-embedding-3-small")
print(f"Cache hit: {cached is not None}")

Client API Compatible avec HolySheep

import requests
import hashlib
from typing import List, Dict
import json

class HolySheepEmbeddingClient:
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def generate_embedding(self, text: str, model: str = "text-embedding-3-small") -> List[float]:
        """Génère un embedding via l'API HolySheep."""
        response = self.session.post(
            f"{self.base_url}/embeddings",
            json={
                "input": text,
                "model": model
            }
        )
        response.raise_for_status()
        data = response.json()
        return data["data"][0]["embedding"]
    
    def batch_generate(self, texts: List[str], model: str = "text-embedding-3-small") -> Dict[str, List[float]]:
        """Génère des embeddings par lots avec mise en cache."""
        results = {}
        for text in texts:
            cache_key = hashlib.md5(text.encode()).hexdigest()
            results[cache_key] = self.generate_embedding(text, model)
        return results

Initialisation du client
client = HolySheepEmbeddingClient(api_key="YOUR_HOLYSHEEP_API_KEY")
embedding = client.generate_embedding("Bonjour le monde")
print(f"Embedding généré: {len(embedding)} dimensions")

Système de Cache Multi-Niveaux

from functools import lru_cache
import hashlib
import time
from collections import OrderedDict
from threading import Lock

class LRUCacheEmbedding:
    """Cache LRU en mémoire pour les embeddings très fréquemment utilisés."""
    
    def __init__(self, capacity: int = 1000):
        self.cache = OrderedDict()
        self.capacity = capacity
        self.lock = Lock()
        self.stats = {"hits": 0, "misses": 0}
    
    def _hash_text(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()
    
    def get(self, text: str) -> tuple:
        """Retourne (embedding, hit)"""
        key = self._hash_text(text)
        
        with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)
                self.stats["hits"] += 1
                return self.cache[key], True
            
            self.stats["misses"] += 1
            return None, False
    
    def put(self, text: str, embedding: list):
        key = self._hash_text(text)
        
        with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)
            self.cache[key] = embedding
            
            if len(self.cache) > self.capacity:
                self.cache.popitem(last=False)
    
    def get_stats(self) -> dict:
        total = self.stats["hits"] + self.stats["misses"]
        hit_rate = self.stats["hits"] / total if total > 0 else 0
        return {
            **self.stats,
            "total_requests": total,
            "hit_rate": f"{hit_rate:.2%}"
        }

Démonstration
cache = LRUCacheEmbedding(capacity=100)
cache.put("Paris", [0.1, 0.2, 0.3])
embedding, hit = cache.get("Paris")
print(f"Cache hit: {hit}, Stats: {cache.get_stats()}")

Stratégies d'Invalidation du Cache

• TTL temporel : Expiration après un délai fixe (24h, 7j)
• Invalidation par version du modèle : Invalider quand le modèle change
• Politique LRU : Éliminer les entrées les moins récemment utilisées
• Cache sémantique : Similarité cosinus pour détecter les requêtes proches

Optimisation pour les Charges de Travail Élevées

Pour les applications manipulant des millions de requêtes, une architecture distribuée devient nécessaire. Le pattern suivant combine un cache local L1 (Redis) avec un cache partagé L2 (Memcached) pour maximiser les performances tout en minimisant la charge sur l'API sous-jacente.

import asyncio
import aioredis
from typing import List, Optional
import numpy as np

class DistributedEmbeddingCache:
    """Cache distribué multi-niveaux pour embeddings."""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis: Optional[aioredis.Redis] = None
        self.redis_url = redis_url
        self.local_cache: LRUCacheEmbedding = LRUCacheEmbedding(capacity=500)
    
    async def connect(self):
        self.redis = await aioredis.create_redis_pool(self.redis_url)
    
    async def get_embedding(self, text: str, model: str) -> Optional[List[float]]:
        # Niveau L1: Cache local
        embedding, hit = self.local_cache.get(text)
        if hit:
            return embedding
        
        # Niveau L2: Redis
        cache_key = f"emb:{model}:{hashlib.sha256(text.encode()).hexdigest()}"
        cached = await self.redis.get(cache_key)
        
        if cached:
            embedding = json.loads(cached)
            self.local_cache.put(text, embedding)  # Populate L1
            return embedding
        
        return None
    
    async def close(self):
        if self.redis:
            self.redis.close()

Exemple d'utilisation asynchrone
async def main():
    cache = DistributedEmbeddingCache()
    await cache.connect()
    
    result = await cache.get_embedding("Ma requête", "text-embedding-3-small")
    print(f"Résultat: {result}")
    
    await cache.close()

asyncio.run(main())

Erreurs courantes et solutions

Erreur 1 : Clé de cache non unique pour textes similaires

# PROBLÈME : Hachage direct sans normalisation
def bad_hash(text):
    return hashlib.md5(text.encode())  # Sensible aux espaces supplémentaires

SOLUTION : Normaliser le texte avant hachage
def good_hash(text):
    normalized = ' '.join(text.lower().split())  # Normalise les espaces
    normalized = normalized.strip()  # Supprime les espaces bords
    return hashlib.sha256(normalized.encode()).hexdigest()

Test
print(good_hash("  Bonjour   le   monde  "))  # Égal à
print(good_hash("Bonjour le monde"))  # Ces deux produisent le même hash

Erreur 2 : Fuite mémoire dans le cache local

# PROBLÈME : Cache sans limite de capacité
class MemoryLeakCache:
    def __init__(self):
        self.cache = {}  # Grandit indéfiniment

SOLUTION : Implémenter une politique d'éviction
class SafeCache:
    def __init__(self, max_size: int = 10000):
        self.cache = OrderedDict()
        self.max_size = max_size
    
    def put(self, key, value):
        if len(self.cache) >= self.max_size:
            # Élimine les 10% les plus anciens
            for _ in range(self.max_size // 10):
                self.cache.popitem(last=False)
        self.cache[key] = value
        self.cache.move_to_end(key)

Erreur 3 : Latence excessive avec Redis non optimisé

# PROBLÈME : Connexion créée pour chaque requête
def bad_approach(texts):
    results = []
    for text in texts:
        r = redis.Redis(host='localhost')  # Nouvelle connexion à chaque fois
        results.append(r.get(text))
    return results

SOLUTION : Pool de connexions et pipeline
import redis
from redis.connection import ConnectionPool

pool = ConnectionPool(host='localhost', max_connections=20)

def good_approach(texts):
    client = redis.Redis(connection_pool=pool)
    pipe = client.pipeline()  # Batch les opérations
    
    for text in texts:
        pipe.get(text)
    
    return pipe.execute()  # Exécute tout en une seule commande réseau

Erreur 4 : Incohérence des dimensions d'embedding

# PROBLÈME : Mixing de modèles avec dimensions différentes
cache = {}
cache[" texte"] = [0.1, 0.2]  # embedding-ada-002 (1536 dims)
cache["texte2"] = [0.1]*1536  # text-embedding-3-small (1536 dims)

SOLUTION : Inclure le nom du modèle dans la clé de cache
def get_embedding_safe(text: str, model: str, client) -> list:
    cache_key = f"emb:{model}:{hashlib.sha256(text.encode()).hexdigest()}"
    
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    embedding = client.generate_embedding(text, model)
    redis_client.setex(cache_key, 86400, json.dumps(embedding))
    return embedding

Bonnes Pratiques de Production

• Surveiller le taux de cache hit : Objectif >80% pour les workloads typiques
• Définir une stratégie de warmed cache : Précharger les embeddings des requêtes fréquentes au démarrage
• Implémenter le fallback gracieux : Calculer l'embedding en cas de cache miss sans bloquer l'utilisateur
• Utiliser la compression : Les embeddings peuvent être compressés (FP16) pour réduire la mémoire
• Définir des TTL par catégorie : Les documents statiques peuvent avoir des TTL plus longs

Les stratégies de mise en cache des embeddings constituent un levier majeur d'optimisation des coûts et des performances. En combinant cache local L1, cache distribué L2 et invalidation intelligente, il est possible de réduire la latence de 80% tout en diminuant significativement les appels API.

👉 Inscrivez-vous sur HolySheep AI — crédits offerts