Créer un chatbot RAG de A à Z [TUTORIEL]

┌─────────────────────────────────────────────────────────────┐
│                    PIPELINE RAG COMPLET                      │
└─────────────────────────────────────────────────────────────┘

1. PRÉPARATION (offline)
   Documents (PDF, MD, Web)
         ↓
   Loading (LangChain loaders)
         ↓
   Chunking (RecursiveCharacterTextSplitter)
         ↓
   Embeddings (OpenAI text-embedding-3-small)
         ↓
   Vector DB (Chroma)

2. CONVERSATION (runtime)
   User Query → Embedding → Similarity Search → Top-K docs
                                                       ↓
                                            Contexte + History
                                                       ↓
                                     GPT-4o (avec prompt + mémoire)
                                                       ↓
                                                  Réponse

mkdir chatbot-rag
cd chatbot-rag
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install langchain==0.1.0 \
            langchain-openai==0.0.5 \
            langchain-community==0.0.13 \
            chromadb==0.4.22 \
            pypdf==4.0.0 \
            gradio==4.16.0 \
            python-dotenv==1.0.0 \
            beautifulsoup4==4.12.0 \
            lxml==5.1.0

# .env
OPENAI_API_KEY=sk-proj-...
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...  # Optionnel : monitoring LangSmith

chatbot-rag/
├── .env
├── requirements.txt
├── data/               # Documents source
│   ├── pdfs/
│   ├── markdown/
│   └── urls.txt
├── chroma_db/          # Base vectorielle (généré)
├── src/
│   ├── 1_ingest.py     # Ingestion documents
│   ├── 2_chatbot.py    # Logique RAG
│   └── 3_interface.py  # Interface Gradio
└── README.md

"""
Ingestion de documents dans Chroma
Supporte : PDF, Markdown, URLs
"""

import os
from dotenv import load_dotenv
from langchain.document_loaders import (
    PyPDFLoader,
    DirectoryLoader,
    TextLoader,
    WebBaseLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# Charger variables d'environnement
load_dotenv()

# Configuration
CHROMA_PATH = "./chroma_db"
DATA_PATH = "./data"

def load_documents():
    """Charge tous les documents depuis data/"""
    documents = []

    # 1. Charger PDFs
    print("📄 Chargement des PDFs...")
    pdf_loader = DirectoryLoader(
        f"{DATA_PATH}/pdfs",
        glob="**/*.pdf",
        loader_cls=PyPDFLoader,
        show_progress=True
    )
    documents.extend(pdf_loader.load())

    # 2. Charger Markdown
    print("📝 Chargement des fichiers Markdown...")
    md_loader = DirectoryLoader(
        f"{DATA_PATH}/markdown",
        glob="**/*.md",
        loader_cls=TextLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    documents.extend(md_loader.load())

    # 3. Charger URLs (depuis urls.txt)
    urls_file = f"{DATA_PATH}/urls.txt"
    if os.path.exists(urls_file):
        print("🌐 Chargement des URLs...")
        with open(urls_file, "r") as f:
            urls = [line.strip() for line in f if line.strip()]

        if urls:
            web_loader = WebBaseLoader(urls)
            web_loader.requests_per_second = 2  # Rate limiting
            documents.extend(web_loader.load())

    print(f"✅ Total: {len(documents)} documents chargés")
    return documents

def split_documents(documents):
    """Découpe les documents en chunks optimaux"""
    print("✂️ Découpage en chunks...")

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,       # Taille cible par chunk
        chunk_overlap=200,     # Overlap 20% pour continuité
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""]
    )

    chunks = text_splitter.split_documents(documents)
    print(f"✅ {len(chunks)} chunks créés")

    return chunks

def create_vector_store(chunks):
    """Crée la base vectorielle Chroma"""
    print("🧠 Création des embeddings et indexation...")

    # Initialiser embeddings OpenAI
    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",  # 1536 dimensions
        chunk_size=1000  # Batch size pour API
    )

    # Créer/mettre à jour Chroma
    # Si chroma_db existe déjà, il sera mis à jour
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CHROMA_PATH,
        collection_name="knowledge_base"
    )

    print(f"✅ Base vectorielle créée : {CHROMA_PATH}")
    print(f"📊 Nombre de vecteurs : {vectorstore._collection.count()}")

    return vectorstore

