Pipeline RAG avec FAISS

On va maintenant combiner ce qu'on a appris pour construire un mini pipeline RAG complet avec FAISS. Pas d'appel API ici : on va utiliser des vecteurs simples pour simuler des embeddings et se concentrer sur la logique du pipeline.

Le pipeline fonctionne en 3 étapes :
1. Indexation : stocker les documents et leurs vecteurs dans FAISS
2. Retrieval : trouver les k documents les plus proches d'une requête
3. Assemblage : construire le prompt RAG avec les documents trouves

Écris une classe RAGPipeline avec :

- __init__(self, dimension) : initialise un index FAISS IndexFlatL2 et une liste vide pour stocker les textes des documents

- add_documents(self, texts, vectors) : ajoute les textes et leurs vecteurs a l'index. texts est une liste de strings, vectors une liste de listes de floats.

- retrieve(self, query_vector, k=3) : cherche les k documents les plus proches du vecteur requête. Renvoie une liste de dicts avec les clés "text", "score" (la distance L2), et "rank" (le rang, en partant de 1).

- build_prompt(self, query_text, query_vector, k=3) : appelle retrieve, puis assemble le prompt au format :
Contexte:
[texte du doc 1]
[texte du doc 2]
...
\n Question: [query_text]

Exemple :

rag = RAGPipeline(dimension=3)
rag.add_documents(["Paris est en France", "Berlin est en Allemagne"],
[[1, 0, 0], [0, 1, 0]])
results = rag.retrieve([1, 0.1, 0], k=1)
results[0]["text"] vaut "Paris est en France"
results[0]["rank"] vaut 1

Tests (1/1)

Tests

rag = RAGPipeline(dimension=3)
rag.add_documents(
    ['Paris est en France', 'Berlin est en Allemagne', 'Tokyo est au Japon'],
    [[1, 0, 0], [0, 1, 0], [0, 0, 1]]
)
# Retrieve
results = rag.retrieve([1, 0.1, 0], k=2)
assert len(results) == 2, f'Doit renvoyer 2 résultats, got {len(results)}'
assert results[0]['text'] == 'Paris est en France', f'Premier résultat incorrect: {results[0]["text"]}'
assert results[0]['rank'] == 1, 'Le premier résultat doit avoir rank 1'
assert results[1]['rank'] == 2
assert isinstance(results[0]['score'], float), 'Le score doit etre un float'
# Build prompt
prompt = rag.build_prompt('Ou est Paris?', [1, 0, 0], k=2)
assert 'Contexte:' in prompt, 'Le prompt doit contenir Contexte:'
assert 'Question: Ou est Paris?' in prompt, 'Le prompt doit contenir la question'
assert 'Paris est en France' in prompt, 'Le contexte doit contenir le document pertinent'
# Ajout incrementiel
rag.add_documents(['Madrid est en Espagne'], [[0.5, 0.5, 0]])
results2 = rag.retrieve([0.5, 0.5, 0], k=1)
assert results2[0]['text'] == 'Madrid est en Espagne'

Indices (3 disponibles)

Solution officielle

import faiss
import numpy as np

class RAGPipeline:
    def __init__(self, dimension: int):
        self.index = faiss.IndexFlatL2(dimension)
        self.documents = []

    def add_documents(self, texts: list[str], vectors: list) -> None:
        self.documents.extend(texts)
        self.index.add(np.array(vectors, dtype='float32'))

    def retrieve(self, query_vector: list, k: int = 3) -> list[dict]:
        k = min(k, len(self.documents))
        distances, indices = self.index.search(
            np.array([query_vector], dtype='float32'), k=k
        )
        results = []
        for rank, (idx, dist) in enumerate(zip(indices[0], distances[0]), 1):
            results.append({
                "text": self.documents[idx],
                "score": float(dist),
                "rank": rank,
            })
        return results

    def build_prompt(self, query_text: str, query_vector: list, k: int = 3) -> str:
        results = self.retrieve(query_vector, k)
        context = "\n".join(r["text"] for r in results)
        return f"Contexte:\n{context}\n\nQuestion: {query_text}"

← Knowledge Graph avec NetworkX Agent LangGraph avec historique →

solution.py

Bravo!

Pipeline RAG avec FAISS

Tests (1/1)

Indices (3 disponibles)

Solution officielle