Modération de contenu à l'échelle — Production, Évals et Optimisation

Protéger vos utilisateurs en production

Dès que votre application est ouverte au public, vous devez gérer les contenus inappropriés — à la fois en entrée (prompts malveillants) et en sortie (réponses problématiques). La double direction surprend souvent : on pense d’abord à filtrer ce que les utilisateurs envoient, alors que le risque réputationnel se concentre sur ce que votre application publie sous votre nom. Un modèle bien aligné produit rarement un contenu choquant de lui-même, mais il le fait parfois lorsqu’il est poussé, et c’est votre marque qui apparaît en haut de la capture d’écran.

Nous irons de l’appel unitaire jusqu’au pipeline asynchrone à haut débit, en montant progressivement en charge.

La première couche : l’API de modération

L’API de modération classe un texte selon plusieurs catégories de risque et retourne à la fois un verdict binaire et des scores détaillés. Elle est rapide et gratuite, ce qui en fait le premier filtre naturel de toute application publique. La fonction ci-dessous ne remonte que les catégories dont le score dépasse 0,3, un seuil bas assumé : à ce stade, on cherche à voir ce qui approche de la limite, pas seulement ce qui la franchit.

import openai

client = openai.OpenAI()

def moderer_contenu(texte: str) -> dict:
    """Vérifie si un texte contient du contenu inapproprié."""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=texte,
    )

    resultat = response.results[0]
    return {
        "flagged": resultat.flagged,
        "categories": {
            cat: score
            for cat, score in resultat.category_scores.__dict__.items()
            if score > 0.3
        },
    }

# Exemple
resultat = moderer_contenu("Comment préparer un gâteau au chocolat ?")
print(f"Flaggé : {resultat['flagged']}")

En production, un verdict binaire ne suffit pas. Selon le contenu et le contexte, vous voudrez laisser passer, reformuler, bloquer, ou transmettre à un humain — quatre décisions différentes qu’il vaut mieux nommer explicitement dans le code plutôt que d’improviser à coups de booléens. Le pipeline ci-dessous formalise ces actions, applique des seuils par catégorie et ajoute une seconde couche de détection d’injection de prompt.

from enum import Enum
from dataclasses import dataclass

class ActionModeration(Enum):
    AUTORISER = "autoriser"
    MODIFIER = "modifier"
    BLOQUER = "bloquer"
    ESCALADER = "escalader"

@dataclass
class ResultatModeration:
    action: ActionModeration
    raison: str | None = None
    contenu_modifie: str | None = None

class PipelineModeration:
    """Pipeline de modération multi-couches."""

    def __init__(self, seuils: dict | None = None):
        self.seuils = seuils or {
            "harassment": 0.7,
            "hate": 0.7,
            "self-harm": 0.5,
            "sexual": 0.7,
            "violence": 0.7,
        }

    def moderer_entree(self, texte: str) -> ResultatModeration:
        """Modère le contenu en entrée (prompt utilisateur)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=texte,
        )

        resultat = moderation.results[0]
        if resultat.flagged:
            categories_problematiques = [
                cat for cat, flagged in resultat.categories.__dict__.items()
                if flagged
            ]
            return ResultatModeration(
                action=ActionModeration.BLOQUER,
                raison=f"Contenu inapproprié : {', '.join(categories_problematiques)}",
            )

        # Détection d'injection de prompt
        if self._detecter_injection(texte):
            return ResultatModeration(
                action=ActionModeration.BLOQUER,
                raison="Tentative d'injection de prompt détectée",
            )

        return ResultatModeration(action=ActionModeration.AUTORISER)

    def moderer_sortie(self, reponse: str) -> ResultatModeration:
        """Modère le contenu en sortie (réponse du LLM)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=reponse,
        )

        if moderation.results[0].flagged:
            return ResultatModeration(
                action=ActionModeration.BLOQUER,
                raison="La réponse générée contient du contenu inapproprié",
            )

        return ResultatModeration(action=ActionModeration.AUTORISER)

    def _detecter_injection(self, texte: str) -> bool:
        """Détecte les tentatives d'injection de prompt."""
        marqueurs = [
            "ignore previous instructions",
            "ignore tes instructions",
            "oublie tes consignes",
            "tu es maintenant",
            "nouveau rôle",
            "system prompt",
            "jailbreak",
        ]
        texte_lower = texte.lower()
        return any(m in texte_lower for m in marqueurs)

Le seuil de 0,5 sur self-harm, plus strict que les 0,7 des autres catégories, illustre le raisonnement à tenir sur chaque seuil : le coût d’un faux positif est ici sans commune mesure avec celui d’un faux négatif, donc on abaisse la barre. Faites cet arbitrage catégorie par catégorie, en fonction de votre public, plutôt que d’appliquer une valeur uniforme.

La détection d’injection par liste de marqueurs demande une mise en garde. Elle attrape les tentatives naïves, celles qui représentent l’essentiel du volume, et rien d’autre : une formulation légèrement différente ou traduite passe au travers. Considérez-la comme un filtre de bruit, pas comme une défense — la véritable protection contre l’injection reste architecturale, en ne donnant jamais au modèle des capacités qu’un prompt détourné pourrait exploiter.

Tenir la charge

Sur une application à fort trafic, la modération synchrone devient un goulot d’étranglement : chaque message attend son verdict avant d’être traité. Le modérateur asynchrone découple les deux temps en plaçant les contenus dans une file consommée par plusieurs workers, et tient au passage des statistiques utiles.

import asyncio