def main():
    """Pipeline complet d'ingestion"""
    print("=" * 50)
    print("🚀 INGESTION DE DOCUMENTS")
    print("=" * 50)

    # 1. Charger documents
    documents = load_documents()

    if not documents:
        print("❌ Aucun document trouvé dans ./data/")
        print("   Ajoutez des PDFs dans ./data/pdfs/")
        print("   Ajoutez des .md dans ./data/markdown/")
        print("   Ajoutez des URLs dans ./data/urls.txt")
        return

    # 2. Découper en chunks
    chunks = split_documents(documents)

    # 3. Créer base vectorielle
    vectorstore = create_vector_store(chunks)

    # 4. Test de recherche
    print("\n🔍 Test de recherche...")
    results = vectorstore.similarity_search("RAG", k=3)
    print(f"Top résultat : {results[0].page_content[:200]}...")

    print("\n✅ Ingestion terminée avec succès!")

if __name__ == "__main__":
    main()

# Introduction au RAG

Le RAG (Retrieval-Augmented Generation) est une technique qui combine :
1. Recherche sémantique dans une base de connaissances
2. Génération de texte par un LLM

Cette approche permet aux LLMs d'accéder à des informations spécifiques
sans fine-tuning, réduisant les hallucinations et améliorant la précision.

## Avantages {#avantages}
- Réponses basées sur vos données
- Pas de fine-tuning nécessaire
- Sources citées
- Mise à jour facile des connaissances

https://naileru.com/ia/rag/
https://naileru.com/ia/embedding/

python src/1_ingest.py

==================================================
🚀 INGESTION DE DOCUMENTS
==================================================
📄 Chargement des PDFs...
📝 Chargement des fichiers Markdown...
🌐 Chargement des URLs...
✅ Total: 15 documents chargés
✂️ Découpage en chunks...
✅ 247 chunks créés
🧠 Création des embeddings et indexation...
✅ Base vectorielle créée : ./chroma_db
📊 Nombre de vecteurs : 247
🔍 Test de recherche...
Top résultat : Le RAG (Retrieval-Augmented Generation) est une technique...
✅ Ingestion terminée avec succès!

"""
Logique principale du chatbot RAG avec mémoire conversationnelle
"""

from dotenv import load_dotenv
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate

load_dotenv()

CHROMA_PATH = "./chroma_db"

class RAGChatbot:
    """Chatbot RAG avec mémoire et sources"""

    def __init__(self):
        self.vectorstore = None
        self.chain = None
        self.chat_history = []
        self._setup()

    def _setup(self):
        """Initialise tous les composants"""
        print("🔧 Initialisation du chatbot...")

        # 1. Charger base vectorielle
        embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Chroma(
            persist_directory=CHROMA_PATH,
            embedding_function=embeddings,
            collection_name="knowledge_base"
        )
        print(f"✅ Base vectorielle chargée ({self.vectorstore._collection.count()} vecteurs)")

        # 2. Créer retriever avec MMR
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",  # Maximum Marginal Relevance
            search_kwargs={
                "k": 5,           # Top 5 résultats finaux
                "fetch_k": 20     # Récupère 20, garde les 5 plus divers
            }
        )

        # 3. LLM
        self.llm = ChatOpenAI(
            model="gpt-4o",
            temperature=0.3,  # Un peu de créativité
            max_tokens=1000
        )

        # 4. Mémoire conversationnelle
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )

        # 5. Prompt personnalisé
        prompt_template = """Tu es un assistant expert qui répond aux questions en te basant sur le contexte fourni.

Contexte pertinent :
{context}

Historique de conversation :
{chat_history}

Question actuelle : {question}

Instructions :
1. Réponds en te basant UNIQUEMENT sur le contexte fourni
2. Si l'information n'est pas dans le contexte, dis "Je n'ai pas cette information dans ma base de connaissances"
3. Cite tes sources en mentionnant les documents pertinents
4. Sois concis mais complet
5. Utilise un ton professionnel et amical
6. Si la question fait référence à la conversation précédente, utilise l'historique

Réponse :"""

        PROMPT = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "chat_history", "question"]
        )

        # 6. Créer la chaîne RAG
        self.chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.retriever,
            memory=self.memory,
            return_source_documents=True,
            combine_docs_chain_kwargs={"prompt": PROMPT},
            verbose=False
        )

        print("✅ Chatbot prêt!\n")

    def ask(self, question: str) -> dict:
        """
        Pose une question au chatbot

        Returns:
            dict avec 'answer', 'sources', 'chat_history'
        """
        # Appeler la chaîne
        result = self.chain({"question": question})

        # Extraire sources
        sources = []
        for doc in result.get("source_documents", []):
            source_info = {
                "content": doc.page_content[:200] + "...",
                "metadata": doc.metadata
            }
            sources.append(source_info)

        return {
            "answer": result["answer"],
            "sources": sources,
            "chat_history": self.chat_history
        }

    def reset_memory(self):
        """Efface l'historique de conversation"""
        self.memory.clear()
        self.chat_history = []
        print("🔄 Mémoire effacée")

