Sensible Daten : Filtern und schützen

Die RAG-Systeme verarbeiten große Datenmengen, von denen einige sensibel sein können. Dieser Leitfaden stellt Techniken vor, um diese kritischen Informationen entlang der gesamten RAG-Pipeline zu identifizieren, zu filtern und zu schützen.

Voraussetzungen : Lesen Sie zuerst die RAG-Grundlagen und unseren Leitfaden zu Sicherheit und Compliance im RAG.

Was sind sensible Daten ?

Definition und Kategorien

Sensible Daten im RAG-Kontext lassen sich in mehrere Kategorien unterteilen:

Angriffspunkte in einer RAG-Pipeline

┌─────────────────────────────────────────────────────────────────┐
│                    RAG-PIPELINE - ANGRIFFSPUNKTE                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. INGESTION        2. SPEICHERUNG      3. RETRIEVAL          │
│  ┌─────────┐         ┌─────────┐         ┌─────────┐          │
│  │Rohdaten-│ ──────► │Embeddings│ ──────► │Relevante│          │
│  │Dokumente│         │Vector DB │         │ Chunks  │          │
│  └────┬────┘         └────┬────┘         └────┬────┘          │
│       │                   │                   │                │
│   RISIKO:              RISIKO:             RISIKO:             │
│   PII in ungefil-      Leck durch          Exposition          │
│   terten Docs          Embedding-          im Kontext          │
│                        Inversion                                │
│                                                                 │
│  4. GENERIERUNG      5. OUTPUT           6. LOGS               │
│  ┌─────────┐         ┌─────────┐         ┌─────────┐          │
│  │   LLM   │ ──────► │Endgültige│ ──────► │Sitzungs-│          │
│  │ Prompt  │         │ Antwort │         │verlauf  │          │
│  └────┬────┘         └────┬────┘         └────┬────┘          │
│       │                   │                   │                │
│   RISIKO:              RISIKO:             RISIKO:             │
│   Prompt               Halluzination       Unverschlüsselte    │
│   Injection            von PII             Speicherung         │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Erkennung sensibler Daten

Erkennung mittels regulärer Ausdrücke

DEVELOPERpython
import re
from typing import Dict, List, Tuple, Any
from dataclasses import dataclass
from enum import Enum

class SensitiveDataType(Enum):
    """Erkennbare Typen sensibler Daten."""
    EMAIL = "email"
    PHONE_FR = "phone_fr"
    PHONE_INTL = "phone_intl"
    CREDIT_CARD = "credit_card"
    IBAN = "iban"
    SSN_FR = "ssn_fr"  # Französische Sozialversicherungsnummer
    SSN_US = "ssn_us"
    PASSPORT = "passport"
    IP_ADDRESS = "ip_address"
    API_KEY = "api_key"
    PASSWORD = "password"
    DATE_OF_BIRTH = "date_of_birth"
    ADDRESS = "address"

@dataclass
class DetectionResult:
    """Ergebnis der Erkennung sensibler Daten."""
    data_type: SensitiveDataType
    value: str
    start_pos: int
    end_pos: int
    confidence: float
    context: str

