RAG pratique : construire un assistant documentation

Vous voulez interroger une documentation technique en langage naturel ? Ce guide vous montre comment construire un RAG complet qui fonctionne sur votre machine, sans GPU, en utilisant uniquement des outils open source.

À la fin de ce guide, vous aurez un assistant capable de répondre à des questions sur la documentation Ansible (ou toute autre documentation web) en citant ses sources.

Ce que vous allez construire

Architecture choisie

Composant	Outil	Pourquoi
Scraping	trafilatura	Extraction propre du contenu web
Embeddings	multilingual-e5-small	384 dims, multilingue, 117M params
Vector DB	ChromaDB	Simple, persistant, métadonnées intégrées
LLM	Ollama (llama3.2:3b)	Local, fonctionne sur CPU, 2 GB

Glissez pour voir

Pourquoi ces choix ?

Cette stack est optimisée pour fonctionner sur un laptop sans GPU :

• multilingual-e5-small : bon compromis qualité/vitesse, supporte le français
• ChromaDB : pas de serveur externe, stockage SQLite
• llama3.2:3b : 2 GB seulement, répond en 10-20s sur CPU

Prérequis

# Python 3.10+
python3 --version

# Ollama installé et démarré
ollama --version
ollama pull llama3.2:3b

Installation des dépendances

# Créer un environnement dédié
python3 -m venv rag-local
source rag-local/bin/activate  # Linux/Mac
# rag-local\Scripts\activate   # Windows

# Installer les dépendances
pip install trafilatura sentence-transformers chromadb requests

Vérifiez que tout fonctionne :

python3 -c "import trafilatura, sentence_transformers, chromadb; print('OK')"

Phase 1 : Scraper la documentation

La première étape consiste à extraire le contenu textuel des pages de documentation.

Pourquoi trafilatura ?

trafilatura excelle pour extraire le contenu utile d’une page web en ignorant les menus, pubs et footers. C’est exactement ce dont on a besoin pour un RAG.

Code de scraping

"""Étape 1 : Scraper la documentation cible"""
import trafilatura


def scraper_page(url: str) -> dict | None:
    """
    Télécharge une page et extrait son contenu textuel.
    Retourne None si l'extraction échoue.
    """
    # Télécharger le HTML
    html = trafilatura.fetch_url(url)
    if not html:
        print(f"Échec téléchargement: {url}")
        return None

    # Extraire le contenu (sans commentaires, avec tableaux)
    texte = trafilatura.extract(
        html,
        include_comments=False,
        include_tables=True,
        output_format="txt"
    )

    if not texte:
        print(f"Pas de contenu extractible: {url}")
        return None

    # Extraire un titre depuis l'URL
    titre = url.split("/")[-1].replace(".html", "").replace("-", " ")

    return {
        "url": url,
        "titre": titre,
        "texte": texte,
        "longueur": len(texte)
    }


# Liste des pages à indexer (documentation Ansible comme exemple)
urls_documentation = [
    "https://docs.ansible.com/ansible/latest/inventory_guide/intro_inventory.html",
    "https://docs.ansible.com/ansible/latest/playbook_guide/playbooks_intro.html",
    "https://docs.ansible.com/ansible/latest/getting_started/index.html",
]


# Scraper toutes les pages
documents_bruts = []
for url in urls_documentation:
    print(f"Scraping: {url.split('/')[-1]}...")
    doc = scraper_page(url)
    if doc:
        documents_bruts.append(doc)
        print(f"  ✓ {doc['longueur']:,} caractères")

print(f"\nTotal: {len(documents_bruts)} pages, "
      f"{sum(d['longueur'] for d in documents_bruts):,} caractères")

Résultat attendu :

Scraping: intro_inventory.html...
  ✓ 30,593 caractères
Scraping: playbooks_intro.html...
  ✓ 7,583 caractères
Scraping: index.html...
  ✓ 550 caractères