# Test du chatbot
if __name__ == "__main__":
    chatbot = RAGChatbot()

    # Conversation de test
    questions = [
        "Qu'est-ce que le RAG ?",
        "Quels sont ses avantages ?",
        "Comment l'implémenter ?"
    ]

    for q in questions:
        print(f"\n{'='*60}")
        print(f"❓ {q}")
        print('='*60)

        response = chatbot.ask(q)

        print(f"\n💬 {response['answer']}\n")

        if response['sources']:
            print("📚 Sources :")
            for i, source in enumerate(response['sources'][:3], 1):
                print(f"   {i}. {source['content']}")

python src/2_chatbot.py

🔧 Initialisation du chatbot...
✅ Base vectorielle chargée (247 vecteurs)
✅ Chatbot prêt!

============================================================
❓ Qu'est-ce que le RAG ?
============================================================

💬 Le RAG (Retrieval-Augmented Generation) est une technique qui combine
la recherche sémantique dans une base de connaissances avec la génération
de texte par un LLM. Cette approche permet aux modèles d'accéder à des
informations spécifiques sans nécessiter de fine-tuning.

📚 Sources :
   1. Le RAG (Retrieval-Augmented Generation) est une technique...
   2. Cette approche permet aux LLMs d'accéder à des informations...

"""
Interface Gradio pour le chatbot RAG
"""

import gradio as gr
from chatbot import RAGChatbot

# Initialiser chatbot global
chatbot = RAGChatbot()

def respond(message, history):
    """
    Fonction appelée par Gradio pour chaque message

    Args:
        message: Question de l'utilisateur
        history: Historique Gradio (liste de tuples)

    Returns:
        str: Réponse du chatbot
    """
    # Obtenir réponse
    response = chatbot.ask(message)

    # Formatter réponse avec sources
    answer = response["answer"]

    if response["sources"]:
        answer += "\n\n---\n**📚 Sources :**\n"
        for i, source in enumerate(response["sources"][:3], 1):
            # Extraire nom de fichier si disponible
            source_name = source["metadata"].get("source", "Document")
            if "/" in source_name:
                source_name = source_name.split("/")[-1]

            answer += f"\n{i}. *{source_name}*"
            if "page" in source["metadata"]:
                answer += f" (page {source['metadata']['page']})"

    return answer

def reset():
    """Reset la conversation"""
    chatbot.reset_memory()
    return None

# Créer interface Gradio
with gr.Blocks(theme=gr.themes.Soft(), title="Chatbot RAG") as demo:
    gr.Markdown(
        """
        # 🤖 Chatbot RAG Intelligent

        Posez vos questions sur la base de connaissances.
        Le chatbot recherche les informations pertinentes et cite ses sources.
        """
    )

    with gr.Row():
        with gr.Column(scale=4):
            chatbot_ui = gr.Chatbot(
                label="Conversation",
                height=500,
                show_copy_button=True
            )

            with gr.Row():
                msg = gr.Textbox(
                    label="Votre question",
                    placeholder="Tapez votre question ici...",
                    lines=2,
                    scale=4
                )
                submit_btn = gr.Button("Envoyer", variant="primary", scale=1)

            with gr.Row():
                clear_btn = gr.Button("🔄 Nouvelle conversation", size="sm")

        with gr.Column(scale=1):
            gr.Markdown(
                """
                ### 💡 Exemples de questions

                - Qu'est-ce que le RAG ?
                - Comment fonctionne l'embedding ?
                - Quels sont les avantages du RAG ?
                - Compare RAG et fine-tuning

                ### ℹ️ Informations

                - **Modèle** : GPT-4o
                - **Embeddings** : text-embedding-3-small
                - **Base vectorielle** : Chroma
                - **Mémoire** : Conversation complète
                """
            )

    # Events
    msg.submit(respond, [msg, chatbot_ui], [chatbot_ui])
    submit_btn.click(respond, [msg, chatbot_ui], [chatbot_ui])
    clear_btn.click(reset, None, chatbot_ui, queue=False)

