Production : scaling et caching

Objectifs

• Optimiser les performances d’un système d’embeddings en production
• Implémenter le caching pour réduire les coûts et la latence
• Gérer le scaling et la haute disponibilité

Les défis de la production

En développement, appeler l’API OpenAI à chaque requête ne pose aucun problème : vous faites dix appels par jour. En production, le même code se heurte à trois murs simultanés. Le coût d’abord, puisque chaque appel est facturé et qu’une réindexation complète d’un corpus se paie intégralement à chaque exécution. La latence ensuite : un aller-retour réseau de 100 à 300 ms par requête s’ajoute au temps de recherche et au temps de génération, et l’utilisateur additionne le tout. Les rate limits enfin, OpenAI limitant le nombre d’appels par minute — la limite que vous ne verrez jamais en développement et qui tombera le jour du lancement.

Les trois problèmes ont largement la même réponse : ne pas appeler l’API quand ce n’est pas nécessaire.

Caching des embeddings

Un embedding est déterministe : le même texte avec le même modèle rend toujours le même vecteur. C’est ce qui rend le cache si rentable ici, bien plus que pour un LLM génératif. Pour une application à trafic modéré, le décorateur lru_cache de la bibliothèque standard suffit et tient en deux lignes. Le retour est converti en tuple parce que lru_cache exige des valeurs hachables ; c’est la seule contorsion à connaître.

from functools import lru_cache
from openai import OpenAI
import hashlib
import json

client = OpenAI()

@lru_cache(maxsize=10_000)
def embedding_cache(texte: str, model: str = "text-embedding-3-large") -> tuple:
    """Cache LRU pour les embeddings (tuple car hashable)."""
    response = client.embeddings.create(input=texte, model=model)
    return tuple(response.data[0].embedding)

# Utilisation - le 2e appel est instantané
emb1 = embedding_cache("Bonjour le monde")  # Appel API
emb2 = embedding_cache("Bonjour le monde")  # Depuis le cache

Ce cache disparaît au moindre redémarrage et n’est partagé par aucun autre processus : dès que vous tournez sur deux instances derrière un load balancer, chacune reconstruit le sien. Redis résout les deux problèmes. La clé est un hachage du couple modèle-texte, ce qui évite qu’un changement de modèle ne serve d’anciens vecteurs, et le TTL de trente jours borne la croissance du stockage.

La méthode get_embeddings_batch mérite un examen attentif, car c’est elle qui fait la différence en ingestion. Elle ne se contente pas de tester le cache texte par texte : elle sépare les manquants, les envoie en un seul appel API, puis réinsère chaque résultat à sa position d’origine grâce à indices_a_calculer. Sur un lot de mille documents dont neuf cents sont déjà connus, vous passez d’un appel par document à un unique appel pour les cent restants.

import redis
import numpy as np
import json

class EmbeddingCacheRedis:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.client = OpenAI()
        self.ttl = 86400 * 30  # 30 jours

    def _cache_key(self, texte: str, model: str) -> str:
        h = hashlib.md5(f"{model}:{texte}".encode()).hexdigest()
        return f"emb:{h}"

    def get_embedding(
        self, texte: str, model: str = "text-embedding-3-large"
    ) -> list[float]:
        key = self._cache_key(texte, model)

        # Chercher dans le cache
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        # Générer et cacher
        response = self.client.embeddings.create(
            input=texte, model=model
        )
        embedding = response.data[0].embedding

        self.redis.setex(key, self.ttl, json.dumps(embedding))
        return embedding

    def get_embeddings_batch(
        self, textes: list[str], model: str = "text-embedding-3-large"
    ) -> list[list[float]]:
        """Batch avec cache partiel."""
        resultats = [None] * len(textes)
        a_calculer = []
        indices_a_calculer = []

        # Vérifier le cache pour chaque texte
        for i, texte in enumerate(textes):
            key = self._cache_key(texte, model)
            cached = self.redis.get(key)
            if cached:
                resultats[i] = json.loads(cached)
            else:
                a_calculer.append(texte)
                indices_a_calculer.append(i)

        # Calculer les manquants en un seul appel
        if a_calculer:
            response = self.client.embeddings.create(
                input=a_calculer, model=model
            )
            for resp_item in sorted(response.data, key=lambda x: x.index):
                idx = indices_a_calculer[resp_item.index]
                texte = a_calculer[resp_item.index]
                embedding = resp_item.embedding

                resultats[idx] = embedding
                key = self._cache_key(texte, model)
                self.redis.setex(key, self.ttl, json.dumps(embedding))

        print(f"Cache: {len(textes) - len(a_calculer)}/{len(textes)} hits")
        return resultats