Total: 3 pages, 38,726 caractères

Respectez les sites

Ne scrapez pas un site entier d’un coup. Espacez vos requêtes (time.sleep(1) entre chaque) et respectez le fichier robots.txt.

Phase 2 : Découper en chunks

Un document complet est trop long pour être utilisé directement. Il faut le découper en chunks (morceaux) de taille raisonnable.

Stratégie de chunking

Paramètre	Valeur	Pourquoi
Taille cible	400 caractères	Bon équilibre contexte/précision
Taille max	600 caractères	Évite les chunks trop longs
Unité de découpe	Phrase	Préserve le sens
Chevauchement	Non	Simplifie le pipeline

Glissez pour voir

Code de chunking

"""Étape 2 : Découper les documents en chunks"""
import re
import unicodedata


def nettoyer_texte(texte: str) -> str:
    """Normalise le texte avant découpage."""
    texte = unicodedata.normalize("NFC", texte)
    texte = re.sub(r'[ \t]+', ' ', texte)  # Espaces multiples
    texte = re.sub(r'\n{3,}', '\n\n', texte)  # Lignes vides excessives
    return texte.strip()


def decouper_en_chunks(
    texte: str,
    taille_cible: int = 400,
    taille_max: int = 600
) -> list[str]:
    """
    Découpe le texte en chunks intelligents.

    - Coupe sur les phrases (pas au milieu d'une phrase)
    - Respecte la taille cible
    - Gère les phrases très longues
    """
    texte = nettoyer_texte(texte)

    # Découper en phrases (regex robuste)
    phrases = re.split(r'(?<=[.!?])\s+', texte)
    phrases = [p.strip() for p in phrases if p.strip() and len(p) > 20]

    chunks = []
    chunk_actuel = []
    taille_actuelle = 0

    for phrase in phrases:
        # Cas spécial : phrase plus longue que taille_max
        if len(phrase) > taille_max:
            # Sauvegarder le chunk en cours
            if chunk_actuel:
                chunks.append(" ".join(chunk_actuel))
                chunk_actuel = []
                taille_actuelle = 0

            # Découper la phrase longue par mots
            mots = phrase.split()
            sous_chunk = []
            for mot in mots:
                sous_chunk.append(mot)
                if len(" ".join(sous_chunk)) > taille_cible:
                    chunks.append(" ".join(sous_chunk))
                    sous_chunk = []
            if sous_chunk:
                chunks.append(" ".join(sous_chunk))
            continue

        # Cas normal : ajouter au chunk si ça rentre
        if taille_actuelle + len(phrase) > taille_cible:
            # Sauvegarder le chunk actuel
            if chunk_actuel:
                chunks.append(" ".join(chunk_actuel))
            chunk_actuel = [phrase]
            taille_actuelle = len(phrase)
        else:
            chunk_actuel.append(phrase)
            taille_actuelle += len(phrase) + 1

    # Dernier chunk
    if chunk_actuel:
        chunks.append(" ".join(chunk_actuel))

    # Filtrer les chunks trop petits (< 50 caractères)
    return [c for c in chunks if len(c) > 50]


# Appliquer le chunking à tous les documents
tous_chunks = []
for doc in documents_bruts:
    chunks = decouper_en_chunks(doc['texte'])
    for i, chunk in enumerate(chunks):
        tous_chunks.append({
            "id": f"{doc['titre'].replace(' ', '_')}_{i:03d}",
            "texte": chunk,
            "source": doc['url'],
            "titre": doc['titre']
        })

# Statistiques
tailles = [len(c['texte']) for c in tous_chunks]
print(f"Chunks créés: {len(tous_chunks)}")
print(f"Taille: min={min(tailles)}, moy={sum(tailles)//len(tailles)}, max={max(tailles)}")

Résultat attendu :

Chunks créés: 117
Taille: min=79, moy=323, max=570

Visualiser vos chunks

Affichez quelques chunks pour vérifier la qualité :