class RegexSensitiveDetector:
    """Detektor für sensible Daten basierend auf Regex."""

    PATTERNS = {
        SensitiveDataType.EMAIL: {
            "pattern": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            "confidence": 0.95
        },
        SensitiveDataType.PHONE_FR: {
            "pattern": r'\b(?:(?:\+33|0033|0)\s?[1-9])(?:[\s.-]?\d{2}){4}\b',
            "confidence": 0.90
        },
        SensitiveDataType.PHONE_INTL: {
            "pattern": r'\b\+?[1-9]\d{1,14}\b',
            "confidence": 0.70
        },
        SensitiveDataType.CREDIT_CARD: {
            "pattern": r'\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14}|3[47][0-9]{13}|6(?:011|5[0-9]{2})[0-9]{12})\b',
            "confidence": 0.98
        },
        SensitiveDataType.IBAN: {
            "pattern": r'\b[A-Z]{2}[0-9]{2}(?:\s?[A-Z0-9]{4}){4,7}\b',
            "confidence": 0.95
        },
        SensitiveDataType.SSN_FR: {
            "pattern": r'\b[12]\s?[0-9]{2}\s?(?:0[1-9]|1[0-2]|[2-9][0-9])\s?(?:0[1-9]|[1-8][0-9]|9[0-5]|2[AB])\s?[0-9]{3}\s?[0-9]{3}\s?(?:[0-9]{2})?\b',
            "confidence": 0.95
        },
        SensitiveDataType.SSN_US: {
            "pattern": r'\b(?!000|666|9\d{2})\d{3}[-\s]?(?!00)\d{2}[-\s]?(?!0000)\d{4}\b',
            "confidence": 0.90
        },
        SensitiveDataType.IP_ADDRESS: {
            "pattern": r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b',
            "confidence": 0.99
        },
        SensitiveDataType.API_KEY: {
            "pattern": r'\b(?:sk|pk|api|key|token|secret|auth)[_-]?[A-Za-z0-9]{20,}\b',
            "confidence": 0.85
        },
        SensitiveDataType.DATE_OF_BIRTH: {
            "pattern": r'\b(?:0[1-9]|[12][0-9]|3[01])[-/.](?:0[1-9]|1[0-2])[-/.](?:19|20)\d{2}\b',
            "confidence": 0.80
        }
    }

    def __init__(self, enabled_types: List[SensitiveDataType] = None):
        """Initialisiert den Detektor mit den zu erkennenden Typen."""
        self.enabled_types = enabled_types or list(SensitiveDataType)
        self._compile_patterns()

    def _compile_patterns(self):
        """Kompiliert die Regex-Patterns für bessere Performance."""
        self.compiled_patterns = {}
        for data_type, config in self.PATTERNS.items():
            if data_type in self.enabled_types:
                self.compiled_patterns[data_type] = {
                    "regex": re.compile(config["pattern"], re.IGNORECASE),
                    "confidence": config["confidence"]
                }

    def detect(self, text: str, context_chars: int = 50) -> List[DetectionResult]:
        """Erkennt sensible Daten in einem Text."""
        results = []

        for data_type, config in self.compiled_patterns.items():
            for match in config["regex"].finditer(text):
                # Extrahiere den Kontext rund um die Erkennung
                start_ctx = max(0, match.start() - context_chars)
                end_ctx = min(len(text), match.end() + context_chars)

                results.append(DetectionResult(
                    data_type=data_type,
                    value=match.group(),
                    start_pos=match.start(),
                    end_pos=match.end(),
                    confidence=config["confidence"],
                    context=text[start_ctx:end_ctx]
                ))

        return results

    def detect_in_documents(
        self,
        documents: List[Dict[str, Any]]
    ) -> Dict[str, List[DetectionResult]]:
        """Erkennt sensible Daten in einer Liste von Dokumenten."""
        results = {}

        for doc in documents:
            doc_id = doc.get("id", doc.get("_id", str(hash(str(doc)))))
            content = doc.get("content", doc.get("text", ""))

            detections = self.detect(content)
            if detections:
                results[doc_id] = detections

        return results

Erkennung mittels Machine Learning (NER)

DEVELOPERpython
from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification
from typing import List, Dict, Any
import torch

