Aller au contenu principal

Validation et stratégies de retry

La validation, dernier rempart avant la production

Même avec les Custom Structured Outputs, la validation reste une étape indispensable. Le schéma garantit la structure, mais pas la cohérence sémantique des données. Un champ price peut contenir -999.0 — c’est un float valide, mais absurde pour un prix.

Validation structurelle vs sémantique

Validation structurelle (automatique)

Pydantic et Zod gèrent automatiquement :

  • La présence des champs requis
  • Le type de chaque champ (string, int, float, bool)
  • Les contraintes d’enum (valeurs parmi un ensemble fini)
  • Les structures imbriquées

Validation sémantique (à votre charge)

Vous devez vérifier :

  • Les bornes des valeurs numériques (prix > 0, score entre 0 et 1)
  • La cohérence entre champs (total = somme des lignes)
  • Le format des chaînes (email valide, date au bon format)
  • La logique métier (date d’échéance après date de facturation)

Validation sémantique avec Pydantic

Pydantic permet d’ajouter des validateurs directement dans le modèle :

from pydantic import BaseModel, Field, field_validator, model_validator

class InvoiceLine(BaseModel):
    description: str
    quantity: float
    unit_price: float
    total: float

    @field_validator("quantity")
    @classmethod
    def quantity_positive(cls, v):
        if v <= 0:
            raise ValueError(f"La quantité doit être positive, reçu : {v}")
        return v

    @field_validator("unit_price")
    @classmethod
    def price_non_negative(cls, v):
        if v < 0:
            raise ValueError(f"Le prix unitaire ne peut pas être négatif : {v}")
        return v

    @model_validator(mode="after")
    def check_total(self):
        expected = round(self.quantity * self.unit_price, 2)
        if abs(self.total - expected) > 0.01:
            raise ValueError(
                f"Total incohérent : {self.total} != {self.quantity} x {self.unit_price} = {expected}"
            )
        return self

En TypeScript avec Zod, utilisez .refine() :

import { z } from "zod";

const InvoiceLineSchema = z
  .object({
    description: z.string(),
    quantity: z.number().positive("La quantité doit être positive"),
    unitPrice: z.number().nonnegative("Le prix ne peut pas être négatif"),
    total: z.number(),
  })
  .refine(
    (data) => Math.abs(data.total - data.quantity * data.unitPrice) < 0.01,
    { message: "Le total ne correspond pas à quantité x prix unitaire" }
  );

Stratégie de retry avec feedback

Quand la validation échoue, le pattern le plus efficace est de renvoyer l’erreur au modèle pour qu’il se corrige :

from mistralai import Mistral
from pydantic import BaseModel, ValidationError
import json

def extract_with_retry(
    client: Mistral,
    messages: list,
    schema,
    max_retries: int = 3,
    model: str = "mistral-large-latest"
):
    """Extraction avec retry et feedback d'erreur structuré."""
    conversation = list(messages)  # Copie pour ne pas modifier l'original

    for attempt in range(max_retries):
        try:
            response = client.chat.parse(
                model=model,
                messages=conversation,
                response_format=schema,
                temperature=0,
                max_tokens=2048
            )

            # Vérifier que la réponse est complète
            if response.choices[0].finish_reason != "stop":
                raise ValueError(
                    f"Réponse incomplète (finish_reason={response.choices[0].finish_reason})"
                )

            parsed = response.choices[0].message.parsed
            if parsed is None:
                raise ValueError("Réponse parsed est None")

            # Validation sémantique supplémentaire
            validate_business_rules(parsed)

            return parsed

        except (ValidationError, ValueError) as e:
            error_msg = str(e)
            print(f"Tentative {attempt + 1}/{max_retries} échouée : {error_msg}")

            if attempt < max_retries - 1:
                # Ajouter le feedback au contexte
                raw = response.choices[0].message.content if response else ""
                conversation.append({"role": "assistant", "content": raw})
                conversation.append({
                    "role": "user",
                    "content": f"Ta réponse contient une erreur : {error_msg}. Corrige-la."
                })
            else:
                raise RuntimeError(
                    f"Échec après {max_retries} tentatives. Dernière erreur : {error_msg}"
                )

