RAG en flux continu : quelle architecture quand les documents et les questions arrivent en même temps

Un RAG de manuel fonctionne ainsi : on rassemble un corpus, on le découpe, on le vectorise, on construit un index, et on interroge. L'index vit des semaines, parfois des mois. La question de la fraîcheur ne se pose pas, celle de la latence se règle en ajustant un paramètre de recherche approchée.

Les systèmes que je rencontre depuis deux ans ne ressemblent plus à cela. Un flux d'événements alimente le corpus en continu — tickets d'incident, journaux applicatifs, dépêches, relevés de capteurs, messages de supervision — et les questions arrivent au même rythme. Entre le moment où un document entre dans le système et celui où une réponse doit s'appuyer dessus, la marge se compte en secondes.

Ce basculement change la nature du problème. Le RAG classique optimise le rappel à corpus figé ; le RAG en flux arbitre en permanence entre la fraîcheur de l'index, la qualité de la recherche et le temps de réponse. Les trois ne montent jamais ensemble.

Ce que le flux change vraiment

Trois propriétés de l'architecture classique tombent d'un coup.

L'index cesse d'être immuable. Un graphe HNSW se construit bien une fois et se lit très vite ; il encaisse mal les insertions massives et encore plus mal les suppressions, qui laissent des pierres tombales jusqu'au prochain compactage. Écrire en continu dans une structure conçue pour la lecture, c'est accepter que la latence de recherche dérive au fil de la journée.

Le découpage perd son contexte global. Découper un corpus complet permet de calculer des statistiques, de dédoublonner, d'ajuster la taille des fragments au document. En flux, chaque document arrive seul : le découpage devient une décision locale, prise sans savoir ce qui suit.

La vectorisation devient un service critique. Tant qu'elle tourne la nuit, un modèle de plongement lent ne gêne personne. Dès qu'elle se place sur le chemin d'ingestion, elle dicte le retard de l'index — et si vous vectorisez aussi la question, elle se place en plus sur le chemin de réponse.

Décomposer le budget de latence

Avant d'ouvrir un catalogue de bases vectorielles, posez le budget. Voici celui d'une plateforme d'assistance à l'exploitation que nous avons livrée l'an dernier : les agents d'astreinte interrogent en langage naturel un corpus alimenté par les alertes, les tickets et les journaux de changement.

Deux enseignements sautent aux yeux. Le réordonnancement pèse à lui seul les deux tiers du budget initial : c'est presque toujours le premier poste à traiter. Et la vectorisation de la question coûtait cent dix millisecondes parce qu'elle passait par le même service que l'ingestion, saturé par les lots — la séparation des deux chemins a suffi à récupérer près de cent millisecondes.

Retenez la règle : le chemin d'ingestion et le chemin de réponse ne partagent jamais le même service de vectorisation. L'un cherche le débit, l'autre la latence. Les deux objectifs se combattent, et le lot gagne toujours.

Trois architectures, trois compromis

Face à un flux, trois familles de solutions tiennent la route. Elles se distinguent moins par les technologies que par la manière dont elles placent la frontière entre ce qui est indexé et ce qui ne l'est pas encore.

Un mot sur la troisième colonne : elle fait référence à l'architecture dite lambda, que le monde des données connaît depuis quinze ans. Le RAG la redécouvre, et les mêmes critiques s'appliquent — deux chemins de traitement, deux fois plus de code, deux fois plus d'occasions de diverger.

Architecture 1 — l'indexation par micro-lots

Le principe tient en une phrase : accumuler les documents pendant un intervalle fixe, puis vectoriser et écrire le lot d'un seul geste.

Cette approche a mauvaise réputation, à tort. Elle offre le meilleur rapport entre la simplicité d'exploitation et la qualité de recherche, et un intervalle de trente secondes satisfait la grande majorité des usages métier. Un assistant documentaire, une base de connaissances d'assistance, un moteur de recherche interne : personne ne remarque qu'un document publié il y a vingt secondes n'apparaît pas encore.