for c in tous_chunks[:3]:
    print(f"[{c['id']}] {c['texte'][:100]}...")
    print()

Phase 3 : Créer les embeddings

Les embeddings transforment le texte en vecteurs numériques qui capturent le sens sémantique. Deux textes avec un sens proche auront des vecteurs proches.

Choix du modèle

Pour un RAG local francophone, multilingual-e5-small est optimal :

Modèle	Dimensions	Taille	Langues	Usage
all-MiniLM-L6-v2	384	80 MB	Anglais	Rapide, anglais uniquement
multilingual-e5-small	384	471 MB	100+	Multilingue, bon compromis
bge-m3	1024	1.1 GB	100+	Meilleure qualité, plus lent

Glissez pour voir

Préfixes E5

Le modèle E5 utilise des préfixes pour distinguer les passages des requêtes :

• "passage: " : pour les documents (chunks)
• "query: " : pour les questions

"""Étape 3 : Générer les embeddings"""
from sentence_transformers import SentenceTransformer

# Charger le modèle (téléchargement automatique au premier lancement)
print("Chargement du modèle d'embeddings...")
model = SentenceTransformer("intfloat/multilingual-e5-small")
print(f"✓ Modèle chargé ({model.get_sentence_embedding_dimension()} dimensions)")


def generer_embedding_passage(texte: str):
    """Génère un embedding pour un document/passage."""
    return model.encode(f"passage: {texte}")


def generer_embedding_requete(texte: str):
    """Génère un embedding pour une question/requête."""
    return model.encode(f"query: {texte}")


# Test rapide
texte_test = "Comment organiser mes serveurs en groupes avec Ansible ?"
embedding = generer_embedding_requete(texte_test)
print(f"Embedding shape: {embedding.shape}")  # (384,)

Phase 4 : Stocker dans ChromaDB

ChromaDB stocke les chunks et leurs embeddings dans une base locale persistante.

Fonction d’embedding personnalisée

ChromaDB peut appeler automatiquement notre fonction d’embedding :

"""Étape 4 : Configurer ChromaDB"""
import chromadb
from chromadb.utils.embedding_functions import EmbeddingFunction


class E5EmbeddingFunction(EmbeddingFunction):
    """Fonction d'embedding pour E5 avec préfixe passage."""

    def __init__(self, model_name: str = "intfloat/multilingual-e5-small"):
        self.model = SentenceTransformer(model_name)

    def __call__(self, input: list[str]) -> list[list[float]]:
        # Ajouter le préfixe "passage:" pour les documents
        texts = [f"passage: {t}" for t in input]
        return self.model.encode(texts, show_progress_bar=False).tolist()


# Créer le client avec persistance
db_path = "./rag_database"
client = chromadb.PersistentClient(path=db_path)

# Créer l'embedding function
embedding_fn = E5EmbeddingFunction()

# Créer ou récupérer la collection
collection = client.get_or_create_collection(
    name="documentation",
    embedding_function=embedding_fn,
    metadata={"description": "Documentation Ansible"}
)

print(f"Collection '{collection.name}' prête")
print(f"Documents existants: {collection.count()}")

Indexer les chunks

L’indexation consiste à stocker chaque chunk dans ChromaDB avec son embedding. ChromaDB appelle automatiquement notre E5EmbeddingFunction pour calculer les vecteurs.

Pourquoi indexer par lots ? Envoyer 100+ chunks d’un coup peut provoquer des timeouts. En découpant en lots de 100, on limite les risques et on peut suivre la progression.

"""Indexer tous les chunks dans ChromaDB"""
import time

# Supprimer les anciens documents (optionnel, pour réindexation)
# Utile quand vous modifiez la stratégie de chunking et voulez repartir de zéro
if collection.count() > 0:
    collection.delete(where={})  # Vider la collection

print(f"Indexation de {len(tous_chunks)} chunks...")
start = time.time()