class ModerateurAsynchrone:
    """Modère le contenu de manière asynchrone pour le haut débit."""

    def __init__(self, nb_workers: int = 10):
        self.pipeline = PipelineModeration()
        self.file: asyncio.Queue = asyncio.Queue()
        self.nb_workers = nb_workers
        self.stats = {"autorise": 0, "bloque": 0, "escalade": 0}

    async def worker(self):
        while True:
            tache = await self.file.get()
            try:
                resultat = self.pipeline.moderer_entree(tache["texte"])
                tache["callback"](resultat)

                match resultat.action:
                    case ActionModeration.AUTORISER:
                        self.stats["autorise"] += 1
                    case ActionModeration.BLOQUER:
                        self.stats["bloque"] += 1
                    case ActionModeration.ESCALADER:
                        self.stats["escalade"] += 1
            finally:
                self.file.task_done()

    async def demarrer(self):
        for _ in range(self.nb_workers):
            asyncio.create_task(self.worker())

    def rapport(self) -> str:
        total = sum(self.stats.values())
        if total == 0:
            return "Aucun contenu modéré."
        return (
            f"Total: {total} | "
            f"Autorisé: {self.stats['autorise']} ({self.stats['autorise']/total:.1%}) | "
            f"Bloqué: {self.stats['bloque']} ({self.stats['bloque']/total:.1%})"
        )

Le taux de blocage produit par rapport est votre principal indicateur de calibrage. S’il grimpe soudainement à 15 % alors qu’il tournait à 0,5 %, deux hypothèses seulement : vous subissez une attaque coordonnée, ou vous venez de casser vos seuils. Dans les deux cas, vous voulez le savoir dans l’heure, pas au prochain bilan mensuel.

Une dernière limite mérite d’être posée : l’API de modération ignore tout de votre métier. Un message parfaitement acceptable en général peut violer vos propres règles — solliciter un conseil médical sur un assistant bancaire, tenter d’obtenir un tarif hors grille, contourner une procédure. La modération contextuelle ajoute cette couche en soumettant le message à un modèle avec vos règles explicites, uniquement après que le filtre standard a laissé passer.

def moderation_contextuelle(
    texte: str,
    contexte_app: str,
    regles_metier: list[str],
) -> ResultatModeration:
    """Modération adaptée au contexte de l'application."""
    pipeline = PipelineModeration()
    resultat_standard = pipeline.moderer_entree(texte)

    if resultat_standard.action == ActionModeration.BLOQUER:
        return resultat_standard

    prompt_verification = (
        f"Contexte de l'application : {contexte_app}\n\n"
        f"Règles métier à vérifier :\n"
        + "\n".join(f"- {r}" for r in regles_metier)
        + f"\n\nMessage de l'utilisateur : {texte}\n\n"
        "Le message respecte-t-il toutes les règles ? "
        "Répondez par oui ou non avec une justification courte."
    )

    response = client.responses.create(
        model="gpt-5.6-terra",
        input=prompt_verification,
        max_output_tokens=100,
        temperature=0.0,
    )

    if "non" in response.output_text.lower()[:10]:
        return ResultatModeration(
            action=ActionModeration.BLOQUER,
            raison=response.output_text,
        )

    return ResultatModeration(action=ActionModeration.AUTORISER)

L’ordre des deux vérifications n’est pas indifférent : le filtre gratuit et rapide s’exécute en premier, et l’appel facturé n’a lieu que pour les messages qui l’ont franchi. Sur un trafic réel, cela représente la quasi-totalité des messages, mais l’inverse — payer un appel de modèle pour un contenu que l’API de modération aurait rejeté en quelques millisecondes — serait un gaspillage.

Garder une trace défendable

Une décision de modération est une décision qui vous sera reprochée, dans un sens ou dans l’autre. Il vous faut donc un journal — mais un journal qui ne devienne pas lui-même un problème. Le commentaire dans le code est la ligne la plus importante du fichier : on enregistre l’action, la raison et la longueur, jamais le texte brut. Stocker les messages des utilisateurs dans un fichier de logs vous exposerait à des obligations de protection des données bien plus lourdes que le service rendu.

import json
import time
from pathlib import Path

class LogModeration:
    """Journalise toutes les décisions de modération pour l'audit."""

    def __init__(self, chemin_log: str = "logs/moderation.jsonl"):
        self.chemin = Path(chemin_log)
        self.chemin.parent.mkdir(parents=True, exist_ok=True)

    def enregistrer(
        self,
        texte: str,
        resultat: ResultatModeration,
        source: str = "entree",
    ):
        entree = {
            "timestamp": time.time(),
            "source": source,
            "action": resultat.action.value,
            "raison": resultat.raison,
            "longueur_texte": len(texte),
            # NE PAS logger le texte brut pour la vie privée
        }

        with open(self.chemin, "a") as f:
            f.write(json.dumps(entree) + "\n")

Si vous devez malgré tout pouvoir retrouver un message précis lors d’une contestation, journalisez une empreinte du texte plutôt que le texte lui-même : elle permet de confirmer qu’un contenu donné correspond bien à une décision enregistrée, sans conserver le contenu de votre côté.

Points clés à retenir

Modérez à la fois les entrées (prompts) et les sorties (réponses)
Utilisez l’API de modération comme première couche rapide et gratuite
Ajustez les seuils catégorie par catégorie selon le coût réel d’un faux négatif
Ajoutez une détection d’injection en sachant qu’elle n’attrape que les tentatives naïves
Journalisez toutes les décisions pour l’audit, sans stocker le contenu brut