Compliance et audit trail

Pour les applications IA en production, la traçabilité est essentielle. Un audit trail complet vous protège en cas de contrôle, de litige ou d’incident. Cette leçon vous montre comment implémenter un système d’audit robuste.

Pourquoi un audit trail ?

Conformité réglementaire : RGPD, AI Act européen, normes sectorielles
Traçabilité : savoir qui a fait quoi, quand, avec quel résultat
Débogage : comprendre les comportements inattendus du modèle
Responsabilité : documenter les décisions prises par l’IA

Structure d’un enregistrement d’audit

from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional
import json
import uuid
import hashlib

@dataclass
class AuditRecord:
    """Enregistrement d'audit pour un appel API."""
    # Identifiants
    audit_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )

    # Contexte
    user_id_hash: str = ""  # Jamais l'ID en clair
    session_id: str = ""
    application: str = ""

    # Requête (sans données sensibles)
    model: str = ""
    prompt_hash: str = ""  # Hash du prompt, pas le prompt
    prompt_length: int = 0

    # Réponse
    response_id: str = ""
    output_length: int = 0
    input_tokens: int = 0
    output_tokens: int = 0

    # Modération
    input_flagged: bool = False
    output_flagged: bool = False

    # Métadonnées
    duration_ms: int = 0
    status: str = "success"
    error: Optional[str] = None

    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)

Système d’audit complet

import time
from openai import OpenAI

class AuditedClient:
    """Client OpenAI avec audit trail intégré."""

    def __init__(self, application: str, audit_store: str = "audit.jsonl"):
        self.client = OpenAI()
        self.application = application
        self.audit_store = audit_store

    def _hash(self, texte: str) -> str:
        """Hash SHA-256 tronqué pour l'audit."""
        return hashlib.sha256(texte.encode()).hexdigest()[:16]

    def _sauvegarder_audit(self, record: AuditRecord):
        """Sauvegarde un enregistrement d'audit (append-only)."""
        with open(self.audit_store, "a", encoding="utf-8") as f:
            f.write(record.to_json() + "\n")

    def create(self, user_id: str, session_id: str,
               model: str, input: str, **kwargs) -> object:
        """Appel API avec audit automatique."""

        record = AuditRecord(
            user_id_hash=self._hash(user_id),
            session_id=session_id,
            application=self.application,
            model=model,
            prompt_hash=self._hash(input),
            prompt_length=len(input),
        )

        start = time.monotonic()

        try:
            # Modération de l'input
            mod = self.client.moderations.create(
                model="omni-moderation-latest",
                input=input
            )
            record.input_flagged = mod.results[0].flagged

            if record.input_flagged:
                record.status = "blocked_input"
                self._sauvegarder_audit(record)
                raise ValueError("Contenu d'entrée inapproprié")

            # Appel API
            response = self.client.responses.create(
                model=model,
                input=input,
                **kwargs
            )

            # Modération de l'output
            mod_out = self.client.moderations.create(
                model="omni-moderation-latest",
                input=response.output_text
            )
            record.output_flagged = mod_out.results[0].flagged

            # Remplir les métriques
            record.response_id = response.id
            record.output_length = len(response.output_text)
            record.input_tokens = response.usage.input_tokens
            record.output_tokens = response.usage.output_tokens
            record.duration_ms = round((time.monotonic() - start) * 1000)
            record.status = "success"

            self._sauvegarder_audit(record)
            return response

        except Exception as e:
            record.duration_ms = round((time.monotonic() - start) * 1000)
            record.status = "error"
            record.error = type(e).__name__
            self._sauvegarder_audit(record)
            raise

# Utilisation
audited = AuditedClient(application="chatbot-support")

response = audited.create(
    user_id="[email protected]",
    session_id="sess-abc123",
    model="gpt-5.3",
    input="Comment retourner un produit ?"
)
print(response.output_text)

Requêtes d’audit

import json