class NERSensitiveDetector:
    """Detektor für sensible Daten basierend auf NER (Named Entity Recognition)."""

    # Mapping der NER-Entitäten zu unseren Typen
    ENTITY_MAPPING = {
        "PER": SensitiveDataType.EMAIL,  # Personen -> potenziell sensibel
        "LOC": SensitiveDataType.ADDRESS,
        "ORG": None,  # Organisationen nicht immer sensibel
        "MISC": None
    }

    def __init__(self, model_name: str = "Jean-Baptiste/camembert-ner"):
        """
        Initialisiert den NER-Detektor.

        Args:
            model_name: Zu verwendendes NER-Modell (camembert für Französisch)
        """
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForTokenClassification.from_pretrained(model_name)
        self.ner_pipeline = pipeline(
            "ner",
            model=self.model,
            tokenizer=self.tokenizer,
            aggregation_strategy="simple"
        )

    def detect(self, text: str, min_confidence: float = 0.8) -> List[Dict[str, Any]]:
        """Erkennt benannte Entitäten, die sensibel sein könnten."""
        entities = self.ner_pipeline(text)

        sensitive_entities = []
        for entity in entities:
            if entity["score"] >= min_confidence:
                entity_type = entity["entity_group"]

                # Prüfen, ob es sich um eine sensible Entität handelt
                if entity_type in ["PER", "LOC"]:  # Personen und Orte
                    sensitive_entities.append({
                        "text": entity["word"],
                        "type": entity_type,
                        "confidence": entity["score"],
                        "start": entity["start"],
                        "end": entity["end"],
                        "is_sensitive": True
                    })

        return sensitive_entities

    def detect_pii_names(self, text: str) -> List[str]:
        """Erkennt speziell Personennamen (PII)."""
        entities = self.ner_pipeline(text)
        return [
            e["word"] for e in entities
            if e["entity_group"] == "PER" and e["score"] > 0.85
        ]


class HybridSensitiveDetector:
    """Kombiniert Regex- und NER-Erkennung für bessere Abdeckung."""

    def __init__(self):
        self.regex_detector = RegexSensitiveDetector()
        self.ner_detector = NERSensitiveDetector()

    def detect(self, text: str) -> Dict[str, Any]:
        """Hybride Erkennung sensibler Daten."""
        # Erkennung per Regex (schnell, präzise für bekannte Patterns)
        regex_results = self.regex_detector.detect(text)

        # Erkennung per NER (für Namen, Orte, Organisationen)
        ner_results = self.ner_detector.detect(text)

        # Zusammenführen und Duplikate entfernen
        all_detections = self._merge_results(regex_results, ner_results)

        return {
            "total_detections": len(all_detections),
            "by_type": self._group_by_type(all_detections),
            "detections": all_detections,
            "risk_level": self._calculate_risk_level(all_detections)
        }

    def _calculate_risk_level(self, detections: List) -> str:
        """Berechnet das allgemeine Risikoniveau."""
        if not detections:
            return "low"

        high_risk_types = {
            SensitiveDataType.CREDIT_CARD,
            SensitiveDataType.SSN_FR,
            SensitiveDataType.SSN_US,
            SensitiveDataType.API_KEY
        }

        for detection in detections:
            if hasattr(detection, 'data_type') and detection.data_type in high_risk_types:
                return "critical"

        if len(detections) > 10:
            return "high"
        elif len(detections) > 3:
            return "medium"
        return "low"

Filter- und Schutztechniken

Schwärzung (Redaction)

DEVELOPERpython
from typing import Dict, List, Callable
import hashlib