Le point de vigilance porte sur la taille du lot. Un lot trop petit gaspille le processeur graphique de vectorisation ; un lot trop gros allonge l'écart de fraîcheur et provoque des à-coups sur l'index. La solution consiste à déclencher l'écriture sur deux conditions concurrentes, la première atteinte l'emportant.

"""Consommateur d'ingestion : accumulation adaptative puis écriture par lot."""
import asyncio
import time
from dataclasses import dataclass, field

import httpx
from aiokafka import AIOKafkaConsumer
from qdrant_client import AsyncQdrantClient, models

TAILLE_MAX_LOT = 64          # plafond imposé par la mémoire du GPU de vectorisation
DELAI_MAX_SECONDES = 5.0     # plafond imposé par l'écart de fraîcheur négocié
URL_VECTORISATION = "http://tei-ingestion.rag.svc.cluster.local/embed"


@dataclass
class Lot:
    """Accumulateur borné en taille et en durée."""
    fragments: list[dict] = field(default_factory=list)
    ouvert_depuis: float = field(default_factory=time.monotonic)

    def pret(self) -> bool:
        return (len(self.fragments) >= TAILLE_MAX_LOT
                or time.monotonic() - self.ouvert_depuis >= DELAI_MAX_SECONDES)

    def vider(self) -> list[dict]:
        fragments, self.fragments = self.fragments, []
        self.ouvert_depuis = time.monotonic()
        return fragments


async def vectoriser(client: httpx.AsyncClient, textes: list[str]) -> list[list[float]]:
    """Un seul aller-retour pour tout le lot : le GPU travaille à plein."""
    reponse = await client.post(URL_VECTORISATION, json={"inputs": textes},
                                timeout=httpx.Timeout(20.0))
    reponse.raise_for_status()
    return reponse.json()


async def ingerer(consommateur: AIOKafkaConsumer, qdrant: AsyncQdrantClient,
                  http: httpx.AsyncClient) -> None:
    lot = Lot()
    async for message in consommateur:
        lot.fragments.extend(decouper(message.value))
        if not lot.pret():
            continue

        fragments = lot.vider()
        vecteurs = await vectoriser(http, [f["texte"] for f in fragments])
        await qdrant.upsert(
            collection_name="corpus",
            wait=False,                       # l'accusé du WAL suffit, l'index suivra
            points=[
                models.PointStruct(
                    id=f["id"], vector=v,
                    payload={"locataire": f["locataire"], "source": f["source"],
                             "horodatage": f["horodatage"], "texte": f["texte"]},
                )
                for f, v in zip(fragments, vecteurs)
            ],
        )
        await consommateur.commit()           # acquittement après écriture, jamais avant

Trois détails méritent l'attention. L'acquittement Kafka intervient après l'écriture dans l'index, ce qui garantit qu'un redémarrage rejoue le lot perdu plutôt que de l'oublier. Le paramètre wait=False rend la main dès l'écriture du journal, sans attendre la construction du graphe. Et la minuterie repart à chaque vidage, pas à chaque message : sans cela, un flux irrégulier déclenche des lots d'un seul élément.

Architecture 2 — la double vue chaude et froide

Quand l'écart de fraîcheur descend sous la seconde, le micro-lot ne suffit plus. La parade consiste à maintenir deux index de nature différente et à fusionner leurs résultats au moment de la requête.

L'index chaud ne contient que les documents des dernières minutes — quelques dizaines de milliers de vecteurs au plus. À cette échelle, la recherche exhaustive par produit scalaire sur un tableau contigu bat n'importe quelle structure approchée : pas de graphe à construire, pas de compactage, une insertion en temps constant et un rappel parfait. Un million de vecteurs de dimension 768 tiennent dans trois gigaoctets et se parcourent en une dizaine de millisecondes avec des instructions vectorielles.

