RAG - Retrieval Augmented Generation

Question de l'utilisateur
           ↓
┌─────────────────────────┐
│   RETRIEVAL (Recherche) │  ← Chercher dans vos documents
└───────────┬─────────────┘
            ↓
     Documents pertinents
            ↓
┌─────────────────────────┐
│   AUGMENTED (Contexte)  │  ← Enrichir le prompt
└───────────┬─────────────┘
            ↓
┌─────────────────────────┐
│  GENERATION (LLM)       │  ← Générer avec contexte
└───────────┬─────────────┘
            ↓
     Réponse + Sources

┌─────────────────────────────────────────────────────────────┐
│                    PIPELINE RAG                              │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  INDEXATION (offline)                                       │
│  ───────────────────                                        │
│  Documents → Chunking → Embedding → Vector DB               │
│                                                              │
│  INFÉRENCE (online)                                         │
│  ──────────────────                                         │
│  Query → Embedding → Similarity Search → Top-K docs         │
│           ↓                                                  │
│  [Query + Retrieved Docs] → LLM → Réponse                   │
│                                                              │
└─────────────────────────────────────────────────────────────┘

# Chunking simple par caractères
from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=1000,      # Taille max
    chunk_overlap=200,    # Chevauchement
    separator="\n"
)
chunks = splitter.split_text(document)

Trop petit (100 tokens) : Perd le contexte
Trop grand (2000 tokens) : Noie l'information

Recommandation : 500-1000 tokens avec 10-20% overlap

from openai import OpenAI

client = OpenAI()

def get_embedding(text):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# Embedder tous les chunks
embeddings = [get_embedding(chunk) for chunk in chunks]

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(chunks)

import chromadb
from chromadb.utils import embedding_functions

# Créer le client
client = chromadb.Client()

# Fonction d'embedding
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="...",
    model_name="text-embedding-3-small"
)

# Créer la collection
collection = client.create_collection(
    name="my_documents",
    embedding_function=openai_ef
)

# Ajouter les documents
collection.add(
    documents=chunks,
    ids=[f"chunk_{i}" for i in range(len(chunks))],
    metadatas=[{"source": "doc1.pdf", "page": i} for i in range(len(chunks))]
)

# Rechercher
results = collection.query(
    query_texts=["Comment fonctionne le RAG ?"],
    n_results=5
)

similarity = cos(query_embedding, doc_embedding)
           = (A · B) / (||A|| × ||B||)

Valeur entre -1 et 1 (1 = très similaire)

# Résultats sémantiques (embeddings)
semantic_results = vector_db.search(query_embedding, k=10)

# Résultats lexicaux (BM25)
keyword_results = bm25_search(query, k=10)

# Fusion des résultats (RRF - Reciprocal Rank Fusion)
final_results = reciprocal_rank_fusion(semantic_results, keyword_results)

from cohere import Client

co = Client("api-key")

# Reranker les résultats
reranked = co.rerank(
    query="Question de l'utilisateur",
    documents=retrieved_docs,
    top_n=5,
    model="rerank-english-v2.0"
)

def build_rag_prompt(query, retrieved_docs):
    context = "\n\n".join([
        f"[Source {i+1}]: {doc}"
        for i, doc in enumerate(retrieved_docs)
    ])

    prompt = f"""Utilise les sources suivantes pour répondre à la question.
Si l'information n'est pas dans les sources, dis-le clairement.
Cite les sources utilisées.

SOURCES :
{context}

QUESTION : {query}

RÉPONSE :"""

    return prompt

from anthropic import Anthropic

client = Anthropic()

def generate_answer(query, retrieved_docs):
    prompt = build_rag_prompt(query, retrieved_docs)

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text

class RAGPipeline:
    def __init__(self):
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        self.vector_db = chromadb.Client()
        self.collection = None
        self.llm = Anthropic()

    def index_documents(self, documents):
        # Chunking
        chunks = self.chunk_documents(documents)

        # Créer collection
        self.collection = self.vector_db.create_collection("docs")

        # Indexer
        embeddings = self.embedder.encode(chunks)
        self.collection.add(
            embeddings=embeddings.tolist(),
            documents=chunks,
            ids=[f"c{i}" for i in range(len(chunks))]
        )

    def query(self, question, top_k=5):
        # Rechercher
        results = self.collection.query(
            query_embeddings=[self.embedder.encode(question).tolist()],
            n_results=top_k
        )

        retrieved_docs = results['documents'][0]

        # Générer
        prompt = self.build_prompt(question, retrieved_docs)
        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "answer": response.content[0].text,
            "sources": retrieved_docs
        }

# Au lieu de chercher avec la question...
# Demander au LLM de générer une réponse hypothétique
hypothetical = llm.generate(f"Réponds brièvement : {query}")

# Utiliser cette réponse pour la recherche
results = vector_db.search(embed(hypothetical))

queries = llm.generate(f"""
Génère 3 reformulations de cette question :
{original_query}
""")

# Rechercher avec chaque reformulation
all_results = []
for q in queries:
    all_results.extend(vector_db.search(q))

# Dédupliquer et fusionner
final_results = deduplicate(all_results)

Document complet
├── Chunk 1 (indexé)  ─┐
├── Chunk 2 (indexé)   ├→ Si match → retourner document complet
├── Chunk 3 (indexé)  ─┘

1. Question → LLM évalue si recherche nécessaire
2. Si oui → Retrieval → Génération avec contexte
3. Si non → Génération directe
4. LLM vérifie sa réponse et peut relancer une recherche

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# Créer le retriever
vectorstore = Chroma.from_documents(documents, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Créer la chaîne RAG
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(),
    retriever=retriever,
    return_source_documents=True
)

result = qa_chain.invoke({"query": "Ma question"})

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# Charger les documents
documents = SimpleDirectoryReader("./data").load_data()

# Créer l'index
index = VectorStoreIndex.from_documents(documents)

# Créer le query engine
query_engine = index.as_query_engine()

# Requête
response = query_engine.query("Ma question")

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy]
)

RAG = Retrieval + Augmentation + Generation

Pipeline :
1. Chunking : Découper les documents
2. Embedding : Vectoriser les chunks
3. Indexation : Stocker dans vector DB
4. Retrieval : Chercher les docs pertinents
5. Augmentation : Enrichir le prompt
6. Generation : Produire la réponse

Avantages :
├── Connaissances à jour
├── Données privées
├── Moins d'hallucinations
├── Sources vérifiables
└── Pas de fine-tuning nécessaire