Modération de contenu à l'échelle

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). Cette leçon couvre les stratégies de modération à l’échelle.

L’API de modération OpenAI

Modération de base

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']}")

Pipeline de modération complet

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)

Modération à l’échelle

Modération asynchrone avec file d’attente

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%})"
        )

Modération contextuelle

Pour une modération plus fine, prenez en compte le contexte de l’application :

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.3",
        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)

Logging et audit

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")

Points clés à retenir

• Modérez à la fois les entrées (prompts) et les sorties (réponses)
• Utilisez l’API de modération OpenAI comme première couche rapide
• Ajoutez une détection d’injection de prompt comme seconde couche
• Adaptez la modération au contexte de votre application
• Journalisez toutes les décisions pour l’audit (sans stocker le contenu brut)