# Utilisation
cache = EmbeddingCacheRedis()
emb = cache.get_embedding("test de cache")

Gestion des rate limits

Le cache réduit le trafic mais ne l’annule pas, et un pic de charge finira par déclencher une RateLimitError. La réponse correcte n’est pas de réessayer immédiatement — ce qui aggrave la congestion — mais d’attendre en doublant le délai à chaque tentative : une seconde, deux, quatre, huit, seize. Cinq tentatives couvrent ainsi une trentaine de secondes, ce qui absorbe l’immense majorité des pics passagers, et la dernière relance l’exception pour que l’incident remonte au lieu d’être silencieusement avalé.

import time
from openai import RateLimitError

class EmbeddingClientRobuste:
    def __init__(self, max_retries: int = 5):
        self.client = OpenAI()
        self.max_retries = max_retries

    def create(
        self, input: str | list[str], model: str = "text-embedding-3-large",
        **kwargs
    ):
        """Appel avec retry exponentiel."""
        for tentative in range(self.max_retries):
            try:
                return self.client.embeddings.create(
                    input=input, model=model, **kwargs
                )
            except RateLimitError as e:
                if tentative == self.max_retries - 1:
                    raise
                wait = 2 ** tentative
                print(f"Rate limit atteint, pause de {wait}s...")
                time.sleep(wait)

Traitement asynchrone