class SensitiveDataRedactor:
    """Schwärzt erkannte sensible Daten."""

    REDACTION_STRATEGIES = {
        "full": lambda x, t: f"[{t.value.upper()}_REDACTED]",
        "partial": lambda x, t: x[:2] + "*" * (len(x) - 4) + x[-2:] if len(x) > 4 else "****",
        "hash": lambda x, t: hashlib.sha256(x.encode()).hexdigest()[:12],
        "placeholder": lambda x, t: f"<{t.value}>",
        "category": lambda x, t: f"[DATEN_{t.value.upper()}]"
    }

    def __init__(
        self,
        detector: RegexSensitiveDetector,
        default_strategy: str = "full"
    ):
        self.detector = detector
        self.default_strategy = default_strategy
        self.type_strategies: Dict[SensitiveDataType, str] = {}

    def set_strategy(self, data_type: SensitiveDataType, strategy: str):
        """Legt eine Schwärzungsstrategie für einen bestimmten Typ fest."""
        if strategy not in self.REDACTION_STRATEGIES:
            raise ValueError(f"Unbekannte Strategie: {strategy}")
        self.type_strategies[data_type] = strategy

    def redact(
        self,
        text: str,
        return_mapping: bool = False
    ) -> Dict[str, Any]:
        """
        Schwärzt sensible Daten in einem Text.

        Args:
            text: Zu verarbeitender Text
            return_mapping: Wenn True, wird das Mapping original -> geschwärzt zurückgegeben

        Returns:
            Dict mit geschwärztem Text und Metadaten
        """
        detections = self.detector.detect(text)

        if not detections:
            return {
                "redacted_text": text,
                "redaction_count": 0,
                "mapping": {} if return_mapping else None
            }

        # Sortieren nach absteigender Position, damit von hinten ersetzt wird
        detections_sorted = sorted(detections, key=lambda x: x.start_pos, reverse=True)

        redacted_text = text
        mapping = {}

        for detection in detections_sorted:
            strategy_name = self.type_strategies.get(
                detection.data_type,
                self.default_strategy
            )
            strategy = self.REDACTION_STRATEGIES[strategy_name]

            replacement = strategy(detection.value, detection.data_type)

            if return_mapping:
                mapping[detection.value] = replacement

            redacted_text = (
                redacted_text[:detection.start_pos] +
                replacement +
                redacted_text[detection.end_pos:]
            )

        return {
            "redacted_text": redacted_text,
            "redaction_count": len(detections),
            "types_redacted": list(set(d.data_type.value for d in detections)),
            "mapping": mapping if return_mapping else None
        }

    def redact_documents(
        self,
        documents: List[Dict[str, Any]],
        content_field: str = "content"
    ) -> List[Dict[str, Any]]:
        """Schwärzt sensible Daten in einer Liste von Dokumenten."""
        redacted_docs = []

        for doc in documents:
            redacted_doc = doc.copy()
            content = doc.get(content_field, "")

            result = self.redact(content)
            redacted_doc[content_field] = result["redacted_text"]
            redacted_doc["_redaction_metadata"] = {
                "count": result["redaction_count"],
                "types": result["types_redacted"]
            }

            redacted_docs.append(redacted_doc)

        return redacted_docs

Selektive 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 json
from typing import Dict, Any, Optional

class SelectiveEncryption:
    """Verschlüsselt sensible Daten selektiv."""

    def __init__(self, master_key: str, salt: bytes = None):
        """
        Initialisiert das Verschlüsselungssystem.

        Args:
            master_key: Master-Key zur Ableitung der Verschlüsselungsschlüssel
            salt: Salt für die Ableitung (wird erzeugt, wenn nicht angegeben)
        """
        self.salt = salt or b'ailog_sensitive_v1'
        self.cipher = self._derive_cipher(master_key)
        self.detector = RegexSensitiveDetector()

    def _derive_cipher(self, master_key: str) -> Fernet:
        """Leitet einen Fernet-Schlüssel aus dem Master-Key ab."""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=self.salt,
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(master_key.encode()))
        return Fernet(key)

    def encrypt_sensitive_fields(
        self,
        document: Dict[str, Any],
        fields_to_check: List[str]
    ) -> Dict[str, Any]:
        """
        Verschlüsselt Felder, die sensible Daten enthalten.

        Args:
            document: Zu verarbeitendes Dokument
            fields_to_check: Zu prüfende Felder

        Returns:
            Dokument mit verschlüsselten sensiblen Feldern
        """
        encrypted_doc = document.copy()
        encrypted_fields = []

        for field in fields_to_check:
            if field not in document:
                continue

            value = document[field]
            if isinstance(value, str):
                detections = self.detector.detect(value)

                if detections:
                    # Verschlüssele das gesamte Feld
                    encrypted_value = self.cipher.encrypt(value.encode()).decode()
                    encrypted_doc[field] = encrypted_value
                    encrypted_fields.append({
                        "field": field,
                        "detection_count": len(detections),
                        "types": [d.data_type.value for d in detections]
                    })

        encrypted_doc["_encryption_metadata"] = {
            "encrypted_fields": encrypted_fields,
            "algorithm": "Fernet (AES-128-CBC)",
            "encrypted_at": datetime.utcnow().isoformat()
        }

        return encrypted_doc

    def decrypt_field(self, encrypted_value: str) -> str:
        """Entschlüsselt ein verschlüsseltes Feld."""
        return self.cipher.decrypt(encrypted_value.encode()).decode()

    def encrypt_inline(self, text: str) -> Dict[str, Any]:
        """
        Verschlüsselt sensible Daten inline in einem Text.

        Nützlich, um den Kontext zu bewahren und gleichzeitig Daten zu schützen.
        """
        detections = self.detector.detect(text)

        if not detections:
            return {"text": text, "encrypted_values": {}}

        detections_sorted = sorted(detections, key=lambda x: x.start_pos, reverse=True)

        encrypted_text = text
        encrypted_mapping = {}

        for i, detection in enumerate(detections_sorted):
            placeholder = f"[[ENCRYPTED_{i}]]"
            encrypted_value = self.cipher.encrypt(detection.value.encode()).decode()
            encrypted_mapping[placeholder] = {
                "encrypted": encrypted_value,
                "type": detection.data_type.value
            }

            encrypted_text = (
                encrypted_text[:detection.start_pos] +
                placeholder +
                encrypted_text[detection.end_pos:]
            )

        return {
            "text": encrypted_text,
            "encrypted_values": encrypted_mapping
        }

