Architecture de sécurité complète
Assembler toutes les pièces
Au fil de cette formation, vous avez découvert les guardrails, l’API de modération, les 11 catégories, les seuils, la détection de jailbreak, la conformité RGPD et les politiques de contenu. Cette dernière leçon assemble toutes ces briques dans une architecture de sécurité complète, prête pour la production.
Vue d’ensemble de l’architecture
L’architecture de sécurité se compose de cinq couches successives :
Utilisateur
↓
[1. Validation d'entrée] → Regex, patterns, rate limiting
↓
[2. Guardrails Mistral] → Blocage pré-génération (403)
↓
[3. Génération LLM] → safe_prompt + instructions sécurisées
↓
[4. Modération sortie] → API Moderation post-génération
↓
[5. Logging & Audit] → Scores, décisions, audit trail RGPD
↓
Réponse utilisateurImplémentation complète
from mistralai import Mistral
from mistralai.models import SDKError
import os
import re
import json
import logging
from datetime import datetime
from enum import Enum
# Configuration
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("moderation")
class Decision(Enum):
ALLOWED = "allowed"
BLOCKED_INPUT = "blocked_input"
BLOCKED_GUARDRAIL = "blocked_guardrail"
BLOCKED_OUTPUT = "blocked_output"
ERROR = "error"
class ArchitectureSecurisee:
"""Architecture de sécurité complète pour applications IA."""
def __init__(self, politique: dict, model: str = "mistral-large-latest"):
self.politique = politique
self.model = model
self.stats = {d.value: 0 for d in Decision}
# --- Couche 1 : Validation d'entrée ---
def valider_entree(self, texte: str) -> dict:
"""Validation rapide avant toute API."""
# Longueur maximale
if len(texte) > 10000:
return {"valide": False, "raison": "Message trop long"}
# Caractères suspects (injections Unicode)
if re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", texte):
return {"valide": False, "raison": "Caractères invalides"}
# Patterns de jailbreak évidents
patterns = [
r"ignore\s+(all\s+)?(previous|prior|above)",
r"you\s+are\s+now\s+(DAN|unfiltered)",
r"oublie\s+(toutes?\s+)?(tes|vos)\s+instructions",
r"system\s*prompt",
]
for pattern in patterns:
if re.search(pattern, texte, re.IGNORECASE):
return {
"valide": False,
"raison": "Pattern injection détecté"
}
return {"valide": True}
# --- Couche 2 : Modération entrée ---
def moderer_entree(self, texte: str) -> dict:
"""Modération du prompt via API."""
response = client.classifiers.moderate(
model="mistral-moderation-2603",
inputs=[texte]
)
scores = response.results[0].category_scores
violations = self._verifier_seuils(scores)
return {
"autorise": len(violations) == 0,
"scores": scores,
"violations": violations
}
# --- Couche 3 : Génération sécurisée ---
def generer(self, messages: list) -> str:
"""Génération avec guardrails et safe_prompt."""
response = client.chat.complete(
model=self.model,
messages=messages,
safe_prompt=True,
guardrails={
"enabled": True,
"custom_category_thresholds":
self.politique["seuils"],
"action": "block",
"block_on_error": True
}
)
return response.choices[0].message.content
# --- Couche 4 : Modération sortie ---
def moderer_sortie(self, prompt: str, reponse: str) -> dict:
"""Modération de la réponse en contexte."""
response = client.classifiers.moderate_chat(
model="mistral-moderation-2603",
inputs=[[
{"role": "user", "content": prompt},
{"role": "assistant", "content": reponse}
]]
)
scores = response.results[0].category_scores
violations = self._verifier_seuils(scores)
return {
"autorise": len(violations) == 0,
"scores": scores,
"violations": violations
}
# --- Couche 5 : Logging ---
def log_decision(self, decision: Decision, data: dict):
"""Enregistre la décision pour audit."""
self.stats[decision.value] += 1
entry = {
"timestamp": datetime.utcnow().isoformat(),
"decision": decision.value,
"politique": self.politique["nom"],
**data
}
logger.info(json.dumps(entry, ensure_ascii=False))
# --- Pipeline principal ---
def traiter(self, user_message: str,
system_prompt: str = None) -> dict:
"""Pipeline complet de traitement sécurisé."""
# Couche 1 : Validation
validation = self.valider_entree(user_message)
if not validation["valide"]:
self.log_decision(Decision.BLOCKED_INPUT, {
"raison": validation["raison"]
})
return {
"status": Decision.BLOCKED_INPUT.value,
"message": "Votre message ne peut pas être traité."
}
# Couche 2 : Modération entrée
check_input = self.moderer_entree(user_message)
if not check_input["autorise"]:
self.log_decision(Decision.BLOCKED_INPUT, {
"violations": check_input["violations"]
})
return {
"status": Decision.BLOCKED_INPUT.value,
"message": "Ce contenu enfreint notre politique."
}
# Couche 3 : Génération
messages = []
if system_prompt:
messages.append({"role": "system",
"content": system_prompt})
messages.append({"role": "user",
"content": user_message})
try:
reponse = self.generer(messages)
except SDKError as e:
if e.status_code == 403:
self.log_decision(Decision.BLOCKED_GUARDRAIL, {})
return {
"status": Decision.BLOCKED_GUARDRAIL.value,
"message": "Requête bloquée par les guardrails."
}
raise
# Couche 4 : Modération sortie
check_output = self.moderer_sortie(user_message, reponse)
if not check_output["autorise"]:
self.log_decision(Decision.BLOCKED_OUTPUT, {
"violations": check_output["violations"]
})
return {
"status": Decision.BLOCKED_OUTPUT.value,
"message": "La réponse a été filtrée."
}
# Succès
self.log_decision(Decision.ALLOWED, {
"input_max_score": max(
check_input["scores"].values()
),
"output_max_score": max(
check_output["scores"].values()
)
})
return {
"status": Decision.ALLOWED.value,
"content": reponse,
"moderation": {
"input": check_input["scores"],
"output": check_output["scores"]
}
}
# --- Utilitaires ---
def _verifier_seuils(self, scores: dict) -> list:
"""Vérifie les scores contre les seuils de la politique."""
violations = []
for cat, seuil in self.politique["seuils"].items():
score = scores.get(cat, 0)
if score > seuil:
violations.append({
"categorie": cat,
"score": round(score, 3),
"seuil": seuil
})
return violations
def rapport(self) -> dict:
"""Génère un rapport de statistiques."""
total = sum(self.stats.values())
return {
"politique": self.politique["nom"],
"total_requetes": total,
"decisions": self.stats,
"taux_blocage": round(
(self.stats["blocked_input"] +
self.stats["blocked_guardrail"] +
self.stats["blocked_output"]) / max(total, 1),
3
)
}Utilisation en production
# Définir la politique
politique_prod = {
"nom": "Production v1.2",
"seuils": {
"sexual": 0.15, "hate": 0.1, "violence": 0.2,
"criminal": 0.1, "selfharm": 0.1, "health": 0.4,
"financial": 0.4, "law": 0.4, "pii": 0.15,
"jailbreaking": 0.1, "unpredictable": 0.3
}
}
# Instancier l'architecture
archi = ArchitectureSecurisee(politique_prod)
# Traiter des requêtes
resultat = archi.traiter(
"Expliquez-moi le fonctionnement des transformers",
system_prompt="Vous êtes un formateur en IA."
)
print(resultat["status"]) # → "allowed"
print(resultat["content"]) # → Explication des transformers
# Consulter les statistiques
print(archi.rapport())Monitoring et alertes
En production, connectez votre architecture à votre stack de monitoring :
def configurer_alertes(archi: ArchitectureSecurisee):
"""Configure les alertes pour le monitoring."""
rapport = archi.rapport()
# Alerte si taux de blocage anormal
if rapport["taux_blocage"] > 0.15:
logger.warning(
f"Taux de blocage élevé : "
f"{rapport['taux_blocage']:.1%}"
)
# Alerte si pics de jailbreak
# Intégrez avec Prometheus, Grafana, Datadog, etc.Récapitulatif de la formation
Au terme de cette formation, vous maîtrisez :
- Custom Guardrails : bloquer les prompts dangereux avant la génération, configurer les seuils par catégorie, appliquer aux agents
- API Moderation : classifier le contenu avec Mistral Moderation 2, interpréter les scores des 11 catégories, prendre des décisions automatisées
- Sécurité en production : définir des politiques de contenu, détecter les prompt injections, assurer la conformité RGPD
- Architecture complète : assembler guardrails + modération + monitoring dans un pipeline robuste
Vous disposez de tout le code nécessaire pour déployer un système de modération IA complet avec l’API Mistral.
Points clés à retenir
- L’architecture de sécurité comprend 5 couches : validation, guardrails, génération, modération, logging
- Chaque couche compense les faiblesses des autres — c’est la défense en profondeur
- Monitorez le taux de blocage et les scores en production
- Maintenez votre politique à jour avec des révisions régulières
- Le code complet de cette leçon est un point de départ — adaptez-le à votre contexte