Validation et stratégies de retry
Mis à jour le 29 juillet 2026
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, pas la cohérence sémantique des données. Un champ price peut contenir -999.0 : c’est un float valide, parfaitement conforme au schéma, et absurde pour un prix.
La frontière est nette. Pydantic et Zod prennent en charge la validation structurelle : présence des champs requis, type de chaque champ, respect des contraintes d’enum, conformité des structures imbriquées. Tout cela est automatique et vous n’avez rien à écrire. Reste la validation sémantique, qui vous incombe entièrement : les bornes des valeurs numériques (un prix supérieur à zéro, un score entre 0 et 1), la cohérence entre champs (le total qui doit égaler la somme des lignes), le format des chaînes (un email réellement valide, une date au bon format) et la logique métier (une date d’échéance postérieure à la date de facturation).
Inscrire les règles métier dans le schéma
Pydantic permet de placer ces contrôles directement dans le modèle, ce qui les rend indissociables de la structure. @field_validator traite un champ isolément — une quantité doit être positive, un prix unitaire ne peut pas être négatif. @model_validator(mode="after") intervient une fois tous les champs remplis et permet les contrôles croisés, ici la vérification que le total annoncé correspond bien au produit quantité × prix, à un centime près.
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 selfZod exprime la même chose avec les contraintes intégrées .positive() et .nonnegative() pour les champs, et .refine() pour la règle croisée appliquée à l’objet complet.
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" }
);Le retry avec feedback
Quand la validation échoue, le pattern le plus efficace consiste à renvoyer l’erreur au modèle pour qu’il se corrige lui-même. La fonction ci-dessous enchaîne les trois contrôles dans l’ordre où ils comptent : réponse complète, objet non nul, règles métier respectées. En cas d’échec, elle ajoute à la conversation la réponse fautive puis le message d’erreur, si bien que la tentative suivante dispose du diagnostic précis. Notez la copie de la liste de messages en début de fonction — sans elle, vous polluez la conversation de l’appelant.
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}"
)Une cascade de plans B
Le retry ne résout pas tout : il arrive qu’un document soit trop atypique pour le schéma. Plutôt que d’échouer sèchement, dégradez progressivement. La première tentative utilise les Custom Structured Outputs, la plus fiable. Si elle échoue, on retombe sur le JSON Mode, plus permissif, dont la sortie est ensuite soumise à model_validate() — vous récupérez ainsi la contrainte du schéma sans la contrainte de génération. En dernier recours, la fonction retourne None, ce qui laisse à l’appelant le soin de décider entre file d’attente manuelle et abandon.
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 NoneCouper le robinet à temps
Dans un pipeline à haute fréquence, une panne d’API transforme le retry en amplificateur : chaque document échoue trois fois, votre facture triple et votre latence explose. Le circuit breaker surveille les échecs sur une fenêtre glissante et bloque les appels dès qu’un seuil est franchi — cinq échecs en soixante secondes dans la configuration proposée. Un succès referme le circuit et vide le compteur.
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()
raiseVoir ce qui se passe
Tout ce dispositif reste aveugle sans traces. Loguez chaque appel avec sa durée et le schéma concerné, en succès comme en échec : c’est ce qui vous permettra plus tard de dire si la dégradation vient d’un modèle, d’un type de document ou d’un changement de prompt.
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}")
raisePoints 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