Implémentation RAG

Maîtrisez la Retrieval-Augmented Generation (RAG) pour construire des applications LLM qui fournissent des réponses précises et fondées en utilisant des sources de connaissances externes.

Quand utiliser cette compétence

Construire des systèmes Q&R sur des documents propriétaires
Créer des chatbots avec des informations actuelles et factuelles
Implémenter une recherche sémantique avec des requêtes en langage naturel
Réduire les hallucinations avec des réponses fondées
Permettre aux LLM d'accéder à des connaissances spécifiques à un domaine
Construire des assistants de documentation
Créer des outils de recherche avec citation des sources

Composants essentiels

1. Bases de données vectorielles

Objectif : Stocker et récupérer efficacement les embeddings de documents

Options :

Pinecone : Gérée, scalable, serverless
Weaviate : Open-source, recherche hybride, GraphQL
Milvus : Haute performance, on-premise
Chroma : Légère, facile à utiliser, développement local
Qdrant : Rapide, recherche filtrée, basée sur Rust
pgvector : Extension PostgreSQL, intégration SQL

2. Embeddings

Objectif : Convertir du texte en vecteurs numériques pour la recherche par similarité

Modèles (2026) : | Modèle | Dimensions | Meilleur pour | |--------|-----------|---------------| | voyage-3-large | 1024 | Applications Claude (recommandé par Anthropic) | | voyage-code-3 | 1024 | Recherche de code | | text-embedding-3-large | 3072 | Applications OpenAI, haute précision | | text-embedding-3-small | 1536 | Applications OpenAI, économique | | bge-large-en-v1.5 | 1024 | Open source, déploiement local | | multilingual-e5-large | 1024 | Support multilingue |

3. Stratégies de récupération

Approches :

Dense Retrieval : Similarité sémantique via embeddings
Sparse Retrieval : Correspondance par mots-clés (BM25, TF-IDF)
Hybrid Search : Combiner dense + sparse avec fusion pondérée
Multi-Query : Générer plusieurs variantes de requête
HyDE : Générer des documents hypothétiques pour une meilleure récupération

4. Reranking

Objectif : Améliorer la qualité de la récupération en réordonnant les résultats

Méthodes :

Cross-Encoders : Reranking basé sur BERT (ms-marco-MiniLM)
Cohere Rerank : Reranking basé sur API
Maximal Marginal Relevance (MMR) : Diversité + pertinence
Basé sur LLM : Utiliser un LLM pour noter la pertinence

Démarrage rapide avec LangGraph

from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing import TypedDict, Annotated

class RAGState(TypedDict):
    question: str
    context: list[Document]
    answer: str

# Initialiser les composants
llm = ChatAnthropic(model="claude-sonnet-4-6")
embeddings = VoyageAIEmbeddings(model="voyage-3-large")
vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Prompt RAG
rag_prompt = ChatPromptTemplate.from_template(
    """Répondez en fonction du contexte ci-dessous. Si vous ne pouvez pas répondre, dites-le.

    Contexte :
    {context}

    Question : {question}

    Réponse :"""
)

async def retrieve(state: RAGState) -> RAGState:
    """Récupérer les documents pertinents."""
    docs = await retriever.ainvoke(state["question"])
    return {"context": docs}

async def generate(state: RAGState) -> RAGState:
    """Générer une réponse à partir du contexte."""
    context_text = "\n\n".join(doc.page_content for doc in state["context"])
    messages = rag_prompt.format_messages(
        context=context_text,
        question=state["question"]
    )
    response = await llm.ainvoke(messages)
    return {"answer": response.content}

# Construire le graphique RAG
builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

rag_chain = builder.compile()

# Utilisation
result = await rag_chain.ainvoke({"question": "What are the main features?"})
print(result["answer"])

Modèles RAG avancés

Modèle 1 : Recherche hybride avec RRF

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# Récupérateur sparse (BM25 pour la correspondance par mots-clés)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10

