RAG for Professionals : Maîtriser le RAG avec LangChain, LangGraph et OpenAI

ia
tutorial
python
langchain
rag
Author

Sylvain Pham

Published

January 27, 2026

Formation Udemy sur le RAG (Retrieval Augmented Generation) avec Python. Je documente ici ce que j’ai appris en suivant le cours RAG for Professionals with LangGraph, Python and OpenAI.

Cours Udemy

Section 1 : Qu’est-ce que le RAG ?

Le RAG (Retrieval Augmented Generation) permet d’enrichir les réponses d’un LLM avec des données externes. Au lieu de se limiter à ses connaissances d’entraînement, le modèle peut chercher dans vos documents.

flowchart LR
    A[Question] --> B[Retriever]
    B --> C[Documents pertinents]
    C --> D[LLM + Contexte]
    D --> E[Réponse enrichie]

    F[(Vector DB)] --> B

Sans RAG : “Quelles sont les règles de mon entreprise ?” → “Je ne connais pas vos règles internes”

Avec RAG : Le LLM récupère les documents pertinents et répond avec des informations précises.

Architecture RAG complète

flowchart TD
    subgraph Ingestion
        A[Documents] --> B[Loader]
        B --> C[Splitter]
        C --> D[Embeddings]
        D --> E[(Vector Store)]
    end

    subgraph Query
        F[Question] --> G[Embed Query]
        G --> H[Similarity Search]
        E --> H
        H --> I[Top-K Documents]
    end

    subgraph Generation
        I --> J[Prompt + Context]
        F --> J
        J --> K[LLM]
        K --> L[Réponse]
    end

Section 4 : LangChain et OpenAI Basics

Lectures 14-26 : ChatBot simple avec LangChain

OpenAI Chat Completions API

from openai import OpenAI

client = OpenAI(api_key="sk-...")

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Tu es un assistant utile."},
        {"role": "user", "content": "Explique le RAG en une phrase."}
    ],
    temperature=0.7
)

print(response.choices[0].message.content)
Paramètre Rôle
model gpt-4o-mini, gpt-4o, etc.
messages Historique de conversation
temperature 0 = déterministe, 1 = créatif
max_tokens Limite de réponse

ChatOpenAI avec LangChain

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
response = llm.invoke("Qu'est-ce que le RAG ?")

Dynamic Prompts avec PromptTemplate

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "Traduis ce texte en {language}: {text}"
)

formatted = prompt.format(language="français", text="Hello world")

ChatPromptTemplate avec Roles

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un expert en {domain}."),
    ("human", "{question}")
])

messages = prompt.invoke({
    "domain": "IA",
    "question": "Qu'est-ce que le RAG ?"
})

LangChain Expression Language (LCEL)

Le pipe | permet de chaîner les composants :

chain = prompt | llm | StrOutputParser()

response = chain.invoke({
    "domain": "IA",
    "question": "Qu'est-ce que le RAG ?"
})

Runnables personnalisés

from langchain_core.runnables import RunnableLambda

def uppercase(text: str) -> str:
    return text.upper()

chain = prompt | llm | StrOutputParser() | RunnableLambda(uppercase)

Output Parsers

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

# Parser simple (string)
chain = prompt | llm | StrOutputParser()

# Parser JSON structuré
class BookRecommendation(BaseModel):
    title: str = Field(description="Titre du livre")
    author: str = Field(description="Auteur")
    reason: str = Field(description="Pourquoi ce livre")

parser = JsonOutputParser(pydantic_object=BookRecommendation)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Recommande un livre. {format_instructions}"),
    ("human", "{query}")
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | llm | parser
result = chain.invoke({"query": "livre sur l'IA"})
# result = {"title": "...", "author": "...", "reason": "..."}

Section 5 : Document Summarization

Lectures 27-49 : Loaders, Splitters, Stratégies de résumé

Document Loaders

flowchart LR
    subgraph Sources
        A[PDF]
        B[CSV]
        C[Web]
        D[YouTube]
        E[TXT]
    end

    subgraph Loaders
        A --> F[PyPDFLoader]
        B --> G[CSVLoader]
        C --> H[WebBaseLoader]
        D --> I[YoutubeLoader]
        E --> J[TextLoader]
    end

    F & G & H & I & J --> K[Documents]


