RAG Souverain : Hebergement France et donnees europeennes

Pour de nombreuses entreprises francaises et europeennes, la souverainete des donnees n'est pas une option mais une obligation. Ce guide explique comment deployer un systeme RAG conforme aux exigences de localisation des donnees et aux reglementations europeennes.

Pourquoi la souverainete compte

Les enjeux pour les entreprises

Le cadre reglementaire

RGPD (depuis 2018) :

• Transferts hors UE tres encadres
• Arret Schrems II invalide Privacy Shield
• Clauses contractuelles types necessaires pour transferts

AI Act (2024-2027) :

• Obligations de transparence
• Documentation des systemes IA
• Potentielles restrictions sur modeles non-europeens

NIS2 (2024) :

• Securite des systemes d'information
• Concerne les OIV et OSE
• Obligations de localisation pour certains secteurs

Architecture souveraine

Infrastructure 100% europeenne

┌─────────────────────────────────────────────────────────────┐
│                    UTILISATEURS                              │
│                 (France / Europe)                           │
└───────────────────────┬─────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│                 HEBERGEMENT FRANCE                           │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                    Frontend                          │    │
│  │              (Serveurs FR - OVH/Scaleway)           │    │
│  └──────────────────────┬──────────────────────────────┘    │
│                         │                                    │
│  ┌──────────────────────▼──────────────────────────────┐    │
│  │                   Backend API                        │    │
│  │              (Serveurs FR - OVH/Scaleway)           │    │
│  └──────────────────────┬──────────────────────────────┘    │
│                         │                                    │
│        ┌────────────────┼────────────────┐                  │
│        ▼                ▼                ▼                  │
│  ┌───────────┐  ┌───────────────┐  ┌───────────────────┐   │
│  │ PostgreSQL│  │ Qdrant Vector │  │ Object Storage    │   │
│  │   (FR)    │  │    DB (FR)    │  │    (FR - S3)      │   │
│  └───────────┘  └───────────────┘  └───────────────────┘   │
│                                                              │
└─────────────────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│                 LLM PROVIDER                                 │
│  Option 1: Self-hosted (Mistral, Llama sur GPUs FR)         │
│  Option 2: API europeenne (Mistral API depuis Paris)        │
│  Option 3: API encadree (OpenAI avec DPA signe)             │
└─────────────────────────────────────────────────────────────┘

Choix d'hebergeur francais

Configuration OVH Cloud

DEVELOPERpython
# Configuration pour deploiement OVH
import os

# Base de donnees managee
DATABASE_CONFIG = {
    "host": os.getenv("OVH_DB_HOST"),  # xxx.db.ovh.net
    "port": 5432,
    "database": "rag_production",
    "user": os.getenv("OVH_DB_USER"),
    "password": os.getenv("OVH_DB_PASSWORD"),
    "sslmode": "require"  # Chiffrement obligatoire
}

# Object Storage S3-compatible
S3_CONFIG = {
    "endpoint_url": "https://s3.gra.io.cloud.ovh.net",  # Gravelines, FR
    "region_name": "gra",
    "aws_access_key_id": os.getenv("OVH_S3_ACCESS_KEY"),
    "aws_secret_access_key": os.getenv("OVH_S3_SECRET_KEY")
}

# Qdrant sur instance privee
QDRANT_CONFIG = {
    "host": os.getenv("QDRANT_HOST"),  # IP privee
    "port": 6333,
    "grpc_port": 6334,
    "api_key": os.getenv("QDRANT_API_KEY")
}

LLMs souverains

Modeles europeens

Mistral AI (France) :

• Mistral 7B, Mixtral 8x7B : open-source
• Mistral Large : API depuis Paris
• Conformite RGPD native

DEVELOPERpython
from mistralai.client import MistralClient

client = MistralClient(api_key=os.getenv("MISTRAL_API_KEY"))

async def generate_with_mistral(prompt: str, context: str) -> str:
    """
    Generation avec Mistral - serveurs en France
    """
    messages = [
        {
            "role": "system",
            "content": "Tu es un assistant RAG. Base tes reponses sur le contexte fourni."
        },
        {
            "role": "user",
            "content": f"Contexte:\n{context}\n\nQuestion: {prompt}"
        }
    ]

    response = client.chat(
        model="mistral-large-latest",
        messages=messages,
        temperature=0.3,
        max_tokens=1000
    )

    return response.choices[0].message.content

Alternatives europeennes :

• Aleph Alpha (Allemagne) : Luminous
• LightOn (France) : Paradigm
• Hugging Face (France) : Hub et Inference API

Self-hosting sur GPUs francais