L'index froid reçoit les mêmes documents avec plusieurs minutes de retard, par micro-lots, dans un HNSW classique parfaitement compacté. Il porte le volume, la durée de vie et la qualité de recherche à grande échelle.

Un processus de bascule retire de l'index chaud tout document dont l'index froid a confirmé la prise en compte. La fenêtre de recouvrement doit rester généreuse : mieux vaut un doublon dédoublonné à la fusion qu'un document momentanément introuvable.

"""Fusion des deux vues au moment de la requête."""
import asyncio

import numpy as np

K_CHAUD, K_FROID, K_FINAL = 30, 50, 12


async def rechercher(question_vecteur: np.ndarray, locataire: str) -> list[dict]:
    """Interroge les deux index en parallèle, fusionne, dédoublonne."""
    chaud, froid = await asyncio.gather(
        index_chaud.chercher(question_vecteur, k=K_CHAUD, locataire=locataire),
        index_froid.chercher(question_vecteur, k=K_FROID, locataire=locataire),
    )

    # Le document le plus récent l'emporte sur son homologue froid.
    par_identifiant: dict[str, dict] = {}
    for resultat in sorted(chaud + froid, key=lambda r: r["horodatage"], reverse=True):
        par_identifiant.setdefault(resultat["id"], resultat)

    candidats = sorted(par_identifiant.values(), key=lambda r: r["score"], reverse=True)
    return candidats[:K_FINAL]

Le coût de cette architecture ne réside pas dans le calcul mais dans l'exploitation. Deux index veulent dire deux schémas à faire évoluer ensemble, deux jeux de métriques, deux modes de panne, et un processus de bascule qui devient le composant le plus délicat de la plateforme. Ne la choisissez que si le métier réclame vraiment la sous-seconde — et faites-lui écrire cette exigence.

Architecture 3 — l'index en écriture continue

Plusieurs moteurs assument désormais l'écriture continue dans leur conception même, ce qui rend la double vue inutile dans la plupart des cas. Le principe est partout le même : un journal d'écriture anticipée accueille les insertions, un segment mutable les sert immédiatement, et une tâche de fond construit puis scelle des segments immuables.

Je retiens Qdrant ou Milvus pour la majorité des plateformes que je conçois. Le premier pour sa simplicité d'exploitation sur Kubernetes et son modèle de filtrage, le second quand le volume dépasse la centaine de millions de vecteurs et que la séparation du calcul et du stockage devient un vrai sujet.

Le réglage décisif chez Qdrant tient au seuil d'indexation. Sous ce seuil, un segment se parcourt intégralement : c'est rapide tant qu'il reste petit, et cela évite de reconstruire un graphe pour trois cents vecteurs. Au-dessus, l'optimiseur bâtit le HNSW en tâche de fond.

# Collection réglée pour un flux soutenu, déployée par Argo CD.
apiVersion: qdrant.io/v1
kind: QdrantCollection
metadata:
  name: corpus
  namespace: rag
spec:
  vectors:
    size: 768
    distance: Cosine
    on_disk: false                  # les vecteurs restent en mémoire, l'index aussi
  hnsw_config:
    m: 24                           # 16 par défaut ; 24 remonte le rappel de 2 à 3 points
    ef_construct: 200
    full_scan_threshold: 20000      # en dessous, recherche exhaustive : plus rapide et exacte
  optimizers_config:
    indexing_threshold: 30000       # construction du graphe seulement au-delà
    memmap_threshold: 200000
    default_segment_number: 8       # un segment par cœur alloué au pod
    flush_interval_sec: 2
    max_optimization_threads: 2     # bride la fusion pour protéger la latence de lecture
  quantization_config:
    scalar:
      type: int8
      quantile: 0.99
      always_ram: true              # divise l'empreinte par quatre, coûte 1 point de rappel
  write_consistency_factor: 1
  replication_factor: 2
  shard_number: 6
  payload_schema:                   # index sur les champs filtrés, sinon le filtrage scanne
    locataire: { type: keyword, is_tenant: true }
    horodatage: { type: integer, range: true }
    source: { type: keyword }