Anonymisierung (K-Anonymität)

DEVELOPERpython
from typing import List, Dict, Any
from collections import defaultdict
import random

class DataAnonymizer:
    """Anonymisiert Daten, um K-Anonymität zu gewährleisten."""

    def __init__(self, k: int = 5):
        """
        Initialisiert den Anonymisierer.

        Args:
            k: Grad der K-Anonymität (jeder Datensatz muss
               von mindestens k-1 anderen nicht unterscheidbar sein)
        """
        self.k = k
        self.generalization_rules = self._default_generalization_rules()

    def _default_generalization_rules(self) -> Dict[str, Callable]:
        """Standard-Regeln zur Generalisierung."""
        return {
            "age": lambda x: f"{(int(x) // 10) * 10}-{(int(x) // 10) * 10 + 9}",
            "zip_code": lambda x: x[:3] + "**" if len(x) >= 3 else "***",
            "date": lambda x: x[:7] if len(x) >= 7 else x[:4],  # YYYY-MM
            "city": lambda x: "Region " + self._get_region(x),
            "salary": lambda x: self._salary_range(float(x))
        }

    def _salary_range(self, salary: float) -> str:
        """Generiert eine Lohnklasse als Bereich."""
        ranges = [
            (0, 25000, "< 25k"),
            (25000, 35000, "25k-35k"),
            (35000, 50000, "35k-50k"),
            (50000, 75000, "50k-75k"),
            (75000, 100000, "75k-100k"),
            (100000, float('inf'), "> 100k")
        ]
        for low, high, label in ranges:
            if low <= salary < high:
                return label
        return "Nicht angegeben"

    def generalize(
        self,
        data: List[Dict[str, Any]],
        quasi_identifiers: List[str]
    ) -> List[Dict[str, Any]]:
        """
        Generalisiert Quasi-Identifier, um K-Anonymität zu erreichen.

        Args:
            data: Liste von Datensätzen
            quasi_identifiers: Felder, die in Kombination identifizieren könnten

        Returns:
            Anonymisierte Daten
        """
        anonymized = []

        for record in data:
            anon_record = record.copy()

            for qi in quasi_identifiers:
                if qi in record and qi in self.generalization_rules:
                    try:
                        anon_record[qi] = self.generalization_rules[qi](record[qi])
                    except (ValueError, TypeError):
                        anon_record[qi] = "[GENERALIZED]"

            anonymized.append(anon_record)

        # Überprüfe K-Anonymität
        if not self._verify_k_anonymity(anonymized, quasi_identifiers):
            # Wende ggf. stärkere Generalisierung an
            anonymized = self._increase_generalization(anonymized, quasi_identifiers)

        return anonymized

    def _verify_k_anonymity(
        self,
        data: List[Dict[str, Any]],
        quasi_identifiers: List[str]
    ) -> bool:
        """Überprüft, ob die Daten K-Anonymität erfüllen."""
        equivalence_classes = defaultdict(list)

        for record in data:
            # Erzeuge einen Schlüssel basierend auf den Quasi-Identifiern
            key = tuple(str(record.get(qi, "")) for qi in quasi_identifiers)
            equivalence_classes[key].append(record)

        # Prüfe, dass jede Klasse mindestens k Elemente hat
        return all(len(records) >= self.k for records in equivalence_classes.values())

    def suppress_outliers(
        self,
        data: List[Dict[str, Any]],
        quasi_identifiers: List[str]
    ) -> List[Dict[str, Any]]:
        """Entfernt Datensätze, die K-Anonymität verletzen."""
        equivalence_classes = defaultdict(list)

        for record in data:
            key = tuple(str(record.get(qi, "")) for qi in quasi_identifiers)
            equivalence_classes[key].append(record)

        # Behalte nur Klassen mit mindestens k Elementen
        valid_data = []
        for key, records in equivalence_classes.items():
            if len(records) >= self.k:
                valid_data.extend(records)

        return valid_data

