RAG en Python : Construire une API de Recherche Documentaire Conforme au RGPD avec Inférence Hébergée dans l'UE

Construisez un système de génération augmentée par la récupération (RAG) prêt pour la production en Python, en gardant toutes les données au sein de l'UE. Ce guide couvre l'ingestion de documents avec PyMuPDF, le stockage vectoriel avec Qdrant, et l'inférence LLM via l'API privée européenne de Juice Factory — le tout encapsulé dans un service FastAPI.

À la fin de ce guide, vous disposerez d'une API de recherche documentaire fonctionnelle qui :

• Extrait le texte des fichiers PDF avec PyMuPDF
• Génère des embeddings et les stocke dans Qdrant
• Répond aux questions en s'appuyant sur le contexte récupéré et l'inférence LLM hébergée dans l'UE
• Ne transmet jamais de données utilisateur en dehors de l'UE

Prérequis

• Python 3.10+
• Docker (pour Qdrant)
• Une clé API Juice Factory (obtenez-la ici)

Vue d'ensemble de l'architecture

┌──────────────┐     ┌───────────────┐     ┌──────────────────┐
│  PDF Upload  │────▶│  PyMuPDF      │────▶│  Qdrant          │
│  (FastAPI)   │     │  Text Extract │     │  Vector Store    │
└──────────────┘     └───────────────┘     └──────────────────┘
                                                    │
┌──────────────┐     ┌───────────────┐              │
│  User Query  │────▶│  Embedding    │──── search ──┘
│  (FastAPI)   │     │  (EU API)     │
└──────────────┘     └───────┬───────┘
                             │
                     ┌───────▼───────┐     ┌──────────────────┐
                     │  Context +    │────▶│  LLM Inference   │
                     │  Query        │     │  (EU-hosted)     │
                     └───────────────┘     └──────────────────┘

Le système suit un pipeline RAG classique, mais chaque composant qui manipule des données utilisateur s'exécute au sein de l'infrastructure européenne. Qdrant est auto-hébergé, et les embeddings comme l'inférence LLM passent par les endpoints européens de Juice Factory.

Étape 1 : Initialisation du projet

Créez le répertoire du projet et installez les dépendances :

mkdir rag-document-search && cd rag-document-search
python -m venv .venv
source .venv/bin/activate

Installez les paquets nécessaires :

pip install fastapi uvicorn pymupdf qdrant-client openai python-multipart

Créez la structure du projet :

rag-document-search/
├── main.py              # Application FastAPI
├── ingest.py            # Pipeline d'ingestion de documents
├── search.py            # Logique de requête et de récupération
├── config.py            # Configuration
└── requirements.txt

requirements.txt :

fastapi==0.115.0
uvicorn==0.30.0
pymupdf==1.24.0
qdrant-client==1.11.0
openai==1.50.0
python-multipart==0.0.9

Étape 2 : Configuration

Configurez le projet avec vos identifiants API Juice Factory :

# config.py
import os

# Juice Factory EU API (OpenAI-compatible)
API_BASE_URL = "https://api.juicefactory.ai/v1"
API_KEY = os.environ.get("JUICEFACTORY_API_KEY", "your-api-key")

# Embedding model
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIMENSIONS = 1536

# Chat model for RAG responses
CHAT_MODEL = "gpt-4"

# Qdrant configuration (self-hosted in EU)
QDRANT_HOST = os.environ.get("QDRANT_HOST", "localhost")
QDRANT_PORT = int(os.environ.get("QDRANT_PORT", "6333"))
COLLECTION_NAME = "documents"

# Chunk settings
CHUNK_SIZE = 500       # tokens per chunk (approximate)
CHUNK_OVERLAP = 50     # overlap between chunks
TOP_K = 5              # number of chunks to retrieve

Étape 3 : Lancer Qdrant avec Docker

Lancez Qdrant en local (ou sur votre serveur européen) :

docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v qdrant_storage:/qdrant/storage \
  qdrant/qdrant:latest

Qdrant stocke toutes les données localement — aucun appel externe, pas de télémétrie, un contrôle total sur la localisation des données.

Étape 4 : Ingestion de documents avec PyMuPDF

Le pipeline d'ingestion extrait le texte des fichiers PDF, le découpe en segments, génère des embeddings via l'API européenne, et stocke le tout dans Qdrant.

# ingest.py
import fitz  # PyMuPDF
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
import config


def get_openai_client() -> OpenAI:
    """Create OpenAI client pointing to Juice Factory EU API."""
    return OpenAI(
        api_key=config.API_KEY,
        base_url=config.API_BASE_URL,
    )


def get_qdrant_client() -> QdrantClient:
    """Create Qdrant client."""
    return QdrantClient(host=config.QDRANT_HOST, port=config.QDRANT_PORT)


def extract_text_from_pdf(pdf_bytes: bytes) -> list[dict]:
    """Extract text from PDF, page by page."""
    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text").strip()
        if text:
            pages.append({
                "page": page_num + 1,
                "text": text,
            })
    doc.close()
    return pages


def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks by word count."""
    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
    return chunks


def generate_embeddings(texts: list[str], client: OpenAI) -> list[list[float]]:
    """Generate embeddings using Juice Factory EU API."""
    response = client.embeddings.create(
        model=config.EMBEDDING_MODEL,
        input=texts,
    )
    return [item.embedding for item in response.data]


def ensure_collection(qdrant: QdrantClient):
    """Create Qdrant collection if it doesn't exist."""
    collections = [c.name for c in qdrant.get_collections().collections]
    if config.COLLECTION_NAME not in collections:
        qdrant.create_collection(
            collection_name=config.COLLECTION_NAME,
            vectors_config=VectorParams(
                size=config.EMBEDDING_DIMENSIONS,
                distance=Distance.COSINE,
            ),
        )


def ingest_pdf(pdf_bytes: bytes, filename: str) -> int:
    """Full ingestion pipeline: PDF → chunks → embeddings → Qdrant."""
    openai_client = get_openai_client()
    qdrant = get_qdrant_client()
    ensure_collection(qdrant)

    # Extract text from PDF
    pages = extract_text_from_pdf(pdf_bytes)

    # Chunk all pages
    all_chunks = []
    for page_data in pages:
        chunks = chunk_text(
            page_data["text"],
            chunk_size=config.CHUNK_SIZE,
            overlap=config.CHUNK_OVERLAP,
        )
        for chunk in chunks:
            all_chunks.append({
                "text": chunk,
                "page": page_data["page"],
                "filename": filename,
            })

    if not all_chunks:
        return 0

    # Generate embeddings (batch)
    texts = [c["text"] for c in all_chunks]
    embeddings = generate_embeddings(texts, openai_client)

    # Store in Qdrant
    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={
                "text": chunk["text"],
                "page": chunk["page"],
                "filename": chunk["filename"],
            },
        )
        for chunk, embedding in zip(all_chunks, embeddings)
    ]

    qdrant.upsert(
        collection_name=config.COLLECTION_NAME,
        points=points,
    )

    return len(points)

Points essentiels :

• PyMuPDF (fitz) extrait le texte sans dépendances externes ni appels cloud
• Les embeddings sont générés via l'API européenne de Juice Factory — même SDK OpenAI, endpoint européen
• Qdrant stocke les vecteurs localement sans aucune télémétrie

Étape 5 : Recherche et requête RAG

Le module de recherche transforme la requête utilisateur en embedding, récupère les segments pertinents, puis les envoie avec la question au LLM.

# search.py
from openai import OpenAI
from qdrant_client import QdrantClient
import config
from ingest import get_openai_client, get_qdrant_client, generate_embeddings