Le paramètre max_optimization_threads mérite une explication. Sans bride, les optimiseurs saturent les cœurs du pod pendant la fusion et la latence de recherche double. En les limitant à deux fils, la fusion s'étale davantage mais le p99 de lecture reste plat — un arbitrage que je refais sur chaque plateforme, jamais une valeur universelle.

Le pipeline d'ingestion, de bout en bout

Voici la chaîne complète, dans l'ordre où les données la traversent.

Réception. Kafka ou Redpanda, partitionné par identifiant de locataire pour garantir l'ordre par client et permettre le parallélisme entre eux. Prévoyez dès le départ une file de messages rejetés : un document mal formé ne doit jamais bloquer une partition.

Découpage. En flux, oubliez le découpage sémantique global qui compare chaque phrase à ses voisines sur tout le corpus. Retenez un découpage par fenêtre glissante avec recouvrement, borné en jetons, et respectant les frontières structurelles du document — titres, paragraphes, blocs de code. Comptez les jetons avec le tokeniseur du modèle de plongement, jamais en caractères.

Dédoublonnage. Les flux d'exploitation répètent énormément : la même alerte revient toutes les trente secondes. Une empreinte par hachage sensible à la localité, conservée dans Redis avec une durée de vie, écarte les quasi-doublons avant la vectorisation. Sur la plateforme d'astreinte évoquée plus haut, ce seul filtre a retiré 38 % du volume entrant et divisé la facture de vectorisation d'autant.

Vectorisation. Un service dédié, distinct de celui de la requête. Le regroupement dynamique côté serveur — celui de Triton ou de Text Embeddings Inference — travaille mieux que le regroupement côté client dès que plusieurs consommateurs coexistent.

Écriture. Dans l'index, sans attendre la construction du graphe, puis acquittement Kafka.

Publication de l'horodatage. Le point le plus souvent oublié : publier, en métrique, l'écart entre l'horodatage d'origine du document et l'instant de son écriture. Sans cette mesure, personne ne sait si la promesse de fraîcheur tient.

La chaîne de réponse : où se cachent les millisecondes

Le chemin de réponse obéit à une règle simple : tout ce qui peut se faire avant l'arrivée de la question doit se faire avant.

Le réordonnancement. Un modèle croisé qui évalue cinquante candidats coûte dix fois plus cher qu'une recherche vectorielle. Trois leviers le ramènent dans le budget : réduire le nombre de candidats à réordonnancer, prendre un modèle distillé de petite taille, et couper le réordonnancement quand l'écart de score entre le premier et le dixième candidat dépasse un seuil — si la recherche a déjà tranché, inutile de payer pour confirmer.

Le cache sémantique. Sur un flux d'assistance, un tiers des questions ressemblent à une question déjà posée. Un cache indexé par le vecteur de la question, avec un seuil de similarité à 0,97 et une durée de vie courte, économise la chaîne complète. Attention au piège : dans un système en flux, une réponse mise en cache vieillit avec le corpus. Invalidez sur l'horodatage du document le plus récent qui a servi à la construire, pas seulement sur une durée.

La recherche hybride. Le lexical et le vectoriel se complètent : le premier trouve les identifiants, les numéros de version et les messages d'erreur exacts, le second saisit l'intention. Leur fusion par rang réciproque coûte quelques millisecondes et fait gagner cinq à dix points de rappel — le meilleur rapport du budget.

La diffusion progressive. Le flux ne concerne pas que l'ingestion. Diffusez les jetons du modèle de génération au fil de l'eau : l'utilisateur perçoit le délai avant le premier jeton, pas la durée totale. Une réponse complète en trois secondes dont le premier mot arrive après deux cents millisecondes passe pour instantanée ; la même réponse livrée d'un bloc après trois secondes passe pour lente.