from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
documents = loader.load()

# Chaque document a: page_content + metadata
for doc in documents:
    print(f"Page {doc.metadata['page']}: {doc.page_content[:100]}...")

Single Mode vs Page Mode

  • Single Mode : Tout le PDF dans un seul Document
  • Page Mode : Un Document par page (défaut)

Lazy Loading

Pour les gros fichiers, charger page par page :

for doc in loader.lazy_load():
    process(doc)  # Traite une page à la fois

Text Splitters

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # Taille max d'un chunk
    chunk_overlap=200,    # Chevauchement entre chunks
    separators=["\n\n", "\n", " ", ""]  # Priorité de découpe
)

chunks = splitter.split_documents(documents)
Splitter Usage
CharacterTextSplitter Découpe simple par caractère
RecursiveCharacterTextSplitter Respecte la structure (paragraphes, phrases)
TokenTextSplitter Découpe par tokens (pour limites API)

TokenTextSplitter

Pour respecter les limites de tokens de l’API :

from langchain.text_splitter import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=500,       # En tokens
    chunk_overlap=50
)

Stratégies de résumé

flowchart TD
    subgraph Stuff
        A1[Doc 1] & A2[Doc 2] & A3[Doc 3] --> B1[Tout dans le prompt]
        B1 --> C1[Un seul appel LLM]
    end

    subgraph MapReduce
        D1[Doc 1] --> E1[Résumé 1]
        D2[Doc 2] --> E2[Résumé 2]
        D3[Doc 3] --> E3[Résumé 3]
        E1 & E2 & E3 --> F1[Résumé final]
    end

    subgraph Refine
        G1[Doc 1] --> H1[Résumé initial]
        H1 --> I1[+ Doc 2 → Résumé amélioré]
        I1 --> J1[+ Doc 3 → Résumé final]
    end

Stratégie Avantage Inconvénient
Stuff Simple, rapide Limité par context window
MapReduce Parallélisable Peut perdre le contexte global
Refine Résumé progressif Lent (séquentiel)

Section 6 : Introduction au RAG

Lectures 50-71 : Embeddings, Vector Stores, Retrieval Strategies

Vectorization & Embedding

Les embeddings convertissent le texte en vecteurs numériques :

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Embed un texte
vector = embeddings.embed_query("Qu'est-ce que le RAG ?")
# vector = [0.023, -0.456, 0.789, ...]  # 1536 dimensions

Vector Store avec FAISS

from langchain_community.vectorstores import FAISS

# Créer et peupler
vectorstore = FAISS.from_documents(chunks, embeddings)

# Recherche par similarité
results = vectorstore.similarity_search("Comment fonctionne le RAG ?", k=3)

Persistance FAISS vs Chroma

# FAISS - Persistance manuelle
vectorstore.save_local("./faiss_index")
loaded = FAISS.load_local("./faiss_index", embeddings)

# ChromaDB - Persistance intégrée
from langchain_community.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
Aspect ChromaDB FAISS
Persistance Intégrée Manuelle
Métadonnées Oui Limitées
Scalabilité Moyenne Haute
Setup Simple Plus complexe

Stratégies de Retrieval

flowchart TD
    A[Query] --> B{Stratégie?}
    B -->|similarity| C[Top-K plus proches]
    B -->|threshold| D[Score > seuil]
    B -->|mmr| E[Diversité maximale]
    C --> F[Résultats]
    D --> F
    E --> F


# Similarity simple (défaut)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# Avec seuil de score (filtre les résultats peu pertinents)
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 4}
)

# MMR - Maximum Marginal Relevance (diversité)
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5}
)
Stratégie Usage Quand l’utiliser
similarity Top-K plus proches Focused Q&A
similarity_score_threshold Filtrer par qualité Éviter les faux positifs
mmr Diversité des résultats Exploratory Q&A, Multi-Aspect

MultiQueryRetriever

Génère plusieurs variantes de la question pour améliorer le recall :

from langchain.retrievers.multi_query import MultiQueryRetriever

retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)
# Génère 3 variantes de la question et fusionne les résultats