# Ajouter par lots de 100 pour éviter les timeouts
batch_size = 100
for i in range(0, len(tous_chunks), batch_size):
    batch = tous_chunks[i:i + batch_size]

    # collection.add() fait 3 choses :
    # 1. Appelle embedding_fn() sur chaque document
    # 2. Stocke les vecteurs dans l'index HNSW
    # 3. Stocke le texte et les métadonnées dans SQLite
    collection.add(
        documents=[c['texte'] for c in batch],
        ids=[c['id'] for c in batch],
        metadatas=[{"source": c['source'], "titre": c['titre']} for c in batch]
    )
    print(f"  Lot {i//batch_size + 1}: {len(batch)} chunks")

print(f"✓ Indexation terminée en {time.time() - start:.1f}s")
print(f"  Total: {collection.count()} documents")

À ce stade, vous avez :

• Une base de données vectorielle locale (./rag_database/)
• 117 chunks indexés avec leurs embeddings
• Les métadonnées (source, titre) pour les citations

Résultat attendu :

Indexation de 117 chunks...
  Lot 1: 100 chunks
  Lot 2: 17 chunks
✓ Indexation terminée en 3.1s
  Total: 117 documents

Phase 5 : Recherche sémantique

Maintenant que les chunks sont indexés, on peut chercher les plus pertinents pour une question.

Fonction de recherche

La recherche sémantique trouve les chunks les plus proches en sens de la question — même sans mots en commun.

Comment ça marche :

1. La question est convertie en vecteur (avec le préfixe query:)
2. ChromaDB compare ce vecteur à tous les vecteurs stockés
3. Les n_resultats chunks les plus proches sont retournés

"""Étape 5 : Recherche sémantique"""


def rechercher(question: str, n_resultats: int = 3) -> list[dict]:
    """
    Recherche les chunks les plus pertinents pour une question.

    Args:
        question: La question de l'utilisateur
        n_resultats: Nombre de chunks à retourner (3 par défaut)

    Returns:
        Liste de dicts avec 'texte', 'source', 'score'
    """
    # 1. Générer l'embedding de la question
    # Le préfixe "query:" est différent de "passage:" utilisé à l'indexation
    # C'est ce qui permet à E5 de comprendre qu'on cherche, pas qu'on stocke
    query_embedding = embedding_fn.model.encode(f"query: {question}").tolist()

    # 2. Rechercher dans ChromaDB
    # ChromaDB utilise HNSW (Hierarchical Navigable Small World)
    # pour trouver les voisins les plus proches en quelques millisecondes
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_resultats
    )

    # 3. Formater les résultats
    # ChromaDB retourne des distances (0 = identique), on convertit en score
    documents = []
    for doc, meta, distance in zip(
        results['documents'][0],
        results['metadatas'][0],
        results['distances'][0]
    ):
        documents.append({
            "texte": doc,
            "source": meta['source'],
            "score": 1 - distance  # Convertir distance en similarité
        })

    return documents

Test de recherche

# Tester la recherche
question = "Comment créer des groupes de serveurs dans l'inventaire Ansible ?"
print(f"Question: {question}\n")

resultats = rechercher(question, n_resultats=3)

for i, r in enumerate(resultats):
    print(f"[{i+1}] Score: {r['score']:.3f}")
    print(f"    {r['texte'][:100]}...")
    print()

Résultat attendu :

Question: Comment créer des groupes de serveurs dans l'inventaire Ansible ?

[1] Score: 0.773
    by datacenter, and each datacenter uses its own NTP server and database server...

[2] Score: 0.767
    Your inventory defines the managed nodes you automate and the variables associ...

[3] Score: 0.755
    Here's the same basic inventory file in YAML format:
ungrouped:
hosts:
mail.exam...

Score de similarité

Un score de 0.7+ indique une bonne correspondance sémantique. En dessous de 0.5, la pertinence est douteuse.

