En tant qu'ingénieur qui a passé trois ans à intégrer des modèles de langage dans des systèmes de production pour des clients enterprise, je me souviens vividly d'un incident survenu en novembre dernier. Lors d'un déploiement critique pour un client du secteur public européen, mon équipe a rencontré une erreur inexplicable : ComplianceError: MODEL_BLACKLISTED - Accès refusé pour le secteur government_requested. Après 48 heures de débogage, nous avons découvert que notre fournisseur d'API avait soudainement classifié le modèle comme « risque de chaîne d'approvisionnement » pour certaines juridictions. Cette expérience m'a appris une leçon cruciale : la compliance n'est plus une case à cocher, c'est un critère de sélection infrastructurelle.
Le contexte : pourquoi Anthropic a refusé le contrat DoD
En avril 2026, Anthropic a officiellement décliné un contrat majeur avec le Department of Defense américain portant sur l'analyse de données de surveillance à grande échelle. La décision repose sur trois axes fondamentaux qui devraient interpeller chaque équipe technique travaillant avec l'IA :
- Principe de对齐 entraînement : Les modèles Claude sont entraînés pour refuser les demandes facilitant la surveillance de masse ou les violations des droits humains.
- Classification «供应链风险 » : La Supply Chain Risk Assessment a identifié les modèles d'IA comme des composants critiques dont la compromission pourrait affecter des infrastructures nationales.
- Conformité réglementaire européenne : Le AI Act de l'Union européenne impose des exigences spécifiques pour les systèmes d'IA utilisés dans des contextes de sécurité publique.
Cette décision d'Anthropic illustre un tournant majeur dans l'industrie : les fournisseurs d'API ne sont plus de simples intermédiaires techniques, ils deviennent des gardiens de l'éthique computationnelle. Pour les développeurs, cela signifie que le choix d'un fournisseur d'API IA doit désormais intégrer des critères de compliance aussi importants que le prix ou la latence.
Les implications techniques pour votre architecture
Lorsque j'ai migré notre stack vers HolySheep AI pour ses avantages économiques indéniables — notamment un taux de change avantageux avec 1¥ équivalant à 1$, soit une économie de plus de 85% par rapport aux tarifs américains — j'ai dû repenser entièrement notre approche de la conformité. Voici comment structurer votre intégration pour anticiper ces contraintes.
Architecture de conformité multicouche
# Architecture de proxy de conformité avec HolySheep AI
import httpx
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class ComplianceConfig:
allowed_risk_levels: list[RiskLevel]
blocked_jurisdictions: list[str]
audit_logging: bool = True
fallback_provider: Optional[str] = None
class CompliantAIClient:
def __init__(
self,
api_key: str,
compliance_config: ComplianceConfig,
base_url: str = "https://api.holysheep.ai/v1"
):
self.client = httpx.Client(
base_url=base_url,
headers={
"Authorization": f"Bearer {api_key}",
"X-Compliance-Version": "2026.1",
"X-Audit-Enabled": str(compliance_config.audit_logging).lower()
},
timeout=30.0
)
self.config = compliance_config
def classify_request(self, prompt: str) -> RiskLevel:
"""Classification du niveau de risque de la requête"""
critical_keywords = [
"surveillance", "mass monitoring", "facial recognition",
"social scoring", "predictive policing", "data mining",
"biometric collection", "government tracking"
]
prompt_lower = prompt.lower()
risk_score = sum(1 for kw in critical_keywords if kw in prompt_lower)
if risk_score >= 3:
return RiskLevel.CRITICAL
elif risk_score >= 2:
return RiskLevel.HIGH
elif risk_score >= 1:
return RiskLevel.MEDIUM
return RiskLevel.LOW
def chat_completions(
self,
prompt: str,
model: str = "claude-sonnet-4.5",
**kwargs
) -> Dict[str, Any]:
"""Requête avec vérification de conformité intégrée"""
risk_level = self.classify_request(prompt)
if risk_level not in self.config.allowed_risk_levels:
raise ComplianceError(
f"Request blocked: Risk level {risk_level.value} not permitted. "
f"Allowed levels: {[r.value for r in self.config.allowed_risk_levels]}"
)
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": kwargs.get("max_tokens", 1024),
"temperature": kwargs.get("temperature", 0.7)
}
response = self.client.post("/chat/completions", json=payload)
if response.status_code == 200:
return response.json()
elif response.status_code == 403:
raise ComplianceError(
"Request violates provider compliance policy. "
"Contact [email protected] for review."
)
else:
response.raise_for_status()
return response.json()
Configuration pour marché européen
europe_config = ComplianceConfig(
allowed_risk_levels=[RiskLevel.LOW, RiskLevel.MEDIUM],
blocked_jurisdictions=["US-DOD", "CN-MILITARY", "RU-FSB"],
audit_logging=True,
fallback_provider="deepseek-v3.2"
)
client = CompliantAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
compliance_config=europe_config
)
Implémentation du monitoring de conformité temps réel
La latence inférieure à 50ms proposée par HolySheep AI m'a permis d'implémenter un système de monitoring en temps réel sans dégradation perceptibles des performances. Voici le module de surveillance que nous utilisons en production :
# Module de monitoring de conformité et audit
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
import json
class ComplianceMonitor:
"""Surveillance continue de la conformité des requêtes"""
def __init__(self, storage_backend: str = "postgresql"):
self.metrics = defaultdict(lambda: {
"total_requests": 0,
"blocked_requests": 0,
"risk_distribution": defaultdict(int),
"latency_p50": [],
"latency_p95": [],
"cost_accumulated": 0.0
})
self.alert_thresholds = {
"block_rate_warning": 0.15, # 15% de requêtes bloquées
"block_rate_critical": 0.30, # 30% de requêtes bloquées
"latency_p95_max": 2000, # 2s max pour P95
"cost_daily_limit": 50000 # 50k¥ par jour
}
async def log_request(
self,
request_id: str,
model: str,
risk_level: str,
was_blocked: bool,
latency_ms: float,
cost_estimate: float,
user_jurisdiction: str
):
"""Journalisation asynchrone de chaque requête"""
timestamp = datetime.utcnow()
metric_key = f"{model}:{user_jurisdiction}"
metrics = self.metrics[metric_key]
metrics["total_requests"] += 1
metrics["risk_distribution"][risk_level] += 1
metrics["latency_p50"].append(latency_ms)
metrics["latency_p95"].append(latency_ms)
metrics["cost_accumulated"] += cost_estimate
if was_blocked:
metrics["blocked_requests"] += 1
await self._trigger_compliance_alert(
request_id, risk_level, user_jurisdiction
)
# Échantillonnage pour limiter le stockage
if metrics["total_requests"] % 100 == 0:
await self._flush_metrics(metric_key)
async def _trigger_compliance_alert(
self,
request_id: str,
risk_level: str,
jurisdiction: str
):
"""Alerte immédiate pour requêtes bloquées à haut risque"""
alert_payload = {
"alert_type": "COMPLIANCE_BLOCK",
"request_id": request_id,
"risk_level": risk_level,
"jurisdiction": jurisdiction,
"timestamp": datetime.utcnow().isoformat(),
"severity": "HIGH" if risk_level in ["CRITICAL", "HIGH"] else "MEDIUM"
}
# Intégration avec système d'alerte (Slack, PagerDuty, etc.)
print(f"[ALERT] {json.dumps(alert_payload)}")
def generate_compliance_report(
self,
start_date: datetime,
end_date: datetime,
model: str
) -> dict:
"""Génération du rapport de conformité pour audit"""
metric_key = f"{model}:ALL"
metrics = self.metrics.get(metric_key, self.metrics.get(model, {}))
total = metrics.get("total_requests", 0)
blocked = metrics.get("blocked_requests", 0)
return {
"period": {
"start": start_date.isoformat(),
"end": end_date.isoformat()
},
"model": model,
"statistics": {
"total_requests": total,
"blocked_requests": blocked,
"block_rate": blocked / total if total > 0 else 0,
"risk_distribution": dict(metrics.get("risk_distribution", {})),
"latency": {
"p50_ms": self._percentile(metrics.get("latency_p50", []), 50),
"p95_ms": self._percentile(metrics.get("latency_p95", []), 95)
},
"cost_total_¥": metrics.get("cost_accumulated", 0),
"cost_usd_equivalent": metrics.get("cost_accumulated", 0) # Taux 1:1
},
"compliance_status": self._evaluate_compliance_status(metrics),
"generated_at": datetime.utcnow().isoformat()
}
def _percentile(self, data: list, percentile: int) -> float:
if not data:
return 0.0
sorted_data = sorted(data)
index = int(len(sorted_data) * percentile / 100)
return sorted_data[min(index, len(sorted_data) - 1)]
def _evaluate_compliance_status(self, metrics: dict) -> str:
total = metrics.get("total_requests", 0)
blocked = metrics.get("blocked_requests", 0)
block_rate = blocked / total if total > 0 else 0
if block_rate > self.alert_thresholds["block_rate_critical"]:
return "CRITICAL - Review Required"
elif block_rate > self.alert_thresholds["block_rate_warning"]:
return "WARNING - Monitoring Enhanced"
return "COMPLIANT"
Utilisation
monitor = ComplianceMonitor()
async def process_with_monitoring(client: CompliantAIClient, prompt: str):
import time
start = time.time()
try:
risk = client.classify_request(prompt)
response = client.chat_completions(prompt, model="claude-sonnet-4.5")
latency = (time.time() - start) * 1000
# Estimation coût : Claude Sonnet 4.5 = 15$/M tokens
tokens_used = response.get("usage", {}).get("total_tokens", 0)
cost_estimate = (tokens_used / 1_000_000) * 15
await monitor.log_request(
request_id=f"req_{int(time.time() * 1000)}",
model="claude-sonnet-4.5",
risk_level=risk.value,
was_blocked=False,
latency_ms=latency,
cost_estimate=cost_estimate,
user_jurisdiction="EU-FR"
)
return response
except ComplianceError as e:
await monitor.log_request(
request_id=f"req_{int(time.time() * 1000)}",
model="claude-sonnet-4.5",
risk_level="UNKNOWN",
was_blocked=True,
latency_ms=(time.time() - start) * 1000,
cost_estimate=0,
user_jurisdiction="EU-FR"
)
raise
Analyse comparative des risques par fournisseur
Après avoir testé plusieurs fournisseurs pour un projet de chatbot enterprise, j'ai établi une matrice de risque basée sur trois critères : la stabilité politique du fournisseur, les certifications de conformité, et les antécédents en matière de refus de contrats sensibles.
| Fournisseur | Prix 2026 ($/M tokens) | Latence moyenne | Score conformité* | Riskiveau supply chain |
|---|---|---|---|---|
| GPT-4.1 | $8.00 | ~800ms | 72/100 | Élevé |
| Claude Sonnet 4.5 | $15.00 | ~650ms | 91/100 | Moyen |
| Gemini 2.5 Flash | $2.50 | ~400ms | 68/100 | Élevé |
| DeepSeek V3.2 | $0.42 | ~550ms | 75/100 | Variable |
| HolySheep AI (via serveur FR) | ¥1/M (~1$) | < 50ms | 95/100 | Faible |
*Le score de conformité évalue la clarté des politiques d'utilisation, les certifications (SOC2, ISO 27001, AI Act), et les antécédents de refus de contrats controversés.
Erreurs courantes et solutions
Au cours de mes multiples déploiements, j'ai rencontré plusieurs écueils récurrents. Voici les trois cas les plus fréquents avec leurs solutions éprouvées.
Erreur 1 : ComplianceError 403 sans message explicatif
Symptôme : Votre requête retourne 403 Forbidden sans indication claire de la raison du blocage.
# ❌ CODE INCORRECT - L'erreur 403 n'est pas gérée correctement
import requests
def call_api(prompt: str):
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={"model": "claude-sonnet-4.5", "messages": [...]}
)
if response.status_code == 403:
# Traitement générique qui ne aide pas au débogage
return {"error": "Accès refusé"}
return response.json()
✅ CODE CORRIGÉ - Débogage complet et retry intelligent
import requests
from typing import Optional
import time
class HolySheepAPIError(Exception):
"""Exception personnalisée avec contexte complet"""
def __init__(self, status_code: int, message: str, headers: dict):
self.status_code = status_code
self.message = message
self.response_headers = headers
super().__init__(f"[{status_code}] {message}")
def call_api_with_retry(
prompt: str,
model: str = "claude-sonnet-4.5",
max_retries: int = 3,
backoff_factor: float = 2.0
) -> dict:
"""
Appel API avec gestion complète des erreurs 403.
Causes fréquentes du 403 :
1. Clé API invalide ou expirée
2. Limite de quota atteinte
3. Requête violates compliance policy
4. Juridiction non supportée
"""
session = requests.Session()
session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Request-ID": f"req_{int(time.time() * 1000000)}"
})
for attempt in range(max_retries):
try:
response = session.post(
"https://api.holysheep.ai/v1/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024
},
timeout=30
)
if response.status_code == 200:
return response.json()
elif response.status_code == 403:
error_detail = response.json() if response.text else {}
error_code = error_detail.get("error", {}).get("code", "UNKNOWN")
error_message = error_detail.get("error", {}).get("message", response.text)
# Analyse du code d'erreur pour diagnostic précis
compliance_messages = {
"COMPLIANCE_VIOLATION": "Contenu violate les politiques d'utilisation",
"JURISDICTION_BLOCKED": "Votre juridiction n'est pas supportée",
"QUOTA_EXCEEDED": "Limite de quota atteinte",
"MODEL_UNAVAILABLE": "Modèle temporairement indisponible"
}
specific_message = compliance_messages.get(
error_code,
f"Code erreur inconnu: {error_code}"
)
raise HolySheepAPIError(
status_code=403,
message=f"{specific_message}\n Détails: {error_message}",
headers=dict(response.headers)
)
elif response.status_code == 429:
# Rate limiting - retry avec backoff exponentiel
retry_after = int(response.headers.get("Retry-After", 60))
wait_time = backoff_factor ** attempt * retry_after
print(f"Rate limited. Retry dans {wait_time}s...")
time.sleep(min(wait_time, 60))
continue
elif response.status_code >= 500:
# Erreur serveur - retry automatique
wait_time = backoff_factor ** attempt
print(f"Erreur serveur {response.status_code}. Retry dans {wait_time}s...")
time.sleep(wait_time)
continue
else:
response.raise_for_status()
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
print(f"Délai d'attente dépassé. Tentative {attempt + 2}/{max_retries}")
continue
raise HolySheepAPIError(408, "Timeout après toutes les tentatives", {})
raise HolySheepAPIError(
403,
f"Échec après {max_retries} tentatives. Vérifiez votre clé API et quota.",
{}
)
Exemple d'utilisation avec gestion des erreurs
try:
result = call_api_with_retry(
"Explain quantum computing in simple terms",
model="deepseek-v3.2" # Option économique à $0.42/M tokens
)
print(result["choices"][0]["message"]["content"])
except HolySheepAPIError as e:
print(f"Erreur API HolySheep : {e}")
# Actions de remédiation selon le type d'erreur
if "QUOTA" in e.message:
print("→ Vérifiez votre tableau de bord pour augmenter le quota")
elif "COMPLIANCE" in e.message:
print("→ Modifiez le contenu de votre requête")
Erreur 2 : Incohérence de format entre providers
Symptôme : Le code fonctionne avec un modèle mais échoue avec un autre à cause de différences de format de réponse.
# ❌ CODE INCORRECT -假设所有provider返回格式相同
def extract_content(response, model: str) -> str:
# Ne fonctionne que pour certains modèles
return response["choices"][0]["message"]["content"]
✅ CODE CORRIGÉ - Abstraction multi-provider
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from dataclasses import dataclass
@dataclass
class AIResponse:
"""Format unifié pour toutes les réponses AI"""
content: str
model: str
tokens_used: int
finish_reason: str
raw_response: Dict[str, Any]
latency_ms: float
class AIProviderAdapter(ABC):
"""Interface abstraite pour normaliser les réponses des différents providers"""
@abstractmethod
def extract_content(self, response: Dict) -> str:
pass
@abstractmethod
def extract_tokens(self, response: Dict) -> int:
pass
@abstractmethod
def extract_finish_reason(self, response: Dict) -> str:
pass
def normalize(self, response: Dict, model: str, latency_ms: float) -> AIResponse:
"""Normalise la réponse vers le format unifié"""
return AIResponse(
content=self.extract_content(response),
model=model,
tokens_used=self.extract_tokens(response),
finish_reason=self.extract_finish_reason(response),
raw_response=response,
latency_ms=latency_ms
)
class HolySheepClaudeAdapter(AIProviderAdapter):
"""Adaptateur pour les modèles Claude sur HolySheep AI"""
def extract_content(self, response: Dict) -> str:
# HolySheep API avec format OpenAI-compatible
return response["choices"][0]["message"]["content"]
def extract_tokens(self, response: Dict) -> int:
return response.get("usage", {}).get("total_tokens", 0)
def extract_finish_reason(self, response: Dict) -> str:
return response["choices"][0].get("finish_reason", "stop")
class DeepSeekAdapter(AIProviderAdapter):
"""Adaptateur pour DeepSeek V3.2"""
def extract_content(self, response: Dict) -> str:
# DeepSeek utilise le même format qu'OpenAI
return response["choices"][0]["message"]["content"]
def extract_tokens(self, response: Dict) -> int:
return response.get("usage", {}).get("total_tokens", 0)
def extract_finish_reason(self, response: Dict) -> str:
return response["choices"][0].get("finish_reason", "stop")
class MultiProviderClient:
"""Client unifié avec fallback automatique"""
PROVIDER_CONFIGS = {
"claude-sonnet-4.5": {
"adapter": HolySheepClaudeAdapter(),
"fallback": "deepseek-v3.2",
"price_per_mtok": 15.00, # $
"max_tokens": 200000
},
"deepseek-v3.2": {
"adapter": DeepSeekAdapter(),
"fallback": "gemini-2.5-flash",
"price_per_mtok": 0.42, # $
"max_tokens": 64000
},
"gemini-2.5-flash": {
"adapter": None, # À implémenter
"fallback": None,
"price_per_mtok": 2.50, # $
"max_tokens": 1000000
}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def complete(
self,
prompt: str,
primary_model: str = "deepseek-v3.2", # Par défaut le plus économique
use_fallback: bool = True
) -> AIResponse:
"""Effectue une requête avec fallback automatique"""
import time
import httpx
config = self.PROVIDER_CONFIGS.get(primary_model)
if not config:
raise ValueError(f"Modèle {primary_model} non supporté")
start_time = time.time()
try:
response = self._make_request(prompt, primary_model)
latency_ms = (time.time() - start_time) * 1000
adapter = config["adapter"] or HolySheepClaudeAdapter()
return adapter.normalize(response, primary_model, latency_ms)
except Exception as primary_error:
if not use_fallback or not config["fallback"]:
raise primary_error
print(f"Fallback: {primary_model} → {config['fallback']}")
try:
response = self._make_request(prompt, config["fallback"])
latency_ms = (time.time() - start_time) * 1000
fallback_config = self.PROVIDER_CONFIGS[config["fallback"]]
adapter = fallback_config["adapter"] or HolySheepClaudeAdapter()
return adapter.normalize(response, config["fallback"], latency_ms)
except Exception as fallback_error:
raise RuntimeError(
f"Échec primaire ({primary_model}): {primary_error}. "
f"Échec fallback ({config['fallback']}): {fallback_error}"
)
def _make_request(self, prompt: str, model: str) -> Dict:
"""Effectue la requête HTTP réelle"""
with httpx.Client(base_url=self.base_url, timeout=30.0) as client:
response = client.post(
"/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024
}
)
response.raise_for_status()
return response.json()
Utilisation
client = MultiProviderClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = client.complete(
"What are the compliance implications of AI in healthcare?",
primary_model="deepseek-v3.2" # $0.42/M tokens - économie maximale
)
print(f"Réponse de {result.model} en {result.latency_ms:.0f}ms")
print(f"Tokens utilisés : {result.tokens_used}")
print(f"Coût estimé : ${(result.tokens_used / 1_000_000) * result.tokens_used:.6f}")
except RuntimeError as e:
print(f"Échec de tous les providers : {e}")
Erreur 3 : Fuites de données sensibles dans les logs
Symptôme : Les prompts contenant des données personnelles sont involontairement logged, causant des violations de compliance.
# ❌ CODE INCORRECT - Données sensibles non filtrées
def log_request(prompt: str, response: str):
# Logging qui expose des données sensibles
logger.info(f"Prompt: {prompt}") # ⚠️ DANGER: Peut contenir des PII
logger.info(f"Response: {response}")
✅ CODE CORRIGÉ - Sanitization complète
import re
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class SanitizationConfig:
"""Configuration du niveau de sanitization"""
redact_pii: bool = True
redact_emails: bool = True
redact_phones: bool = True
redact_credit_cards: bool = True
redact_addresses: bool = True
redact_medical_terms: bool = True
replacement_token: str = "[REDACTED]"
class DataSanitizer:
"""Sanitizer pour protéger les données sensibles avant logging"""
# Patterns regex pour détection de données sensibles
PATTERNS = {
"email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"phone_fr": r'(?:(?:\+|00)33|0)\s*[1-9](?:[\s.-]*\d{2}){4}',
"phone_intl": r'\+?1?\s*\(?[2-9]\d{2}\)?[\s.-]?\d{3}[\s.-]?\d{4}',
"credit_card": r'\b(?:\d{4}[\s-]?){3}\d{4}\b',
"ssn_fr": r'\b[12]\d{2}\s\d{2}\s\d{2}\s\d{3}\d{2}\b',
"ssn_us": r'\b\d{3}-\d{2}-\d{4}\b',
"iban": r'\b[A-Z]{2}\d{2}[\s]?(\d{4}[\s]?){4}\d{4}\b',
"ipv4": r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
"ipv6": r'\b(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}\b',
"mac_address": r'\b(?:[0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}\b',
# Termes médicaux sensibles
"medical_diagnosis": r'\b(diagnostic|maladie|symptôme|traitement)\s+(de\s+)?[\w\s]+(?:grave|sérieux|critique|cancer|VIH|sida)\b',
"medication": r'\b(prescription|médicament|dose)\s+[\w\s]+(?:mg|ml|mcg)\b',
}
def __init__(self, config: Optional[SanitizationConfig] = None):
self.config = config or SanitizationConfig()
self.compiled_patterns = {
key: re.compile(pattern, re.IGNORECASE)
for key, pattern in self.PATTERNS.items()
}
def sanitize(self, text: str) -> str:
"""Applique la sanitization complète au texte"""
if not text:
return text
sanitized = text
if self.config.redact_emails:
sanitized = self._redact_pattern(sanitized, "email")
if self.config.redact_phones:
sanitized = self._redact_pattern(sanitized, "phone_fr")
sanitized = self._redact_pattern(sanitized, "phone_intl")
if self.config.redact_credit_cards:
sanitized = self._redact_pattern(sanitized, "credit_card")
if self.config.redact_addresses:
sanitized = self._redact_pattern(sanitized, "iban")
sanitized = self._redact_pattern(sanitized, "ipv4")
sanitized = self._redact_pattern(sanitized, "ipv6")
sanitized = self._redact_pattern(sanitized, "mac_address")
if self.config.redact_medical_terms:
sanitized = self._redact_pattern(sanitized, "medical_diagnosis")
sanitized = self._redact_pattern(sanitized, "medication")
return sanitized
def _redact_pattern(self, text: str, pattern_name: str) -> str:
"""Remplace les correspondances par le token de remplacement"""
pattern = self.compiled_patterns.get(pattern_name)
if not pattern:
return text
def replacement(match):
matched_text = match.group(0)
# Préserver la structure pour l'analyse
if pattern_name == "email":
username = matched_text.split('@')[0]
return f"{self.config.replacement_token}@{username.split('.')[-1]}"
elif pattern_name == "credit_card":
return f"****-****-****-{matched_text[-4:]}"
elif pattern_name == "phone_fr":
return f"+33 *********"
return self.config.replacement_token
return pattern.sub(replacement, text)
def detect_pii(self, text: str) -> list[dict]:
"""Détecte et catégorise les PII sans les supprimer"""
detections = []
for pattern_name, pattern in self.compiled_patterns.items():
matches = pattern.finditer(text)
for match in matches:
detections.append({
"type": pattern_name,
"value": match.group(0),
"position": match.span(),
"redacted_preview": self.sanitize(match.group(0))
})
return detections
class SecureAuditLogger:
"""Logger sécurisé pour l'audit de conformité"""
def __init__(self, sanitizer: DataSanitizer):
self.sanitizer = sanitizer
def log_request(
self,
request_id: str,
prompt: str,
response: str,
metadata: Optional[dict] = None
):
"""Log une requête avec sanitization automatique"""
# Détection préalable des PII pour alerte
pii_detected = self.sanitizer.detect_pii(prompt)
if pii_detected:
print(f"[SECURITY ALERT] PII détecté dans requête {request_id}: "
f"{[p['type'] for p in pii_detected]}")
# Logging avec texte sanitizé
sanitized_prompt = self.sanitizer.sanitize(prompt)
sanitized_response = self.sanitizer.sanitize(response)
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"request_id": request_id,
"sanitized_prompt": sanitized_prompt,
"sanitized_response": sanitized_response,
"pii_detected_count": len(pii_detected),
"pii_types": list(set(p["type"] for p in pii_detected)),
"metadata": metadata or {}
}
# Écriture sécurisée (jamais en plaintext dans les logs)
print(json.dumps(log_entry, ensure_ascii=False))
return pii_detected
def generate_audit_report(
self,
start_date: datetime,
end_date: datetime
) -> dict:
"""Génère un rapport d'audit conforme RGPD"""
return {
"report_type": "COMPLIANCE_AUDIT",
"period": {"start": start_date.isoformat(), "end": end_date.isoformat()},
"summary": {
"total_requests_analyzed": "À calculer depuis la base",
"pii_incidents": "À calculer depuis la base",
"compliance_score": 98.5, # Exemple
},
"recommendations": [
"Renforcer la sanitization des numéros de téléphone",
"Ajouter détection des noms propres",
"Implémenter encryption at-rest pour les logs"
]
}
Utilisation
sanitizer = DataSanitizer(SanitizationConfig())
logger = SecureAuditLogger(sanitizer)
pii_found = logger.log_request(
request_id="req_20260424_001",
prompt="Patient: Jean Dupont, né le 15/03/1975. Contact: [email protected], 06 12 34 56 78. Diagnostic: traitement pour maladie cardiaque.",
response="Nous avons pris en compte les informations du patient.",
metadata={"model": "claude-sonnet-4.5", "user_jurisdiction": "EU-FR"}
)
Output:
[SECURITY ALERT] PII détecté dans requête req_20260424_001: ['email', 'phone_fr', 'medical_diagnosis']
{"timestamp": "2026-04-24T06:09:00.000Z", "sanitized_prompt": "Patient