RAG Chain complète

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# Prompt RAG
prompt = ChatPromptTemplate.from_template("""
Réponds à la question en utilisant uniquement le contexte suivant :

Contexte : {context}

Question : {question}

Réponse :
""")

# Chain
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
)

response = rag_chain.invoke("Quelles sont les étapes du RAG ?")

Sécurité des données

flowchart TD
    subgraph "Niveau 1 - OpenAI API"
        A[Données] --> B[API OpenAI]
        B --> C[Serveurs OpenAI]
    end

    subgraph "Niveau 2 - Azure OpenAI"
        D[Données] --> E[Azure OpenAI]
        E --> F[Votre tenant Azure]
    end

    subgraph "Niveau 3 - Local"
        G[Données] --> H[Ollama/vLLM]
        H --> I[Votre serveur]
    end

Niveau Solution Données
Basique OpenAI API Transitent par OpenAI
Entreprise Azure OpenAI Restent dans votre tenant
Maximum Ollama, vLLM 100% local

Section 7 : LangGraph Basics

Lectures 72-91 : StateGraph, Reducers, Memory

TypedDict pour le State

from typing import TypedDict, Annotated
from operator import add

class RAGState(TypedDict):
    question: str
    context: list
    messages: Annotated[list, add]  # Reducer: accumule les messages
    response: str

StateGraph simple

flowchart TD
    A[START] --> B[Retrieve]
    B --> C[Generate]
    C --> D[END]


from langgraph.graph import StateGraph, END

def retrieve_node(state: RAGState) -> RAGState:
    docs = retriever.invoke(state["question"])
    return {"context": docs}

def generate_node(state: RAGState) -> RAGState:
    response = llm.invoke(f"Context: {state['context']}\nQ: {state['question']}")
    return {"response": response.content}

# Build graph
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()

LangGraph Reducers

Les reducers définissent comment combiner les valeurs :

from operator import add
from typing import Annotated

class State(TypedDict):
    count: int                      # Écrasement (défaut)
    messages: Annotated[list, add]  # Accumulation avec add

add_messages Reducer

Pour les conversations :

from langgraph.graph import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

MessagesState (raccourci)

from langgraph.graph import MessagesState

# Équivalent à:
# class State(TypedDict):
#     messages: Annotated[list, add_messages]

Mémoire persistante avec MemorySaver

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Utiliser avec un thread_id
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke({"question": "Qu'est-ce que le RAG?"}, config)

# Le prochain appel avec le même thread_id conserve l'historique
result2 = app.invoke({"question": "Peux-tu préciser?"}, config)

Gestion de la mémoire conversationnelle

Les LLM ont une context window limitée. Stratégies :

from langchain_core.messages import trim_messages

# Truncation par tokens
trimmer = trim_messages(
    max_tokens=1000,
    strategy="last",           # Garder les derniers
    token_counter=len,         # Ou tiktoken pour comptage précis
    include_system=True,       # Toujours garder le system prompt
    allow_partial=False        # Pas de messages tronqués
)

trimmed = trimmer.invoke(messages)

Conditional Edges

flowchart TD
    A[START] --> B[Retrieve]
    B --> C[Generate]
    C --> D[Validate]
    D --> E{OK?}
    E -->|Non| F[Refine]
    F --> D
    E -->|Oui| G[END]


def should_continue(state: RAGState) -> str:
    return "end" if state["is_valid"] else "refine"

workflow.add_conditional_edges(
    "validate",
    should_continue,
    {"end": END, "refine": "refine"}
)

Section 8 : Multiple Documents et File Types

Lectures 92-107 : DirectoryLoader, UnstructuredLoader, types de fichiers

PyPDFDirectoryLoader

from langchain_community.document_loaders import PyPDFDirectoryLoader

loader = PyPDFDirectoryLoader("./docs/")
documents = loader.load()

DirectoryLoader (flexible)

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader

# PDFs seulement
loader = DirectoryLoader(
    "./docs/",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader
)

# Avec show_progress
loader = DirectoryLoader(
    "./docs/",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    show_progress=True
)

UnstructuredLoader (multi-format)

from langchain_community.document_loaders import UnstructuredFileLoader