Pour les pipelines d’ingestion, le goulot n’est pas le calcul mais l’attente réseau : votre programme passe l’essentiel de son temps à ne rien faire. L’asynchrone permet de garder plusieurs requêtes en vol. Le sémaphore borne volontairement ce parallélisme, car lancer cent appels simultanés ne fait que provoquer les rate limits que vous venez d’apprendre à gérer — trois à cinq requêtes concurrentes constituent un réglage sain. Les textes sont d’abord découpés en lots de cent, et l’indice de départ de chaque lot voyage avec la tâche pour que as_completed, qui rend les résultats dans le désordre, puisse quand même les replacer correctement.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def generer_embeddings_async(
    textes: list[str],
    model: str = "text-embedding-3-large",
    max_concurrent: int = 5
) -> list[list[float]]:
    """Génère des embeddings en parallèle avec sémaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)
    resultats = [None] * len(textes)

    async def traiter_batch(batch_idx, batch):
        async with semaphore:
            response = await async_client.embeddings.create(
                input=batch, model=model
            )
            return batch_idx, [
                d.embedding
                for d in sorted(response.data, key=lambda x: x.index)
            ]

    # Découper en lots de 100
    taches = []
    batch_size = 100
    for i in range(0, len(textes), batch_size):
        batch = textes[i:i + batch_size]
        taches.append(traiter_batch(i, batch))

    for coro in asyncio.as_completed(taches):
        batch_idx, embeddings = await coro
        for j, emb in enumerate(embeddings):
            resultats[batch_idx + j] = emb

    return resultats

# Utilisation
embeddings = asyncio.run(
    generer_embeddings_async(mes_textes, max_concurrent=3)
)

Optimiser le stockage

Reste la facture d’infrastructure. Dix mille vecteurs de 3 072 dimensions en float32 occupent environ 122 Mo ; un million en occupe douze gigaoctets, qu’il faut tenir en mémoire pour que la recherche reste rapide. La quantification en int8 ramène chaque coordonnée sur 256 valeurs entières, divisant l’empreinte par quatre pour tomber autour de 30 Mo dans l’exemple ci-dessous. La perte de précision existe mais reste marginale au regard des écarts de similarité que vous manipulez.

import numpy as np

def quantifier_int8(embeddings: np.ndarray) -> tuple:
    """Quantifié les embeddings en int8 (4x moins de mémoire)."""
    min_val = embeddings.min(axis=1, keepdims=True)
    max_val = embeddings.max(axis=1, keepdims=True)
    scale = (max_val - min_val) / 255.0

    quantifie = ((embeddings - min_val) / scale).astype(np.int8)
    return quantifie, min_val, scale

def dequantifier_int8(quantifie, min_val, scale):
    """Restaure les embeddings depuis int8."""
    return quantifie.astype(np.float32) * scale + min_val

# Comparaison taille
embeddings = np.random.randn(10000, 3072).astype(np.float32)
print(f"float32 : {embeddings.nbytes / 1e6:.1f} Mo")  # ~122 Mo

quantifie, min_val, scale = quantifier_int8(embeddings)
print(f"int8 :    {quantifie.nbytes / 1e6:.1f} Mo")    # ~30 Mo

Le levier le plus simple reste toutefois le paramètre dimensions de l’API, qui produit directement des vecteurs plus courts sans aucun post-traitement de votre part : 3 072 dimensions pèsent environ 24 Ko par vecteur, 1 024 dimensions descendent à 8 Ko, et 256 dimensions à 2 Ko.

# 3072 dim -> ~24 Ko par vecteur
# 1024 dim -> ~8 Ko par vecteur  (réduction API)
# 256 dim  -> ~2 Ko par vecteur  (réduction API)

response = client.embeddings.create(
    input="texte",
    model="text-embedding-3-large",
    dimensions=1024  # 3x moins de stockage
)

Architecture de production

Assemblées, ces briques donnent un pipeline sobre : le cache Redis en façade, le client robuste pour les appels qui passent malgré tout. L’ingestion emprunte la voie batch, la recherche met en cache jusqu’à l’embedding de la question — et ce dernier point n’est pas anecdotique, car les requêtes des utilisateurs se répètent bien plus qu’on ne l’imagine.

class PipelineProduction:
    """Pipeline RAG optimisé pour la production."""

    def __init__(self):
        self.cache = EmbeddingCacheRedis()
        self.client_robuste = EmbeddingClientRobuste()

    def ingerer(self, documents: list[dict]):
        """Ingestion avec cache et retry."""
        textes = [d["texte"] for d in documents]
        embeddings = self.cache.get_embeddings_batch(textes)
        # Stocker dans la base vectorielle...

    def rechercher(self, question: str, k: int = 5):
        """Recherche avec cache de la requête."""
        query_emb = self.cache.get_embedding(question)
        # Rechercher dans la base vectorielle...
        return query_emb

Checklist de mise en production

Deux lignes de ce tableau ne produisent aucun gain de performance et sont pourtant les plus importantes. Le monitoring du taux de cache hit est votre indicateur de santé économique : s’il s’effondre, votre facture triple avant que quiconque s’en aperçoive. Et la sauvegarde des embeddings vous évite, le jour d’un incident, de devoir repayer et réattendre l’encodage complet du corpus.

Résumé

• Le cache Redis réduit les coûts et la latence de 90%+
• Le retry exponentiel gère les rate limits d’OpenAI
• Le traitement asynchrone accélère l’ingestion de gros corpus
• La quantification int8 réduit le stockage de 4x
• La réduction de dimensions à l’API est le levier le plus simple

Testez vos connaissances

Vingt-deux leçons d’embeddings et de RAG : l’essentiel tient en cinq questions.

Le fil du cours : le sens devient mesurable — recherche, classement, recommandation n’en sont que des applications. La production (scaling, caching) fait le reste.