TP : Créer un Chatbot RAG de A à Z

mkdir mon-chatbot-rag && cd mon-chatbot-rag
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

pip install openai chromadb streamlit pypdf2 tiktoken

mon-chatbot-rag/
├── documents/          # Vos fichiers source (PDF, txt, md)
├── ingest.py          # Script d''ingestion des documents
├── chat.py            # Logique du chatbot RAG
├── app.py             # Interface Streamlit
└── requirements.txt

import os
from pathlib import Path
import chromadb
from openai import OpenAI
from PyPDF2 import PdfReader

client = OpenAI()  # Utilise OPENAI_API_KEY env var

# --- 1. Lecture des documents ---
def load_documents(folder: str) -> list[dict]:
    """Charge tous les documents d''un dossier."""
    docs = []
    for file_path in Path(folder).rglob("*"):
        if file_path.suffix == ".pdf":
            reader = PdfReader(str(file_path))
            text = "\n".join(page.extract_text() for page in reader.pages)
        elif file_path.suffix in (".txt", ".md"):
            text = file_path.read_text(encoding="utf-8")
        else:
            continue

        docs.append({
            "text": text,
            "source": file_path.name,
        })
        print(f"Chargé : {file_path.name} ({len(text)} caractères)")

    return docs

# --- 2. Chunking ---
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Découpe un texte en morceaux avec chevauchement."""
    words = text.split()
    chunks = []
    start = 0

    while start < len(words):
        end = start + chunk_size
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start = end - overlap  # Chevauchement pour ne pas couper le contexte

    return chunks

# --- 3. Embedding + Stockage ---
def embed_and_store(docs: list[dict]):
    """Transforme les documents en vecteurs et les stocke."""
    # Initialiser ChromaDB
    chroma = chromadb.PersistentClient(path="./chroma_db")
    collection = chroma.get_or_create_collection(
        name="mes_documents",
        metadata={"hnsw:space": "cosine"},
    )

    all_chunks = []
    all_ids = []
    all_metadata = []

    for doc in docs:
        chunks = chunk_text(doc["text"])
        for i, chunk in enumerate(chunks):
            chunk_id = f"{doc[''source'']}_{i}"
            all_chunks.append(chunk)
            all_ids.append(chunk_id)
            all_metadata.append({"source": doc["source"], "chunk_index": i})

    # Embedding par batch de 100
    batch_size = 100
    for i in range(0, len(all_chunks), batch_size):
        batch_texts = all_chunks[i:i + batch_size]
        batch_ids = all_ids[i:i + batch_size]
        batch_meta = all_metadata[i:i + batch_size]

        # Appel API OpenAI pour les embeddings
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=batch_texts,
        )
        embeddings = [item.embedding for item in response.data]

        collection.upsert(
            ids=batch_ids,
            embeddings=embeddings,
            documents=batch_texts,
            metadatas=batch_meta,
        )
        print(f"Indexé {i + len(batch_texts)}/{len(all_chunks)} chunks")

    print(f"\nTotal : {len(all_chunks)} chunks indexés dans ChromaDB")

# --- Exécution ---
if __name__ == "__main__":
    docs = load_documents("./documents")
    embed_and_store(docs)

import chromadb
from openai import OpenAI

client = OpenAI()

def get_collection():
    """Connecte à la collection ChromaDB existante."""
    chroma = chromadb.PersistentClient(path="./chroma_db")
    return chroma.get_collection("mes_documents")

def search_context(query: str, n_results: int = 5) -> list[dict]:
    """Recherche les chunks les plus pertinents."""
    collection = get_collection()

    # Embedding de la question
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=[query],
    )
    query_embedding = response.data[0].embedding

    # Recherche dans ChromaDB
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
    )

    contexts = []
    for i in range(len(results["documents"][0])):
        contexts.append({
            "text": results["documents"][0][i],
            "source": results["metadatas"][0][i]["source"],
            "distance": results["distances"][0][i],
        })

    return contexts

def chat(query: str, history: list[dict] = None) -> str:
    """Génère une réponse RAG."""
    # 1. Rechercher le contexte pertinent
    contexts = search_context(query)

    # 2. Construire le prompt enrichi
    context_text = "\n\n---\n\n".join(
        f"[Source: {c[''source'']}]\n{c[''text'']}" for c in contexts
    )

    system_prompt = f"""Tu es un assistant qui répond aux questions
en te basant UNIQUEMENT sur le contexte fourni ci-dessous.

Si le contexte ne contient pas l''information, dis-le clairement.
Cite toujours la source de tes informations.

--- CONTEXTE ---
{context_text}
--- FIN CONTEXTE ---"""

    messages = [{"role": "system", "content": system_prompt}]
    if history:
        messages.extend(history)
    messages.append({"role": "user", "content": query})

    # 3. Appel au LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.3,  # Basse pour des réponses factuelles
    )

    return response.choices[0].message.content

import streamlit as st
from chat import chat

st.set_page_config(page_title="Mon Chatbot RAG", page_icon="🤖")
st.title("Mon Chatbot RAG")
st.caption("Posez des questions sur vos documents")

# Historique de conversation
if "messages" not in st.session_state:
    st.session_state.messages = []

# Afficher l''historique
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Input utilisateur
if prompt := st.chat_input("Votre question..."):
    # Afficher la question
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # Générer la réponse
    with st.chat_message("assistant"):
        with st.spinner("Recherche en cours..."):
            response = chat(
                prompt,
                history=st.session_state.messages[:-1],
            )
            st.markdown(response)

    st.session_state.messages.append({"role": "assistant", "content": response})

# 1. D''abord, indexer vos documents
python ingest.py

# 2. Lancer l''interface
streamlit run app.py

# Après la recherche vectorielle, reranker les résultats
from cohere import Client

cohere = Client(api_key="...")

def rerank_results(query: str, contexts: list[dict], top_n: int = 3):
    """Reranke les résultats avec Cohere."""
    docs = [c["text"] for c in contexts]
    response = cohere.rerank(
        model="rerank-v3.5",
        query=query,
        documents=docs,
        top_n=top_n,
    )
    return [contexts[r.index] for r in response.results]

import re

def chunk_by_headers(text: str) -> list[str]:
    """Découpe par titres markdown (## ou ###)."""
    sections = re.split(r'\n(?=#{2,3} )', text)
    return [s.strip() for s in sections if len(s.strip()) > 50]

system_prompt = """...
IMPORTANT : Pour chaque information que tu donnes, cite la source
entre crochets comme ceci : [Source: nom_du_fichier.pdf]
Si plusieurs sources se contredisent, mentionne-le."""