RAG de A à Z : construire un système de question-réponse sur vos documents

from sentence_transformers import SentenceTransformer
import numpy as np

# Charger un modèle d'embedding multilingue
model = SentenceTransformer("BAAI/bge-m3")

# Exemples de textes
textes = [
    "Comment configurer un serveur Nginx en reverse proxy ?",
    "La mise en place d'un proxy inverse avec Nginx",
    "Recette de la tarte aux pommes traditionnelle",
    "Les bases de données vectorielles pour l'IA",
]

# Générer les embeddings
embeddings = model.encode(textes, normalize_embeddings=True)

# Calculer la matrice de similarité cosinus
similarity_matrix = np.dot(embeddings, embeddings.T)

print("Matrice de similarité cosinus :")
for i, t1 in enumerate(textes):
    for j, t2 in enumerate(textes):
        if i < j:
            print(f"  [{i}] vs [{j}]: {similarity_matrix[i][j]:.4f}")
            # [0] vs [1]: 0.8234 — très similaires (même sujet)
            # [0] vs [2]: 0.1123 — très différents
            # [0] vs [3]: 0.3456 — légèrement liés (tech)

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    TokenTextSplitter,
)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings


# ============================================================
# Stratégie 1 : Découpage récursif par caractères (recommandé pour commencer)
# ============================================================
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,           # Taille cible en caractères
    chunk_overlap=200,         # Chevauchement entre chunks
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""],  # Hiérarchie de séparateurs
    is_separator_regex=False,
)

text = """
La tokenisation est le processus de découpage d'un texte en unités élémentaires 
appelées tokens. Les tokens peuvent être des mots, des sous-mots ou des caractères 
selon l'algorithme utilisé.

Les algorithmes les plus courants sont BPE (Byte-Pair Encoding), WordPiece et 
SentencePiece. BPE est utilisé par GPT, WordPiece par BERT, et SentencePiece 
par LLaMA et T5.

Le choix de la tokenisation impacte directement les performances du modèle. Un 
vocabulaire trop petit force le modèle à découper les mots en nombreux sous-tokens, 
augmentant la longueur des séquences. Un vocabulaire trop grand augmente la taille 
des embeddings et risque de sous-entraîner les tokens rares.
"""

chunks = recursive_splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i} ({len(chunk)} chars): {chunk[:80]}...")


# ============================================================
# Stratégie 2 : Découpage par en-têtes Markdown (structurel)
# ============================================================
headers_to_split = [
    ("#", "Titre principal"),
    ("##", "Section"),
    ("###", "Sous-section"),
]

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split,
    strip_headers=False,
)

markdown_doc = """
# Guide de déploiement

## Prérequis
Il faut installer Docker et docker-compose sur votre serveur.
La version minimale de Docker est 20.10.

## Installation
### Étape 1 : Cloner le repository
Utilisez git clone pour récupérer le code source.

### Étape 2 : Configuration
Copiez le fichier .env.example en .env et modifiez les variables.
"""

md_chunks = markdown_splitter.split_text(markdown_doc)
for chunk in md_chunks:
    print(f"Section: {chunk.metadata} → {chunk.page_content[:60]}...")


# ============================================================
# Stratégie 3 : Découpage sémantique (avancé)
# ============================================================
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=75,  # Seuil de changement de sujet
)

# Le chunker sémantique détecte automatiquement les changements
# de sujet dans le texte et coupe aux frontières sémantiques
sem_chunks = semantic_splitter.split_text(text)


# ============================================================
# Stratégie 4 : Chunking personnalisé avec métadonnées enrichies
# ============================================================
from dataclasses import dataclass
from typing import List, Dict
import hashlib


@dataclass
class EnrichedChunk:
    content: str
    metadata: Dict
    embedding: List[float] = None
    
    @property
    def id(self) -> str:
        return hashlib.md5(self.content.encode()).hexdigest()


