RAG en Python simple : segmentation en blocs, TF-IDF, similarité cosinus et génération de prompts

Un tutoriel typique sur RAG vous fait installer cinq bibliothèques et appeler la méthode .from_documents(). Dix minutes plus tard, une démonstration fonctionne, mais rien n’a été compris. Lorsque la récupération des données produit du contenu inutile ou que le modèle génère des hallucinations, vous êtes bloqué, car c’est la bibliothèque qui a effectué le travail à votre place.

Ici, le pipeline est construit de manière inverse, en Python pur, en utilisant uniquement le module math ainsi qu’une seule appel à un LLM de texte brut à la fin. Aucun modèle d’incorporation, aucun FAISS, aucune Pinecone. Comprendre les mécanismes permet de savoir ce qui se passe réellement lorsque l’on utilise par la suite LangChain ou une véritable base de données vectorielle. À la fin, on dispose d’un pipeline fonctionnel et, ce qui est encore plus utile, d’une compréhension précise de chaque valeur qu’il génère.

Le problème résolu par RAG

Les connaissances d’un modèle de langage sont figées une fois l’entraînement terminé et résident entièrement dans ses poids. Pour qu’il puisse répondre à des questions sur des documents internes de l’entreprise, un ensemble de codes ou les actualités du matin, les deux options évidentes déçoivent toutes les deux :

1. Reformer le modèle. C’est lent et coûteux, et les modifications du lendemain nécessitent une nouvelle série d’entraînements.
2. Coller tout le contenu dans la requête. Cette méthode n’est pas scalable. Même une fenêtre de un million de tokens ne peut pas contenir 50 000 documents, et les recherches sur les modèles à contexte long rapportent à plusieurs reprises un effet de « perte d’informations au milieu », où les données enfouies profondément dans une longue requête sont utilisées de manière moins efficace, même si elles y tiennent. De plus, davantage de tokens entraînent une latence accrue et des coûts plus élevés par appel.

Le véritable mécanisme du RAG est plus restreint qu’il n’y paraît. Les documents restent dans un stockage externe, et au moment de la réponse, seuls les quelques fragments pertinents pour cette question sont récupérés et transmis au modèle. Le LLM ne contient jamais vos données ; il reçoit à chaque fois un fragment frais et pertinent.

Ainsi, tout système RAG est en réalité la combinaison de deux systèmes : la récupération, qui trouve le texte adéquat dans une collection bien plus vaste pour une requête donnée, et la génération, qui transmet ce texte ainsi que la requête à un LLM pour obtenir une réponse. La suite de cet article explique comment construire ces deux composants à partir d’éléments de base.

Étape 1 : segmentation en fragments

Transformer un PDF de 50 pages en un seul vecteur élimine presque tous les signaux utiles, car les détails permettant de répondre à une question sont moyennés et deviennent insignifiants. La première étape consiste donc à diviser les documents en petits fragments ciblés.

La méthode la plus simple consiste en une fenêtre glissante de taille fixée, comptée en mots, avec un chevauchement entre les blocs voisins :

def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")

    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap  # step back so chunks share context
    return chunks

Cette clause de protection n’est pas une simple décoration. Si overlap atteint ou dépasse chunk_size, la position de départ cesse d’avancer et la boucle ne s’arrête jamais. Ajuster la valeur de overlap à une valeur plus élevée « pour obtenir plus de contexte » rend cette erreur facile à commettre ; par conséquent, la fonction refuse de s’exécuter au lieu de rester bloquée en silence.

L’effet de l’overlap est plus important qu’il n’y paraît. Sans lui, une phrase répondant à une question pourrait se trouver à cheval sur la frontière entre deux blocs et être divisée, de sorte qu’aucun des blocs ne contiendra une idée complète pouvant être récupérée. Avec l’overlap, la région frontalière reste intacte au moins dans un des blocs.

