RAG : Connecter les LLMs à vos données
Le Retrieval-Augmented Generation (RAG) est devenu l'architecture standard pour déployer des LLMs en production. Pourquoi ? Parce que les modèles comme GPT-4 ou Claude ont une connaissance figée (training cutoff) et ne peuvent pas accéder à vos données propriétaires. Le RAG résout ce problème en permettant au LLM de récupérer dynamiquement les informations pertinentes avant de générer sa réponse.
Principe RAG : Au lieu de se fier uniquement à sa mémoire (paramètres), le LLM interroge une base de données vectorielle pour trouver les documents/passages les plus pertinents, puis génère une réponse groundée sur ces sources.
Avantages RAG :
- Données à jour : Documents ajoutés → disponibles immédiatement
- Sources citées : Traçabilité des réponses (compliance, trust)
- Pas de retraining : Ajouter connaissances sans fine-tuner modèle
- Réduction hallucinations : Réponses basées sur faits réels
- Scalabilité : Millions documents indexables (vs contexte LLM limité)
- Coût optimisé : Pas besoin contexte géant (128k tokens coûteux)
Adoption RAG
Selon Gartner, 78% des déploiements LLM entreprise en 2025 utilisent RAG. Les cas d'usage incluent chatbots support (42%), knowledge management (31%), analyse documentaire (18%) et code generation (9%). Les vector databases connaissent une croissance de 340% YoY (Pinecone, Weaviate, Qdrant).
Architecture RAG complète
Pipeline RAG : Indexation et Query
2 phases distinctes :
- Indexation (offline) : Convertir documents en embeddings vectoriels
- Query (online) : Rechercher contexte pertinent + générer réponse
PHASE 1: INDEXATION (Exécutée périodiquement)
═══════════════════════════════════════════════
[Documents sources]
├── PDF: Documentation produit (1200 fichiers)
├── Confluence: Wiki interne (8400 pages)
├── SharePoint: Procédures (3200 docs)
├── Database: FAQ clients (15k Q&A)
└── Web scraping: Site web public
↓
[1. Document Parsing]
├── PDF: PyPDF2, pdfplumber
├── HTML: BeautifulSoup
├── Docs: python-docx
└── Extraction texte brut
↓
[2. Chunking] (Découpage en morceaux)
├── Stratégie: Recursive character splitter
├── Taille chunk: 500-1000 tokens
├── Overlap: 50-100 tokens (préserve contexte)
└── Exemple:
Document 10k mots → 25 chunks 400 tokens
↓
[3. Embedding Generation] (Vectorisation)
├── Modèle: text-embedding-ada-002 (OpenAI)
│ ou sentence-transformers (open source)
├── Dimension: 1536 (OpenAI) ou 768/384
├── Output: Chunk → Vector [0.023, -0.145, 0.891, ...]
└── Batch processing: 100 chunks/sec
↓
[4. Vector Storage]
├── Vector Database: Pinecone, Weaviate, ChromaDB
├── Index: HNSW (Hierarchical Navigable Small World)
├── Metadata: {source, page, timestamp, category}
└── Total: 500k chunks = 500k vectors indexed
════════════════════════════════════════════════
PHASE 2: QUERY (Temps réel - Latence <2s)
═══════════════════════════════════════════════
[User Query]
"Quelle est la procédure de remboursement produit défectueux?"
↓
[1. Query Embedding]
├── Même modèle qu'indexation (consistency!)
├── Query → Vector [0.034, -0.156, 0.782, ...]
└── Latency: 50ms
↓
[2. Vector Similarity Search]
├── Recherche ANN (Approximate Nearest Neighbor)
├── Metric: Cosine similarity
├── Top-K: Retrieve 5-10 chunks les plus similaires
├── Latency: 100-200ms (index optimisé)
└── Résultats:
• Chunk #23487 (similarity: 0.94) - "Procédure_Remboursement.pdf, p.3"
• Chunk #8921 (similarity: 0.91) - "FAQ_SAV.md"
• Chunk #15673 (similarity: 0.88) - "Conditions_Generales.pdf, p.12"
• [7 autres chunks...]
↓
[3. Context Assembly] (Reranking optionnel)
├── Reranker model (Cohere, Sentence-BERT)
├── Rerank top-10 → Keep top-3 meilleurs
├── Concatenation chunks + metadata
└── Context final: ~2000 tokens
↓
[4. LLM Prompt Construction]
┌────────────────────────────────────────────────┐
│ System: Tu es assistant SAV. Réponds en │
│ français basé UNIQUEMENT sur contexte fourni. │
│ Cite sources. Si info manquante, dis-le. │
│ │
│ CONTEXTE: │
│ [Chunk 1 - Procédure_Remboursement.pdf, p.3] │
│ "Les produits défectueux peuvent être │
│ retournés sous 30 jours. Procédure: │
│ 1. Contactez SAV (sav@company.com) │
│ 2. Obtenez numéro RMA │
│ 3. Renvoyez produit (frais port remboursés) │
│ 4. Remboursement sous 7-10 jours ouvrés..." │
│ │
│ [Chunk 2 - FAQ_SAV.md] │
│ "Q: Qui paie les frais de retour? │
│ R: Si défaut avéré, frais remboursés par │
│ entreprise. Joindre justificatif envoi." │
│ │
│ [Chunk 3 - Conditions_Generales.pdf, p.12] │
│ "Garantie légale 2 ans. Produits défectueux │
│ remplacés ou remboursés à choix client..." │
│ │
│ USER QUERY: │
│ "Quelle est la procédure de remboursement │
│ produit défectueux?" │
└────────────────────────────────────────────────┘
↓
[5. LLM Generation]
├── Modèle: GPT-4 Turbo, Claude 3.5, Llama 4 70B
├── Temperature: 0.2 (factuel, pas créatif)
├── Max tokens: 500
└── Latency: 800-1200ms
↓
[Response]
"Voici la procédure de remboursement pour produit défectueux:
1. **Contactez le SAV** : Envoyez email à sav@company.com en
décrivant le défaut.
2. **Obtenez numéro RMA** : Le SAV vous fournira un numéro de
retour (RMA - Return Merchandise Authorization).
3. **Renvoyez le produit** : Emballez produit + RMA, envoyez à
l'adresse fournie. Les **frais de port sont remboursés** si
défaut avéré (joindre justificatif envoi).
4. **Remboursement** : Vous recevrez remboursement sous **7-10 jours
ouvrés** après réception produit.
**Délais** : Cette procédure doit être effectuée dans les **30 jours**
suivant achat.
**Alternative** : Selon garantie légale (2 ans), vous pouvez aussi
choisir **remplacement** du produit au lieu du remboursement.
**Sources**:
• Procédure_Remboursement.pdf (page 3)
• FAQ_SAV.md
• Conditions_Generales.pdf (page 12)"
LATENCY TOTALE: 1.2-1.8 secondes ✓
Code implementation complet
# RAG Pipeline complet (Python)
from langchain.document_loaders import DirectoryLoader, UnstructuredPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.llms import ChatOpenAI
from langchain.chains import RetrievalQA
import pinecone
# ════════════════════════════════════════════════════════════
# PHASE 1: INDEXATION (Run once / périodiquement)
# ════════════════════════════════════════════════════════════
class RAGIndexer:
def __init__(self):
# Embedding model
self.embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002",
openai_api_key="sk-..."
)
# Pinecone setup
pinecone.init(
api_key="pc-...",
environment="gcp-starter"
)
self.index_name = "company-knowledge-base"
def load_documents(self, source_dir="./documents"):
"""Charge documents depuis répertoire"""
# Loaders pour différents formats
loaders = [
DirectoryLoader(
source_dir,
glob="**/*.pdf",
loader_cls=UnstructuredPDFLoader
),
DirectoryLoader(
source_dir,
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader
),
DirectoryLoader(
source_dir,
glob="**/*.txt",
loader_cls=TextLoader
)
]
documents = []
for loader in loaders:
documents.extend(loader.load())
print(f"Loaded {len(documents)} documents")
return documents
def chunk_documents(self, documents):
"""Découpe documents en chunks"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 1000 caractères par chunk
chunk_overlap=100, # Overlap 100 chars
separators=["\n\n", "\n", " ", ""] # Hiérarchie séparateurs
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")
return chunks
def create_vectorstore(self, chunks):
"""Crée/met à jour vector store Pinecone"""
# Création index si pas existe
if self.index_name not in pinecone.list_indexes():
pinecone.create_index(
self.index_name,
dimension=1536, # OpenAI embeddings
metric="cosine"
)
# Indexation chunks
vectorstore = Pinecone.from_documents(
chunks,
self.embeddings,
index_name=self.index_name
)
print(f"Indexed {len(chunks)} chunks to Pinecone")
return vectorstore
def index_all(self):
"""Pipeline complet indexation"""
docs = self.load_documents()
chunks = self.chunk_documents(docs)
vectorstore = self.create_vectorstore(chunks)
return vectorstore
# ════════════════════════════════════════════════════════════
# PHASE 2: QUERY (Temps réel)
# ════════════════════════════════════════════════════════════
class RAGQueryEngine:
def __init__(self, index_name="company-knowledge-base"):
# Embeddings (même modèle qu'indexation!)
self.embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002"
)
# Vectorstore
pinecone.init(
api_key="pc-...",
environment="gcp-starter"
)
self.vectorstore = Pinecone(
index=pinecone.Index(index_name),
embedding_function=self.embeddings.embed_query,
text_key="text"
)
# LLM
self.llm = ChatOpenAI(
model="gpt-4-turbo",
temperature=0.2, # Factuel
max_tokens=500
)
# Retrieval QA chain
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # "stuff" = concat all context
retriever=self.vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5} # Top 5 chunks
),
return_source_documents=True # Retourne sources
)
def query(self, question):
"""Exécute query RAG"""
result = self.qa_chain({"query": question})
response = {
"answer": result["result"],
"sources": [
{
"content": doc.page_content,
"metadata": doc.metadata
}
for doc in result["source_documents"]
]
}
return response
def query_with_filter(self, question, metadata_filter):
"""Query avec filtrage metadata"""
# Exemple: Chercher seulement dans docs FAQ
retriever = self.vectorstore.as_retriever(
search_type="similarity",
search_kwargs={
"k": 5,
"filter": metadata_filter # {"category": "FAQ"}
}
)
qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
retriever=retriever,
return_source_documents=True
)
return qa_chain({"query": question})
# ════════════════════════════════════════════════════════════
# USAGE
# ════════════════════════════════════════════════════════════
# Indexation (1 fois ou périodiquement)
indexer = RAGIndexer()
indexer.index_all()
# Query (temps réel)
engine = RAGQueryEngine()
result = engine.query(
"Quelle est la procédure de remboursement produit défectueux?"
)
print("ANSWER:", result["answer"])
print("\nSOURCES:")
for i, source in enumerate(result["sources"]):
print(f"{i+1}. {source['metadata'].get('source', 'Unknown')}")
print(f" Excerpt: {source['content'][:200]}...")
Vector databases : Comparaison et choix
Top 5 vector databases 2025
COMPARAISON VECTOR DATABASES:
┌─────────────────────────────────────────────────────────────────┐
│ Pinecone Weaviate Qdrant ChromaDB Milvus │
├─────────────────────────────────────────────────────────────────┤
│ Type Managed Hybrid OSS OSS OSS │
│ Scale ∞ 100M+ 50M+ 10M+ 1B+ │
│ Latency <50ms <100ms <80ms <120ms <60ms │
│ Price $0.096/hr Self/Mng Free Free Self/Mng │
│ Ease ★★★★★ ★★★★☆ ★★★☆☆ ★★★★★ ★★☆☆☆ │
│ Filters ★★★★★ ★★★★★ ★★★★☆ ★★★☆☆ ★★★★★ │
│ Hybrid ✓ ✓ ✓ ✗ ✓ │
└─────────────────────────────────────────────────────────────────┘
Hybrid search: Vector + keyword (BM25) combinés
Recommandations par use case :
PINECONE:
✓ Startup/PME: Pas infra à gérer
✓ Scaling rapide: Serverless auto-scale
✓ Besoin simplicité: Setup 5 minutes
✗ Coût élevé: $70-500/mois selon volume
Use case: Chatbots SaaS, applications prototypes
WEAVIATE:
✓ Enterprise: Features avancées (GraphQL, multi-tenancy)
✓ Hybrid search: Vector + keyword excellent
✓ Flexible: Cloud managed ou self-hosted
Use case: Search engines, recommendation systems
QDRANT:
✓ Performance: Rust (très rapide)
✓ Open source: Pas vendor lock-in
✓ Filtering: Metadata filters puissants
Use case: Production mid-scale, souveraineté données
CHROMADB:
✓ Simplicité: Embedded database (SQLite-like)
✓ Développement: Local testing facile
✗ Scale limité: Pas production large scale
Use case: Prototypes, applications locales, dev/test
MILVUS:
✓ Scale extreme: Milliards de vectors
✓ GPU support: Accélération matérielle
✗ Complexité: Setup/maintenance lourds
Use case: Large enterprises, recherche académique
Exemple setup Pinecone (Managed)
import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings
# 1. Initialize Pinecone
pinecone.init(
api_key="YOUR_PINECONE_API_KEY",
environment="gcp-starter" # ou us-west1-gcp, etc.
)
# 2. Create index
index_name = "my-rag-index"
if index_name not in pinecone.list_indexes():
pinecone.create_index(
name=index_name,
dimension=1536, # OpenAI ada-002 embeddings
metric="cosine", # ou "euclidean", "dotproduct"
pods=1, # 1 pod = $0.096/hr
pod_type="p1.x1" # Performance tier
)
# 3. Connect vectorstore
embeddings = OpenAIEmbeddings()
vectorstore = Pinecone(
index=pinecone.Index(index_name),
embedding_function=embeddings.embed_query,
text_key="text"
)
# 4. Add documents
from langchain.schema import Document
docs = [
Document(
page_content="Python est un langage de programmation interprété...",
metadata={"source": "wiki_python.txt", "category": "programming"}
),
Document(
page_content="JavaScript est le langage du web...",
metadata={"source": "wiki_js.txt", "category": "programming"}
),
# ... more documents
]
vectorstore.add_documents(docs)
# 5. Query
results = vectorstore.similarity_search(
"Qu'est-ce que Python?",
k=3,
filter={"category": "programming"} # Metadata filtering
)
for doc in results:
print(doc.page_content)
print(doc.metadata)
Techniques avancées RAG
1. Hybrid Search (Vector + Keyword)
Combine semantic search (embeddings) et keyword search (BM25).
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever
# Vector retriever
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# Keyword retriever (BM25)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# Ensemble (combine results)
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4] # 60% vector, 40% keyword
)
# Query hybrid
results = ensemble_retriever.get_relevant_documents(
"Python programming language"
)
# Résultats:
# - Vector catch: "Langage programmation orienté objet..." (paraphrase)
# - BM25 catch: "Python est..." (exact match "Python")
# → Best of both worlds!
Quand utiliser :
- Queries avec termes techniques précis (noms produits, acronymes)
- Domaines où exactitude terminologie compte (juridique, médical)
2. Reranking (Améliore pertinence)
Problème : Similarity search parfois retourne faux positifs. Solution : Reranker model affine top results.
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
# Base retriever (get top 20)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
# Reranker (Cohere API)
reranker = CohereRerank(
cohere_api_key="YOUR_COHERE_KEY",
top_n=5 # Keep only top 5 after reranking
)
# Compression retriever (retrieval + rerank)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=base_retriever
)
# Query (retourne top 5 reranked)
results = compression_retriever.get_relevant_documents(
"How to deploy Python application?"
)
# Amélioration typique: +12-18% précision vs vector search seul
Alternatives reranking :
- Cohere Rerank : API managed, excellent performance
- Sentence-BERT CrossEncoder : Open source, self-hosted
- Custom reranker : Fine-tuned sur votre domaine
3. Query Expansion (Reformulation)
Problème : User query mal formulée → miss résultats pertinents. Solution : LLM reformule query en variantes.
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# Prompt expansion
expansion_prompt = PromptTemplate(
input_variables=["query"],
template="""
Given this user query, generate 3 alternative formulations
to improve search results:
Original query: {query}
Alternative formulations:
1.
2.
3.
"""
)
expansion_chain = LLMChain(llm=llm, prompt=expansion_prompt)
# Expand query
original_query = "deploy python app"
expansions = expansion_chain.run(query=original_query)
# Output:
# 1. "How to deploy Python application to production?"
# 2. "Python deployment best practices"
# 3. "Hosting Python web services"
# Search with all queries
all_queries = [original_query] + parse_expansions(expansions)
all_results = []
for query in all_queries:
results = vectorstore.similarity_search(query, k=3)
all_results.extend(results)
# Deduplicate + rerank
final_results = deduplicate_and_rerank(all_results, top_k=5)
4. Parent-Child Chunking
Problème : Chunks petits perdent contexte, chunks grands dilue signal. Solution : Indexer chunks petits, mais retourner chunks larges.
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
# Parent splitter (chunks larges, 2000 chars)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
# Child splitter (chunks petits, 400 chars - pour indexation)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
# Document store (stocke parents)
docstore = InMemoryStore()
# Retriever
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=docstore,
child_splitter=child_splitter,
parent_splitter=parent_splitter
)
retriever.add_documents(documents)
# Query: Recherche sur petits chunks (précis), retourne parents (contexte)
results = retriever.get_relevant_documents("Python deployment")
# → Chunks retournés: 2000 chars (contexte complet)
Production best practices
Monitoring et observabilité
# Logging et metrics (production)
import logging
from prometheus_client import Counter, Histogram
import time
# Prometheus metrics
rag_query_counter = Counter('rag_queries_total', 'Total RAG queries')
rag_latency = Histogram('rag_query_latency_seconds', 'RAG query latency')
rag_errors = Counter('rag_errors_total', 'RAG errors')
class ProductionRAG:
def query(self, question, user_id=None):
start_time = time.time()
try:
# Increment counter
rag_query_counter.inc()
# Log query
logging.info(f"RAG query from user {user_id}: {question[:100]}")
# Execute RAG
result = self.qa_chain({"query": question})
# Record latency
latency = time.time() - start_time
rag_latency.observe(latency)
# Log sources retrieved
logging.info(f"Retrieved {len(result['source_documents'])} sources")
logging.info(f"Latency: {latency:.2f}s")
# Quality check
if self._is_low_quality(result):
logging.warning("Low quality response detected")
# Trigger alert, fallback, etc.
return result
except Exception as e:
# Increment error counter
rag_errors.inc()
# Log error
logging.error(f"RAG error: {str(e)}", exc_info=True)
# Fallback response
return {
"answer": "Désolé, erreur technique. Contactez support.",
"sources": [],
"error": True
}
def _is_low_quality(self, result):
"""Détecte réponses basse qualité"""
# Heuristiques:
# - Réponse trop courte (<50 chars)
# - Pas de sources retrieved
# - Similarity score trop bas
if len(result["result"]) < 50:
return True
if len(result["source_documents"]) == 0:
return True
# Check similarity scores (si disponibles)
# ...
return False
Caching pour performance
from functools import lru_cache
import hashlib
class CachedRAG:
def __init__(self):
self.cache = {} # ou Redis pour distributed cache
def query(self, question):
# Cache key (hash question)
cache_key = hashlib.md5(question.encode()).hexdigest()
# Check cache
if cache_key in self.cache:
logging.info("Cache HIT")
return self.cache[cache_key]
# Cache MISS - execute RAG
logging.info("Cache MISS")
result = self.qa_chain({"query": question})
# Store in cache (TTL 1h)
self.cache[cache_key] = result
# Pour Redis: redis.setex(cache_key, 3600, json.dumps(result))
return result
# Amélioration latency typique: -60% (1.2s → 0.5s) avec cache
Coûts et optimisation
ANALYSE COÛTS RAG (100k queries/mois):
Embeddings (indexation):
├── Documents: 50k chunks
├── Model: text-embedding-ada-002
├── Coût: $0.0001 / 1k tokens
├── Tokens: 50k chunks × 400 tokens = 20M tokens
└── TOTAL: $2 / mois (one-time + updates)
Embeddings (queries):
├── Queries: 100k/mois
├── Tokens: 100k × 20 tokens = 2M tokens
└── TOTAL: $0.20 / mois
Vector Database:
├── Pinecone: 50k vectors, 1 pod
└── TOTAL: $70 / mois
LLM Generation:
├── Queries: 100k
├── Input: 100k × 2k tokens (context) = 200M tokens
├── Output: 100k × 500 tokens = 50M tokens
├── GPT-4 Turbo pricing: $10/1M in, $30/1M out
└── TOTAL: (200 × $10) + (50 × $30) = $2000 + $1500 = $3500/mois
TOTAL RAG STACK: $3572/mois (100k queries)
→ $0.036 / query
OPTIMISATIONS:
1. Cache (30% queries) → -$1050/mois (-30%)
2. Llama 4 70B self-hosted → -$2800/mois (vs GPT-4)
3. Qdrant self-hosted → -$70/mois (vs Pinecone)
→ TOTAL optimisé: $652/mois (82% économie!)
Articles connexes
- GPT-5 : OpenAI dévoile son modèle révolutionnaire pour 2026
- Fine-tuning LLMs : Guide pratique 2025 pour adapter vos modèles
- Meta Llama 4 : L'open source qui défie les modèles propriétaires
Conclusion : RAG devient le standard
Le RAG est désormais l'architecture de référence pour connecter LLMs aux données entreprise. En permettant aux modèles d'accéder dynamiquement à des connaissances à jour, le RAG résout les limitations fondamentales des LLMs (knowledge cutoff, hallucinations, données propriétaires).
Forces RAG :
- Données toujours à jour (pas besoin retraining)
- Sources citées (traçabilité, compliance)
- Scalable (millions documents indexables)
- Réduction hallucinations (-60-80% vs LLM seul)
Production considerations :
- Vector DB choice : Managed (Pinecone) vs self-hosted (Qdrant)
- Chunking strategy : 500-1000 tokens, overlap 10-20%
- Advanced techniques : Hybrid search, reranking, query expansion
- Monitoring : Latency, quality, errors (Prometheus, logs)
Coûts :
- Stack complet : $0.03-0.04/query (GPT-4)
- Optimisé : $0.006/query (Llama 4 + self-hosted)
- Break-even vs fine-tuning : >10k queries/mois
2026 : 90% déploiements LLM entreprise utiliseront RAG selon Gartner. Les vector databases deviennent infrastructure critique, au même titre que SQL databases.