def smart_chunk_document(
    text: str,
    source: str,
    chunk_size: int = 800,
    chunk_overlap: int = 150,
) -> List[EnrichedChunk]:
    """Découpe un document avec métadonnées enrichies."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    
    raw_chunks = splitter.split_text(text)
    enriched = []
    
    for i, chunk in enumerate(raw_chunks):
        enriched.append(EnrichedChunk(
            content=chunk,
            metadata={
                "source": source,
                "chunk_index": i,
                "total_chunks": len(raw_chunks),
                "char_count": len(chunk),
                "word_count": len(chunk.split()),
                "position": "start" if i == 0 else "end" if i == len(raw_chunks)-1 else "middle",
            }
        ))
    
    return enriched

import chromadb
from chromadb.config import Settings
from chromadb.utils import embedding_functions

# ============================================================
# Mode 1 : Client embeddé (développement/prototypage)
# ============================================================
client = chromadb.PersistentClient(path="./chroma_db")

# Configurer la fonction d'embedding
# Option A : Sentence Transformers (local, gratuit)
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-m3",
    device="cuda",  # ou "cpu"
)

# Option B : OpenAI (API, payant mais simple)
# embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
#     api_key="sk-...",
#     model_name="text-embedding-3-small",
# )

# Créer une collection
collection = client.get_or_create_collection(
    name="documentation_technique",
    embedding_function=embedding_fn,
    metadata={
        "hnsw:space": "cosine",       # Métrique de distance
        "hnsw:M": 32,                  # Connexions par nœud (qualité ↑)
        "hnsw:construction_ef": 200,   # Effort de construction (qualité ↑)
    }
)

# ============================================================
# Insérer des documents
# ============================================================
documents = [
    "Docker est une plateforme de conteneurisation qui permet d'empaqueter des applications.",
    "Kubernetes orchestre les conteneurs Docker à grande échelle en production.",
    "FastAPI est un framework web Python moderne, rapide et facile à utiliser.",
    "Les embeddings transforment le texte en vecteurs numériques pour la recherche sémantique.",
    "Le monitoring avec Prometheus et Grafana est essentiel pour les applications en production.",
    "CI/CD signifie Continuous Integration / Continuous Deployment.",
    "Les microservices sont une architecture où l'application est divisée en services indépendants.",
    "gRPC est un framework RPC haute performance développé par Google.",
]

metadatas = [
    {"source": "docker_guide.md", "category": "infrastructure"},
    {"source": "k8s_intro.md", "category": "infrastructure"},
    {"source": "fastapi_docs.md", "category": "backend"},
    {"source": "ml_basics.md", "category": "machine_learning"},
    {"source": "monitoring.md", "category": "devops"},
    {"source": "cicd_guide.md", "category": "devops"},
    {"source": "architecture.md", "category": "backend"},
    {"source": "grpc_guide.md", "category": "backend"},
]

ids = [f"doc_{i}" for i in range(len(documents))]

# Insertion par batch
collection.add(
    documents=documents,
    metadatas=metadatas,
    ids=ids,
)
print(f"✅ {collection.count()} documents insérés")

# ============================================================
# Recherche par similarité
# ============================================================
results = collection.query(
    query_texts=["comment déployer une application Python ?"],
    n_results=3,
    include=["documents", "metadatas", "distances"],
)

print("\n🔍 Résultats de recherche :")
for i in range(len(results["ids"][0])):
    print(f"  [{i+1}] Score: {1 - results['distances'][0][i]:.4f}")
    print(f"      Source: {results['metadatas'][0][i]['source']}")
    print(f"      Texte: {results['documents'][0][i][:80]}...")

# ============================================================
# Recherche avec filtres sur les métadonnées
# ============================================================
filtered_results = collection.query(
    query_texts=["monitoring et observabilité"],
    n_results=3,
    where={"category": "devops"},  # Filtrer par catégorie
    include=["documents", "metadatas", "distances"],
)

# ============================================================
# Mode 2 : Client-serveur (production)
# ============================================================
# Démarrer le serveur : chroma run --host 0.0.0.0 --port 8000
# client_prod = chromadb.HttpClient(host="localhost", port=8000)

from langchain.retrievers import (
    ContextualCompressionRetriever,
    MultiQueryRetriever,
    EnsembleRetriever,
)
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings


# ============================================================
# Retriever hybride (Dense + BM25) avec Re-ranking
# ============================================================

# 1. Retriever dense (vectoriel)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="documentation_technique",
)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 2. Retriever sparse (BM25)
# Charger tous les documents pour BM25
all_docs = vectorstore.get()
bm25_retriever = BM25Retriever.from_texts(
    texts=all_docs["documents"],
    metadatas=all_docs["metadatas"],
)
bm25_retriever.k = 10

# 3. Ensemble retriever (fusion des résultats)
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4],  # Pondération dense/sparse
)

# 4. Re-ranker (cross-encoder)
reranker_model = HuggingFaceCrossEncoder(
    model_name="cross-encoder/ms-marco-MiniLM-L-12-v2"
)
reranker = CrossEncoderReranker(
    model=reranker_model,
    top_n=5,  # Garder les 5 meilleurs après re-ranking
)

# 5. Pipeline complet : recherche + re-ranking
retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=ensemble_retriever,
)

# Utilisation
query = "Comment configurer un healthcheck Docker ?"
results = retriever.invoke(query)

for i, doc in enumerate(results):
    print(f"[{i+1}] {doc.page_content[:100]}...")
    print(f"    Source: {doc.metadata.get('source', 'N/A')}\n")


# ============================================================
# Multi-query retriever (reformulation automatique)
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

multi_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm,
)

# Le LLM génère automatiquement des reformulations :
# "Comment configurer un healthcheck Docker ?"
# → "Docker healthcheck configuration"
# → "Vérification de santé conteneur Docker"
# → "Docker HEALTHCHECK instruction Dockerfile"
multi_results = multi_retriever.invoke(query)

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough


# Template de prompt RAG optimisé
RAG_PROMPT_TEMPLATE = """Tu es un assistant technique expert. Réponds à la question 
en te basant UNIQUEMENT sur le contexte fourni ci-dessous.