# Récupérateur dense (embeddings pour la recherche sémantique)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# Combiner avec les poids Reciprocal Rank Fusion
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, dense_retriever],
    weights=[0.3, 0.7]  # 30% mots-clés, 70% sémantique
)

Modèle 2 : Récupération Multi-Query

from langchain.retrievers.multi_query import MultiQueryRetriever

# Générer plusieurs perspectives de requête pour une meilleure couverture
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm
)

# Une requête → plusieurs variantes → résultats combinés
results = await multi_query_retriever.ainvoke("What is the main topic?")

Modèle 3 : Compression contextuelle

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# Le compresseur extrait uniquement les portions pertinentes
compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# Retourne uniquement les parties pertinentes des documents
compressed_docs = await compression_retriever.ainvoke("specific query")

Modèle 4 : Récupérateur de document parent

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Petits chunks pour une récupération précise, gros chunks pour le contexte
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# Stockage pour les documents parents
docstore = InMemoryStore()

parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

# Ajouter les documents (divise les enfants, stocke les parents)
await parent_retriever.aadd_documents(documents)

# La récupération retourne les documents parents avec le contexte complet
results = await parent_retriever.ainvoke("query")

Modèle 5 : HyDE (Hypothetical Document Embeddings)

from langchain_core.prompts import ChatPromptTemplate

class HyDEState(TypedDict):
    question: str
    hypothetical_doc: str
    context: list[Document]
    answer: str

hyde_prompt = ChatPromptTemplate.from_template(
    """Écrivez un passage détaillé qui répondrait à cette question :

    Question : {question}

    Passage :"""
)

async def generate_hypothetical(state: HyDEState) -> HyDEState:
    """Générer un document hypothétique pour une meilleure récupération."""
    messages = hyde_prompt.format_messages(question=state["question"])
    response = await llm.ainvoke(messages)
    return {"hypothetical_doc": response.content}

async def retrieve_with_hyde(state: HyDEState) -> HyDEState:
    """Récupérer en utilisant le document hypothétique."""
    # Utiliser le doc hypothétique pour la récupération au lieu de la requête originale
    docs = await retriever.ainvoke(state["hypothetical_doc"])
    return {"context": docs}

# Construire le graphique RAG HyDE
builder = StateGraph(HyDEState)
builder.add_node("hypothetical", generate_hypothetical)
builder.add_node("retrieve", retrieve_with_hyde)
builder.add_node("generate", generate)
builder.add_edge(START, "hypothetical")
builder.add_edge("hypothetical", "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

hyde_rag = builder.compile()

Stratégies de segmentation de documents

Recursive Character Text Splitter

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""]  # Essayer dans l'ordre
)

chunks = splitter.split_documents(documents)

Segmentation basée sur les tokens

from langchain_text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    encoding_name="cl100k_base"  # Encodage OpenAI tiktoken
)

Segmentation sémantique

from langchain_experimental.text_splitter import SemanticChunker

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

Markdown Header Splitter

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False
)

Configurations de Vector Store

Pinecone (Serverless)

from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore

# Initialiser le client Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