Un cas particulier à connaître : les fenêtres de mots fixes ignorent la structure des phrases et des paragraphes, ce qui peut diviser une phrase ou une ligne de tableau en deux. La segmentation tenant compte des phrases ou des paragraphes évite ce problème, mais crée des tailles d’chunks inégales, car une phrase de 10 mots et une phrase de 150 mots deviennent chacune un seul chunk. Cela altère le budget de tokens et l’influence relative d’un chunk donné sur les calculs de récupération ultérieurs. Aucune méthode de segmentation n’est exempte de compromis, et le choix dépend vraiment du type d’erreur que l’on préfère diagnostiquer.

Étape 2 : transformer du texte en nombres avec TF-IDF

Un ordinateur ne peut pas comparer directement la similarité de deux chaînes de caractères. Chaque étape suivante nécessite des vecteurs, c’est-à-dire des listes de nombres sur lesquelles on peut effectuer des opérations arithmétiques. La méthode classique, ne reposant pas sur de dépendances, pour les créer est TF-IDF, c’est-à-dire la fréquence des termes multipliée par la fréquence inverse de document.

• TF indique la fréquence à laquelle un mot apparaît dans ce fragment. Un score élevé pour le mot « kidney » suggère que le fragment concerne les reins.
• IDF mesure à quel point un mot est rare dans l’ensemble de la collection. Des mots tels que « the » et « is » se trouvent partout et ne donnent aucune indication sur le fragment que l’on lit, ce qui fait qu’ils obtiennent un score proche de zéro. Un mot rare dans un fragment constitue un signal bien plus fort.

Le code crée un vocabulaire à partir de tous les fragments, calcule l’IDF pour chaque terme, puis convertit chaque fragment en un vecteur TF-IDF :

import math
from collections import Counter

def build_vocabulary(chunks: list[str]) -> list[str]:
    vocab = set()
    for chunk in chunks:
        vocab.update(chunk.lower().split())
    return sorted(vocab)

def compute_idf(chunks: list[str], vocab: list[str]) -> dict[str, float]:
    num_chunks = len(chunks)
    idf = {}
    for word in vocab:
        containing = sum(1 for c in chunks if word in c.lower().split())
        idf[word] = math.log(num_chunks / (1 + containing))
    return idf

def compute_tf(chunk: str, vocab: list[str]) -> dict[str, float]:
    words = chunk.lower().split()
    counts = Counter(words)
    total = len(words)
    return {word: counts.get(word, 0) / total for word in vocab}

def vectorize(chunk: str, vocab: list[str], idf: dict[str, float]) -> list[float]:
    tf = compute_tf(chunk, vocab)
    return [tf[word] * idf[word] for word in vocab]

Essayez-le sur trois segments et des choses deviendront claires. Un mot présent dans presque tous les segments, comme « the », obtient un IDF proche de zéro, de sorte que sa répétition n’ajoute rien au vecteur. Un mot rare, spécifique à un thème, reçoit un IDF clairement positif. Le TF-IDF est littéralement une formule permettant de déterminer si un mot vous apporte des informations distinctives sur un segment.

L’inconvénient réside dans l’échelle. Avec 100 000 documents, le vocabulaire peut contenir plus de 500 000 mots uniques, ce qui fait que chaque vecteur de segment, même pour un segment de trois mots, compte 500 000 nombres, pour la plupart nuls. Ces vecteurs spars sont coûteux à stocker et à comparer. C’est la raison principale pour laquelle les systèmes en production passent à des embeddings neuronaux denses de 384 à 1536 dimensions, contenant presque aucun zéro.

Étape 3 : mesurer la pertinence avec la similarité cosinus

Avec les fragments en tant que vecteurs, la question suivante est de savoir dans quelle mesure un fragment est pertinent par rapport à une requête. La distance brute n’est pas l’outil adéquat, car pour le texte, l’angle entre les vecteurs est plus important que leur longueur. Un fragment qui mentionne « chat » une fois et un autre qui le répète vingt fois traitent toujours du même sujet, et la similarité cosinus les traite de la même manière sans se soucier duquel est le plus marquant.