Phase 6 : Génération avec Ollama

L’étape finale : envoyer la question + contexte à un LLM local pour obtenir une réponse.

Vérifier Ollama

# Vérifier que le service tourne
curl http://localhost:11434/api/tags

# Vérifier que le modèle est disponible
ollama list

Fonction de génération

"""Étape 6 : Génération avec Ollama"""
import requests


def generer_reponse(
    question: str,
    contexte: list[dict],
    model: str = "llama3.2:3b"
) -> str:
    """
    Génère une réponse en utilisant le contexte récupéré.

    Args:
        question: La question de l'utilisateur
        contexte: Liste de chunks pertinents
        model: Modèle Ollama à utiliser

    Returns:
        Réponse générée
    """
    # Formater le contexte
    contexte_formate = "\n\n---\n\n".join([
        f"[Source: {c['source']}]\n{c['texte']}"
        for c in contexte
    ])

    # Construire le prompt RAG
    prompt = f"""Tu es un assistant technique spécialisé dans la documentation.
Réponds à la question en te basant UNIQUEMENT sur le contexte fourni.
Si le contexte ne contient pas l'information, dis-le clairement.
Réponds en français de manière concise et précise.

CONTEXTE:
{contexte_formate}

QUESTION: {question}

RÉPONSE:"""

    # Appeler Ollama
    try:
        response = requests.post(
            "http://localhost:11434/api/generate",
            json={
                "model": model,
                "prompt": prompt,
                "stream": False,
                "options": {
                    "temperature": 0.2,  # Faible pour plus de factualité
                    "num_predict": 300,  # Limiter la longueur
                }
            },
            timeout=300  # 5 minutes max (CPU peut être lent)
        )

        if response.status_code == 200:
            return response.json()["response"]
        else:
            return f"Erreur HTTP {response.status_code}: {response.text}"

    except requests.exceptions.Timeout:
        return "Erreur: Timeout - la génération a pris trop de temps"
    except Exception as e:
        return f"Erreur: {e}"

Test complet

# Pipeline RAG complet
question = "Comment organiser mes serveurs en groupes avec Ansible ?"

print(f"Question: {question}")
print("-" * 50)

# 1. Rechercher le contexte
print("Recherche du contexte pertinent...")
contexte = rechercher(question, n_resultats=3)
print(f"  ✓ {len(contexte)} chunks trouvés")

# 2. Générer la réponse
print("Génération de la réponse (peut prendre 10-30s sur CPU)...")
reponse = generer_reponse(question, contexte)

print("\nRÉPONSE:")
print("=" * 50)
print(reponse)
print("=" * 50)

Exemple de réponse :

Question: Comment organiser mes serveurs en groupes avec Ansible ?
--------------------------------------------------
Recherche du contexte pertinent...
  ✓ 3 chunks trouvés
Génération de la réponse (peut prendre 10-30s sur CPU)...

RÉPONSE:
==================================================
Pour organiser vos serveurs en groupes avec Ansible, vous devez créer
un fichier d'inventaire. Voici la syntaxe de base :

[webservers]
web1.example.com
web2.example.com

[dbservers]
db1.example.com
db2.example.com

Les noms entre crochets ([webservers], [dbservers]) définissent les
groupes. Chaque serveur listé sous un groupe en fait partie. Un même
serveur peut appartenir à plusieurs groupes.
==================================================

Pipeline complet

Voici le code complet, prêt à copier-coller :

Le code

#!/usr/bin/env python3
"""
Assistant RAG local - Documentation Ansible
Fonctionne sans GPU avec Ollama + ChromaDB
"""
import os
import re
import shutil
import time
import unicodedata
import requests
import trafilatura
import chromadb
from sentence_transformers import SentenceTransformer
from chromadb.utils.embedding_functions import EmbeddingFunction


# ============================================================
# CONFIGURATION
# ============================================================
DB_PATH = "./rag_database"
OLLAMA_MODEL = "llama3.2:3b"
EMBEDDING_MODEL = "intfloat/multilingual-e5-small"