def search_documents(query: str, top_k: int = None) -> list[dict]:
    """Search for relevant document chunks."""
    if top_k is None:
        top_k = config.TOP_K

    openai_client = get_openai_client()
    qdrant = get_qdrant_client()

    # Embed the query
    query_embedding = generate_embeddings([query], openai_client)[0]

    # Search Qdrant
    results = qdrant.search(
        collection_name=config.COLLECTION_NAME,
        query_vector=query_embedding,
        limit=top_k,
    )

    return [
        {
            "text": hit.payload["text"],
            "page": hit.payload["page"],
            "filename": hit.payload["filename"],
            "score": hit.score,
        }
        for hit in results
    ]


def rag_query(question: str) -> dict:
    """Full RAG pipeline: embed query → retrieve context → generate answer."""
    # Retrieve relevant chunks
    chunks = search_documents(question)

    if not chunks:
        return {
            "answer": "No relevant documents found. Please upload documents first.",
            "sources": [],
        }

    # Build context from retrieved chunks
    context_parts = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(
            f"[Source {i}: {chunk['filename']}, page {chunk['page']}]\n{chunk['text']}"
        )
    context = "\n\n".join(context_parts)

    # Generate answer using EU-hosted LLM
    openai_client = get_openai_client()
    response = openai_client.chat.completions.create(
        model=config.CHAT_MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a document assistant. Answer questions based on the "
                    "provided context. Always cite which source and page number your "
                    "answer comes from. If the context doesn't contain enough "
                    "information to answer, say so clearly."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}",
            },
        ],
        temperature=0.1,
        max_tokens=1000,
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [
            {
                "filename": c["filename"],
                "page": c["page"],
                "score": round(c["score"], 4),
                "excerpt": c["text"][:200] + "..." if len(c["text"]) > 200 else c["text"],
            }
            for c in chunks
        ],
        "model": response.model,
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
        },
    }

La fonction rag_query constitue le coeur du système :

1. Elle transforme la question de l'utilisateur en embedding via l'API européenne
2. Elle récupère les top-K segments les plus pertinents depuis Qdrant
3. Elle envoie le contexte et la question au LLM hébergé dans l'UE
4. Elle renvoie la réponse accompagnée des citations de sources

Étape 6 : Application FastAPI

Assemblez le tout dans un service FastAPI :

# main.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
from ingest import ingest_pdf
from search import rag_query, search_documents

app = FastAPI(
    title="GDPR-Safe Document Search API",
    description="RAG-powered document search with EU-hosted inference",
    version="1.0.0",
)


class QueryRequest(BaseModel):
    question: str
    top_k: int = 5


class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    model: str | None = None
    usage: dict | None = None


@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """Upload a PDF document for indexing."""
    if not file.filename.lower().endswith(".pdf"):
        raise HTTPException(status_code=400, detail="Only PDF files are supported")

    pdf_bytes = await file.read()
    if len(pdf_bytes) > 50 * 1024 * 1024:  # 50MB limit
        raise HTTPException(status_code=400, detail="File too large (max 50MB)")

    num_chunks = ingest_pdf(pdf_bytes, file.filename)

    return {
        "filename": file.filename,
        "chunks_indexed": num_chunks,
        "status": "indexed",
    }


@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    """Ask a question about uploaded documents."""
    if not request.question.strip():
        raise HTTPException(status_code=400, detail="Question cannot be empty")

    result = rag_query(request.question)
    return QueryResponse(**result)


@app.post("/search")
async def search_only(request: QueryRequest):
    """Search for relevant chunks without generating an answer."""
    results = search_documents(request.question, top_k=request.top_k)
    return {"results": results}


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "ok", "data_residency": "EU"}

Étape 7 : Lancement et tests

Démarrez le serveur API :

export JUICEFACTORY_API_KEY="your-api-key"
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Envoyer un document

curl -X POST http://localhost:8000/upload \
  -F "file=@contract.pdf"

Réponse :

{
  "filename": "contract.pdf",
  "chunks_indexed": 47,
  "status": "indexed"
}

