Données sensibles : Filtrer et protéger les informations

Les systèmes RAG manipulent des volumes importants de données, dont certaines peuvent être sensibles. Ce guide présente les techniques pour identifier, filtrer et protéger ces informations critiques tout au long du pipeline RAG.

Prérequis : Consultez d'abord les fondamentaux du RAG et notre guide sur la sécurité et conformité RAG.

Qu'est-ce qu'une donnée sensible ?

Définition et catégories

Les données sensibles dans un contexte RAG se divisent en plusieurs catégories :

Points d'exposition dans un pipeline RAG

┌─────────────────────────────────────────────────────────────────┐
│                    PIPELINE RAG - POINTS D'EXPOSITION           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. INGESTION        2. STOCKAGE         3. RETRIEVAL          │
│  ┌─────────┐         ┌─────────┐         ┌─────────┐          │
│  │Documents│ ──────► │Embeddings│ ──────► │ Chunks  │          │
│  │ bruts   │         │Vector DB │         │pertinents│          │
│  └────┬────┘         └────┬────┘         └────┬────┘          │
│       │                   │                   │                │
│   RISQUE:              RISQUE:             RISQUE:            │
│   PII dans docs        Fuite via          Exposition          │
│   non filtrés         embedding          dans contexte        │
│                       inversion                               │
│                                                                │
│  4. GÉNÉRATION       5. OUTPUT           6. LOGS              │
│  ┌─────────┐         ┌─────────┐         ┌─────────┐          │
│  │   LLM   │ ──────► │ Réponse │ ──────► │Historique│          │
│  │ prompt  │         │  finale │         │ sessions │          │
│  └────┬────┘         └────┬────┘         └────┬────┘          │
│       │                   │                   │                │
│   RISQUE:              RISQUE:             RISQUE:            │
│   Prompt               Hallucination       Stockage           │
│   injection            de PII              non chiffré        │
│                                                                │
└─────────────────────────────────────────────────────────────────┘

Détection des données sensibles

Détection par expressions régulières

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

class SensitiveDataType(Enum):
    """Types de données sensibles détectables."""
    EMAIL = "email"
    PHONE_FR = "phone_fr"
    PHONE_INTL = "phone_intl"
    CREDIT_CARD = "credit_card"
    IBAN = "iban"
    SSN_FR = "ssn_fr"  # Numéro de sécurité sociale français
    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:
    """Résultat de détection de données sensibles."""
    data_type: SensitiveDataType
    value: str
    start_pos: int
    end_pos: int
    confidence: float
    context: str

class RegexSensitiveDetector:
    """Détecteur de données sensibles basé sur les 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):
        """Initialise le détecteur avec les types à détecter."""
        self.enabled_types = enabled_types or list(SensitiveDataType)
        self._compile_patterns()

    def _compile_patterns(self):
        """Compile les patterns regex pour 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]:
        """Détecte les données sensibles dans un texte."""
        results = []

        for data_type, config in self.compiled_patterns.items():
            for match in config["regex"].finditer(text):
                # Extraire le contexte autour de la détection
                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]]:
        """Détecte les données sensibles dans une liste de documents."""
        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

Détection par Machine Learning (NER)

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

class NERSensitiveDetector:
    """Détecteur de données sensibles basé sur NER (Named Entity Recognition)."""

    # Mapping des entités NER vers nos types
    ENTITY_MAPPING = {
        "PER": SensitiveDataType.EMAIL,  # Personnes -> potentiellement sensible
        "LOC": SensitiveDataType.ADDRESS,
        "ORG": None,  # Organisations pas toujours sensibles
        "MISC": None
    }

    def __init__(self, model_name: str = "Jean-Baptiste/camembert-ner"):
        """
        Initialise le détecteur NER.

        Args:
            model_name: Modèle NER à utiliser (camembert pour français)
        """
        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]]:
        """Détecte les entités nommées sensibles."""
        entities = self.ner_pipeline(text)

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

                # Vérifier si c'est une entité sensible
                if entity_type in ["PER", "LOC"]:  # Personnes et lieux
                    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]:
        """Détecte spécifiquement les noms de personnes."""
        entities = self.ner_pipeline(text)
        return [
            e["word"] for e in entities
            if e["entity_group"] == "PER" and e["score"] > 0.85
        ]


class HybridSensitiveDetector:
    """Combine détection regex et NER pour une meilleure couverture."""

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

    def detect(self, text: str) -> Dict[str, Any]:
        """Détection hybride de données sensibles."""
        # Détection par regex (rapide, précis pour patterns connus)
        regex_results = self.regex_detector.detect(text)

        # Détection par NER (pour noms, lieux, organisations)
        ner_results = self.ner_detector.detect(text)

        # Fusionner et dédupliquer
        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:
        """Calcule le niveau de risque global."""
        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"

Techniques de filtrage et protection

Masquage (Redaction)

DEVELOPERpython
from typing import Dict, List, Callable
import hashlib