# ============================================================
# SCRAPING
# ============================================================
def scraper_page(url: str) -> dict | None:
    """Extrait le contenu textuel d'une page web."""
    html = trafilatura.fetch_url(url)
    if not html:
        return None
    texte = trafilatura.extract(html, include_comments=False, include_tables=True)
    if not texte:
        return None
    return {
        "url": url,
        "titre": url.split("/")[-1].replace(".html", ""),
        "texte": texte
    }


# ============================================================
# CHUNKING
# ============================================================
def decouper_en_chunks(texte: str, taille_cible: int = 400) -> list[str]:
    """Découpe le texte en chunks de taille optimale."""
    texte = unicodedata.normalize("NFC", texte)
    texte = re.sub(r'[ \t]+', ' ', texte)

    phrases = re.split(r'(?<=[.!?])\s+', texte)
    phrases = [p.strip() for p in phrases if p.strip() and len(p) > 20]

    chunks = []
    chunk_actuel = []
    taille_actuelle = 0

    for phrase in phrases:
        if len(phrase) > 600:  # Phrase trop longue
            if chunk_actuel:
                chunks.append(" ".join(chunk_actuel))
                chunk_actuel = []
                taille_actuelle = 0
            mots = phrase.split()
            sous_chunk = []
            for mot in mots:
                sous_chunk.append(mot)
                if len(" ".join(sous_chunk)) > taille_cible:
                    chunks.append(" ".join(sous_chunk))
                    sous_chunk = []
            if sous_chunk:
                chunks.append(" ".join(sous_chunk))
            continue

        if taille_actuelle + len(phrase) > taille_cible and chunk_actuel:
            chunks.append(" ".join(chunk_actuel))
            chunk_actuel = [phrase]
            taille_actuelle = len(phrase)
        else:
            chunk_actuel.append(phrase)
            taille_actuelle += len(phrase) + 1

    if chunk_actuel:
        chunks.append(" ".join(chunk_actuel))

    return [c for c in chunks if len(c) > 50]


# ============================================================
# EMBEDDINGS
# ============================================================
class E5EmbeddingFunction(EmbeddingFunction):
    def __init__(self):
        self.model = SentenceTransformer(EMBEDDING_MODEL)

    def __call__(self, input: list[str]) -> list[list[float]]:
        texts = [f"passage: {t}" for t in input]
        return self.model.encode(texts, show_progress_bar=False).tolist()


