Créer un pipeline RAG local en Python sans frais d’API
Un tutoriel pratique pour créer une génération augmentée par la récupération d’informations sur CPU à l’aide de sentence-transformers, numpy et Flan-T5, incluant le découpage des données, la recherche et un test A/B.
Sources: Build a RAG System From Scratch — Hands-On, No API Costs
Un pipeline complet de génération augmentée par récupération d’informations peut être mis en place et exécuté entièrement sur un ordinateur personnel, uniquement avec une CPU, à l’aide de modèles open source gratuits, sans clé OpenAI, sans carte de crédit et sans limites de vitesse. L’objectif est de comprendre chaque composant plutôt que de copier une boîte noire.
La meilleure façon d’apprendre est par la pratique : gardez un terminal et un fichier Python nommé rag.py ouverts, exécutez chaque bloc et prédisez la sortie avant de l’exécuter réellement. De courtes exercices apparaissent après la plupart des étapes, et c’est là que se produit la majeure partie de l’apprentissage.
Le problème que RAG résout
Un modèle de langage ne connaît que ce qui figurait dans ses données d’entraînement. Des notes privées, une wiki d’entreprise, la réunion de la semaine dernière ou tout document inconnu lui restent invisibles. Lorsqu’on lui pose des questions à leur sujet, il admet son ignorance ou, pire encore, invente quelque chose avec assurance.
RAG divise cette tâche en deux parties :
- Récupérer : trouver les extraits de vos propres documents qui sont pertinents pour la question. Il s’agit d’un problème de recherche.
- Générer : fournir ces extraits au modèle en lui donnant l’instruction de répondre à partir d’eux.
Le modèle cesse de deviner et commence à lire. Tout le reste concerne les méthodes pour bien effectuer ces deux étapes.
question ──► [embed] ──► [search your docs] ──► top chunks ──┐
▼
[LLM: "answer using this context"] ──► answer
Le kit d’outils gratuit
Trois composants couvrent l’ensemble du système. sentence-transformers transforme le texte en vecteurs, en utilisant un petit modèle qui fonctionne sur la CPU et peut être téléchargé une seule fois. numpy gère les calculs de similarité, et toute la « base de données vectorielle » tient en environ cinq lignes, ce qui permet de voir comment elle fonctionne. transformers avec Flan-T5 écrit la réponse, car il est adapté aux instructions, fonctionne sur la CPU et n’a pas besoin d’API.
Installez tout une seule fois :
pip install sentence-transformers transformers torch numpy
Lors du premier chargement, chaque modèle télécharge ses poids depuis Hugging Face. L’embedder pèse environ 80 MB et flan-t5-base environ 1 GB. Par la suite, ces données sont mémorisées en cache et permettent de fonctionner hors ligne. En cas de capacité disque ou de RAM limitée, flan-t5-small (environ 300 MB) peut remplacer le modèle de base, avec une qualité inférieure mais une vitesse bien supérieure.
Étape 1 : un corpus que le modèle ne peut pas connaître
RAG est le plus facile à comprendre avec des documents que aucun modèle n’aurait pu voir ; c’est pourquoi le script commence par une petite wiki fictive d’entreprise. Créez rag.py et collez-y les documents :
# rag.py
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
Le wiki continue avec davantage d’entrées :
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""", """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""", """The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""", """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""", """Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]
Avant de construire quoi que ce soit, prouvez d’abord que le problème existe. Dans une shell Python, interrogez un modèle simple au sujet de Nimbus :
from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])
Cela produira quelque chose, mais ce quelque chose sera incorrect ou vide, car il s’agit d’une estimation. Gardez cette réponse en tête, car le reste du tutoriel la corrige.
Étape 2 : segmentation en blocs
Les documents entiers ne sont pas envoyés au modèle. Ils sont divisés en de plus petits blocs pour deux raisons : la recherche devient plus précise, car le modèle trouve le paragraphe qui répond à la question plutôt qu’un manuel entier, et la fenêtre de contexte du modèle est limitée.
Ajoutez une fonction de segmentation en blocs :
def chunk_text(text, chunk_size=60, overlap=15):
"""Split text into overlapping chunks of `chunk_size` words."""
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 by `overlap` so context isn't cut
return chunks
Ensuite, créez la liste des blocs en enregistrant quel document chaque bloc provient :
# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
print("-", c["text"][:70], "...")
L’overlap existe parce qu’un fait peut se situer à la frontière entre deux critères, comme une politique de remboursement définie entre « dans 5 » et « jours ouvrés ». Sans overlap, aucune partie ne contient l’ensemble du fait. En répétant quelques mots à chaque point de séparation, les faits peuvent survivre à cette coupure.
Exercice : relancez l’opération avec chunk_size=20 pour obtenir des parties plus petites, puis essayez chunk_size=500, où chaque document court devient une seule partie. Avant de lancer, devinez quel paramètre permet d’obtenir une précision plus élevée mais risque de faire perdre le contexte environnant. Les petites parties sont précises mais manquent de contexte, et cette tension est au cœur du processus de chunking. Il n’existe pas de taille universellement correcte.
Étape 3 : embeddings
Un embedding est un vecteur de nombres qui encode le sens d’un texte. Les fragments de texte ayant un sens similaire se transforment en vecteurs pointant dans la même direction, même en l’absence de mots communs, de sorte que « Combien ça coûte ? » se trouve à côté de « Quel est le prix ? »
Importez le modèle d’embedding :
from sentence_transformers import SentenceTransformer
Puis insérez chaque morceau :
embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape) # (num_chunks, 384)
Chaque bloc correspond maintenant à un vecteur de 384 dimensions, où 384 représente simplement le nombre de valeurs que ce modèle spécifique utilise pour exprimer des significations.
Exercice : observez le fonctionnement de cette représentation par le biais d’une petite expérience :
import numpy as np
pairs = embedder.encode(
["the price of the pro plan", "how much does it cost", "the weather in Paris"],
normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3)) # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW
L’expression a @ b représente un produit scalaire, qui pour des vecteurs unitaires équivaut à la similitude cosinus. Une valeur de 1,0 indique une même direction tandis que 0 signifie qu’il n’y a pas de relation. Le premier couple devrait obtenir un score bien supérieur au second, même si les phrases ne partagent aucun mot ; c’est là l’essence de la recherche sémantique. Essayez de deviner les deux valeurs avec une précision inférieure à 0,2 avant d’exécuter le traitement.
Étape 4 : récupération
Pour l’étape de recherche, incorporez la question, comparez-la avec chaque vecteur de chunk et conservez les meilleurs résultats. Comme les vecteurs sont normalisés, la « comparaison » se réduit à une multiplication matricielle.
import numpy as np
La fonction de récupération :
def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec # cosine similarity to every chunk
top_idx = np.argsort(scores)[::-1][:k] # indices of the k highest scores
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
print(f"[{score:.3f}] {text[:80]}...")
Le bloc relatif aux prix doit figurer en tête avec la meilleure note, même si la question mentionnait « coût » tandis que le document indiquait « 8 dollars par mois ». Il s’agit d’une correspondance de sens et non d’une correspondance de mots-clés.
Exercice : essayez trois questions pour voir ce qui apparaît :
retrieve("Puis-je récupérer mon argent ?")doit renvoyer le bloc relatif au remboursement, bien que le mot « remboursement » n’ait jamais été utilisé dans la question.retrieve("Mes données sont-elles sécurisées ?")doit afficher le bloc relatif au chiffrement.retrieve("Fonctionne-t-il encore sans connexion Internet ?")devrait restituer le bloc relatif à la synchronisation hors ligne.
Beaucoup de gens sous-estiment cette étape. Lorsque la récupération fournit des fragments incorrects, aucune incitation astucieuse ne permet de retrouver la bonne réponse. Ce qui est donné en entrée se retrouve en sortie, c’est pourquoi les efforts de débogage doivent d’abord se concentrer ici.
Étape 5 : génération
Les deux parties se rencontrent maintenant. Il s’agit de créer une incitation qui transmette les fragments récupérés à Flan-T5 et lui indique de ne répondre qu’à partir d’eux.
from transformers import pipeline
La fonction de génération :
generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved) prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:""" out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
print(f" [{score:.3f}] {text[:70]}...")
Comparez le résultat avec la réponse incorrecte ou vide de l’Étape 1. Le modèle et la question sont identiques, et la seule différence est que le texte pertinent a été fourni pour lecture. Cet écart constitue toute la valeur de RAG.
Exercice : posez une question que les documents ne couvrent pas, par exemple concernant un client de bureau Linux pour Nimbus, et appelez rag_answer avec cette question. Une configuration correctement conçue admet qu’elle ne sait pas. Si votre système invente une réponse, il s’agit d’un véritable mode de défaillance appelé hallucination malgré la récupération d’informations, et une section ultérieure aborde ce problème.
Le script final
Voici l’intégralité de rag.py sous forme de programme exécutable unique avec une boucle interactive. Il commence par les importations :
# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
Le corps entier suit :
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""",
"""Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""",
"""The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""",
"""Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""",
"""Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
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
return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
[c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec
top_idx = np.argsort(scores)[::-1][:k]
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved)
prompt = (
"Answer the question using only the context below. "
"If the answer is not in the context, say you don't know.\n\n"
f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
)
out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip()if __name__ == "__main__":
print("RAG ready. Ask about Nimbus (or type 'quit').\n")
while True:
q = input("You: ").strip()
if q.lower() in {"quit", "exit", ""}:
break
print("Nimbus bot:", rag_answer(q), "\n")
Commencez par :
python rag.py
Essayez ensuite des questions telles que la période de remboursement, le fait de savoir si les utilisateurs gratuits ont accès au chat en direct ou encore quelle est la capacité de stockage gratuite.
Prouver la valeur par un test A/B
Une expérience permet de clarifier cela en posant la même question avec et sans contexte fourni. Tout d’abord, une fonction de référence :
def no_rag(question):
out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
return out[0]["generated_text"].strip()
Puis la comparaison :
q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context :", rag_answer(q))
La réponse sans contexte n’est qu’une supposition, tandis que celle qui prend en compte le contexte doit indiquer que la messagerie en direct est réservée aux abonnés Pro et que les utilisateurs gratuits reçoivent des e-mails. Essayez trois questions différentes. Cette démonstration A/B est la plus convaincante qui existe pour RAG, alors gardez-la pour expliquer l’idée aux autres.
Mises à niveau selon leur utilité
Avec un système fonctionnel en main, ces mises à niveau sont les plus importantes, approximativement dans l’ordre de leur utilité :
- Afficher les sources. La fonction renvoie déjà la valeur
retrieved, il faut donc afficher les extraits sous chaque réponse. C’est sur ce point que les vrais produits RAG réussissent ou échouent. Une réponse citée est fiable, tandis qu’une réponse non citée n’est qu’une supposition présentée dans un format attrayant.
k à 5 et fixez une note minimale, en éliminant tout résultat inférieur à environ 0,25. Lorsqu’aucun chunk ne satisfait les critères, répondez "Je ne sais pas" et évitez complètement d’appeler le modèle. Cela élimine la plupart des hallucinations comme celle observée dans l’exercice du pas 5.chunk_vectors @ q_vec parcourt chaque chunk, ce qui est acceptable pour des centaines de données mais lent pour des millions. Chroma (pip install chromadb) indexe les vecteurs pour des recherches rapides et les enregistre sur disque afin qu’aucun embedding ne soit recréé au démarrage. Les concepts d’embedding, de stockage et de recherche cosinus restent les mêmes, Chroma se contentant de les appliquer à grande échelle.generator(...) par une requête vers son API locale. Un niveau API gratuit de Gemini ou Groq de Google fonctionne également en ne modifiant que la ligne de génération dans rag_answer. Une seule fonction change tandis que le mécanisme de récupération reste inchangé.rag_answer et vérifiez leur exactitude. C’est une méthode peu glamour, mais c’est ce qui distingue un outil expérimental d’un produit réel, car ce qui n’est pas mesuré ne peut être amélioré.Diagnostic des problèmes
RAG est une recherche associée à un assistant d’écriture, les deux éléments étant liés, et presque tous les problèmes trouvent leur origine dans l’un ou l’autre de ces deux aspects. Une réponse incorrecte avec le bon fragment récupéré relève d’un problème de génération ou de prompt. Une réponse incorrecte où le bon fragment n’est jamais apparu correspond à un problème de récupération lié au découpage en fragments, aux embeddings ou au paramètre k. Déterminez d’abord quel aspect est défaillant avant de modifier quoi que ce soit. Le fait de conserver retrieve() et rag_answer() en tant que fonctions distinctes permet précisément d’examiner chaque aspect séparément, et cette séparation constitue ici l’habitude la plus importante.
Résolution de problèmes
- Téléchargement lent ou échoué : les modèles sont téléchargés depuis Hugging Face lors de la première exécution et stockés en cache sous
~/.cache/huggingface. Une fois en cache, ils fonctionnent hors ligne.
google/flan-t5-small avec une valeur de chunk_size égale à 40. Ce modèle d’incorporation est suffisamment petit pour ne pas ralentir le système.pip install torch --index-url https://download.pytorch.org/whl/cpu.max_new_tokens, car Flan-T5 a tendance à s’arrêter prématurément. Un générateur plus puissant, comme celui fourni par Ollama, est plus efficace que n’importe quel ajustement.En travaillant sur ces exercices, on acquiert une compréhension plus profonde du RAG que ce que la plupart des gens obtiennent en se contentant d’utiliser des bibliothèques préfabriquées.