class SensitiveDataRedactor:
    """Masque les données sensibles détectées."""

    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"[DONNÉES_{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):
        """Définit une stratégie de masquage pour un type spécifique."""
        if strategy not in self.REDACTION_STRATEGIES:
            raise ValueError(f"Stratégie inconnue: {strategy}")
        self.type_strategies[data_type] = strategy

    def redact(
        self,
        text: str,
        return_mapping: bool = False
    ) -> Dict[str, Any]:
        """
        Masque les données sensibles d'un texte.

        Args:
            text: Texte à traiter
            return_mapping: Si True, retourne le mapping original -> masqué

        Returns:
            Dict avec le texte masqué et métadonnées
        """
        detections = self.detector.detect(text)

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

        # Trier par position décroissante pour remplacer de la fin au début
        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]]:
        """Masque les données sensibles d'une liste de documents."""
        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

Chiffrement sélectif

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:
    """Chiffre sélectivement les données sensibles."""

    def __init__(self, master_key: str, salt: bytes = None):
        """
        Initialise le système de chiffrement.

        Args:
            master_key: Clé maître pour dériver les clés de chiffrement
            salt: Salt pour la dérivation (généré si non fourni)
        """
        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:
        """Dérive une clé Fernet à partir de la clé maître."""
        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]:
        """
        Chiffre les champs contenant des données sensibles.

        Args:
            document: Document à traiter
            fields_to_check: Champs à analyser

        Returns:
            Document avec champs sensibles chiffrés
        """
        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:
                    # Chiffrer le champ entier
                    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:
        """Déchiffre un champ chiffré."""
        return self.cipher.decrypt(encrypted_value.encode()).decode()

    def encrypt_inline(self, text: str) -> Dict[str, Any]:
        """
        Chiffre les données sensibles inline dans un texte.

        Utile pour préserver le contexte tout en protégeant les données.
        """
        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
        }

Anonymisation (K-anonymity)

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

class DataAnonymizer:
    """Anonymise les données pour garantir k-anonymity."""

    def __init__(self, k: int = 5):
        """
        Initialise l'anonymiseur.

        Args:
            k: Niveau de k-anonymité (chaque enregistrement doit être
               indistinguable d'au moins k-1 autres)
        """
        self.k = k
        self.generalization_rules = self._default_generalization_rules()

    def _default_generalization_rules(self) -> Dict[str, Callable]:
        """Règles de généralisation par défaut."""
        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: "Région " + self._get_region(x),
            "salary": lambda x: self._salary_range(float(x))
        }

    def _salary_range(self, salary: float) -> str:
        """Généralise un salaire en fourchette."""
        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 "Non spécifié"

    def generalize(
        self,
        data: List[Dict[str, Any]],
        quasi_identifiers: List[str]
    ) -> List[Dict[str, Any]]:
        """
        Généralise les quasi-identifiants pour atteindre k-anonymité.

        Args:
            data: Liste d'enregistrements
            quasi_identifiers: Champs qui, combinés, pourraient identifier

        Returns:
            Données anonymisées
        """
        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)

        # Vérifier k-anonymité
        if not self._verify_k_anonymity(anonymized, quasi_identifiers):
            # Appliquer une généralisation plus agressive si nécessaire
            anonymized = self._increase_generalization(anonymized, quasi_identifiers)

        return anonymized

    def _verify_k_anonymity(
        self,
        data: List[Dict[str, Any]],
        quasi_identifiers: List[str]
    ) -> bool:
        """Vérifie si les données respectent k-anonymité."""
        equivalence_classes = defaultdict(list)

        for record in data:
            # Créer une clé basée sur les quasi-identifiants
            key = tuple(str(record.get(qi, "")) for qi in quasi_identifiers)
            equivalence_classes[key].append(record)

        # Vérifier que chaque classe a au moins k éléments
        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]]:
        """Supprime les enregistrements qui violent k-anonymité."""
        equivalence_classes = defaultdict(list)

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

        # Garder seulement les classes avec k+ éléments
        valid_data = []
        for key, records in equivalence_classes.items():
            if len(records) >= self.k:
                valid_data.extend(records)

        return valid_data

Protection dans le pipeline RAG

Filtrage à l'ingestion

DEVELOPERpython
class SecureDocumentIngestion:
    """Pipeline d'ingestion sécurisé avec filtrage des données sensibles."""

    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]:
        """
        Ingère un document en appliquant les politiques de sécurité.

        Args:
            document: Document à ingérer
            force: Si True, ignore les avertissements

        Returns:
            Résultat de l'ingestion
        """
        content = document.get("content", "")

        # Étape 1: Détection
        detection_result = self.detector.detect(content)

        if detection_result["total_detections"] == 0:
            # Pas de données sensibles, ingestion normale
            return await self._ingest_clean(document)

        # Étape 2: Appliquer la politique
        if self.policy == "reject":
            return {
                "status": "rejected",
                "reason": "Données sensibles détectées",
                "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": "Données sensibles détectées, utiliser force=True pour continuer",
                    "detections": detection_result
                }
            return await self._ingest_clean(document)

        elif self.policy == "encrypt":
            # Chiffrer les parties sensibles
            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]:
        """Ingère un document nettoyé."""
        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]:
        """Ingestion en masse avec gestion des documents sensibles."""
        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

