Préparer les Données
La qualité du RAG dépend des données
Un pipeline RAG ne peut pas être meilleur que les données qu’il indexe. Cette leçon couvre les trois étapes de préparation : récupérer les données, les nettoyer, puis les découper en chunks optimaux pour l’indexation.
Récupérer les données
Les sources de données peuvent être variées. Voici comment gérer les plus courantes :
Fichiers texte
import os
def load_text_files(directory, extensions=(".txt", ".md", ".rst")):
"""Charger tous les fichiers texte d'un répertoire."""
documents = []
for root, dirs, files in os.walk(directory):
for f in files:
if any(f.endswith(ext) for ext in extensions):
filepath = os.path.join(root, f)
with open(filepath, "r", encoding="utf-8") as fh:
content = fh.read()
documents.append({
"content": content,
"source": filepath,
"type": "text",
})
return documents
docs = load_text_files("./data/")
print(f"{len(docs)} documents chargés")
Pages web
import requests
def fetch_web_page(url):
"""Récupérer le contenu textuel d'une page web."""
response = requests.get(url)
response.raise_for_status()
# Extraction basique du texte (pour la production, utilisez BeautifulSoup)
text = response.text
# Supprimer les balises HTML
import re
text = re.sub(r'<[^>]+>', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
return {
"content": text,
"source": url,
"type": "web",
}
# Exemple
page = fetch_web_page("https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt")
print(f"Contenu: {len(page['content'])} caractères")
PDF (avec PyPDF2)
# pip install PyPDF2
from PyPDF2 import PdfReader
def load_pdf(filepath):
"""Extraire le texte d'un PDF."""
reader = PdfReader(filepath)
pages = []
for i, page in enumerate(reader.pages):
text = page.extract_text()
if text.strip():
pages.append({
"content": text,
"source": f"{filepath}#page={i+1}",
"type": "pdf",
"page": i + 1,
})
return pages
Nettoyer les données
Le nettoyage est crucial pour la qualité du retrieval :
import re
def clean_text(text):
"""Nettoyer un texte pour l'indexation RAG."""
# Supprimer les caractères de contrôle
text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
# Normaliser les espaces et sauts de ligne
text = re.sub(r'\n{3,}', '\n\n', text)
text = re.sub(r' {2,}', ' ', text)
# Supprimer les en-têtes/pieds de page répétitifs (PDF)
text = re.sub(r'Page \d+ of \d+', '', text)
# Supprimer les URLs très longues
text = re.sub(r'https?://\S{100,}', '[URL]', text)
return text.strip()
# Appliquer le nettoyage
for doc in docs:
doc["content"] = clean_text(doc["content"])
Stratégies de chunking
Le chunking est l’étape la plus importante de la préparation. La taille et la méthode de découpage impactent directement la qualité du retrieval.
Par taille fixe avec overlap
def chunk_fixed_size(text, chunk_size=2048, overlap=200):
"""Découpage par taille fixe avec chevauchement."""
chunks = []
start = 0
while start < len(text):
end = min(start + chunk_size, len(text))
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start += chunk_size - overlap
return chunks
Par paragraphes (recommandé pour la documentation)
def chunk_by_paragraphs(text, max_size=1500, min_size=100):
"""Découpage intelligent par paragraphes."""
paragraphs = text.split("\n\n")
chunks = []
current = ""
for para in paragraphs:
para = para.strip()
if not para:
continue
if len(current) + len(para) + 2 <= max_size:
current = current + "\n\n" + para if current else para
else:
if len(current) >= min_size:
chunks.append(current)
current = para
if current and len(current) >= min_size:
chunks.append(current)
return chunks
Par sections Markdown
def chunk_by_markdown_sections(text, max_size=2000):
"""Découpage par titres Markdown (## et ###)."""
sections = re.split(r'\n(?=##\s)', text)
chunks = []
for section in sections:
section = section.strip()
if not section:
continue
if len(section) <= max_size:
chunks.append(section)
else:
# Sous-découper les sections trop longues
sub_chunks = chunk_by_paragraphs(section, max_size=max_size)
chunks.extend(sub_chunks)
return chunks
Pipeline complet de préparation
def prepare_documents(raw_docs, chunk_size=1500, overlap=200):
"""Pipeline complet : nettoyer, chunker, enrichir de métadonnées."""
all_chunks = []
for doc in raw_docs:
# 1. Nettoyer
clean = clean_text(doc["content"])
# 2. Chunker selon le type
if doc.get("source", "").endswith(".md"):
chunks = chunk_by_markdown_sections(clean, max_size=chunk_size)
else:
chunks = chunk_by_paragraphs(clean, max_size=chunk_size)
# 3. Enrichir de métadonnées
for i, chunk in enumerate(chunks):
all_chunks.append({
"text": chunk,
"source": doc["source"],
"chunk_index": i,
"total_chunks": len(chunks),
})
return all_chunks
# Exécution
chunks = prepare_documents(docs)
print(f"{len(docs)} documents -> {len(chunks)} chunks")
print(f"Taille moyenne: {sum(len(c['text']) for c in chunks) / len(chunks):.0f} caractères")
Points clés à retenir
- La qualité du RAG dépend directement de la qualité des données et du chunking
- Nettoyez les données avant l’indexation (caractères parasites, URLs longues, doublons)
- Le découpage par paragraphes ou sections Markdown préserve la cohérence sémantique
- Un chevauchement de 10-15 % entre les chunks évite de couper des idées
- Conservez les métadonnées (source, page, position) pour la traçabilité des réponses