Schutz in der RAG-Pipeline

Filterung bei der Ingestion

DEVELOPERpython
class SecureDocumentIngestion:
    """Sichere Ingest-Pipeline mit Filterung sensibler Daten."""

    def __init__(
        self,
        vector_store,
        detector: HybridSensitiveDetector,
        redactor: SensitiveDataRedactor,
        policy: str = "redact"  # "redact", "reject", "encrypt", "warn"
    ):
        self.vector_store = vector_store
        self.detector = detector
        self.redactor = redactor
        self.policy = policy

    async def ingest_document(
        self,
        document: Dict[str, Any],
        force: bool = False
    ) -> Dict[str, Any]:
        """
        Nimmt ein Dokument auf und wendet Sicherheitsrichtlinien an.

        Args:
            document: Zu ingestierendes Dokument
            force: Wenn True, werden Warnungen ignoriert

        Returns:
            Ergebnis der Ingestion
        """
        content = document.get("content", "")

        # Schritt 1: Erkennung
        detection_result = self.detector.detect(content)

        if detection_result["total_detections"] == 0:
            # Keine sensiblen Daten, normale Ingestion
            return await self._ingest_clean(document)

        # Schritt 2: Richtlinie anwenden
        if self.policy == "reject":
            return {
                "status": "rejected",
                "reason": "Sensible Daten erkannt",
                "detections": detection_result["total_detections"],
                "risk_level": detection_result["risk_level"]
            }

        elif self.policy == "redact":
            redacted = self.redactor.redact(content)
            document_clean = document.copy()
            document_clean["content"] = redacted["redacted_text"]
            document_clean["_original_had_sensitive"] = True

            result = await self._ingest_clean(document_clean)
            result["redaction_applied"] = True
            result["redaction_count"] = redacted["redaction_count"]
            return result

        elif self.policy == "warn":
            if not force:
                return {
                    "status": "warning",
                    "message": "Sensible Daten erkannt, force=True verwenden, um fortzufahren",
                    "detections": detection_result
                }
            return await self._ingest_clean(document)

        elif self.policy == "encrypt":
            # Verschlüssele die sensiblen Teile
            encrypted = self.encryption.encrypt_inline(content)
            document_enc = document.copy()
            document_enc["content"] = encrypted["text"]
            document_enc["_encrypted_values"] = encrypted["encrypted_values"]

            return await self._ingest_clean(document_enc)

    async def _ingest_clean(self, document: Dict[str, Any]) -> Dict[str, Any]:
        """Ingestiert ein bereinigtes Dokument."""
        doc_id = await self.vector_store.add_document(document)
        return {
            "status": "success",
            "document_id": doc_id,
            "indexed_at": datetime.utcnow().isoformat()
        }

    async def bulk_ingest(
        self,
        documents: List[Dict[str, Any]],
        on_sensitive: str = "skip"  # "skip", "redact", "fail"
    ) -> Dict[str, Any]:
        """Massen-Ingestion mit Handling sensibler Dokumente."""
        results = {
            "successful": 0,
            "skipped": 0,
            "redacted": 0,
            "failed": 0,
            "details": []
        }

        for doc in documents:
            try:
                doc_result = await self.ingest_document(doc)

                if doc_result["status"] == "success":
                    results["successful"] += 1
                    if doc_result.get("redaction_applied"):
                        results["redacted"] += 1
                elif doc_result["status"] == "rejected":
                    results["skipped"] += 1
                else:
                    results["failed"] += 1

                results["details"].append({
                    "doc_id": doc.get("id"),
                    "result": doc_result["status"]
                })

            except Exception as e:
                results["failed"] += 1
                results["details"].append({
                    "doc_id": doc.get("id"),
                    "result": "error",
                    "error": str(e)
                })

        return results