DEVELOPERpython
# Deploiement Mistral sur GPU OVH/Scaleway
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class LocalMistralLLM:
    def __init__(self, model_name: str = "mistralai/Mistral-7B-Instruct-v0.2"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    async def generate(self, prompt: str, max_tokens: int = 500) -> str:
        inputs = self.tokenizer(prompt, return_tensors="pt").to("cuda")

        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=max_tokens,
                temperature=0.3,
                do_sample=True,
                pad_token_id=self.tokenizer.eos_token_id
            )

        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)

# Configuration GPU Scaleway
# Instance GPU-3070-S : ~0.8 EUR/h
# Instance L4 : ~1.2 EUR/h (meilleur rapport qualite/prix)

Embeddings souverains

DEVELOPERpython
from sentence_transformers import SentenceTransformer

class SovereignEmbedder:
    def __init__(self):
        # Modeles multilingues performants en francais
        # Self-hosted, pas d'appel API externe
        self.model = SentenceTransformer('intfloat/multilingual-e5-large')

    def embed(self, texts: list[str]) -> list[list[float]]:
        # Prefixe pour E5
        prefixed = [f"passage: {t}" for t in texts]
        return self.model.encode(prefixed).tolist()

    def embed_query(self, query: str) -> list[float]:
        return self.model.encode(f"query: {query}").tolist()

# Alternative : CamemBERT pour le francais specifiquement
# from transformers import CamembertModel, CamembertTokenizer

Conformite RGPD technique

Chiffrement bout en bout

DEVELOPERpython
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

class GDPRCompliantStorage:
    def __init__(self, master_key: bytes):
        # Derivation de cle avec sel aleatoire
        salt = os.urandom(16)
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(master_key))
        self.cipher = Fernet(key)

    def encrypt_pii(self, data: str) -> bytes:
        """Chiffre les donnees personnelles"""
        return self.cipher.encrypt(data.encode())

    def decrypt_pii(self, encrypted: bytes) -> str:
        """Dechiffre les donnees personnelles"""
        return self.cipher.decrypt(encrypted).decode()

    def pseudonymize(self, user_id: str) -> str:
        """Pseudonymise un identifiant utilisateur"""
        import hashlib
        return hashlib.sha256(user_id.encode() + self.salt).hexdigest()[:16]

Journalisation conforme

DEVELOPERpython
import logging
from datetime import datetime
import json

class GDPRLogger:
    def __init__(self, log_file: str):
        self.logger = logging.getLogger("gdpr_audit")
        handler = logging.FileHandler(log_file)
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_access(
        self,
        user_id: str,
        action: str,
        data_category: str,
        purpose: str,
        legal_basis: str
    ):
        """
        Journalise un acces aux donnees personnelles
        """
        entry = {
            "timestamp": datetime.now().isoformat(),
            "user_id_hash": self._hash_id(user_id),  # Ne pas logger l'ID en clair
            "action": action,
            "data_category": data_category,
            "purpose": purpose,
            "legal_basis": legal_basis,
            "retention_days": self._get_retention(data_category)
        }

        self.logger.info(json.dumps(entry))

    def log_deletion(self, user_id: str, data_deleted: list):
        """
        Journalise une suppression (droit a l'oubli)
        """
        entry = {
            "timestamp": datetime.now().isoformat(),
            "action": "RIGHT_TO_ERASURE",
            "user_id_hash": self._hash_id(user_id),
            "data_categories_deleted": data_deleted,
            "confirmation": True
        }

        self.logger.info(json.dumps(entry))

Gestion du consentement

DEVELOPERpython
from datetime import datetime, timedelta
from enum import Enum

class ConsentPurpose(Enum):
    RAG_INDEXING = "rag_indexing"
    CONVERSATION_HISTORY = "conversation_history"
    ANALYTICS = "analytics"
    PERSONALIZATION = "personalization"

class ConsentManager:
    def __init__(self, db):
        self.db = db

    async def request_consent(
        self,
        user_id: str,
        purpose: ConsentPurpose,
        data_description: str
    ) -> dict:
        """
        Prepare une demande de consentement
        """
        return {
            "consent_id": self._generate_id(),
            "purpose": purpose.value,
            "data_description": data_description,
            "legal_text": self._get_legal_text(purpose),
            "revocable": True,
            "valid_until": (datetime.now() + timedelta(days=365)).isoformat()
        }

    async def record_consent(
        self,
        user_id: str,
        consent_id: str,
        granted: bool,
        method: str  # "explicit_click", "api", etc.
    ):
        """
        Enregistre le consentement de l'utilisateur
        """
        await self.db.insert("consents", {
            "user_id": user_id,
            "consent_id": consent_id,
            "granted": granted,
            "method": method,
            "timestamp": datetime.now(),
            "ip_address_hash": self._hash_ip(),
            "user_agent_hash": self._hash_ua()
        })

    async def verify_consent(
        self,
        user_id: str,
        purpose: ConsentPurpose
    ) -> bool:
        """
        Verifie si l'utilisateur a donne son consentement
        """
        consent = await self.db.find_one("consents", {
            "user_id": user_id,
            "purpose": purpose.value,
            "granted": True,
            "revoked": False,
            "valid_until": {"$gte": datetime.now()}
        })

        return consent is not None

    async def revoke_consent(
        self,
        user_id: str,
        purpose: ConsentPurpose = None
    ):
        """
        Revoque le consentement
        """
        query = {"user_id": user_id}
        if purpose:
            query["purpose"] = purpose.value

        await self.db.update_many(
            "consents",
            query,
            {"$set": {"revoked": True, "revoked_at": datetime.now()}}
        )

