RAG pour les nuls : donner de la mémoire à son IA

Sans RAG :
   Question --> LLM --> Réponse (basée sur ses connaissances générales)

Avec RAG :
   Question --> Recherche dans vos docs --> Documents pertinents
                                                    |
                                                    v
   Question + Documents pertinents --> LLM --> Réponse précise et sourcée

# Conceptuellement, un embedding transforme du texte en nombres
"Le chat dort sur le canapé"  -->  [0.23, -0.45, 0.78, 0.12, ...]
"Le félin se repose au salon" -->  [0.21, -0.43, 0.76, 0.14, ...]
"Python est un langage"       -->  [-0.67, 0.34, -0.12, 0.89, ...]

import httpx
import os

async def get_embedding(text: str) -> list[float]:
    """Transforme du texte en vecteur via OpenAI."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.openai.com/v1/embeddings",
            headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
            json={
                "model": "text-embedding-3-small",
                "input": text,
            },
        )
        response.raise_for_status()
        return response.json()["data"][0]["embedding"]

import numpy as np

def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
    """Calcule la similarite cosinus entre deux vecteurs."""
    a = np.array(vec_a)
    b = np.array(vec_b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

# Exemple
sim = cosine_similarity(
    embedding_chat,     # "Le chat dort sur le canape"
    embedding_felin     # "Le felin se repose au salon"
)
# sim ≈ 0.95 (tres similaire!)

sim = cosine_similarity(
    embedding_chat,     # "Le chat dort sur le canape"
    embedding_python    # "Python est un langage"
)
# sim ≈ 0.12 (pas du tout similaire)

import faiss
import numpy as np

# Creer un index FAISS
dimension = 1536  # Taille des embeddings OpenAI
index = faiss.IndexFlatL2(dimension)

# Ajouter des vecteurs
vectors = np.array(embeddings_list).astype('float32')
index.add(vectors)

# Rechercher les 5 plus proches
query_vector = np.array([query_embedding]).astype('float32')
distances, indices = index.search(query_vector, k=5)

print(f"Documents les plus pertinents : {indices[0]}")
print(f"Distances : {distances[0]}")

import chromadb

# Creer/ouvrir une collection
client = chromadb.PersistentClient(path="/home/deploy/data/chroma")
collection = client.get_or_create_collection(
    name="mes_documents",
    metadata={"hnsw:space": "cosine"}
)

# Ajouter des documents (ChromaDB genere les embeddings automatiquement)
collection.add(
    documents=[
        "Guide complet du self-hosting",
        "Comment configurer un reverse proxy",
        "Les bases de la securite serveur",
    ],
    ids=["doc1", "doc2", "doc3"],
    metadatas=[
        {"source": "blog", "date": "2025-01"},
        {"source": "blog", "date": "2025-02"},
        {"source": "wiki", "date": "2025-01"},
    ],
)

# Rechercher
results = collection.query(
    query_texts=["comment securiser mon serveur"],
    n_results=3,
)
print(results["documents"])
# --> [["Les bases de la securite serveur", ...]]

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

# Connexion
client = QdrantClient(host="localhost", port=6333)

# Creer une collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
    ),
)

# Ajouter des documents
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=embedding_1,
            payload={"text": "Guide du self-hosting", "source": "blog"},
        ),
        PointStruct(
            id=2,
            vector=embedding_2,
            payload={"text": "Config reverse proxy", "source": "blog"},
        ),
    ],
)

# Rechercher
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=5,
)
for result in results:
    print(f"Score: {result.score:.3f} | {result.payload['text']}")

1. INGESTION        2. CHUNKING         3. EMBEDDING
┌──────────┐       ┌──────────┐       ┌──────────┐
│ Documents │──────>│ Découper │──────>│ Vectoriser│
│ (PDF, MD, │      │ en chunks│       │ chaque    │
│  TXT...)  │      │ de ~500  │       │ chunk     │
└──────────┘       │ tokens   │       └─────┬─────┘
                   └──────────┘             │
                                            v
4. STOCKAGE         5. RETRIEVAL        6. GENERATION
┌──────────┐       ┌──────────┐       ┌──────────┐
│ Vector DB │<──── │ Chercher │──────>│ LLM +    │
│ (Qdrant,  │      │ les plus │       │ contexte │
│  Chroma)  │─────>│ proches  │       │ = Reponse│
└──────────┘       └──────────┘       └──────────┘

from pathlib import Path

def load_documents(directory: str) -> list[dict]:
    """Charge tous les documents d'un repertoire."""
    docs = []
    for path in Path(directory).rglob("*"):
        if path.suffix in (".md", ".txt", ".py", ".json"):
            content = path.read_text(encoding="utf-8")
            docs.append({
                "content": content,
                "source": str(path),
                "type": path.suffix,
            })
    return docs

