plateforme RAG open-source : transformer des documents en base de données vectorielle
Les modèles de langage (LLM) échouent systématiquement lorsqu’on les interroge sur des données privées ou récentes non présentes dans leur corpus d’entraînement. Une plateforme RAG open-source résout ce problème en injectant du contexte pertinent directement dans le prompt via une recherche sémantique préalable.
Le mécanisme repose sur la vectorisation de textes et une recherche de proximité dans un espace multidimensionnel. En utilisant des outils comme ChromaDB ou FAISS, on réduit la latence de récupération de l’information de plusieurs ordres de grandeur par rapport à une recherche textuelle classique sur des fichiers texte.
Ce guide détaille la mise en place d’un pipeline complet, de l’ingestion de documents PDF à la génération de réponses par un modèle local via Ollama.
🛠️ Prérequis
L’environnement doit être configuré avec les versions suivantes pour garantir la compatibilité des types et des dépendances C-extensions.
- Python 3.12+ (pour profiter des améliorations de performance de la gestion des types et de l’asyncio)
- Ollama 0.1.30+ (pour l’exécution locale des LLM)
- Docker 24.0+ (si utilisation de ChromaDB en conteneur)
- pip install langchain langchain-community chromadb sentence-transformers pypdf
📚 Comprendre plateforme RAG open-source
Le concept de plateforme RAG open-source repose sur trois piliers mathématiques et informatiques : l’embedding, le chunking et l’indexation vectorielle.
L’Embedding : Transformer un token en un vecteur de dimension $d$ (souvent 768 ou 1536). On utilise la similarité cosinus pour mesurer la distance entre deux vecteurs $A$ et $B$ : $\text{sim}(A, B) = \frac{A \cdot B}{\|A\| \|B\|}$. Si le résultat est proche de 1, les concepts sont sémantiquement proches.
Le Chunking : Découper un document en segments (chunks). Un chunk trop petit perd le contexte ; un chunk trop grand dilue l’information et augmente le coût en tokens. La stratégie standard utilise un recouvrement (overlap) pour maintenir une continuité sémantique entre les segments.
L’Indexation (HNSW) : Pour éviter une recherche linéaire $O(N)$ qui devient prohibitive sur des millions de vecteurs, on utilise des structures comme le Hierarchical Navigable Small World (HNSW). C’est un graphe de couches successives permettant une recherche de type ‘proximity search’ en complexité logarithmique.
Comparaison des approches de recherche :
- Recherche BM25 (Lexicale) : Basée sur la fréquence des mots. Efficace pour les noms propres, nulle pour la sémantique.
- Recherche Vectorielle (Dense) : Capture l’intention. Sensible au bruit si le modèle d’embedding est de faible dimension.
🐍 Le code — plateforme RAG open-source
📖 Explication
Dans le code_source, l’utilisation de RecursiveCharacterTextSplitter avec une liste de separators est intentionnelle. On descend dans la hiérarchie des délimiteurs pour préserver l’unité sémantique. Si le split sur \n\n échoue à respecter la taille cible, l’algorithme tente le split sur \n, puis sur le point, et enfin sur l’espace.
Attention au piège classique : l’utilisation de Chroma.from_documents sans spécifier un persist_directory. Par défaut, les données sont stockées en RAM. Si votre processus Python s’arrête, votre index vectoriel disparaît. Pour une plateforme RAG open-source pérenne, utilisez toujours un chemin de persistance sur disque.
Sur le plan du typage, l’utilisation de List[str] pour le retour de la recherche permet une manipulation facile des chaînes de caractères lors de la concaténation du prompt final. L’utilisation de async dans code_source_2 prépare l’application à gérer plusieurs requêtes simultanées sans bloquer la boucle d’événements (Event Loop) de Python, ce qui est crucial pour un serveur d’API.
Documentation officielle Python
🔄 Second exemple
▶️ Exemple d’utilisation
Voici comment orchestrer les deux snippets pour effectuer une requête sur un document PDF local.
import asyncio
from my_rag_module import RAGPipeline, retrieve_context
async def main():
# Initialisation du pipeline
pipeline = RAGPipeline()
# Ingestion du document
vectorstore = pipeline.ingest_pdf("rapport_annuel_2023.pdf")
# Question utilisateur
query = "Quel est le chiffre d'affaires déclaré ?"
# Récupération du contexte
context_chunks = await retrieve_context(query, vectorstore)
# Construction du prompt final
prompt = f"Contexte: {chunky_join(context_chunks)}\n\nQuestion: {query}"
print(f"Prompt généré :\n{prompt}")
if __name__ == "__main__":
asyncio.run(main())