Transferts de donnees encadres

Utilisation d'APIs americaines avec precautions

Si l'utilisation d'OpenAI ou autre API US est necessaire :

DEVELOPERpython
class SecureAPIWrapper:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    async def generate_safe(
        self,
        prompt: str,
        context: str,
        pii_filter: bool = True
    ) -> str:
        """
        Appel API avec protection des donnees
        """
        # 1. Detecter et masquer les PII
        if pii_filter:
            context = self._mask_pii(context)
            prompt = self._mask_pii(prompt)

        # 2. Appel API
        response = await self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "..."},
                {"role": "user", "content": f"{context}\n\n{prompt}"}
            ]
        )

        # 3. Demasquer les PII dans la reponse si necessaire
        answer = response.choices[0].message.content

        return answer

    def _mask_pii(self, text: str) -> str:
        """
        Masque les donnees personnelles avant envoi
        """
        import re

        # Emails
        text = re.sub(
            r'\b[\w.-]+@[\w.-]+\.\w+\b',
            '[EMAIL]',
            text
        )

        # Numeros de telephone FR
        text = re.sub(
            r'\b(?:(?:\+33|0033|0)[1-9](?:[\s.-]?\d{2}){4})\b',
            '[PHONE]',
            text
        )

        # Noms propres (simplifie)
        # En production, utiliser un NER

        return text

Documentation des transferts

DEVELOPERpython
class TransferRegistry:
    """
    Registre des transferts hors UE (obligation RGPD)
    """

    def __init__(self, db):
        self.db = db

    async def register_transfer(
        self,
        destination_country: str,
        recipient: str,
        data_categories: list,
        legal_basis: str,
        safeguards: list
    ):
        """
        Enregistre un transfert hors UE
        """
        await self.db.insert("transfer_registry", {
            "timestamp": datetime.now(),
            "destination": destination_country,
            "recipient": recipient,
            "data_categories": data_categories,
            "legal_basis": legal_basis,  # "SCC", "adequacy_decision", etc.
            "safeguards": safeguards,
            "documented_by": "system",
            "review_date": datetime.now() + timedelta(days=365)
        })

    async def generate_report(self) -> dict:
        """
        Genere le rapport pour le DPO
        """
        transfers = await self.db.find("transfer_registry")

        return {
            "report_date": datetime.now().isoformat(),
            "total_transfers": len(transfers),
            "by_country": self._group_by_country(transfers),
            "by_legal_basis": self._group_by_basis(transfers),
            "pending_reviews": self._get_pending_reviews(transfers)
        }

Checklist souverainete

Infrastructure

• Hebergeur francais/europeen avec certifications (ISO 27001, HDS si sante)
• Donnees stockees exclusivement en France/UE
• Sauvegardes sur territoire europeen
• Chiffrement au repos et en transit (TLS 1.3)
• Acces restreints avec authentification forte

LLM et IA

• Modele europeen (Mistral) ou self-hosted
• Si API US : DPA signe, PII masquees
• Documentation des traitements IA

RGPD

• Registre des traitements a jour
• Consentements traces
• Mecanismes d'exercice des droits
• Politique de retention definie
• Logs d'audit conserves

Contractuel

• DPA avec tous les sous-traitants
• Clauses contractuelles types si transfert hors UE
• Analyse d'impact (DPIA) si donnees sensibles

Pour aller plus loin

• Securite et Conformite RAG - Guide complet securite
• RGPD et Chatbots - Conformite chatbots
• AI Act et RAG - Implications AI Act

RAG Souverain cle en main avec Ailog

Construire une infrastructure RAG souveraine demande des competences variees. Avec Ailog, beneficiez d'une solution 100% francaise :

• Hebergement OVH Cloud a Gravelines et Roubaix
• Donnees jamais transferees hors UE
• Modeles Mistral pour la generation
• Embeddings self-hosted sans appel API externe
• Conformite RGPD native avec DPA fourni
• Support francais et documentation en francais

Testez Ailog gratuitement et deployez un RAG souverain en toute serenite.