Poser une question

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What are the payment terms in the contract?"}'

Réponse :

{
  "answer": "According to the contract (Source 1, page 4), payment terms are Net 30 from the date of invoice. Late payments accrue interest at 1.5% per month as specified in Section 5.2.",
  "sources": [
    {
      "filename": "contract.pdf",
      "page": 4,
      "score": 0.9234,
      "excerpt": "Payment Terms. The Client shall pay all invoices within thirty (30) days..."
    }
  ],
  "model": "gpt-4-0125-preview",
  "usage": {
    "prompt_tokens": 847,
    "completion_tokens": 89
  }
}

Checklist de conformité RGPD

Cette architecture satisfait les exigences du RGPD à chaque couche :

Garanties essentielles :

• Les requêtes utilisateur ne quittent jamais l'UE
• Aucune donnée n'est utilisée pour l'entraînement de modèles
• Qdrant ne stocke que les embeddings (pas les requêtes brutes)
• L'inférence LLM est sans état — les requêtes ne sont pas conservées
• Vous contrôlez toutes les politiques de journalisation et de rétention des données

Considérations pour la production

Mise à l'échelle de Qdrant

Pour les déploiements en production avec des collections de documents volumineuses :

# Run Qdrant with persistent storage and resource limits
docker run -d \
  --name qdrant \
  -p 6333:6333 \
  --memory=4g \
  -v /data/qdrant:/qdrant/storage \
  qdrant/qdrant:latest

Pour les collections dépassant 10 millions de vecteurs, envisagez le mode distribué de Qdrant avec du sharding réparti sur plusieurs noeuds hébergés dans l'UE.

Stratégie de découpage

Le découpage simple par nombre de mots présenté dans ce guide fonctionne pour la plupart des documents. Pour de meilleurs résultats avec des documents structurés :

• Découpage sémantique : découpez aux frontières des paragraphes ou des sections
• Fenêtre glissante : utilisez des segments avec chevauchement pour éviter de couper le contexte
• Enrichissement des métadonnées : incluez les titres de section, les titres de documents et les dates dans les métadonnées de chaque segment

Gestion des erreurs

Ajoutez une logique de réessai pour les appels API et gérez les défaillances de connexion à Qdrant :

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def generate_embeddings_with_retry(texts, client):
    return generate_embeddings(texts, client)

Authentification

Ajoutez une authentification par clé API à vos endpoints FastAPI pour la production :

from fastapi import Depends, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != os.environ.get("APP_API_KEY"):
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.post("/query", dependencies=[Depends(verify_api_key)])
async def query_documents(request: QueryRequest):
    ...

Résumé

Ce guide présente un pipeline RAG complet qui maintient la conformité RGPD de bout en bout :

1. Ingestion de documents : PyMuPDF extrait le texte localement, sans dépendance cloud
2. Embeddings : générés via l'API européenne de Juice Factory sans rétention de données
3. Stockage vectoriel : Qdrant auto-hébergé conserve toutes les données indexées sous votre contrôle
4. Inférence LLM : traitement sans état hébergé dans l'UE, sans stockage des requêtes
5. Couche API : FastAPI vous offre un contrôle total sur les accès, la journalisation et le traitement des données

L'ensemble du système peut être déployé sur une infrastructure européenne sans qu'aucune donnée ne quitte la région. Migrer depuis une configuration non conforme est simple : remplacez l'URL de base de l'API, redirigez les embeddings vers l'endpoint européen, et auto-hébergez votre stockage vectoriel.

Guides connexes

• GDPR-Safe AI Inference — Guide d'architecture pour des applications IA conformes avec RAG
• Replacing OpenAI with EU Infrastructure — Guide de migration pour changer de fournisseur d'API
• n8n + Private AI Automation — Automatisation de workflows avec inférence hébergée dans l'UE
• Cursor AI BYOK Setup — Utiliser Juice Factory comme fournisseur BYOK dans Cursor