# Charge PDF, DOCX, PPTX, TXT, etc.
loader = UnstructuredFileLoader("document.docx")
documents = loader.load()

Loaders par type de fichier

Type Loader Package
PDF PyPDFLoader pypdf
TXT TextLoader -
DOCX Docx2txtLoader docx2txt
PPTX UnstructuredPowerPointLoader unstructured
CSV CSVLoader - 
from langchain_community.document_loaders import (
    TextLoader,
    Docx2txtLoader,
    UnstructuredPowerPointLoader,
    CSVLoader
)

# TXT
loader = TextLoader("notes.txt")

# Word
loader = Docx2txtLoader("document.docx")

# PowerPoint
loader = UnstructuredPowerPointLoader("slides.pptx")

# CSV (un Document par ligne)
loader = CSVLoader("data.csv")

Chargement mixte avec DirectoryLoader

from langchain_community.document_loaders import DirectoryLoader

# Mapping type -> loader
loaders = {
    "**/*.pdf": PyPDFLoader,
    "**/*.txt": TextLoader,
    "**/*.docx": Docx2txtLoader,
}

all_docs = []
for glob, loader_cls in loaders.items():
    loader = DirectoryLoader("./docs/", glob=glob, loader_cls=loader_cls)
    all_docs.extend(loader.load())

Section 9 : Dynamic Vector Database avec Chroma

Lectures 108-127 : CRUD, Metadata filtering, Query syntax

Créer une Chroma DB vide

from langchain_community.vectorstores import Chroma

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

# Ajouter des documents
vectorstore.add_documents(documents, ids=["doc1", "doc2", "doc3"])

Charger une DB existante

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

CRUD Operations

# Get by ID
result = vectorstore.get(ids=["doc1"])

# Delete by ID
vectorstore.delete(ids=["doc1"])

# Count
count = vectorstore._collection.count()

Filtrage par métadonnées

# Ajouter des métadonnées
documents = [
    Document(
        page_content="Contenu du chapitre 1",
        metadata={"source": "livre.pdf", "chapter": 1, "author": "Dupont"}
    )
]

# Filtrer lors de la recherche
results = vectorstore.similarity_search(
    "question",
    k=3,
    filter={"chapter": 1}
)

Query Syntax avancée

# Opérateurs de comparaison
results = vectorstore.similarity_search(
    "question",
    filter={"chapter": {"$gte": 2}}  # chapter >= 2
)

# Opérateurs logiques
results = vectorstore.similarity_search(
    "question",
    filter={
        "$and": [
            {"author": {"$eq": "Dupont"}},
            {"chapter": {"$gte": 2}}
        ]
    }
)

# $or
results = vectorstore.similarity_search(
    "question",
    filter={
        "$or": [
            {"source": "livre1.pdf"},
            {"source": "livre2.pdf"}
        ]
    }
)
Opérateur Signification
$eq Égal
$ne Différent
$gt, $gte Plus grand (ou égal)
$lt, $lte Plus petit (ou égal)
$in Dans la liste
$nin Pas dans la liste
$and, $or Combinaison logique

Filtrage par contenu (where_document)

# Recherche dans le contenu du document
results = vectorstore._collection.query(
    query_texts=["question"],
    where_document={"$contains": "RAG"}
)

Timestamps et dates

import time

# Ajouter un timestamp
doc.metadata["created_at"] = int(time.time())

# Filtrer par date
one_week_ago = int(time.time()) - 7*24*60*60
results = vectorstore.similarity_search(
    "question",
    filter={"created_at": {"$gte": one_week_ago}}
)

Section 10 : Automated VectorDB Updates

Lectures 128-135 : Détection de changements, Hashing

Détection de changements avec Hash

flowchart TD
    A[Nouveaux docs] --> B[Calculer hash]
    B --> C{Hash existe?}
    C -->|Non| D[Ajouter au vector store]
    C -->|Oui| E{Hash identique?}
    E -->|Non| F[Mettre à jour]
    E -->|Oui| G[Ignorer]
    D --> H[Sauver manifest]
    F --> H


import hashlib
import json

def compute_hash(content: str) -> str:
    return hashlib.md5(content.encode()).hexdigest()