Filterung in den LLM-Antworten

DEVELOPERpython
class OutputSanitizer:
    """Filtert sensible Daten in den LLM-Antworten."""

    def __init__(self, detector: RegexSensitiveDetector):
        self.detector = detector
        self.hallucination_patterns = self._compile_hallucination_patterns()

    def _compile_hallucination_patterns(self) -> Dict[str, re.Pattern]:
        """Patterns für potenziell halluzinierte Daten."""
        return {
            "fake_email": re.compile(r'[a-z]+\.(exemple|test|demo)@'),
            "placeholder_phone": re.compile(r'01\s?23\s?45\s?67\s?89'),
            "example_data": re.compile(r'(exemple|sample|test|dummy|fake)', re.I)
        }

    def sanitize_response(
        self,
        response: str,
        context_documents: List[str] = None
    ) -> Dict[str, Any]:
        """
        Bereinigt eine LLM-Antwort von sensiblen Daten.

        Args:
            response: LLM-Antwort
            context_documents: Quell-Dokumente zur Validierung

        Returns:
            Bereinigte Antwort mit Metadaten
        """
        # Schritt 1: Erkenne sensible Daten
        detections = self.detector.detect(response)

        # Schritt 2: Prüfen, ob die Daten aus dem Kontext stammen
        if context_documents and detections:
            hallucinated = self._identify_hallucinated_data(
                detections,
                context_documents
            )
        else:
            hallucinated = detections

        # Schritt 3: Filtere halluzinierte oder sensible Daten
        sanitized = response
        for detection in hallucinated:
            # Ersetze durch einen informativen Platzhalter
            replacement = self._get_safe_replacement(detection)
            sanitized = (
                sanitized[:detection.start_pos] +
                replacement +
                sanitized[detection.end_pos:]
            )

        # Schritt 4: Zusätzliche Prüfungen
        sanitized = self._remove_potential_hallucinations(sanitized)

        return {
            "sanitized_response": sanitized,
            "original_response": response,
            "modifications": len(detections),
            "hallucinated_data_removed": len(hallucinated),
            "is_modified": sanitized != response
        }

    def _identify_hallucinated_data(
        self,
        detections: List[DetectionResult],
        context_docs: List[str]
    ) -> List[DetectionResult]:
        """Identifiziert Daten, die nicht im Kontext vorkommen."""
        hallucinated = []
        context_combined = " ".join(context_docs).lower()

        for detection in detections:
            # Wenn die Daten nicht im Kontext vorkommen, sind sie potenziell halluziniert
            if detection.value.lower() not in context_combined:
                hallucinated.append(detection)

        return hallucinated

    def _get_safe_replacement(self, detection: DetectionResult) -> str:
        """Erzeugt eine sichere Ersetzung für eine sensible Angabe."""
        replacements = {
            SensitiveDataType.EMAIL: "[E-Mail nicht verfügbar]",
            SensitiveDataType.PHONE_FR: "[Nummer nicht verfügbar]",
            SensitiveDataType.CREDIT_CARD: "[Bankdaten verborgen]",
            SensitiveDataType.SSN_FR: "[vertrauliche Nummer]",
            SensitiveDataType.API_KEY: "[Schlüssel verborgen]"
        }
        return replacements.get(detection.data_type, "[Daten verborgen]")

    def _remove_potential_hallucinations(self, text: str) -> str:
        """Entfernt Muster häufiger halluzinierter Daten."""
        for pattern_name, pattern in self.hallucination_patterns.items():
            # Nur warnen, nicht automatisch löschen
            if pattern.search(text):
                # Log für die Analyse
                pass
        return text