# Lancer l'application
if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",  # Accessible depuis réseau local
        server_port=7860,
        share=False,  # True pour créer lien public temporaire
        show_error=True
    )

python src/3_interface.py

pip install cohere

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# Dans _setup(), après création du retriever de base
base_retriever = self.vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 20}  # Récupérer plus de candidats
)

# Ajouter reranker
compressor = CohereRerank(
    cohere_api_key=os.getenv("COHERE_API_KEY"),
    model="rerank-multilingual-v2.0",  # Support français
    top_n=5  # Garder top 5 après reranking
)

self.retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

def smart_split_documents(documents):
    """Chunking adapté au type de document"""
    all_chunks = []

    for doc in documents:
        # Markdown : découper par headers
        if doc.metadata.get("source", "").endswith(".md"):
            headers_to_split_on = [
                ("#", "Header 1"),
                ("##", "Header 2"),
                ("###", "Header 3"),
            ]

            md_splitter = MarkdownHeaderTextSplitter(
                headers_to_split_on=headers_to_split_on
            )
            md_chunks = md_splitter.split_text(doc.page_content)

            # Découper davantage si chunks trop gros
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000,
                chunk_overlap=200
            )
            chunks = text_splitter.split_documents(md_chunks)

        # PDF : chunking standard
        else:
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000,
                chunk_overlap=200
            )
            chunks = text_splitter.split_documents([doc])

        all_chunks.extend(chunks)

    return all_chunks

# Dans chatbot.ask(), avant d'appeler la chaîne
# Détecter si question contient filtre temporel
import re
from datetime import datetime, timedelta

def extract_time_filter(question):
    """Extrait filtre temporel de la question"""
    # Exemples: "cette semaine", "ce mois", "2024"

    if "cette semaine" in question.lower():
        week_ago = datetime.now() - timedelta(days=7)
        return {"date": {"$gte": week_ago.isoformat()}}

    if "ce mois" in question.lower():
        month_ago = datetime.now() - timedelta(days=30)
        return {"date": {"$gte": month_ago.isoformat()}}

    # Extraction année
    year_match = re.search(r'\b(20\d{2})\b', question)
    if year_match:
        year = year_match.group(1)
        return {"year": year}

    return None

# Utilisation
time_filter = extract_time_filter(question)
if time_filter:
    # Récupérer avec filtre
    docs = self.vectorstore.similarity_search(
        question,
        k=5,
        filter=time_filter
    )

pip install diskcache

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore

# Dans _setup()
store = LocalFileStore("./cache/embeddings")

cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    document_embedding_cache=store,
    namespace="openai-embeddings"
)

# Utiliser cached_embeddings au lieu de embeddings

[
  {
    "question": "Qu'est-ce que le RAG ?",
    "expected_keywords": ["retrieval", "augmented", "generation", "recherche", "llm"],
    "should_have_sources": true
  },
  {
    "question": "Quels sont les avantages du RAG ?",
    "expected_keywords": ["hallucinations", "connaissances", "précision", "sources"],
    "should_have_sources": true
  },
  {
    "question": "Qui a inventé la relativité ?",
    "expected_keywords": ["pas cette information", "base de connaissances"],
    "should_have_sources": false
  }
]

"""Évaluation du chatbot RAG"""

import json
from src.chatbot import RAGChatbot