Stratégie de fallback

Quand le retry échoue, vous avez besoin d’un plan B :

def extract_with_fallback(client, text, schema, system_prompt):
    """Extraction avec cascade de fallbacks."""

    # Tentative 1 : Custom Structured Output (meilleur)
    try:
        return extract_with_retry(
            client,
            [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": text}
            ],
            schema,
            max_retries=2,
            model="mistral-large-latest"
        )
    except RuntimeError:
        print("Custom parse échoué, fallback vers JSON Mode")

    # Tentative 2 : JSON Mode (plus flexible)
    try:
        response = client.chat.complete(
            model="mistral-large-latest",
            messages=[
                {"role": "system", "content": system_prompt + "\nRetourne un JSON valide."},
                {"role": "user", "content": text}
            ],
            response_format={"type": "json_object"},
            temperature=0
        )
        raw = json.loads(response.choices[0].message.content)
        return schema.model_validate(raw)  # Tenter la validation Pydantic
    except Exception:
        print("JSON Mode échoué, fallback vers extraction manuelle")

    # Tentative 3 : Retourner un objet partiel ou une erreur
    return None

Pattern de circuit breaker

Pour les pipelines à haute fréquence, implémentez un circuit breaker qui arrête les appels quand le taux d’erreur est trop élevé :

from collections import deque
from time import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, window_seconds=60):
        self.failures = deque()
        self.threshold = failure_threshold
        self.window = window_seconds
        self.is_open = False

    def record_failure(self):
        now = time()
        self.failures.append(now)
        # Nettoyer les erreurs hors fenêtre
        while self.failures and self.failures[0] < now - self.window:
            self.failures.popleft()
        if len(self.failures) >= self.threshold:
            self.is_open = True

    def record_success(self):
        self.is_open = False
        self.failures.clear()

    def can_proceed(self) -> bool:
        if not self.is_open:
            return True
        # Vérifier si la fenêtre est passée
        now = time()
        while self.failures and self.failures[0] < now - self.window:
            self.failures.popleft()
        if len(self.failures) < self.threshold:
            self.is_open = False
            return True
        return False

# Utilisation
breaker = CircuitBreaker(failure_threshold=5, window_seconds=60)

def safe_extract(client, text, schema):
    if not breaker.can_proceed():
        raise RuntimeError("Circuit ouvert — trop d'erreurs récentes")
    try:
        result = extract_with_retry(client, [...], schema)
        breaker.record_success()
        return result
    except RuntimeError:
        breaker.record_failure()
        raise

Logging et monitoring

En production, loguez chaque appel pour diagnostiquer les problèmes :

import logging
import time

logger = logging.getLogger("structured_output")

def monitored_extract(client, text, schema, **kwargs):
    start = time.time()
    try:
        result = extract_with_retry(client, [...], schema, **kwargs)
        duration = time.time() - start
        logger.info(f"Extraction réussie en {duration:.2f}s — schema={schema.__name__}")
        return result
    except Exception as e:
        duration = time.time() - start
        logger.error(f"Extraction échouée en {duration:.2f}s — schema={schema.__name__} — {e}")
        raise

Points clés à retenir

  • La validation structurelle (types) est automatique, la validation sémantique (logique) est à votre charge
  • Le retry avec feedback est le pattern le plus efficace : renvoyez l’erreur au modèle
  • Implémentez une cascade de fallbacks : Custom > JSON Mode > valeur par défaut
  • En production haute fréquence, utilisez un circuit breaker pour éviter les cascades d’erreurs
  • Loguez chaque appel pour le monitoring et le diagnostic
← Lecon precedenteCas d'usage avancésLecon suivante →Bonnes pratiques et récapitulatif