def load_manifest(path: str) -> dict:
    try:
        with open(path, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

def save_manifest(manifest: dict, path: str):
    with open(path, "w") as f:
        json.dump(manifest, f)

Sync incrémentale

def sync_documents(documents, vectorstore, manifest_path="manifest.json"):
    manifest = load_manifest(manifest_path)

    for doc in documents:
        doc_id = doc.metadata.get("source", str(hash(doc.page_content)))
        new_hash = compute_hash(doc.page_content)

        if doc_id not in manifest:
            # Nouveau document
            vectorstore.add_documents([doc], ids=[doc_id])
            manifest[doc_id] = new_hash
        elif manifest[doc_id] != new_hash:
            # Document modifié
            vectorstore.delete([doc_id])
            vectorstore.add_documents([doc], ids=[doc_id])
            manifest[doc_id] = new_hash
        # Sinon: inchangé, on skip

    save_manifest(manifest, manifest_path)

Détection avec métadonnées

Alternative : stocker le hash dans les métadonnées Chroma :

# Lors de l'ajout
doc.metadata["content_hash"] = compute_hash(doc.page_content)

# Pour vérifier si changé
existing = vectorstore.get(where={"source": doc.metadata["source"]})
if existing and existing["metadatas"][0]["content_hash"] != new_hash:
    # Le document a changé
    pass

Section 11 : Application RAG complète

Lectures 136-143 : TKinter, Desktop App

GUI avec TKinter

import tkinter as tk
from tkinter import scrolledtext

class RAGChatApp:
    def __init__(self, rag_chain):
        self.rag_chain = rag_chain
        self.window = tk.Tk()
        self.window.title("RAG Chat")

        # Zone de chat
        self.chat_area = scrolledtext.ScrolledText(self.window, wrap=tk.WORD)
        self.chat_area.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

        # Zone de saisie
        self.input_field = tk.Entry(self.window)
        self.input_field.pack(padx=10, pady=5, fill=tk.X)
        self.input_field.bind("<Return>", self.send_message)

        # Bouton envoyer
        self.send_button = tk.Button(self.window, text="Envoyer", command=self.send_message)
        self.send_button.pack(pady=5)

    def send_message(self, event=None):
        question = self.input_field.get()
        if question:
            self.chat_area.insert(tk.END, f"Vous: {question}\n")
            response = self.rag_chain.invoke(question)
            self.chat_area.insert(tk.END, f"RAG: {response}\n\n")
            self.input_field.delete(0, tk.END)

    def run(self):
        self.window.mainloop()

# Usage
app = RAGChatApp(rag_chain)
app.run()

Desktop App avec PyInstaller

# Installation
pip install pyinstaller

# Créer l'exécutable
pyinstaller --onefile --windowed rag_app.py

Stack technique

Composant Rôle
OpenAI API LLM (gpt-4o-mini) + Embeddings
LangChain Orchestration, Loaders, Chains
ChromaDB Vector store persistant
LangGraph Workflows avec état
TKinter Interface utilisateur desktop

Comparaison avec le cours AI Agents

Aspect AI Agents Course RAG Course
Focus Tool Calling, Memory Document Retrieval
LLM Ollama (local) OpenAI API
Workflow LangGraph (optionnel) LangGraph (avancé)
Mémoire Chat history Document embeddings

Bilan

Ce cours couvre en profondeur :

Section Concepts clés
Section 4 ChatOpenAI, PromptTemplate, LCEL, Runnables, Output Parsers
Section 5 Loaders, Splitters, Stuff/MapReduce/Refine
Section 6 Embeddings, FAISS/Chroma, Retrieval Strategies, Sécurité
Section 7 TypedDict, StateGraph, Reducers, Memory, Conditional Edges
Section 8 DirectoryLoader, UnstructuredLoader, multi-formats
Section 9 CRUD Chroma, Metadata filtering, Query syntax
Section 10 Change detection, Hashing, Dynamic sync
Section 11 TKinter GUI, Desktop App avec PyInstaller

Le RAG est essentiel pour créer des assistants qui peuvent répondre sur vos données privées sans fine-tuning coûteux.

Liens