documents = load_documents("/home/deploy/knowledge")
print(f"{len(documents)} documents charges")

def chunk_text(
    text: str,
    chunk_size: int = 500,
    overlap: int = 50
) -> list[str]:
    """Decoupe un texte en chunks avec chevauchement."""
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk.strip():
            chunks.append(chunk)

    return chunks

# Exemple
text = "Un tres long document..."
chunks = chunk_text(text, chunk_size=500, overlap=50)
print(f"{len(chunks)} chunks crees")

async def embed_chunks(chunks: list[str]) -> list[list[float]]:
    """Genere les embeddings pour une liste de chunks."""
    embeddings = []
    # Traiter par batch de 100 (limite API)
    for i in range(0, len(chunks), 100):
        batch = chunks[i:i+100]
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.openai.com/v1/embeddings",
                headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
                json={
                    "model": "text-embedding-3-small",
                    "input": batch,
                },
                timeout=30.0,
            )
            response.raise_for_status()
            batch_embeddings = [
                item["embedding"]
                for item in response.json()["data"]
            ]
            embeddings.extend(batch_embeddings)
    return embeddings

collection.add(
    documents=chunks,
    embeddings=embeddings,
    ids=[f"chunk_{i}" for i in range(len(chunks))],
    metadatas=[
        {"source": doc["source"], "chunk_index": i}
        for i, doc in enumerate(chunk_metadata)
    ],
)

async def retrieve(question: str, n_results: int = 5) -> list[str]:
    """Retrouve les chunks les plus pertinents pour une question."""
    results = collection.query(
        query_texts=[question],
        n_results=n_results,
    )
    return results["documents"][0]

# Exemple
chunks = await retrieve("Comment securiser mon VPS ?")
# --> ["Les bases de la securite serveur...", "UFW et fail2ban...", ...]

async def rag_answer(question: str) -> str:
    """Repond a une question en utilisant le RAG."""
    # 1. Retrouver les documents pertinents
    relevant_chunks = await retrieve(question, n_results=5)

    # 2. Construire le prompt avec le contexte
    context = "\n\n---\n\n".join(relevant_chunks)
    prompt = f"""Reponds a la question en te basant UNIQUEMENT
sur le contexte fourni. Si le contexte ne contient pas
l'information, dis-le clairement.

CONTEXTE :
{context}

QUESTION : {question}

REPONSE :"""

    # 3. Appeler le LLM
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://openrouter.ai/api/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {os.getenv('OPENROUTER_API_KEY')}",
            },
            json={
                "model": "anthropic/claude-sonnet-4-20250514",
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": 1000,
            },
            timeout=60.0,
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]

# Utilisation
answer = await rag_answer("Comment securiser mon VPS ?")
print(answer)

# Charger le contexte depuis des fichiers
context_files = [
    "MEMORY.md",       # Memoire long-terme
    "PROJECT.md",      # Info projet
    "NOTES.md",        # Notes diverses
]

context = ""
for f in context_files:
    with open(f) as fh:
        context += f"\n## {f}\n{fh.read()}\n"

prompt = f"""Contexte :\n{context}\n\nQuestion : {question}"""

# MEMORY.md

## Preferences utilisateur
- Prefere les reponses concises
- Utilise Python 3.11
- VPS chez Hostinger, Ubuntu 24.04

## Projets en cours
- AI-master.dev : site de guides IA en francais
- API interne : FastAPI + SQLite

## Decisions prises
- 2025-02-20 : Choisi Caddy plutot que Nginx
- 2025-02-22 : Migration vers ChromaDB pour le RAG

import re

def memory_search(query: str, documents: list[dict], top_k: int = 5) -> list[dict]:
    """Recherche simple par mots-cles dans les documents."""
    keywords = set(query.lower().split())
    scored = []

    for doc in documents:
        text = doc["content"].lower()
        # Compter les mots-cles presents
        score = sum(1 for kw in keywords if kw in text)
        if score > 0:
            scored.append((score, doc))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [doc for _, doc in scored[:top_k]]