# ============================================================
# CLASSE PRINCIPALE
# ============================================================
class RAGAssistant:
    """Assistant RAG pour documentation."""

    def __init__(self, reset_db: bool = False):
        # Réinitialiser la DB si demandé
        if reset_db and os.path.exists(DB_PATH):
            shutil.rmtree(DB_PATH)

        # Initialiser les embeddings
        print("Chargement du modèle d'embeddings...")
        self.embedding_fn = E5EmbeddingFunction()

        # Initialiser ChromaDB
        self.client = chromadb.PersistentClient(path=DB_PATH)
        self.collection = self.client.get_or_create_collection(
            name="documentation",
            embedding_function=self.embedding_fn
        )
        print(f"✓ Base de données prête ({self.collection.count()} documents)")

    def indexer_urls(self, urls: list[str]) -> int:
        """Scrape et indexe une liste d'URLs."""
        tous_chunks = []

        for url in urls:
            print(f"  Scraping: {url.split('/')[-1]}...", end=" ")
            doc = scraper_page(url)
            if not doc:
                print("✗")
                continue

            chunks = decouper_en_chunks(doc['texte'])
            for i, chunk in enumerate(chunks):
                tous_chunks.append({
                    "id": f"{doc['titre']}_{i:03d}",
                    "texte": chunk,
                    "source": url
                })
            print(f"✓ ({len(chunks)} chunks)")

        if tous_chunks:
            self.collection.add(
                documents=[c['texte'] for c in tous_chunks],
                ids=[c['id'] for c in tous_chunks],
                metadatas=[{"source": c['source']} for c in tous_chunks]
            )

        return len(tous_chunks)

    def rechercher(self, question: str, n: int = 3) -> list[dict]:
        """Recherche les chunks pertinents."""
        emb = self.embedding_fn.model.encode(f"query: {question}").tolist()
        res = self.collection.query(query_embeddings=[emb], n_results=n)

        return [
            {"texte": d, "source": m["source"], "score": 1 - dist}
            for d, m, dist in zip(
                res['documents'][0],
                res['metadatas'][0],
                res['distances'][0]
            )
        ]

    def repondre(self, question: str) -> str:
        """Répond à une question en utilisant le RAG."""
        # Rechercher le contexte
        contexte = self.rechercher(question)

        # Formater le contexte
        ctx = "\n\n---\n\n".join([c['texte'] for c in contexte])

        # Construire le prompt
        prompt = f"""Tu es un assistant technique. Réponds en français en te basant
UNIQUEMENT sur le contexte. Si l'information n'est pas présente, dis-le.

CONTEXTE:
{ctx}

QUESTION: {question}

RÉPONSE:"""

        # Appeler Ollama
        try:
            resp = requests.post(
                "http://localhost:11434/api/generate",
                json={
                    "model": OLLAMA_MODEL,
                    "prompt": prompt,
                    "stream": False,
                    "options": {"temperature": 0.2, "num_predict": 300}
                },
                timeout=300
            )
            if resp.status_code == 200:
                return resp.json()["response"]
            return f"Erreur HTTP {resp.status_code}"
        except Exception as e:
            return f"Erreur: {e}"


# ============================================================
# UTILISATION
# ============================================================
if __name__ == "__main__":
    # Créer l'assistant (reset_db=True pour réindexer)
    assistant = RAGAssistant(reset_db=True)

    # Indexer la documentation
    urls = [
        "https://docs.ansible.com/ansible/latest/inventory_guide/intro_inventory.html",
        "https://docs.ansible.com/ansible/latest/playbook_guide/playbooks_intro.html",
    ]
    print("\nIndexation de la documentation:")
    n = assistant.indexer_urls(urls)
    print(f"  ✓ {n} chunks indexés")

    # Poser une question
    question = "Comment créer des groupes de serveurs avec Ansible ?"
    print(f"\n{'='*50}")
    print(f"Question: {question}")
    print(f"{'='*50}")

    print("\nRecherche et génération...")
    reponse = assistant.repondre(question)

    print(f"\nRÉPONSE:\n{reponse}")

Test de validation

Pour valider que votre RAG fonctionne correctement, exécutez ce test :

"""Test de validation du pipeline RAG"""