Règles strictes :
1. Si l'information n'est pas dans le contexte, dis-le explicitement
2. Cite tes sources en mentionnant le document d'origine entre [crochets]
3. Structure ta réponse avec des titres et des listes si approprié
4. Sois précis et concis — pas de blabla
5. Si plusieurs sources se contredisent, mentionne les différentes perspectives

Contexte :
{context}

Question : {question}

Réponse structurée :"""

prompt = ChatPromptTemplate.from_template(RAG_PROMPT_TEMPLATE)

# Modèle de génération
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,    # Basse température pour la précision factuelle
    max_tokens=1500,
)

# Fonction de formatage du contexte
def format_docs(docs):
    formatted = []
    for i, doc in enumerate(docs):
        source = doc.metadata.get("source", "inconnu")
        formatted.append(f"[Source {i+1}: {source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(formatted)

# Chain RAG complète avec LCEL
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

# Utilisation
response = rag_chain.invoke("Comment monitorer une application FastAPI en production ?")
print(response)

"""
Pipeline RAG complet — Prêt pour la production
Auteur: Sophie Laurent | Février 2026
"""

import os
import logging
from pathlib import Path
from typing import List, Optional, Dict, Any
from dataclasses import dataclass

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    DirectoryLoader,
    CSVLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.documents import Document


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class RAGConfig:
    """Configuration centralisée du pipeline RAG."""
    # Chemins
    documents_dir: str = "./documents"
    chroma_dir: str = "./chroma_db"
    collection_name: str = "rag_collection"
    
    # Chunking
    chunk_size: int = 800
    chunk_overlap: int = 150
    
    # Retrieval
    dense_k: int = 10
    bm25_k: int = 10
    dense_weight: float = 0.6
    rerank_top_n: int = 5
    
    # Embedding
    embedding_model: str = "text-embedding-3-small"
    
    # Generation
    llm_model: str = "gpt-4o"
    llm_temperature: float = 0.1
    max_tokens: int = 2000
    
    # Re-ranking
    reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"


class RAGPipeline:
    """Pipeline RAG complet avec recherche hybride et re-ranking."""
    
    def __init__(self, config: RAGConfig = None):
        self.config = config or RAGConfig()
        self.embeddings = OpenAIEmbeddings(model=self.config.embedding_model)
        self.llm = ChatOpenAI(
            model=self.config.llm_model,
            temperature=self.config.llm_temperature,
            max_tokens=self.config.max_tokens,
        )
        self.vectorstore = None
        self.retriever = None
        self._all_docs = []
    
    # ========================================================
    # Phase 1 : Indexation
    # ========================================================
    
    def load_documents(self, directory: str = None) -> List[Document]:
        """Charge les documents depuis un répertoire (multi-format)."""
        doc_dir = directory or self.config.documents_dir
        logger.info(f"📂 Chargement des documents depuis {doc_dir}")
        
        loaders = {
            "**/*.pdf": PyPDFLoader,
            "**/*.txt": TextLoader,
            "**/*.md": UnstructuredMarkdownLoader,
            "**/*.csv": CSVLoader,
        }
        
        all_docs = []
        for glob_pattern, loader_cls in loaders.items():
            try:
                loader = DirectoryLoader(
                    doc_dir,
                    glob=glob_pattern,
                    loader_cls=loader_cls,
                    show_progress=True,
                    use_multithreading=True,
                )
                docs = loader.load()
                all_docs.extend(docs)
                logger.info(f"  ✅ {len(docs)} documents {glob_pattern}")
            except Exception as e:
                logger.warning(f"  ⚠️ Erreur {glob_pattern}: {e}")
        
        logger.info(f"📊 Total: {len(all_docs)} documents chargés")
        return all_docs
    
    def chunk_documents(self, documents: List[Document]) -> List[Document]:
        """Découpe les documents en chunks."""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.config.chunk_size,
            chunk_overlap=self.config.chunk_overlap,
            separators=["\n\n", "\n", ". ", ", ", " ", ""],
            length_function=len,
        )
        
        chunks = splitter.split_documents(documents)
        
        # Enrichir les métadonnées
        for i, chunk in enumerate(chunks):
            chunk.metadata["chunk_id"] = i
            chunk.metadata["char_count"] = len(chunk.page_content)
        
        logger.info(f"✂️ {len(documents)} documents → {len(chunks)} chunks")
        return chunks
    
    def index_documents(self, chunks: List[Document]):
        """Indexe les chunks dans ChromaDB."""
        logger.info("🔄 Indexation dans ChromaDB...")
        
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.config.chroma_dir,
            collection_name=self.config.collection_name,
        )
        
        self._all_docs = chunks
        logger.info(f"✅ {len(chunks)} chunks indexés")
    
    def build_index(self, directory: str = None):
        """Pipeline d'indexation complet."""
        docs = self.load_documents(directory)
        chunks = self.chunk_documents(docs)
        self.index_documents(chunks)
        self._setup_retriever()
        return len(chunks)
    
    # ========================================================
    # Phase 2 : Retrieval
    # ========================================================
    
    def _setup_retriever(self):
        """Configure le retriever hybride avec re-ranking."""
        # Dense retriever
        dense_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": self.config.dense_k}
        )
        
        # BM25 retriever
        bm25_retriever = BM25Retriever.from_documents(self._all_docs)
        bm25_retriever.k = self.config.bm25_k
        
        # Ensemble (hybride)
        ensemble = EnsembleRetriever(
            retrievers=[dense_retriever, bm25_retriever],
            weights=[self.config.dense_weight, 1 - self.config.dense_weight],
        )
        
        # Re-ranker
        reranker = CrossEncoderReranker(
            model=HuggingFaceCrossEncoder(model_name=self.config.reranker_model),
            top_n=self.config.rerank_top_n,
        )
        
        self.retriever = ContextualCompressionRetriever(
            base_compressor=reranker,
            base_retriever=ensemble,
        )
        
        logger.info("🔍 Retriever hybride + re-ranking configuré")
    
    def load_existing_index(self):
        """Charge un index existant."""
        self.vectorstore = Chroma(
            persist_directory=self.config.chroma_dir,
            embedding_function=self.embeddings,
            collection_name=self.config.collection_name,
        )
        
        # Récupérer les docs pour BM25
        data = self.vectorstore.get(include=["documents", "metadatas"])
        self._all_docs = [
            Document(page_content=doc, metadata=meta)
            for doc, meta in zip(data["documents"], data["metadatas"])
        ]
        
        self._setup_retriever()
        logger.info(f"✅ Index chargé: {len(self._all_docs)} chunks")
    
    # ========================================================
    # Phase 3 : Generation
    # ========================================================
    
    def _format_context(self, docs: List[Document]) -> str:
        """Formate les documents récupérés en contexte."""
        formatted = []
        for i, doc in enumerate(docs):
            source = doc.metadata.get("source", "inconnu")
            source_name = Path(source).name if source != "inconnu" else source
            formatted.append(
                f"[Source {i+1}: {source_name}]\n{doc.page_content}"
            )
        return "\n\n---\n\n".join(formatted)
    
    def query(self, question: str) -> Dict[str, Any]:
        """Exécute une requête RAG complète."""
        if not self.retriever:
            raise RuntimeError("Index non chargé. Appelez build_index() ou load_existing_index().")
        
        # Récupérer les documents pertinents
        docs = self.retriever.invoke(question)
        context = self._format_context(docs)
        
        # Prompt de génération
        prompt = ChatPromptTemplate.from_template(
            """Tu es un assistant technique expert. Réponds à la question 
en te basant UNIQUEMENT sur le contexte fourni.

Règles :
1. Base ta réponse exclusivement sur le contexte
2. Cite tes sources avec [Source X]
3. Si l'info n'est pas dans le contexte, dis-le
4. Sois précis et structuré

Contexte :
{context}

Question : {question}

Réponse :"""
        )
        
        # Générer la réponse
        chain = prompt | self.llm | StrOutputParser()
        answer = chain.invoke({"context": context, "question": question})
        
        return {
            "question": question,
            "answer": answer,
            "sources": [
                {
                    "content": doc.page_content[:200],
                    "source": doc.metadata.get("source", "inconnu"),
                    "chunk_id": doc.metadata.get("chunk_id"),
                }
                for doc in docs
            ],
            "num_sources": len(docs),
        }
    
    def query_stream(self, question: str):
        """Exécute une requête RAG avec streaming."""
        docs = self.retriever.invoke(question)
        context = self._format_context(docs)
        
        prompt = ChatPromptTemplate.from_template(
            """Contexte : {context}\n\nQuestion : {question}\n\nRéponse :"""
        )
        
        chain = prompt | self.llm | StrOutputParser()
        
        for chunk in chain.stream({"context": context, "question": question}):
            yield chunk