Dimensionner sur Kubernetes

La plateforme se décompose en quatre familles de charges aux besoins franchement divergents, qui appellent quatre pools de nœuds distincts.

L'index n'entre jamais dans une mise à l'échelle automatique. Ajouter une réplique déclenche une copie de segments qui pèse sur la latence au moment précis où la charge monte. Dimensionnez pour le pic, et redimensionnez à froid.

Pour le reste, KEDA lit le retard Kafka et ajuste les consommateurs. Le point subtil tient au choix de la métrique : le nombre de messages en attente réagit trop tard sur un flux irrégulier. Je préfère piloter sur le retard exprimé en secondes, qui traduit directement la promesse de fraîcheur.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: ingestion-corpus
  namespace: rag
spec:
  scaleTargetRef:
    name: ingestion-corpus
  minReplicaCount: 2               # jamais zéro : le démarrage à froid coûte 25 s
  maxReplicaCount: 24              # plafonné au nombre de partitions du sujet
  pollingInterval: 10
  cooldownPeriod: 120
  advanced:
    horizontalPodAutoscalerConfig:
      behavior:
        scaleUp:
          stabilizationWindowSeconds: 0     # monter vite
          policies: [{ type: Percent, value: 100, periodSeconds: 30 }]
        scaleDown:
          stabilizationWindowSeconds: 300   # redescendre lentement
          policies: [{ type: Pods, value: 1, periodSeconds: 120 }]
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: redpanda.flux.svc.cluster.local:9093
        consumerGroup: ingestion-corpus
        topic: documents-entrants
        lagThreshold: "2000"
        activationLagThreshold: "50"
        offsetResetPolicy: earliest
      authenticationRef:
        name: redpanda-mtls
    - type: prometheus                       # second déclencheur : la fraîcheur elle-même
      metadata:
        serverAddress: http://prometheus.observabilite.svc.cluster.local:9090
        threshold: "8"
        query: |
          histogram_quantile(0.95,
            sum by (le) (rate(rag_ecart_fraicheur_secondes_bucket[2m])))

Le second déclencheur ferme la boucle : même si le retard Kafka reste bas, un ralentissement de la vectorisation dégrade la fraîcheur, et la plateforme réagit à la conséquence plutôt qu'à la cause.

Cloisonnement multi-locataires et droit à l'oubli

Deux sujets que les démonstrations ignorent et que la production impose.

Le filtrage doit précéder la recherche approchée, pas la suivre. Filtrer après coup semble équivalent et ne l'est pas : si les cinquante plus proches voisins appartiennent tous à un autre locataire, la réponse finale est vide alors que des documents pertinents existaient. Pire, la latence explose quand le moteur élargit la recherche pour compenser. Qdrant traite ce cas avec un index de charge utile marqué is_tenant, qui regroupe physiquement les vecteurs d'un même locataire dans les mêmes segments ; Milvus propose les partitions par clé. Dans les deux cas, déclarez le cloisonnement à la création de la collection — le rattraper ensuite oblige à tout réécrire.

La suppression d'un document ne suffit pas. Retirer un vecteur d'un index le rend introuvable, mais il subsiste dans le journal d'écriture, dans les segments non compactés, dans les instantanés de sauvegarde et — point aveugle le plus fréquent — dans le cache sémantique des réponses déjà produites. Une procédure d'effacement complète enchaîne quatre gestes : suppression dans l'index, compactage forcé du segment concerné, purge des entrées de cache dont la provenance cite le document, et vérification par une requête de contrôle. Automatisez-la, documentez-la, et testez-la à intervalle régulier : le jour où la demande arrive, le délai légal court déjà.

Dans la même logique, le flux d'ingestion transporte des données que personne n'a relues. Un détecteur de données personnelles placé avant la vectorisation, qui masque ou écarte, coûte quelques millisecondes et évite qu'un numéro de sécurité sociale trouvé dans un journal applicatif finisse dans une réponse générée.