def test_rag():
    """Vérifie que tous les composants fonctionnent."""
    print("=" * 50)
    print("TEST DE VALIDATION DU PIPELINE RAG")
    print("=" * 50)

    erreurs = []

    # Test 1: Scraping
    print("\n[1/5] Test scraping...", end=" ")
    try:
        doc = scraper_page(
            "https://docs.ansible.com/ansible/latest/inventory_guide/intro_inventory.html"
        )
        if doc and len(doc['texte']) > 1000:
            print("✓")
        else:
            erreurs.append("Scraping: contenu insuffisant")
            print("✗")
    except Exception as e:
        erreurs.append(f"Scraping: {e}")
        print("✗")

    # Test 2: Chunking
    print("[2/5] Test chunking...", end=" ")
    try:
        texte_test = "Phrase un. " * 50 + "Phrase deux. " * 50
        chunks = decouper_en_chunks(texte_test)
        if len(chunks) >= 2 and all(len(c) < 700 for c in chunks):
            print("✓")
        else:
            erreurs.append("Chunking: taille incorrecte")
            print("✗")
    except Exception as e:
        erreurs.append(f"Chunking: {e}")
        print("✗")

    # Test 3: Embeddings
    print("[3/5] Test embeddings...", end=" ")
    try:
        emb = embedding_fn.model.encode("test")
        if len(emb) == 384:
            print("✓")
        else:
            erreurs.append(f"Embeddings: dimension {len(emb)} != 384")
            print("✗")
    except Exception as e:
        erreurs.append(f"Embeddings: {e}")
        print("✗")

    # Test 4: Recherche
    print("[4/5] Test recherche...", end=" ")
    try:
        if collection.count() > 0:
            results = rechercher("test", n_resultats=1)
            if results and 'texte' in results[0]:
                print("✓")
            else:
                erreurs.append("Recherche: pas de résultats")
                print("✗")
        else:
            print("⊘ (base vide)")
    except Exception as e:
        erreurs.append(f"Recherche: {e}")
        print("✗")

    # Test 5: Ollama
    print("[5/5] Test Ollama...", end=" ")
    try:
        resp = requests.get("http://localhost:11434/api/tags", timeout=5)
        if resp.status_code == 200:
            print("✓")
        else:
            erreurs.append(f"Ollama: HTTP {resp.status_code}")
            print("✗")
    except Exception as e:
        erreurs.append(f"Ollama: {e}")
        print("✗")

    # Résumé
    print("\n" + "=" * 50)
    if erreurs:
        print(f"ÉCHEC: {len(erreurs)} erreur(s)")
        for e in erreurs:
            print(f"  - {e}")
        return False
    else:
        print("SUCCÈS: Tous les tests passent ✓")
        return True


# Exécuter le test
test_rag()

Résultat attendu si tout fonctionne :

==================================================
TEST DE VALIDATION DU PIPELINE RAG
==================================================

[1/5] Test scraping... ✓
[2/5] Test chunking... ✓
[3/5] Test embeddings... ✓
[4/5] Test recherche... ✓
[5/5] Test Ollama... ✓

==================================================
SUCCÈS: Tous les tests passent ✓

Dépannage

Ollama ne répond pas

# Vérifier que le service tourne
systemctl status ollama

# Démarrer si nécessaire
ollama serve &

# Vérifier qu'un modèle est installé
ollama list

# Installer un modèle si absent
ollama pull llama3.2:3b

ChromaDB trop lent

# Utiliser l'indexation par lots
batch_size = 50  # Réduire si trop lent

for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i + batch_size]
    collection.add(...)
    time.sleep(0.1)  # Pause entre les lots

Génération timeout

# Réduire la taille du contexte
contexte = rechercher(question, n_resultats=2)  # 2 au lieu de 3

# Limiter la longueur de chaque chunk
ctx = "\n".join([c['texte'][:300] for c in contexte])

# Augmenter le timeout
timeout=600  # 10 minutes

Mauvaise qualité des réponses

1. Vérifier les chunks : affichez quelques exemples pour voir s’ils sont cohérents
2. Tester la recherche : les bons chunks sont-ils trouvés ?
3. Ajuster le prompt : soyez plus précis sur le format attendu
4. Essayer un modèle plus grand : llama3.2:8b si vous avez la mémoire

À retenir

Point clé	Détail
Architecture	trafilatura → chunks → embeddings → ChromaDB → Ollama
Embeddings	multilingual-e5-small avec préfixes passage: / query:
Chunking	400 caractères cible, découpe sur les phrases
Recherche	Top 3 chunks, score > 0.7 = pertinent
Génération	Prompt RAG explicite, température basse (0.2)

Glissez pour voir

Prochaines étapes

RAG avancé Recherche hybride, reranking et contextual retrieval

RAG en production Monitoring, cache et optimisation des coûts

ChromaDB en détail Collections, métadonnées et filtres avancés

Ollama avancé Modèles, quantization et API en détail

×

📖 Voir la documentation →