# ============================================================
# Utilisation
# ============================================================
if __name__ == "__main__":
    # Configuration
    config = RAGConfig(
        documents_dir="./mes_documents",
        chunk_size=800,
        chunk_overlap=150,
        llm_model="gpt-4o",
        llm_temperature=0.1,
    )
    
    # Initialiser le pipeline
    rag = RAGPipeline(config)
    
    # Indexer les documents (première fois)
    rag.build_index()
    
    # Ou charger un index existant
    # rag.load_existing_index()
    
    # Poser une question
    result = rag.query("Comment configurer le monitoring Prometheus ?")
    
    print(f"\n📝 Question: {result['question']}")
    print(f"\n💬 Réponse:\n{result['answer']}")
    print(f"\n📚 Sources ({result['num_sources']}):")
    for s in result['sources']:
        print(f"  - {s['source']}: {s['content'][:80]}...")

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset


# Préparer le dataset d'évaluation
eval_data = {
    "question": [
        "Comment configurer un healthcheck Docker ?",
        "Quelle est la différence entre CMD et ENTRYPOINT ?",
        "Comment optimiser une image Docker ?",
    ],
    "answer": [
        rag.query("Comment configurer un healthcheck Docker ?")["answer"],
        rag.query("Quelle est la différence entre CMD et ENTRYPOINT ?")["answer"],
        rag.query("Comment optimiser une image Docker ?")["answer"],
    ],
    "contexts": [
        [doc.page_content for doc in retriever.invoke("Comment configurer un healthcheck Docker ?")],
        [doc.page_content for doc in retriever.invoke("CMD vs ENTRYPOINT Docker")],
        [doc.page_content for doc in retriever.invoke("optimiser image Docker")],
    ],
    "ground_truth": [
        "Le healthcheck Docker se configure avec l'instruction HEALTHCHECK dans le Dockerfile...",
        "CMD définit la commande par défaut, ENTRYPOINT définit l'exécutable...",
        "Utiliser des images de base légères, multi-stage builds, minimiser les layers...",
    ],
}

dataset = Dataset.from_dict(eval_data)

# Évaluer
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 
#  'context_precision': 0.85, 'context_recall': 0.78}