Mesurer : les six métriques qui comptent

Un RAG en flux se surveille comme n'importe quelle plateforme distribuée : des métriques, des seuils, des alertes. Six suffisent à savoir si le système tient.

La quatrième mérite un mot. Le rappel ne se mesure pas en production sans référence : mettez en place un jeu de sondes — cinquante questions dont vous connaissez les documents attendus — rejouées toutes les dix minutes par un travail périodique. C'est le seul garde-fou contre la dégradation silencieuse, celle où le système répond vite et mal sans qu'aucune alerte technique ne se déclenche.

groups:
  - name: rag-flux
    rules:
      - record: rag:fraicheur:p95
        expr: |
          histogram_quantile(0.95,
            sum by (le, collection) (rate(rag_ecart_fraicheur_secondes_bucket[5m])))

      - alert: FraicheurIndexDegradee
        expr: rag:fraicheur:p95 > 10
        for: 5m
        labels: { severity: majeure, equipe: plateforme-ia }
        annotations:
          resume: "Écart de fraîcheur de {{ $value | humanizeDuration }} sur {{ $labels.collection }}"
          action: "Vérifier le retard Kafka, puis la file du service de vectorisation."

      - alert: RappelEnBaisse
        expr: rag_sondes_rappel < 0.90
        for: 15m
        labels: { severity: majeure, equipe: plateforme-ia }
        annotations:
          resume: "Rappel mesuré à {{ $value }} — le système répond, mais mal."
          action: "Comparer aux segments non optimisés ; suspecter un compactage en retard."

      - alert: CompactageEnRetard
        expr: increase(qdrant_collection_segments_non_optimises[30m]) > 0
                and qdrant_collection_segments_non_optimises > 12
        for: 30m
        labels: { severity: mineure, equipe: plateforme-ia }
        annotations:
          resume: "Les optimiseurs ne suivent plus le rythme d'écriture."
          action: "Relever max_optimization_threads ou ralentir l'ingestion."

Les erreurs que j'ai vues

Confondre fraîcheur et latence. Ce sont deux budgets distincts, portés par deux chemins distincts. Une équipe optimisait sans relâche la latence de recherche alors que sa vraie plainte tenait à des documents invisibles pendant quatre minutes. Deux semaines de travail sur le mauvais problème.

Partager le service de vectorisation entre l'ingestion et la requête. Le lot d'ingestion occupe le processeur graphique pendant que la question attend. C'est la cause numéro un des latences erratiques que je rencontre, et la correction prend une demi-journée.

Laisser les optimiseurs libres. Un compactage non bridé transforme un p99 de vingt millisecondes en quatre-vingts pendant dix minutes, plusieurs fois par heure. La courbe ressemble à un peigne, et personne ne comprend pourquoi jusqu'à ce qu'on superpose les deux séries.

Filtrer après la recherche approchée. Cela fonctionne en démonstration avec trois locataires et casse dès que l'un d'eux pèse 80 % du corpus. Le cloisonnement se déclare à la création de la collection, jamais après.

Oublier le cache dans la procédure d'effacement. Le document disparaît de l'index, et le système continue de le citer pendant des heures parce qu'une réponse mise en cache le contient. Auditez la chaîne complète, pas seulement l'index.

Négliger le dédoublonnage en amont. Les flux d'exploitation se répètent énormément. Vectoriser dix fois la même alerte coûte dix fois le prix et dégrade la recherche, parce que les dix copies occupent les dix premières places de chaque résultat.

Croire qu'une base vectorielle suffit. La recherche hybride — lexicale et vectorielle — fait gagner cinq à dix points de rappel pour quelques millisecondes. Aucun réglage de HNSW ne rattrape l'absence d'un BM25 sur un corpus truffé de références exactes.