# Créer l'index si nécessaire
if "my-index" not in pc.list_indexes().names():
    pc.create_index(
        name="my-index",
        dimension=1024,  # dimensions de voyage-3-large
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

# Créer le vector store
index = pc.Index("my-index")
vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

Weaviate

import weaviate
from langchain_weaviate import WeaviateVectorStore

client = weaviate.connect_to_local()  # ou connect_to_weaviate_cloud()

vectorstore = WeaviateVectorStore(
    client=client,
    index_name="Documents",
    text_key="content",
    embedding=embeddings
)

Chroma (Développement local)

from langchain_chroma import Chroma

vectorstore = Chroma(
    collection_name="my_collection",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

pgvector (PostgreSQL)

from langchain_postgres.vectorstores import PGVector

connection_string = "postgresql+psycopg://user:pass@localhost:5432/vectordb"

vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="documents",
    connection=connection_string,
)

Optimisation de la récupération

1. Filtrage par métadonnées

from langchain_core.documents import Document

# Ajouter les métadonnées lors de l'indexation
docs_with_metadata = []
for doc in documents:
    doc.metadata.update({
        "source": doc.metadata.get("source", "unknown"),
        "category": determine_category(doc.page_content),
        "date": datetime.now().isoformat()
    })
    docs_with_metadata.append(doc)

# Filtrer lors de la récupération
results = await vectorstore.asimilarity_search(
    "query",
    filter={"category": "technical"},
    k=5
)

2. Maximal Marginal Relevance (MMR)

# Équilibrer la pertinence avec la diversité
results = await vectorstore.amax_marginal_relevance_search(
    "query",
    k=5,
    fetch_k=20,  # Récupérer 20, retourner les 5 meilleurs diversifiés
    lambda_mult=0.5  # 0=max diversité, 1=max pertinence
)

3. Reranking avec Cross-Encoder

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

async def retrieve_and_rerank(query: str, k: int = 5) -> list[Document]:
    # Obtenir les résultats initiaux
    candidates = await vectorstore.asimilarity_search(query, k=20)

    # Reranker
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = reranker.predict(pairs)

    # Trier par score et prendre les k premiers
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:k]]

4. Cohere Rerank

from langchain.retrievers import CohereRerank
from langchain_cohere import CohereRerank

reranker = CohereRerank(model="rerank-english-v3.0", top_n=5)

# Enrouler le récupérateur avec reranking
reranked_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

Ingénierie des prompts pour RAG

Prompt contextuel avec citations

rag_prompt = ChatPromptTemplate.from_template(
    """Répondez à la question en fonction du contexte ci-dessous. Incluez les citations en utilisant [1], [2], etc.

    Si vous ne pouvez pas répondre en fonction du contexte, dites « Je n'ai pas assez d'informations ».

    Contexte :
    {context}

    Question : {question}

    Instructions :
    1. Utilisez uniquement les informations du contexte
    2. Citez les sources au format [1], [2]
    3. En cas de doute, exprimez l'incertitude

    Réponse (avec citations) :"""
)

Sortie structurée pour RAG

from pydantic import BaseModel, Field

class RAGResponse(BaseModel):
    answer: str = Field(description="The answer based on context")
    confidence: float = Field(description="Confidence score 0-1")
    sources: list[str] = Field(description="Source document IDs used")
    reasoning: str = Field(description="Brief reasoning for the answer")

# Utiliser avec la sortie structurée
structured_llm = llm.with_structured_output(RAGResponse)

Métriques d'évaluation

from typing import TypedDict

class RAGEvalMetrics(TypedDict):
    retrieval_precision: float  # Docs pertinents / docs récupérés
    retrieval_recall: float     # Docs pertinents récupérés / total pertinents
    answer_relevance: float     # La réponse adresse la question
    faithfulness: float         # La réponse est fondée sur le contexte
    context_relevance: float    # Le contexte est pertinent pour la question

async def evaluate_rag_system(
    rag_chain,
    test_cases: list[dict]
) -> RAGEvalMetrics:
    """Évaluer le système RAG sur des cas de test."""
    metrics = {k: [] for k in RAGEvalMetrics.__annotations__}

    for test in test_cases:
        result = await rag_chain.ainvoke({"question": test["question"]})

        # Métriques de récupération
        retrieved_ids = {doc.metadata["id"] for doc in result["context"]}
        relevant_ids = set(test["relevant_doc_ids"])

        precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
        recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)

        metrics["retrieval_precision"].append(precision)
        metrics["retrieval_recall"].append(recall)

        # Utiliser LLM-as-judge pour les métriques de qualité
        quality = await evaluate_answer_quality(
            question=test["question"],
            answer=result["answer"],
            context=result["context"],
            expected=test.get("expected_answer")
        )
        metrics["answer_relevance"].append(quality["relevance"])
        metrics["faithfulness"].append(quality["faithfulness"])
        metrics["context_relevance"].append(quality["context_relevance"])

    return {k: sum(v) / len(v) for k, v in metrics.items()}

rag-implementation