def evaluate():
    """Évalue le chatbot sur dataset de test"""

    # Charger dataset
    with open("tests/eval_dataset.json", "r") as f:
        dataset = json.load(f)

    chatbot = RAGChatbot()
    results = []

    print("🧪 ÉVALUATION DU CHATBOT")
    print("=" * 60)

    for i, test_case in enumerate(dataset, 1):
        question = test_case["question"]
        expected_keywords = test_case["expected_keywords"]
        should_have_sources = test_case["should_have_sources"]

        print(f"\n{i}. {question}")

        # Obtenir réponse
        response = chatbot.ask(question)
        answer = response["answer"].lower()
        has_sources = len(response["sources"]) > 0

        # Vérifier keywords
        keywords_found = sum(1 for kw in expected_keywords if kw.lower() in answer)
        keyword_score = keywords_found / len(expected_keywords)

        # Vérifier sources
        source_check = has_sources == should_have_sources

        # Score global
        score = (keyword_score + (1 if source_check else 0)) / 2

        result = {
            "question": question,
            "answer": response["answer"],
            "keyword_score": keyword_score,
            "source_check": source_check,
            "score": score
        }
        results.append(result)

        print(f"   Keywords: {keywords_found}/{len(expected_keywords)}")
        print(f"   Sources: {'✅' if source_check else '❌'}")
        print(f"   Score: {score:.1%}")

    # Résumé
    avg_score = sum(r["score"] for r in results) / len(results)
    print(f"\n{'='*60}")
    print(f"📊 RÉSULTATS")
    print(f"   Score moyen: {avg_score:.1%}")
    print(f"   Tests passés: {sum(1 for r in results if r['score'] >= 0.7)}/{len(results)}")

    # Sauvegarder résultats
    with open("tests/results.json", "w") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)

    print(f"\n✅ Résultats sauvegardés dans tests/results.json")

if __name__ == "__main__":
    evaluate()

python tests/evaluate.py

langchain==0.1.0
langchain-openai==0.0.5
chromadb==0.4.22
gradio==4.16.0
pypdf==4.0.0
python-dotenv==1.0.0

from src.interface import demo

if __name__ == "__main__":
    demo.launch()

FROM python:3.10-slim

WORKDIR /app

# Installer dépendances
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier application
COPY . .

# Exposer port Gradio
EXPOSE 7860

# Lancer interface
CMD ["python", "src/3_interface.py"]

docker build -t chatbot-rag .
docker run -p 7860:7860 \
  -e OPENAI_API_KEY=sk-... \
  -v $(pwd)/chroma_db:/app/chroma_db \
  chatbot-rag

pip install modal

import modal

stub = modal.Stub("chatbot-rag")

@stub.function(
    image=modal.Image.debian_slim().pip_install(
        "langchain", "langchain-openai", "chromadb", "gradio"
    ),
    secrets=[modal.Secret.from_name("openai-secret")],
    mounts=[modal.Mount.from_local_dir("./chroma_db", remote_path="/chroma_db")]
)
@modal.web_endpoint(method="POST")
def ask(question: str):
    from src.chatbot import RAGChatbot
    chatbot = RAGChatbot()
    response = chatbot.ask(question)
    return response

@stub.function()
@modal.asgi_app()
def app():
    from src.interface import demo
    return demo.app

modal deploy deploy_modal.py

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...
LANGCHAIN_PROJECT=chatbot-rag-prod

import logging
import json
from datetime import datetime

# Configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('chatbot.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# Dans chatbot.ask()
def ask(self, question: str) -> dict:
    start_time = datetime.now()

    logger.info(json.dumps({
        "event": "question_received",
        "question": question,
        "timestamp": start_time.isoformat()
    }))

    try:
        result = self.chain({"question": question})

        duration = (datetime.now() - start_time).total_seconds()

        logger.info(json.dumps({
            "event": "response_generated",
            "duration_sec": duration,
            "num_sources": len(result.get("source_documents", [])),
            "timestamp": datetime.now().isoformat()
        }))

        return result

    except Exception as e:
        logger.error(json.dumps({
            "event": "error",
            "error": str(e),
            "question": question,
            "timestamp": datetime.now().isoformat()
        }))
        raise

import os
from openai import OpenAI

client = OpenAI()

def check_usage():
    """Vérifie utilisation quotidienne OpenAI"""
    # Note: API usage nécessite org admin access

    # Estimer depuis logs LangSmith
    # Ou implémenter compteur local

    DAILY_BUDGET = 10.00  # $10/jour
    current_spend = get_daily_spend()  # À implémenter

    if current_spend > DAILY_BUDGET * 0.8:
        send_alert(f"⚠️ 80% du budget quotidien atteint : ${current_spend:.2f}")

    if current_spend > DAILY_BUDGET:
        send_alert(f"🚨 BUDGET DÉPASSÉ : ${current_spend:.2f}")
        # Optionnel : désactiver temporairement

def send_alert(message):
    """Envoie alerte (email, Slack, etc.)"""
    # Implémenter selon votre stack
    print(message)