L’implémentation nécessite un produit scalaire, une magnitude ainsi que le rapport entre ces deux valeurs :

def dot_product(a, b):
    return sum(x * y for x, y in zip(a, b))

def magnitude(v):
    return math.sqrt(sum(x * x for x in v))

def cosine_similarity(a, b):
    denom = magnitude(a) * magnitude(b)
    return 0.0 if denom == 0 else dot_product(a, b) / denom

En principe, la similarité cosinus varie de -1 à 1, et avec cette configuration TF-IDF, elle reste presque toujours dans la plage de 0 à 1 :

• 1,0 signifie que les vecteurs pointent exactement dans la même direction, ce qui représente une similarité maximale.
• 0,0 signifie qu’ils sont orthogonaux et ne partagent aucune dimension significative ; en termes simples, cela veut dire que la requête et le fragment ne possèdent aucun mot commun non trivial.

Un score de 0 n’est pas anormal. Cela indique simplement que le système rapporte correctement que le fragment n’a aucun lien avec la requête.

Un point de précision : la formule IDF lissée peut attribuer une valeur légèrement négative à un mot qui apparaît dans tous les fragments, ce qui fait que la similarité cosinus peut parfois descendre légèrement en dessous de zéro. Pour du texte réel, cela a peu d’importance, mais affirmer qu’elle n’est jamais négative serait exagéré.

Le point faible du TF-IDF avec la similarité cosinus est l’existence de synonymes. Une requête contenant le mot « automobile » obtient un score proche de zéro face à un document utilisant le mot « car », car seuls les chevauchements de vocabulaire littéral sont pris en compte, sans aucune considération du sens. Les modèles d’incorporation dense, qui placent les synonymes à proximité les uns des autres, existent justement pour combler cette lacune.

Étape 4 : le récupérateur

Lorsque la vectorisation et la mesure de similarité sont prêtes, le récupérateur consiste à vectoriser chaque morceau de texte une fois, à vectoriser la requête entrante, à évaluer chaque morceau par rapport à elle, puis à trier et conserver les k meilleurs résultats :

class VectorStore:
    def __init__(self):
        self.chunks, self.vocab, self.idf, self.vectors = [], [], {}, []

    def index(self, documents, chunk_size=50, overlap=10):
        all_chunks = []
        for doc in documents:
            all_chunks.extend(chunk_text(doc, chunk_size, overlap))
        self.chunks = all_chunks
        self.vocab = build_vocabulary(all_chunks)
        self.idf = compute_idf(all_chunks, self.vocab)
        self.vectors = [vectorize(c, self.vocab, self.idf) for c in all_chunks]

    def search(self, query, top_k=3):
        query_vector = vectorize(query, self.vocab, self.idf)
        scored = [(c, cosine_similarity(query_vector, v))
                  for c, v in zip(self.chunks, self.vectors)]
        scored.sort(key=lambda pair: pair[1], reverse=True)
        return scored[:top_k]

Un détail peut ruiner tout le travail en silence. La requête doit être convertie à l’aide du vocabulaire et des valeurs IDF calculées à partir des documents, et il est erroné d’utiliser un vocabulaire dérivé uniquement de la requête elle-même. La similarité cosinus suppose des vecteurs de même longueur, où la position i correspond au même mot dans chacun d’eux. Si l’on attribue à la requête son propre vocabulaire, la position 5 pourrait être « heart » dans un vecteur et « and » dans l’autre, ce qui rend le score de similarité structurellement sans signification, et non simplement légèrement erroné.

Étape 5 : enrichissement du prompt

Cette étape ne nécessite aucune calcul mathématique, seulement de la discipline. Vous combinez les extraits obtenus avec la question de l’utilisateur pour former un prompt qui indique au modèle exactement comment se comporter :