Filtrage dans les réponses LLM

DEVELOPERpython
class OutputSanitizer:
    """Filtre les données sensibles dans les réponses du LLM."""

    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 de données potentiellement hallucinées."""
        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]:
        """
        Nettoie une réponse LLM des données sensibles.

        Args:
            response: Réponse du LLM
            context_documents: Documents source pour validation

        Returns:
            Réponse nettoyée avec métadonnées
        """
        # Étape 1: Détecter les données sensibles
        detections = self.detector.detect(response)

        # Étape 2: Vérifier si les données viennent du contexte
        if context_documents and detections:
            hallucinated = self._identify_hallucinated_data(
                detections,
                context_documents
            )
        else:
            hallucinated = detections

        # Étape 3: Filtrer les données hallucinées ou sensibles
        sanitized = response
        for detection in hallucinated:
            # Remplacer par un placeholder informatif
            replacement = self._get_safe_replacement(detection)
            sanitized = (
                sanitized[:detection.start_pos] +
                replacement +
                sanitized[detection.end_pos:]
            )

        # Étape 4: Vérifications supplémentaires
        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]:
        """Identifie les données qui n'apparaissent pas dans le contexte."""
        hallucinated = []
        context_combined = " ".join(context_docs).lower()

        for detection in detections:
            # Si la donnée n'est pas dans le contexte, elle est potentiellement hallucinée
            if detection.value.lower() not in context_combined:
                hallucinated.append(detection)

        return hallucinated

    def _get_safe_replacement(self, detection: DetectionResult) -> str:
        """Génère un remplacement sûr pour une donnée sensible."""
        replacements = {
            SensitiveDataType.EMAIL: "[email non disponible]",
            SensitiveDataType.PHONE_FR: "[numéro non disponible]",
            SensitiveDataType.CREDIT_CARD: "[données bancaires masquées]",
            SensitiveDataType.SSN_FR: "[numéro confidentiel]",
            SensitiveDataType.API_KEY: "[clé masquée]"
        }
        return replacements.get(detection.data_type, "[donnée masquée]")

    def _remove_potential_hallucinations(self, text: str) -> str:
        """Supprime les patterns de données hallucinées courantes."""
        for pattern_name, pattern in self.hallucination_patterns.items():
            # Juste avertir, ne pas supprimer automatiquement
            if pattern.search(text):
                # Log pour analyse
                pass
        return text

Bonnes pratiques et checklist

Architecture sécurisée

┌─────────────────────────────────────────────────────────────────┐
│                  ARCHITECTURE RAG SÉCURISÉE                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  COUCHE DONNÉES                                                │
│  ├── Documents originaux (chiffrés, accès restreint)          │
│  ├── Documents masqués (pour indexation)                       │
│  └── Métadonnées de masquage (mapping réversible si besoin)   │
│                                                                 │
│  COUCHE TRAITEMENT                                             │
│  ├── Détection à l'ingestion (obligatoire)                    │
│  ├── Détection à la retrieval (optionnel)                     │
│  └── Détection à la génération (obligatoire)                  │
│                                                                 │
│  COUCHE ACCÈS                                                  │
│  ├── Authentification forte                                    │
│  ├── Autorisation basée sur les rôles                         │
│  └── Audit trail complet                                       │
│                                                                 │
│  COUCHE MONITORING                                             │
│  ├── Alertes sur détection de données sensibles               │
│  ├── Dashboard de conformité                                   │
│  └── Rapports périodiques                                      │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Checklist de sécurité données sensibles

Détection :

• Détecteur regex pour patterns connus (email, téléphone, carte, etc.)
• Détecteur NER pour noms et lieux
• Détection hybride en production
• Patterns personnalisés pour données métier

Protection à l'ingestion :

• Scan obligatoire avant indexation
• Politique de traitement définie (redact/reject/encrypt)
• Conservation des métadonnées de masquage
• Logs des documents rejetés

Protection à la génération :

• Filtrage des réponses LLM
• Détection des hallucinations de PII
• Validation contre les documents source

Conformité :

• Documentation des types de données traitées
• Procédures de réponse aux demandes d'accès
• Rétention et purge automatiques
• Audit trail des accès aux données sensibles

Conclusion

La protection des données sensibles dans un système RAG nécessite une approche multicouche : détection robuste, filtrage adapté au contexte, et monitoring continu. Les techniques présentées dans ce guide vous permettent de construire un pipeline sécurisé tout en préservant l'utilité de votre assistant.

Points clés :

1. Détection hybride - Combinez regex et ML pour une couverture maximale
2. Politique claire - Définissez comment traiter chaque type de donnée
3. Filtrage à chaque étape - Ingestion, retrieval et génération
4. Audit continu - Surveillez et améliorez constamment

Pour aller plus loin

• RGPD et chatbots IA - Conformité réglementaire
• Audit trail RAG - Traçabilité des opérations
• Cloud souverain pour le RAG - Hébergement conforme