Souveräner RAG: Hosting in Frankreich und europäische Daten

Für viele französische und europäische Unternehmen ist Datensouveränität keine Option, sondern eine Pflicht. Dieser Leitfaden erklärt, wie man ein RAG-System einsetzt, das den Anforderungen an Datenlokalisierung und europäische Vorschriften entspricht.

Warum Souveränität wichtig ist

Die Herausforderungen für Unternehmen

Der regulatorische Rahmen

DSGVO (seit 2018):

• Übermittlungen außerhalb der EU sind stark reglementiert
• Das Schrems-II-Urteil hat den Privacy Shield für ungültig erklärt
• Standardvertragsklauseln sind für Übermittlungen erforderlich

AI Act (2024-2027):

• Transparenzpflichten
• Dokumentation der KI-Systeme
• Mögliche Einschränkungen bei nicht-europäischen Modellen

NIS2 (2024):

• Sicherheit von Informationssystemen
• Betrifft kritische Infrastrukturen und wesentliche Dienste
• Lokalisierungspflichten für bestimmte Branchen

Souveräne Architektur

100 % europäische Infrastruktur

┌─────────────────────────────────────────────────────────────┐
│                    BENUTZER                                  │
│                 (Frankreich / Europa)                        │
└───────────────────────┬─────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│                 HOSTING FRANKREICH                            │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                    Frontend                          │    │
│  │              (Server FR - OVH/Scaleway)             │    │
│  └──────────────────────┬──────────────────────────────┘    │
│                         │                                    │
│  ┌──────────────────────▼──────────────────────────────┐    │
│  │                   Backend API                        │    │
│  │              (Server FR - OVH/Scaleway)             │    │
│  └──────────────────────┬──────────────────────────────┘    │
│                         │                                    │
│        ┌────────────────┼────────────────┐                  │
│        ▼                ▼                ▼                  │
│  ┌───────────┐  ┌───────────────┐  ┌───────────────────┐   │
│  │ PostgreSQL│  │ Qdrant Vector │  │ Object Storage    │   │
│  │   (FR)    │  │    DB (FR)    │  │    (FR - S3)      │   │
│  └───────────┘  └───────────────┘  └───────────────────┘   │
│                                                              │
└─────────────────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│                 LLM-ANBIETER                                 │
│  Option 1: Self-hosted (Mistral, Llama auf FR-GPUs)          │
│  Option 2: Europäische API (Mistral-API aus Paris)           │
│  Option 3: Regulierte API (OpenAI mit unterzeichnetem DPA)   │
└─────────────────────────────────────────────────────────────┘

Wahl eines französischen Hosters

OVH-Cloud-Konfiguration

DEVELOPERpython
# Konfiguration für OVH-Deployment
import os

# Verwaltete Datenbank
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"  # Verschlüsselung obligatorisch
}

# S3-kompatibler Object Storage
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 auf privater Instanz
QDRANT_CONFIG = {
    "host": os.getenv("QDRANT_HOST"),  # Private IP
    "port": 6333,
    "grpc_port": 6334,
    "api_key": os.getenv("QDRANT_API_KEY")
}

Souveräne LLMs

Europäische Modelle

Mistral AI (Frankreich):

• Mistral Small 4, Mistral Large 3: Open Source (Apache 2.0)
• Mistral Large: API aus Paris
• Native DSGVO-Konformität

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:
    """
    Generierung mit Mistral - Server in Frankreich
    """
    messages = [
        {
            "role": "system",
            "content": "Du bist ein RAG-Assistent. Stütze deine Antworten auf den bereitgestellten Kontext."
        },
        {
            "role": "user",
            "content": f"Kontext:\n{context}\n\nFrage: {prompt}"
        }
    ]

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

    return response.choices[0].message.content

Europäische Alternativen:

• Aleph Alpha (Deutschland): PhariaAI
• LightOn (Frankreich): Paradigm
• Hugging Face (Frankreich): Hub und Inference API

Self-Hosting auf französischen GPUs

DEVELOPERpython
# Bereitstellung von Mistral auf OVH/Scaleway-GPU
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)

# Scaleway-GPU-Konfiguration
# Instanz GPU-3070-S: ~0,8 EUR/h
# Instanz L4: ~0,79 EUR/h (bestes Preis-Leistungs-Verhältnis)

Souveräne Embeddings

DEVELOPERpython
from sentence_transformers import SentenceTransformer

class SovereignEmbedder:
    def __init__(self):
        # Leistungsstarke mehrsprachige Modelle für Deutsch/Französisch
        # Self-hosted, kein externer API-Aufruf
        self.model = SentenceTransformer('intfloat/multilingual-e5-large')

    def embed(self, texts: list[str]) -> list[list[float]]:
        # Präfix für 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 speziell für Französisch
# from transformers import CamembertModel, CamembertTokenizer