def lire_audit(fichier: str = "audit.jsonl") -> list[dict]:
    """Lit tous les enregistrements d'audit."""
    records = []
    try:
        with open(fichier, "r", encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    records.append(json.loads(line))
    except FileNotFoundError:
        pass
    return records

def rapport_audit(fichier: str = "audit.jsonl") -> dict:
    """Génère un rapport d'audit."""
    records = lire_audit(fichier)

    if not records:
        return {"message": "Aucun enregistrement"}

    total = len(records)
    erreurs = sum(1 for r in records if r["status"] == "error")
    bloques = sum(1 for r in records if r["status"] == "blocked_input")
    tokens_total = sum(
        r.get("input_tokens", 0) + r.get("output_tokens", 0)
        for r in records
    )
    durees = [r["duration_ms"] for r in records if r["duration_ms"] > 0]

    return {
        "periode": {
            "debut": records[0]["timestamp"],
            "fin": records[-1]["timestamp"],
        },
        "total_appels": total,
        "succes": total - erreurs - bloques,
        "erreurs": erreurs,
        "bloques_moderation": bloques,
        "tokens_consommes": tokens_total,
        "latence_moyenne_ms": round(sum(durees) / max(len(durees), 1)),
        "utilisateurs_uniques": len(set(r["user_id_hash"] for r in records)),
    }

print(json.dumps(rapport_audit(), indent=2, ensure_ascii=False))

Rétention et archivage

from datetime import datetime, timedelta
import shutil
import os

def archiver_audit(fichier: str = "audit.jsonl",
                   retention_jours: int = 90):
    """Archive et purge les enregistrements anciens."""
    records = lire_audit(fichier)
    limite = datetime.now() - timedelta(days=retention_jours)

    recents = []
    archives = []

    for record in records:
        date_record = datetime.fromisoformat(record["timestamp"])
        if date_record > limite:
            recents.append(record)
        else:
            archives.append(record)

    if archives:
        # Archiver les anciens
        archive_name = f"audit_archive_{datetime.now().strftime('%Y%m%d')}.jsonl"
        with open(archive_name, "w", encoding="utf-8") as f:
            for record in archives:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

        # Réécrire le fichier principal avec les récents
        with open(fichier, "w", encoding="utf-8") as f:
            for record in recents:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

        print(f"Archivé {len(archives)} enregistrements dans {archive_name}")
        print(f"Conservé {len(recents)} enregistrements récents")

AI Act européen : ce qui change

Le Règlement européen sur l’IA (AI Act) impose des obligations selon le niveau de risque de votre application :

# Évaluer le niveau de risque de votre application
NIVEAUX_RISQUE = {
    "minimal": {
        "exemples": ["chatbot FAQ", "résumé de texte", "traduction"],
        "obligations": ["Transparence basique"]
    },
    "limité": {
        "exemples": ["chatbot service client", "analyse de sentiment"],
        "obligations": ["Informer l'utilisateur qu'il interagit avec une IA"]
    },
    "élevé": {
        "exemples": ["scoring crédit", "recrutement", "diagnostic médical"],
        "obligations": [
            "Évaluation de conformité",
            "Système de gestion des risques",
            "Données d'entraînement documentées",
            "Logging et traçabilité complète",
            "Supervision humaine obligatoire",
            "Précision et robustesse démontrées"
        ]
    },
    "inacceptable": {
        "exemples": ["scoring social", "manipulation subliminale"],
        "obligations": ["INTERDIT"]
    }
}

Checklist compliance

Audit trail append-only avec horodatage
Hash des données personnelles dans les logs (jamais en clair)
Modération systématique des entrées et sorties
Registre des traitements RGPD à jour
Politique de rétention et d’archivage définie
Évaluation du niveau de risque AI Act effectuée
Informer les utilisateurs qu’ils interagissent avec une IA
Supervision humaine pour les décisions à impact

Points clés à retenir

Un audit trail est obligatoire pour la conformité réglementaire
Loguez en append-only : ne modifiez jamais les enregistrements existants
Hashez toutes les données personnelles dans les logs d’audit
L’AI Act européen classe les applications par niveau de risque
Implémentez la modération automatique ET la supervision humaine
Définissez une politique de rétention et archivez régulièrement