Best Practices und Checkliste

Sichere Architektur

┌─────────────────────────────────────────────────────────────────┐
│                  SICHERE RAG-ARCHITEKTUR                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  DATENSCHICHT                                                  │
│  ├── Originaldokumente (verschlüsselt, eingeschränkter Zugriff)│
│  ├── Geschwärzte Dokumente (für die Indexierung)                │
│  └── Schwärzungs-Metadaten (reversibles Mapping bei Bedarf)    │
│                                                                 │
│  VERARBEITUNGSSCHICHT                                          │
│  ├── Erkennung bei der Ingestion (obligatorisch)                │
│  ├── Erkennung beim Retrieval (optional)                        │
│  └── Erkennung bei der Generierung (obligatorisch)              │
│                                                                 │
│  ZUGRIFFSSCHICHT                                                │
│  ├── Starke Authentifizierung                                   │
│  ├── Rollenbasierte Autorisierung                                │
│  └── Vollständiger Audit-Trail                                  │
│                                                                 │
│  MONITORING-SCHICHT                                             │
│  ├── Warnmeldungen bei Erkennung sensibler Daten                │
│  ├── Compliance-Dashboard                                       │
│  └── Regelmäßige Berichte                                       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Checkliste für sensible Daten

Erkennung :

• Regex-Detektor für bekannte Patterns (E-Mail, Telefon, Karte usw.)
• NER-Detektor für Namen und Orte
• Hybride Erkennung im Produktivbetrieb
• Angepasste Patterns für Geschäftsdaten

Schutz bei der Ingestion :

• Obligatorischer Scan vor der Indexierung
• Definierte Verarbeitungsrichtlinie (redact/reject/encrypt)
• Aufbewahrung der Schwärzungs-Metadaten
• Protokolle abgelehnter Dokumente

Schutz bei der Generierung :

• Filterung der LLM-Antworten
• Erkennung von PII-Halluzinationen
• Validierung gegen die Quelldokumente

Konformität :

• Dokumentation der verarbeiteten Datentypen
• Verfahren zur Beantwortung von Auskunftsersuchen
• Automatische Aufbewahrung und Löschung
• Audit-Trail der Zugriffe auf sensible Daten

Fazit

Der Schutz sensibler Daten in einem RAG-System erfordert einen mehrschichtigen Ansatz: robuste Erkennung, kontextangepasstes Filtern und kontinuierliches Monitoring. Die in diesem Leitfaden vorgestellten Techniken ermöglichen den Aufbau einer sicheren Pipeline, ohne die Nutzbarkeit Ihres Assistenten zu beeinträchtigen.

Wichtige Punkte :

1. Hybride Erkennung - Kombinieren Sie Regex und ML für maximale Abdeckung
2. Klare Richtlinie - Definieren Sie, wie jeder Datentyp behandelt wird
3. Filterung bei jedem Schritt - Ingestion, Retrieval und Generierung
4. Kontinuierlicher Audit - Überwachen und kontinuierlich verbessern

Zum Weiterlesen

• DSGVO und KI-Chatbots - Regulatorische Konformität
• Audit-Trail RAG - Nachvollziehbarkeit der Vorgänge
• Souveräne Cloud für RAG - Konformes Hosting