Technische DSGVO-Konformität

End-to-End-Verschlüsselung

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):
        # Schlüsselableitung mit zufälligem Salt
        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:
        """Verschlüsselt personenbezogene Daten"""
        return self.cipher.encrypt(data.encode())

    def decrypt_pii(self, encrypted: bytes) -> str:
        """Entschlüsselt personenbezogene Daten"""
        return self.cipher.decrypt(encrypted).decode()

    def pseudonymize(self, user_id: str) -> str:
        """Pseudonymisiert eine Benutzer-ID"""
        import hashlib
        return hashlib.sha256(user_id.encode() + self.salt).hexdigest()[:16]

Konforme Protokollierung

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
    ):
        """
        Protokolliert einen Zugriff auf personenbezogene Daten
        """
        entry = {
            "timestamp": datetime.now().isoformat(),
            "user_id_hash": self._hash_id(user_id),  # ID nicht im Klartext loggen
            "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):
        """
        Protokolliert eine Löschung (Recht auf Vergessenwerden)
        """
        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))

Einwilligungsmanagement

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:
        """
        Bereitet eine Einwilligungsanfrage vor
        """
        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", usw.
    ):
        """
        Speichert die Einwilligung des Nutzers
        """
        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:
        """
        Prüft, ob der Nutzer seine Einwilligung gegeben hat
        """
        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
    ):
        """
        Widerruft die Einwilligung
        """
        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()}}
        )

Geregelte Datenübermittlungen

Nutzung von US-APIs mit Vorsichtsmaßnahmen

Falls die Nutzung von OpenAI oder einer anderen US-API notwendig ist:

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:
        """
        API-Aufruf mit Datenschutz
        """
        # 1. PII erkennen und maskieren
        if pii_filter:
            context = self._mask_pii(context)
            prompt = self._mask_pii(prompt)

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

        # 3. PII in der Antwort bei Bedarf demaskieren
        answer = response.choices[0].message.content

        return answer

    def _mask_pii(self, text: str) -> str:
        """
        Maskiert personenbezogene Daten vor dem Versand
        """
        import re

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

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

        # Eigennamen (vereinfacht)
        # In Produktion einen NER-Ansatz verwenden

        return text

Dokumentation der Übermittlungen

DEVELOPERpython
class TransferRegistry:
    """
    Register der Übermittlungen außerhalb der EU (DSGVO-Pflicht)
    """

    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
    ):
        """
        Erfasst eine Übermittlung außerhalb der EU
        """
        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", usw.
            "safeguards": safeguards,
            "documented_by": "system",
            "review_date": datetime.now() + timedelta(days=365)
        })

    async def generate_report(self) -> dict:
        """
        Erstellt den Bericht für den Datenschutzbeauftragten
        """
        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)
        }

Souveränitäts-Checkliste

Infrastruktur

• Französischer/europäischer Hoster mit Zertifizierungen (ISO 27001, HDS im Gesundheitswesen)
• Daten ausschließlich in Frankreich/der EU gespeichert
• Backups auf europäischem Territorium
• Verschlüsselung im Ruhezustand und bei der Übertragung (TLS 1.3)
• Eingeschränkter Zugriff mit starker Authentifizierung

LLM und KI

• Europäisches Modell (Mistral) oder Self-Hosting
• Bei US-API: DPA unterzeichnet, PII maskiert
• Dokumentation der KI-Verarbeitungen

DSGVO

• Verarbeitungsverzeichnis aktuell
• Einwilligungen nachvollziehbar dokumentiert
• Mechanismen zur Ausübung der Betroffenenrechte
• Definierte Aufbewahrungsrichtlinie
• Audit-Logs aufbewahrt

Vertraglich

• DPA mit allen Auftragsverarbeitern
• Standardvertragsklauseln bei Übermittlung außerhalb der EU
• Datenschutz-Folgenabschätzung (DPIA) bei sensiblen Daten

Weiterführende Themen

• Sicherheit und Konformität von RAG - Vollständiger Sicherheitsleitfaden
• DSGVO und Chatbots - Konformität für Chatbots
• AI Act und RAG - Auswirkungen des AI Act

Souveräner RAG schlüsselfertig mit Ailog

Der Aufbau einer souveränen RAG-Infrastruktur erfordert vielfältige Kompetenzen. Mit Ailog profitieren Sie von einer 100 % französischen Lösung:

• Hosting bei OVH Cloud in Gravelines und Roubaix
• Daten werden niemals außerhalb der EU übertragen
• Mistral-Modelle für die Generierung
• Self-hosted Embeddings ohne externen API-Aufruf
• Native DSGVO-Konformität mit bereitgestelltem DPA
• Deutschsprachiger Support und Dokumentation

Testen Sie Ailog kostenlos und setzen Sie einen souveränen RAG ganz beruhigt ein.