def build_augmented_prompt(query: str, retrieved_chunks: list[str]) -> str:
    context = "\n\n".join(
        f"[Source {i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)
    )
    return f"""Answer the question using ONLY the context below.
If the answer is not contained in the context, say "I don't know based on the provided context."
Cite which source number(s) you used.

Context:
{context}

Question: {query}

Answer:"""

Deux détails ici ne sont pas purement esthétiques.

« Utiliser UNIQUEMENT le contexte ci-dessous. » Sans cette règle explicite, le modèle mélange les extraits obtenus avec ce dont il se souvient de sa formation, et il devient impossible de distinguer quelle partie d’une réponse provient de documents fiables et laquelle est le fruit d’une supposition. Fournie avec assurance, une telle combinaison cause plus de dommages que de passer complètement à côté de la méthode RAG, car elle donne l’impression d’avoir une source alors qu’en réalité ce n’est pas le cas.

Des étiquettes numérotées pour les sources et des instructions de citation. Celles-ci permettent de tracer les réponses. Lorsqu’une réponse est incorrecte – ce qui arrivera – les sources étiquetées indiquent quel extrait a induit en erreur le modèle, que l’extrait lui-même était obsolète ou erroné, ou que le modèle a mal interprété un extrait correct. Sans étiquettes, retrouver la cause d’une mauvaise réponse relève en grande partie du hasard.

Étape 6 : les relier entre eux

Le flux complet consiste d’abord en la récupération des données, puis en leur augmentation, et enfin en une appel à un LLM. La décision de conception clé consiste à conserver l’indexation et la consultation comme deux cycles de vie distincts, plutôt qu’une seule fonction effectuant les deux tâches :

class RagPipeline:
    def __init__(self, client, model="openai/gpt-oss-20b"):
        self.store = VectorStore()
        self.client = client
        self.model = model

    def build_index(self, documents: list[str]):
        """Run once, whenever documents are loaded or updated."""
        self.store.index(documents, chunk_size=50, overlap=10)

    def answer(self, query: str, top_k: int = 3) -> str:
        """Run on every user query — never re-indexes."""
        results = self.store.search(query, top_k=top_k)
        chunks = [c for c, _ in results]
        prompt = build_augmented_prompt(query, chunks)

        response = self.client.chat.completions.create(
            model=self.model,
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

Si l’indexation était effectuée à l’intérieur de la fonction par requête, chaque requête diviserait à nouveau le texte, reconstruirait le vocabulaire et recalculerait tous les vecteurs TF-IDF pour l’ensemble de la collection, alors que rien ne change entre les requêtes. Avec 10 000 documents et 1 000 requêtes, cela entraîne 999 reconstructions identiques inutiles. Construire l’index une seule fois puis effectuer de nombreuses requêtes est la méthode recommandée par FAISS, Chroma et Pinecone ; c’est l’architecture correcte dès le départ, et non une optimisation prématurée.

Quel est le problème avec ce système

Ces limites méritent d’être abordées franchement, car ce ne sont pas des bugs. Elles découlent directement des choix de conception mentionnés ci-dessus.

• Aucune compréhension sémantique. « Car » et « automobile » ne correspondent jamais, car TF-IDF ne prend en compte que les chevauchements littéraux du vocabulaire.

Vecteurs espacés et très larges. La taille du vocabulaire évolue avec le corpus, ce qui fait que chaque vecteur s’élargit en même temps. Au-delà de quelques milliers de documents, cela devient difficile à gérer.

Rien dans le code ne vérifie que la récupération a trouvé quelque chose d’utile. Même lorsque tous les fragments du top-k obtiennent des scores proches de zéro, le pipeline les transmet quand même au modèle. La demande formulée dans l’instruction d’admettre son ignorance repose entièrement sur la coopération du LLM, de sorte qu’une récupération insuffisante peut encore être transmise à l’étape de génération sans contrôle.