Structured outputs 2026

TL;DR En 2026, sortir du JSON valide d'un LLM n'est plus un défi — c'est une API native chez les 4 majors (OpenAI strict json_schema, Anthropic tool use, Mistral response_format, Gemini structured output). Mais "JSON valide" ≠ "données justes" : tu valides le schema et la métier (Pydantic v2 + custom validators), tu retry intelligemment, et tu prévois les escape hatches. Pour les modèles open self-hosted : Outlines, Guidance, lm-format-enforcer forcent le grammaire au niveau du token sampling. En TypeScript : Zod + Vercel AI SDK gèrent tout. Maîtriser ça = livrer des pipelines fiables qui se branchent direct sur la DB du client.

Mental model

Trois niveaux de garantie pour le JSON :

Niveau 0 (prompt-only) : "renvoie du JSON" → ça marche 92-97% du temps, ça casse à 3 AM en prod.
Niveau 1 (json_object) : le provider garantit JSON syntaxiquement valide mais le schema n'est pas forcé. 99%.
Niveau 2 (json_schema strict) : le provider force le schema exact au moment du sampling. 100% (sauf bugs).

text

                  TEXT FREE-FORM (niveau 0)
                  ┌─────────────────────────┐
                  │ "Voici les données      │
                  │  {"nom": ...}"          │
                  └──────────┬──────────────┘
                             │ parsing fragile
                             ▼
                  ┌─────────────────────────┐
   prompt:        │   JSON OBJECT (niv 1)   │
   "Réponds en    │   {"nom": "Dupont"}     │
    JSON"         │   {"nom": null}         │
                  └──────────┬──────────────┘
                             │ valid JSON, schema not forced
                             ▼
                  ┌─────────────────────────┐
   schema:        │   JSON SCHEMA STRICT    │
   {              │   (niveau 2)            │
     type:object, │                         │
     props:{...}, │   Validated at token    │
     required:[], │   sampling time         │
     additional   │                         │
     Props: false │                         │
   }              │   → format garanti      │
                  └──────────┬──────────────┘
                             │
                             ▼
                  ┌─────────────────────────┐
                  │   PYDANTIC VALIDATION   │
                  │   (niveau 3 = métier)   │
                  │  ✔ types                │
                  │  ✔ ranges               │
                  │  ✔ custom validators    │
                  │  ✔ business rules       │
                  └─────────────────────────┘

Analogie : niveau 0 = post-it manuscrit. Niveau 1 = formulaire papier rempli au stylo. Niveau 2 = formulaire web avec champs typés. Niveau 3 = formulaire web + validation côté serveur. Tu veux le 3 pour la prod.

Le réflexe de staff : "garanti" est un mot piégé. Le niveau 2 (strict) garantit la forme (le JSON parse, le schema est respecté), jamais la vérité (la valeur est juste) ni la complétude (le doc n'a pas été tronqué à max_tokens). Un pipeline d'extraction qui ne distingue pas ces trois garanties shippe des hallucinations bien formées droit dans la DB du client. Les trois s'empilent : strict (forme) → Pydantic (métier) → confiance + flag humain (vérité incertaine). En entretien, c'est exactement la distinction qu'on cherche à te faire verbaliser.

Le mental model du décodeur (ce qui te distingue d'un junior)

Côté Anthropic, en 2026, il existe deux surfaces pour la même garantie de forme, et savoir laquelle prendre est un signal de séniorité :

Surface	Quand	Mécanisme
`client.messages.parse(output_config={"format": Schema})`	Un schema garanti par appel (extraction, classification)	Le SDK envoie le schema, l'API force le format au décodage, le SDK valide Pydantic au retour → `resp.parsed_output`
`tool_choice={"type": "tool", "name": ...}` ou `{"type": "any"}`	Le modèle doit router parmi N schemas (dispatcher), ou tu veux le résultat dans un `tool_use` block pour ton agent loop	Un tool par branche, `input_schema` = ton schema, tu re-valides Pydantic toi-même

Règle : messages.parse() pour un schema ; tool_choice quand le modèle choisit la branche. Ne fais pas du tool_choice détourné quand messages.parse() suffit — c'est plus de code, plus de fragilité, et tu perds la validation SDK gratuite.

Code minimal

OpenAI strict json_schema :

python

from openai import OpenAI
from pydantic import BaseModel
from typing import Literal

client = OpenAI()

class Devis(BaseModel):
    prix_eur: float
    delai_jours: int
    options: list[str]
    contact_email: str

resp = client.responses.parse(
    model="gpt-5",
    instructions="Tu es un commercial. Construis un devis structuré.",
    input="Demande : assurance habitation, 80m2, Lyon, 2 personnes",
    text_format=Devis,
)
print(resp.output_parsed.prix_eur)

Anthropic — messages.parse() natif (la bonne approche en 2026) :

python

import anthropic
from pydantic import BaseModel

client = anthropic.Anthropic()  # AsyncAnthropic pour un serveur — voir plus bas

class Devis(BaseModel):
    prix_eur: float
    delai_jours: int
    options: list[str]
    contact_email: str

# La voie canonique 2026 : output_config.format + Pydantic, validé côté SDK.
resp = client.messages.parse(
    model="claude-opus-4-8",  # flagship 2026 (5$/25$ par M tok, contexte 1M)
    max_tokens=1024,
    messages=[{"role": "user", "content": "Demande : assurance habitation, 80m2, Lyon"}],
    output_config={"format": Devis},  # schema strict appliqué + validé Pydantic
)
devis: Devis | None = resp.parsed_output  # None si refusal / max_tokens → TOUJOURS guarder
if devis is None:
    # stop_reason == "refusal" (safety) ou "max_tokens" (tronqué) → pas de JSON exploitable
    raise ValueError(f"Extraction échouée, stop_reason={resp.stop_reason}")
print(resp.usage.input_tokens, resp.usage.output_tokens)  # logge le coût

Piège 2026 #1 — ne PAS confondre extended thinking et structured outputs. Sur claude-opus-4-8, le thinking historique avec budget_tokens renvoie un HTTP 400 (paramètre supprimé, comme temperature/top_p/top_k) : utilise thinking={"type": "adaptive"} + output_config={"effort": "low"|"medium"|"high"}. Pour de l'extraction stricte, laisse l'effort bas (low/medium) — tu ne paies pas du raisonnement inutile, et tu gagnes en latence.
Piège 2026 #2 — pas de temperature=0 sur Opus 4.8. L'astuce historique "temperature 0 pour du déterministe" est morte : le paramètre est rejeté (400). Le déterminisme se pilote désormais par effort: "low" + un prompt serré. De toute façon temperature=0 n'a jamais garanti des sorties identiques bit-à-bit.
Piège 2026 #3 — structured outputs ⊥ citations. output_config.format + citations → 400. Si tu as besoin des deux (extraction sourcée), fais deux passes ou utilise le tool_choice forcé.
Piège 2026 #4 — refusal avant de lire content. Les classifieurs safety peuvent renvoyer un HTTP 200 avec stop_reason="refusal" et un content vide. resp.parsed_output est alors None. Lire resp.content[0] directement → IndexError. Toujours brancher sur resp.parsed_output is None / stop_reason.

Anthropic — tool_choice forcé (l'ancienne voie, à connaître pour les legacy / le routing multi-tools). Avant messages.parse(), on détournait un tool dont l'input_schema était le schema cible. Toujours valide quand tu veux que le modèle choisisse parmi plusieurs schemas (dispatcher) :

python

tools = [{
    "name": "produce_devis",
    "description": "Produit un devis structuré",
    "input_schema": Devis.model_json_schema(),
}]
resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "tool", "name": "produce_devis"},  # force ce tool
    messages=[{"role": "user", "content": "Demande..."}],
)
tool_block = next(b for b in resp.content if b.type == "tool_use")
devis = Devis(**tool_block.input)  # re-valide TOUJOURS, même via tool_use

Règle de staff : messages.parse() pour un schema garanti ; tool_choice quand le modèle doit router vers l'un de N schemas (voir Discriminated Unions plus bas — côté Anthropic c'est un tool par branche + tool_choice={"type": "any"}).

Mistral :

python

from mistralai import Mistral
client = Mistral(api_key=...)

resp = client.chat.complete(
    model="mistral-large-2-2026",
    messages=[{"role": "user", "content": "Demande..."}],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "Devis",
            "schema": Devis.model_json_schema(),
            "strict": True,
        },
    },
)
devis = Devis.model_validate_json(resp.choices[0].message.content)

TypeScript / Zod via Vercel AI SDK :

import { generateObject } from "ai";
import { openai } from "@ai-sdk/openai";
import { z } from "zod";

const devisSchema = z.object({
  prixEur: z.number(),
  delaiJours: z.number().int(),
  options: z.array(z.string()),
  contactEmail: z.string().email(),
});

const { object: devis } = await generateObject({
  model: openai("gpt-5"),
  schema: devisSchema,
  prompt: "Demande : assurance habitation, 80m2, Lyon",
});
console.log(devis.prixEur);

Comment le "strict" fonctionne vraiment (mental model staff)

Un junior croit que strict: true = "le provider relit le JSON et le rejette s'il est mauvais". Faux. Le strict mode est appliqué au moment du décodage, pas après. À chaque token généré, le moteur d'inférence applique un masque de logits : il met à -inf la probabilité de tout token qui rendrait le JSON invalide vis-à-vis du schema. Concrètement :

Le schema JSON est compilé en automate à états finis (FSM) / grammaire (CFG selon le provider).
À chaque position, l'état de l'automate détermine l'ensemble des tokens légaux. On ne sample QUE dedans.
Résultat : il est mathématiquement impossible de produire du JSON hors-schema. Pas de retry, pas de parse fragile.

text

   logits bruts du modèle        masque FSM (état: "après {")        sample
   ┌──────────────┐              ┌──────────────┐                   ┌──────────┐
   │ "prix"  0.4  │              │ "prix"   ok  │                   │          │
   │ "Bonjour"0.3 │   ─────►     │ "Bonjour"-inf│   ─────►          │  "prix"  │
   │ "{"     0.2  │              │ "{"     -inf │                   │          │
   │ ...          │              │ ...          │                   │          │
   └──────────────┘              └──────────────┘                   └──────────┘

Les trois conséquences que tu DOIS connaître en entretien :

Coût de compilation au premier appel. Compiler la grammaire d'un schema neuf prend du temps (latence ajoutée sur la 1ʳᵉ requête). Les providers cachent la grammaire compilée ~24h. Donc : un schema stable = cache chaud ; un schema généré dynamiquement par requête = cache toujours froid → latence p99 qui explose. Garde tes schemas figés.
Le strict garantit la FORME, jamais la VÉRITÉ. Le masque force prix_eur à être un float. Il ne peut PAS forcer que ce soit le BON prix. Une hallucination bien formée passe le strict mode sans broncher. D'où le niveau 3 Pydantic (ranges, cross-field, confiance).
Énum/contraintes non supportées par l'automate (regex pattern, minLength, minimum…). La plupart des providers strict ne compilent PAS ces contraintes dans le FSM — elles sont silencieusement droppées côté API et re-validées côté SDK/Pydantic. Si tu comptes dessus pour la sécurité, valide-les toi-même, ne suppose pas que le provider le fait.

Limites du strict Anthropic à connaître (à citer de mémoire en entretien) : pas de schemas récursifs, pas de contraintes numériques (minimum/maximum/multipleOf), pas de contraintes de longueur de string (minLength/maxLength), pas de contraintes complexes d'array. additionalProperties: false est requis sur chaque objet. Sont supportés : types de base (object/array/string/integer/number/boolean/null), enum, const, anyOf, allOf, $ref/$def, et les formats string (date-time, date, time, duration, email, uri, uuid, ipv4/ipv6, hostname). Les SDK Python/TS retirent les contraintes non supportées du schema envoyé et les revalident client-side — donc ça "marche", mais le forcing token-level ne s'applique qu'aux types/enum/structure. Conséquence sécurité : si tu comptes sur pattern: "^\d{9}$" (SIREN) pour garantir un format, le FSM ne le force PAS — c'est Pydantic qui l'attrape côté SDK. Ne suppose jamais que l'API a validé ta regex.

Modèles supportés (2026) : structured outputs (output_config.format + strict tool use) tournent sur claude-opus-4-8, claude-sonnet-4-6, claude-haiku-4-5, et claude-fable-5. Choix de staff : extraction de masse → claude-haiku-4-5 (1 $/5 $ par M tok) si le schema est simple et le doc propre ; extraction délicate (KYC, juridique, multi-pièces) → claude-opus-4-8 (5 $/25 $) avec effort: "low". Le strict mode marche identiquement sur les trois ; ce que tu paies en montant en gamme, c'est la qualité de la valeur extraite, pas la garantie de forme.

Production : async, caching, observabilité

Le code minimal ci-dessus est synchrone et naïf. Voici ce qu'un staff engineer met réellement en prod sur un pipeline d'extraction Anthropic à volume :

python

import asyncio
from anthropic import AsyncAnthropic, APITimeoutError, RateLimitError, APIStatusError
from pydantic import BaseModel

# 1) Client async + retries SDK + timeout par appel
client = AsyncAnthropic(max_retries=4)  # backoff exponentiel + jitter intégrés

SYSTEM = [  # 2) Préfixe STABLE → prompt caching. Mets les instructions + le schema ici.
    {
        "type": "text",
        "text": "Tu es un extracteur. Règles d'extraction : ...",  # long, figé
        "cache_control": {"type": "ephemeral"},  # ~0.1x en lecture sur les appels suivants
    }
]

class Facture(BaseModel):
    montant_eur: float
    devise: str
    confiance: float

async def extract(text: str) -> Facture | None:
    try:
        resp = await client.messages.parse(
            model="claude-opus-4-8",
            max_tokens=1024,
            system=SYSTEM,                       # stable d'abord (cache)
            messages=[{"role": "user", "content": text}],  # volatile ensuite
            output_config={"format": Facture, "effort": "low"},
            timeout=30.0,                        # 3) deadline dure par appel
        )
    except (APITimeoutError, RateLimitError, APIStatusError) as e:
        # 4) exceptions TYPÉES, jamais de match sur le message
        log.warning("extract_failed", error=type(e).__name__)
        return None
    # 5) observabilité du coût + du cache
    u = resp.usage
    log.info("extract_ok",
             in_tok=u.input_tokens,
             out_tok=u.output_tokens,
             cache_read=u.cache_read_input_tokens,   # si 0 sur des appels répétés → invalidateur silencieux
             cost_usd=(u.input_tokens*5 + u.output_tokens*25)/1e6)
    return resp.parsed_output

async def extract_batch(docs: list[str]) -> list[Facture | None]:
    sem = asyncio.Semaphore(8)  # 6) borne la concurrence — sinon tu te fais rate-limit
    async def bounded(d):
        async with sem:
            return await extract(d)
    return await asyncio.gather(*[bounded(d) for d in docs])

Les six leviers de prod, dans l'ordre où ils te sauvent la vie :

Levier	Pourquoi	Métrique à surveiller
`AsyncAnthropic` + `asyncio.gather`	Parallélise les extractions I/O-bound	Débit (docs/s)
`max_retries` + exceptions typées	Survit aux 429/529/timeout transitoires	Taux d'erreur après retries
`timeout` par appel	Empêche un appel pendu de bloquer le pool	Latence p99
Prompt caching (système+schema figés)	Le préfixe stable se lit à ~0.1x	`cache_read_input_tokens` / appel
`effort: "low"` sur extraction	Pas de raisonnement inutile	Output tokens / doc
`Semaphore`	Borne la concurrence sous le rate limit	429 / minute

Le piège du batch latence-insensible : si l'extraction n'est PAS temps réel (nightly run, backfill), passe par la Batches API (50% du prix, async, complète en général en < 1h, max 24h, 100 000 req/batch). Sur 1M docs/mois, c'est la différence entre une facture à 4 chiffres et une à 5. La Batches API supporte les structured outputs sans surcoût de code.

Le budget de latence (comment un staff raisonne sur le p99)

Un junior optimise le coût ; un staff optimise le p99 de bout en bout, parce que c'est lui qui casse les SLA et fait timeout les endpoints amont. Décompose la latence d'un appel d'extraction strict :

text

   t_réseau  +  t_compilation_grammaire  +  t_inférence  +  t_validation_pydantic
   (~50ms)      (0ms si cache chaud,         (domine :        (~1ms, négligeable)
                ~100-800ms si cache froid)    output tokens)

t_compilation_grammaire : payé UNIQUEMENT au 1ᵉʳ appel d'un schema neuf (compilation FSM/grammaire), puis caché ~24h côté provider. Schema figé → 0ms. Schema généré dynamiquement par requête → tu repaies ça à chaque appel et ton p99 explose. C'est la cause #1 de p99 surprise en prod.
t_inférence ∝ output tokens. Le strict mode ne réduit PAS les output tokens (le modèle écrit autant de JSON). Ce qui les réduit : effort: "low" (pas de raisonnement inutile), un schema plat (moins de nesting = moins de tokens de structure), et virer les champs description_longue que personne ne lit.
Streaming : pour un gros schema (> ~16K tokens de sortie), streame (max_tokens peut monter à 128K sur Opus 4.8, mais au-delà de ~16K le SDK timeout en non-streaming). Le messages.parse() SDK gère le partial parse ; tu peux afficher les champs au fur et à mesure pour l'UX.

Métrique à instrumenter en plus de celles du tableau : schema_compile_rate = fraction d'appels où cache_read_input_tokens == 0 sur un schema censé être stable. Si ça dérive vers le haut, quelqu'un a introduit un invalidateur silencieux (timestamp, UUID, sérialisation non déterministe du schema) dans le préfixe.

Failure modes (ce qui casse en prod, pas en démo)

Failure mode	Symptôme	Cause racine	Fix staff
Hallucination bien formée	JSON valide, valeur fausse	Strict ne valide que la forme	Champ `confiance` + cross-field validators + flag humain sous seuil
Troncature	JSON coupé, `parsed_output=None`	`stop_reason="max_tokens"`	Augmenter `max_tokens` ; streamer ; raccourcir le schema
Refusal safety	`content` vide, 200	Classifieur déclenché	Brancher sur `stop_reason="refusal"` AVANT de lire `content`
Cache toujours froid	p99 latence élevée, `cache_read=0`	Schema/system généré dynamiquement, timestamp dans le préfixe	Figer le préfixe ; sérialiser le schema de façon déterministe
Énum à 200 valeurs	Latence + qualité qui chutent	Le FSM explose, le modèle "cherche" dans l'énum	Extraction libre + post-validation, ou hiérarchiser l'énum
Schema drift silencieux	La DB casse après un déploiement	Quelqu'un a changé le Pydantic sans migration	Schema-hash test en CI (voir section Testing)
`additionalProperties` oublié	Le modèle invente des champs	Défaut JSON schema = `true`	`model_config = ConfigDict(extra="forbid")`

Cas d'usage concrets

RH — Parsing CV → Pydantic schema, ATS éditeur recrutement IT Paris (40 consultants)

Problème : cabinet de recrutement IT, 4 500 CV reçus/mois (LinkedIn, mail, Jobboards). Chaque consultant doit lire, qualifier, indexer dans Bullhorn/ATS. 7 min/CV × 4500 = 525h/mois = 3.3 ETP juste pour la saisie. Les CV en PDF / DOCX / sans format standard → erreurs.

Solution : parsing CV → Pydantic schema strict → matching automatique contre fiches de poste actives → ranking par fit + push ATS via API.

python

from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Literal
from datetime import date

class Experience(BaseModel):
    poste: str
    entreprise: str
    secteur: str | None
    date_debut: date
    date_fin: date | None
    en_cours: bool
    technos: list[str]
    description_short: str = Field(max_length=500)

class Diplome(BaseModel):
    intitule: str
    ecole: str
    annee_obtention: int
    niveau: Literal["bac", "bac+2", "bac+3", "bac+5", "doctorat"]

class CV(BaseModel):
    nom: str
    prenom: str
    email: EmailStr
    telephone: str | None
    linkedin: str | None
    titre_pro: str
    seniorite: Literal["junior", "confirmé", "senior", "lead", "manager"]
    annees_experience: float = Field(ge=0, le=50)
    technos_majeures: list[str] = Field(max_length=20)
    soft_skills: list[str] = Field(max_length=10)
    experiences: list[Experience]
    diplomes: list[Diplome]
    langues: list[dict]  # [{"langue": "Anglais", "niveau": "C1"}]
    mobilite: list[str]  # villes
    pretentions_eur: int | None
    confiance_extraction: float = Field(ge=0, le=1)

    @field_validator("annees_experience")
    @classmethod
    def coherent_with_experiences(cls, v: float, info) -> float:
        # Sanity check : somme expériences ~ années annoncées
        return v

def parse_cv(file_path: str) -> CV:
    text = extract_text(file_path)  # pypdf / python-docx
    resp = client.responses.parse(
        model="gpt-5",
        instructions=CV_EXTRACTION_PROMPT,
        input=f"CV à parser :\n\n{text}",
        text_format=CV,
        temperature=0.0,
    )
    cv = resp.output_parsed
    if cv.confiance_extraction < 0.85:
        flag_for_human(file_path, cv)
    return cv

Gains chiffrés :

Saisie CV : 7 min → 20 sec auto + 1 min validation
3.3 ETP → 0.6 ETP (les autres font du sourcing actif et placements)
Coût IA : ~$0.04/CV × 4500 = $180/mois (~165€)
Capacité placements +35% (consultants libérés)
TJM mission : 1 300€/j × 25 j = 32 500€ + 1 500€/mois MCO

E-commerce — Enrichissement catalogue produit auto (marketplace mode 25 000 SKU)

Problème : marketplace mode, 25 000 SKU + 800 nouveautés/mois. Les fiches produits arrivent en bordel (titre marchand court, photos, données techniques manquantes). SEO catastrophique, conversion -. Une équipe de 4 content managers retravaille les fiches à 30 min/SKU.

Solution : Pydantic schema riche (titre SEO, meta-description, attributs faceted, tags, description longue, alt-text images). Vision GPT-5 sur les photos pour extraire couleurs/coupe/matière. Output direct dans Shopify via API.

python

class FicheProduit(BaseModel):
    titre_court: str = Field(max_length=80, description="Pour listing")
    titre_seo: str = Field(max_length=70, description="<title> tag")
    meta_description: str = Field(max_length=160, description="<meta description>")
    description_courte: str = Field(max_length=300, description="Pour fiche produit haut")
    description_longue_md: str = Field(description="Markdown, 600-1200 chars, storytelling brand")

    couleur_principale: str
    couleurs_secondaires: list[str]
    matiere_principale: str
    coupe: Literal["slim", "regular", "loose", "oversized"] | None
    style: list[Literal["casual", "formel", "sport", "vintage", "minimaliste"]]

    tailles_disponibles: list[str]
    saisons: list[Literal["printemps", "été", "automne", "hiver"]]
    occasions: list[Literal["quotidien", "soirée", "bureau", "sport", "voyage"]]

    tags_seo: list[str] = Field(max_length=15)
    alt_texts_images: list[str]  # 1 par image
    confiance: float = Field(ge=0, le=1)

Gains chiffrés :

Enrichissement fiche : 30 min → 45 sec auto + 3 min validation
4 content managers → 1 (les 3 autres font du brief créatif, shooting)
SEO traffic organique : +47% sur 6 mois
Conversion : +18% (meilleures fiches, alt-text images = accessibilité + SEO)
Coût IA : ~$0.08/SKU × 800 nouveautés + 25k rebuild = $2 200 one-shot + $64/mois
TJM mission : 1 200€/j × 30 j = 36 000€ + 1 800€/mois MCO

FinTech — KYC complet : selfie + pièce d'identité → profil client (néobanque pro B2B)

Problème : néobanque B2B (TPE), KYC dirigeant + KYB entreprise + RBE (Registre des Bénéficiaires Effectifs). Pile compliance : 8 formulaires différents, 12 jours moyens de traitement, 28% taux abandon par le créateur d'entreprise.

Solution : pipeline structured outputs en 4 étapes (CNI dirigeant, KBIS entreprise, justif domicile, RBE) → consolidation dans un schéma unique → décision auto (green/amber/red).

python

class IdentitePersonne(BaseModel):
    nom: str
    prenom: str
    date_naissance: date
    lieu_naissance: str
    nationalite: str
    numero_piece: str
    type_piece: Literal["CNI", "PASSEPORT", "TITRE_SEJOUR"]
    date_expiration: date

class Entreprise(BaseModel):
    raison_sociale: str
    siren: str = Field(pattern=r"^\d{9}$")
    forme_juridique: str
    capital_eur: float
    date_immatriculation: date
    adresse_siege: str
    code_naf: str

class BeneficiaireEffectif(BaseModel):
    nom: str
    prenom: str
    date_naissance: date
    pourcentage_detention: float = Field(ge=0, le=100)
    nature_controle: Literal["capital", "vote", "direction_de_fait"]

class KYCComplet(BaseModel):
    dirigeant: IdentitePersonne
    entreprise: Entreprise
    beneficiaires_effectifs: list[BeneficiaireEffectif]
    pep_check: bool = Field(description="Politiquement Exposé Person")
    sanctions_check: bool = Field(description="Sur liste UE/US/OFAC")
    coherence_data: bool = Field(description="Données cohérentes entre docs")
    raisons_alerte: list[str]
    decision_auto: Literal["green", "amber", "red"]
    confiance_globale: float = Field(ge=0, le=1)

Gains chiffrés :

Délai onboarding : 12 j → 18h (la majorité)
Taux abandon : 28% → 11%
Capacité KYC : 80/j → 350/j sans staffing additionnel
Coût IA : ~$0.30/dossier complet × 80/j = $24/j (720€/mois)
ROI : +17% conversion × 350 dossiers/j × 80€ MRR LTV = significatif
TJM mission : 1 500€/j × 45 j = 67 500€ + 3 500€/mois MCO

Exemple end-to-end

Pipeline RH complet : 100 CV → parsing Pydantic → match contre 10 fiches de poste → ranking + explication.

python

"""
hr_matching_pipeline.py — Pipeline RH end-to-end.

Workflow:
1. Ingest 100 CV (PDF/DOCX) depuis S3
2. Extract → Pydantic CV (parallèle, GPT-5 + structured outputs)
3. Load 10 fiches de poste (Pydantic JobPosting)
4. Pour chaque (CV, Poste) : score de match + explication structurée
5. Ranking final, push CRM, mail recruteur
"""

from __future__ import annotations
import os
import json
import asyncio
from pathlib import Path
from datetime import date, datetime
from typing import Literal
from pydantic import BaseModel, Field, EmailStr
from openai import AsyncOpenAI
from pypdf import PdfReader
from docx import Document
import psycopg
import boto3

aclient = AsyncOpenAI()
s3 = boto3.client("s3")

# ---------- 1. Schemas ----------

class Experience(BaseModel):
    poste: str
    entreprise: str
    annees: float
    technos: list[str]
    description: str = Field(max_length=400)

class CV(BaseModel):
    nom: str
    prenom: str
    email: EmailStr
    telephone: str | None
    titre_pro: str
    seniorite: Literal["junior", "confirmé", "senior", "lead", "manager"]
    annees_experience: float = Field(ge=0, le=50)
    technos_majeures: list[str]
    soft_skills: list[str]
    experiences: list[Experience]
    formation_max: Literal["bac", "bac+2", "bac+3", "bac+5", "doctorat"]
    langues: list[dict]
    mobilite: list[str]
    pretentions_eur_annuelles: int | None
    confiance_extraction: float = Field(ge=0, le=1)

class JobPosting(BaseModel):
    job_id: str
    titre: str
    seniorite_min: Literal["junior", "confirmé", "senior", "lead", "manager"]
    annees_exp_min: float
    technos_obligatoires: list[str]
    technos_bonus: list[str]
    formation_min: Literal["bac", "bac+2", "bac+3", "bac+5", "doctorat"]
    langues_requises: list[str]
    localisation: str
    salaire_cible_eur: int
    description_court: str

class MatchScore(BaseModel):
    technos_score: float = Field(ge=0, le=1)
    seniorite_score: float = Field(ge=0, le=1)
    experience_score: float = Field(ge=0, le=1)
    soft_skills_score: float = Field(ge=0, le=1)
    salaire_compatible: bool
    mobilite_ok: bool
    score_global: float = Field(ge=0, le=1)
    points_forts: list[str] = Field(max_length=5)
    points_faibles: list[str] = Field(max_length=5)
    decision: Literal["fortement_recommandé", "à_considérer", "non_adapté"]
    raisonnement: str = Field(max_length=400)

# ---------- 2. Extract text from CV file ----------

def read_cv_file(path: Path) -> str:
    if path.suffix.lower() == ".pdf":
        return "\n".join((p.extract_text() or "") for p in PdfReader(str(path)).pages)
    if path.suffix.lower() in (".docx", ".doc"):
        return "\n".join(p.text for p in Document(str(path)).paragraphs)
    return path.read_text()

# ---------- 3. Parse CV ----------

CV_PROMPT = """Tu es un expert RH spécialisé tech. Extrais les données du CV avec rigueur.

Règles :
- Calcule annees_experience comme somme des durées de poste (post-bac+5)
- seniorite : junior <2 ans, confirmé 2-5, senior 5-10, lead 10+, manager si encadrement
- technos_majeures : top 10 technos vraiment maîtrisées (pas tout listé)
- confiance_extraction : 1.0 si CV très clair, baisser si infos manquantes
"""

async def parse_cv_file(s3_key: str) -> CV | None:
    tmp = Path(f"/tmp/{s3_key.replace('/', '_')}")
    s3.download_file(os.environ["S3_BUCKET"], s3_key, str(tmp))
    text = read_cv_file(tmp)
    if len(text) < 200:
        print(f"[!] CV trop court: {s3_key}")
        return None
    try:
        resp = await aclient.responses.parse(
            model="gpt-5",
            instructions=CV_PROMPT,
            input=text[:30000],  # cap input
            text_format=CV,
            temperature=0.0,
        )
        return resp.output_parsed
    except Exception as e:
        print(f"[!] Parse failed for {s3_key}: {e}")
        return None

# ---------- 4. Match score ----------

MATCH_PROMPT = """Tu es un consultant RH senior. Évalue le fit entre un CV et une fiche de poste.

Méthode :
- technos_score : (intersection technos_obligatoires) / max(1, len(technos_obligatoires))
- seniorite_score : ordinal match (1.0 si pile, -0.25 par cran d'écart)
- experience_score : min(1.0, cv.annees_experience / poste.annees_exp_min)
- soft_skills_score : interprétation
- score_global : moyenne pondérée (40% technos, 20% seniorite, 20% expérience, 10% soft, 10% formation)
- decision : >0.75 fortement_recommandé, 0.55-0.75 à_considérer, <0.55 non_adapté
"""

async def score_match(cv: CV, job: JobPosting) -> MatchScore:
    payload = json.dumps({"cv": cv.model_dump(mode="json"), "poste": job.model_dump(mode="json")}, default=str)
    resp = await aclient.responses.parse(
        model="gpt-5",
        instructions=MATCH_PROMPT,
        input=payload,
        text_format=MatchScore,
        temperature=0.0,
    )
    return resp.output_parsed

# ---------- 5. Persistence ----------

def save_match(cv: CV, job_id: str, score: MatchScore):
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO matches
                (cv_email, job_id, score_global, decision, payload, created_at)
                VALUES (%s, %s, %s, %s, %s, now())
                ON CONFLICT (cv_email, job_id) DO UPDATE
                SET score_global=EXCLUDED.score_global, decision=EXCLUDED.decision, payload=EXCLUDED.payload""",
                (cv.email, job_id, score.score_global, score.decision, score.model_dump_json()),
            )

# ---------- 6. Pipeline ----------

async def run_pipeline(cv_keys: list[str], jobs: list[JobPosting]):
    sem = asyncio.Semaphore(8)

    async def bounded_parse(k):
        async with sem:
            return await parse_cv_file(k)

    print(f"[+] Parsing {len(cv_keys)} CVs...")
    cvs = await asyncio.gather(*[bounded_parse(k) for k in cv_keys])
    valid_cvs = [c for c in cvs if c is not None]
    print(f"[+] {len(valid_cvs)} CVs parsed successfully")

    print(f"[+] Scoring {len(valid_cvs)}x{len(jobs)} pairs...")

    async def bounded_score(cv, job):
        async with sem:
            try:
                return cv, job, await score_match(cv, job)
            except Exception as e:
                print(f"[!] Score failed {cv.email}/{job.job_id}: {e}")
                return cv, job, None

    pairs = [(cv, job) for cv in valid_cvs for job in jobs]
    results = await asyncio.gather(*[bounded_score(cv, job) for cv, job in pairs])

    # Save + rank
    by_job: dict[str, list] = {j.job_id: [] for j in jobs}
    for cv, job, score in results:
        if score is None:
            continue
        save_match(cv, job.job_id, score)
        by_job[job.job_id].append((cv, score))

    # Top 10 par job
    report = {}
    for job_id, pairs in by_job.items():
        pairs.sort(key=lambda x: x[1].score_global, reverse=True)
        report[job_id] = [
            {
                "candidat": f"{cv.prenom} {cv.nom}",
                "email": cv.email,
                "score": round(s.score_global, 2),
                "decision": s.decision,
                "points_forts": s.points_forts,
                "raisonnement": s.raisonnement,
            }
            for cv, s in pairs[:10]
        ]
    return report

# ---------- 7. Demo run ----------

async def main():
    # 1. List CV files in S3
    response = s3.list_objects_v2(Bucket=os.environ["S3_BUCKET"], Prefix="cvs/inbox/")
    cv_keys = [o["Key"] for o in response.get("Contents", [])][:100]

    # 2. Load 10 fiches de poste from DB
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT payload FROM job_postings WHERE status='active' LIMIT 10")
            jobs = [JobPosting.model_validate_json(r[0]) for r in cur.fetchall()]

    report = await run_pipeline(cv_keys, jobs)
    print(json.dumps(report, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())

Coût observé : ~$0.07 par CV (parsing) + ~$0.05 par couple (CV × Job) = sur 100 CV × 10 jobs, ~$57. Soit < 60€ pour un batch complet, vs des dizaines d'heures consultant.

Patterns courants

1. Pydantic Discriminated Unions pour outputs polymorphes

python

from typing import Annotated, Literal
from pydantic import BaseModel, Field

class ReponseDevis(BaseModel):
    type: Literal["devis"]
    prix_eur: float
    delai_jours: int

class ReponseRefus(BaseModel):
    type: Literal["refus"]
    motif: str

class ReponseRedirection(BaseModel):
    type: Literal["redirection"]
    service_cible: str

ReponseLLM = Annotated[
    ReponseDevis | ReponseRefus | ReponseRedirection,
    Field(discriminator="type"),
]

Force le modèle à choisir une branche explicite. Parfait pour les routeurs / dispatchers.

2. Retry avec correction ("re-ask")

Quand la validation Pydantic métier échoue (mais le JSON est syntaxiquement valide) :

python

from pydantic import ValidationError

def call_with_retry(input_text: str, schema, max_retries: int = 2):
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            if attempt == 0:
                user = input_text
            else:
                user = f"{input_text}\n\nTa précédente réponse a échoué la validation : {last_error}. Corrige et renvoie."
            resp = client.responses.parse(model="gpt-5", input=user, text_format=schema)
            return resp.output_parsed
        except ValidationError as e:
            last_error = str(e)
    raise RuntimeError(f"Échec après {max_retries+1} tentatives : {last_error}")

3. Escape hatch : champ "unable_to_extract"

python

class Facture(BaseModel):
    montant_eur: float | None = None
    unable_to_extract: bool = False
    reason_if_unable: str | None = None

Donne au modèle une porte de sortie propre. Évite qu'il invente des valeurs pour satisfaire le schema.

4. Streaming structured outputs (partial parse)

python

# OpenAI Responses API stream
stream = client.responses.parse(
    model="gpt-5", input="...", text_format=Devis, stream=True,
)
for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    if event.type == "response.completed":
        final: Devis = event.response.output_parsed

Utile pour UX progressive (afficher les champs au fur et à mesure).

5. Outlines / lm-format-enforcer pour modèles open self-hosted

python

# pip install outlines
import outlines
from pydantic import BaseModel

class Devis(BaseModel):
    prix_eur: float
    delai_jours: int

model = outlines.models.transformers("mistralai/Mistral-Small-3-Instruct")
generator = outlines.generate.json(model, Devis)
result: Devis = generator("Demande devis...")

Outlines force le grammaire au niveau du token sampling → garanti même sur les petits modèles open. Indispensable quand tu n'as pas d'API structured output (Llama self-hosted brut).

6. Instructor (cross-provider Pydantic-first)

python

# pip install instructor
import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())
devis: Devis = client.chat.completions.create(
    model="gpt-5",
    response_model=Devis,  # <- magic
    messages=[{"role": "user", "content": "Demande..."}],
)

Instructor wrappe OpenAI, Anthropic, Mistral, Gemini, Ollama, etc., en une API Pydantic unifiée + retries + validation.

Versions & écosystème 2026

Outil	Version mai 2026	Force
OpenAI strict json_schema	GA	Le plus mûr, niveau prod
Anthropic `messages.parse()` + `output_config.format`	GA	Voie canonique 2026 (Opus 4.8 / Sonnet 4.6 / Haiku 4.5 / Fable 5) ; `tool_choice` forcé reste pour le routing N-schemas
Mistral response_format	GA	json_schema strict supporté
Gemini structured output	GA	Performant sur Pro 2.5
Pydantic	2.10+	Validation Python standard
Zod	3.24+	Validation TS standard
Instructor	1.10+	Pydantic cross-provider
Outlines	0.1.x	Self-host, grammar-level
Guidance	0.2.x	Templates + grammars
lm-format-enforcer	0.11+	Léger, hooke n'importe quel model HF
Vercel AI SDK	4.x	Zod + streaming TS-first
LlamaIndex	0.12+	Pydantic-first orchestration

Tendance 2026 : tout le monde a convergé vers du JSON schema strict côté API + Pydantic/Zod côté SDK. Les approches "grammar-level" (Outlines, Guidance) survivent pour le self-host.

Pitfalls

additionalProperties: true (par défaut JSON schema) → strict mode pas vraiment strict. Avec Pydantic 2, mets model_config = ConfigDict(extra="forbid").
Optional[X] mal configuré → strict mode rejette. En Pydantic, préfère X | None = None avec default explicite.
Union complexes sans discriminator → modèle confus, mauvais matching. Utilise Annotated[A | B, Field(discriminator="type")].
Numpy float / Decimal dans le schema → OpenAI strict ne supporte que les types JSON standard. Use float, pas Decimal.
description= ignorées dans certaines libs → toujours mettre des description= dans tes Field(), ça améliore la qualité de l'output (le modèle les voit comme partie du schema).
Énums longues (>50 valeurs) → augmente la latence ET tend à dégrader la qualité. Préfère extraction libre + post-validation.
Schemas récursifs → certains providers ne supportent pas. Test avant prod. Si non, "aplatis" en 2 niveaux.
Retry sans backoff → tu DDoS le provider sur erreur transitoire. Toujours exponential backoff avec jitter.
Validation Pydantic côté serveur seulement → tu acceptes du JSON dans la base parce que "le modèle l'a généré". Toujours re-valider Pydantic à la persistence.
temperature > 0 sur extraction → variabilité non nécessaire. Sur OpenAI/Mistral/Gemini → temperature=0 pour l'extraction stricte, 0.3-0.7 pour la création. Sur Anthropic Opus 4.8 / Sonnet 4.6, temperature (et top_p/top_k) sont supprimés (HTTP 400) — le déterminisme se pilote par effort: "low" + prompt serré, pas par un sampling param.

Pricing / ROI client

Le ROI des structured outputs vient de l'élimination de la couche "parsing fragile + retries + corrections humaines". Sur un pipeline d'extraction :

Étape	Sans structured	Avec structured strict
Coût LLM	$0.02/doc	$0.022/doc (+10% car json_schema overhead)
Code parsing	200 LoC fragiles	5 LoC `model_validate_json`
Taux d'échec parse	2-5%	<0.1%
Retries moyens	0.15	0.02
Coût final	$0.023 + erreurs humaines	$0.022 + erreurs métier seulement

Sur 100 000 docs/mois : ~30 erreurs/mois en moins, chacune coûtant 5-15 min à un humain = ~10h économisées = ~500€/mois. Pour un client à 1M docs/mois, ça monte vite.

Pitch : "Aujourd'hui votre pipeline doit gérer les exceptions JSON, ça vous coûte X. Je remplace par un schema strict avec validation Pydantic, vous descendez à <0.1% d'erreurs format, vous n'avez plus que des erreurs métier à traiter."

Testing / Eval

1. Tests unitaires Pydantic

python

def test_cv_schema_validation():
    cv_json = {
        "nom": "Dupont", "prenom": "Jean", "email": "[email protected]",
        "titre_pro": "Dev", "seniorite": "senior",
        "annees_experience": 8.0,
        # ...
    }
    cv = CV.model_validate(cv_json)
    assert cv.seniorite == "senior"

def test_cv_schema_rejects_invalid():
    with pytest.raises(ValidationError):
        CV.model_validate({"nom": "X", "annees_experience": -1})

2. Eval sur dataset annoté

python

def test_pipeline_extraction():
    dataset = load_annotated_cvs()  # 50 CV + ground truth
    scores = []
    for cv_path, expected in dataset:
        cv = asyncio.run(parse_cv_file(cv_path))
        scores.append(field_accuracy(cv, expected))
    avg_acc = sum(scores) / len(scores)
    assert avg_acc > 0.92  # seuil régression

3. Schema diffing en CI

python

import hashlib
import json

def test_schema_no_breaking_change():
    new_schema = CV.model_json_schema()
    new_hash = hashlib.sha256(json.dumps(new_schema, sort_keys=True).encode()).hexdigest()
    known_hash = open("schema_hash.txt").read().strip()
    assert new_hash == known_hash, "Schema CV a changé : bump version + migration DB"

Quand utiliser / éviter

Utilise structured outputs strict quand :

Tu pushes vers une DB / API qui attend un format strict
Tu fais de l'extraction (facture, CV, KYC, fiche produit)
Tu fais du routing / classification
Tu construis un pipeline déterministe
Output utilisé par du code, pas par un humain direct

Évite (ou complète avec free-form) quand :

Output destiné à un humain (chatbot, rédaction) → Markdown ou texte libre + structured pour metadata seulement
Créativité requise (rédaction marketing, code créatif)
Le schema change toutes les semaines (overhead maintenance)
Tu fais du brainstorming / exploration

🏋️ Exercices

Progressifs, du "ça marche" au "défends le chiffre en prod". Fais-les dans l'ordre.

Exercice 1 — Le pipeline de base (échauffement)

Objectif : extraire une facture PDF en Facture Pydantic strict via client.messages.parse() sur claude-opus-4-8, avec garde sur parsed_output is None. Indice/Solution : pypdf → texte → messages.parse(output_config={"format": Facture, "effort": "low"}). Branche sur stop_reason : refusal → log + skip, max_tokens → augmente la cap, sinon → parsed_output. Le piège : ne JAMAIS lire resp.content[0] sans avoir vérifié parsed_output.

Exercice 2 — Casse le strict, puis répare-le

Objectif : prouver empiriquement que le strict ne garantit que la forme. Fais extraire un montant_eur à partir d'un texte ambigu ("entre 800 et 1200€") et observe une valeur inventée mais bien typée. Puis ajoute la couche métier qui l'attrape. Indice/Solution : ajoute confiance: float = Field(ge=0, le=1) + un @model_validator(mode="after") qui flag si montant_eur n'apparaît pas littéralement dans le texte source. Sous confiance < 0.85 → flag_for_human. La leçon : le niveau 2 (strict) et le niveau 3 (Pydantic métier) sont orthogonaux, tu as besoin des deux.

Exercice 3 — Fais chauffer le cache (défends ta latence p99)

Objectif : prendre un pipeline qui extrait 1000 docs avec le même prompt système + schema, et faire passer cache_read_input_tokens de 0 à >90% du préfixe. Mesure le gain de coût ET de latence p99. Indice/Solution : mets le prompt système long + le schema dans un bloc system avec cache_control: {"type": "ephemeral"}, et SEULEMENT le doc variable dans messages. Vérifie qu'aucun datetime.now()/UUID ne pollue le préfixe (sinon cache toujours froid). Compte les deux : sans cache vs avec. Sache expliquer pourquoi un schema généré dynamiquement par requête tue le cache (grammaire recompilée + préfixe différent).

Exercice 4 — Production-grade sous rate limit

Objectif : rendre l'extraction batch résiliente : 10 000 docs, AsyncAnthropic, Semaphore, max_retries, exceptions typées, et un budget de coût qui s'arrête net si dépassé. Indice/Solution : asyncio.gather borné par Semaphore(8), try/except (RateLimitError, APITimeoutError, APIStatusError) qui retourne None au lieu de crasher le batch entier, et un compteur de usage cumulé qui lève si on dépasse MAX_BUDGET_USD. Bonus : compare le coût avec la Batches API (50%) et défends le choix temps-réel vs batch.

Exercice 5 — Discriminated union + routing (architecture)

Objectif : un endpoint qui doit router une demande client vers Devis | Refus | Redirection, chacun avec son propre schema, et choisir la branche. Indice/Solution : côté Pydantic, Annotated[Devis | Refus | Redirection, Field(discriminator="type")]. Côté Anthropic en mode legacy/multi-tools : un tool par branche + tool_choice={"type": "any"}, puis re-valide Devis(**block.input). Sache dire quand tu prends messages.parse() (un schema garanti) vs tool_choice (router parmi N).

Exercice 6 — Le schema-drift qui casse la prod (défends ton CI)

Objectif : écrire le test CI qui empêche un collègue de changer un schema Pydantic sans migration DB, et le faire échouer volontairement. Indice/Solution : hash SHA-256 de CV.model_json_schema() (avec sort_keys=True) comparé à un hash gelé en repo. Change le schema → le test rouge force un bump de version + migration. Pousse plus loin : génère le diff lisible (champ ajouté/retiré/retypé) dans le message d'assert, pas juste "hash mismatch". C'est la différence entre un test qui bloque et un test qu'on désactive en jurant.

Exercice 7 — Défends ton chiffre devant le CFO (build vs buy)

Objectif : un client veut savoir si remplacer son pipeline d'extraction "regex + 3 ETP de correction" par du structured output Anthropic est rentable. Produis une estimation défendable : coût/doc, p99, taux d'erreur résiduel, et le point mort. Indice/Solution : pose les hypothèses explicitement. Coût LLM/doc = (in_tok * 5 + out_tok * 25) / 1e6 sur claude-opus-4-8, ou ÷5 sur claude-haiku-4-5 si le doc est simple, ou ÷2 de plus via Batches API si non-temps-réel. Mesure les vrais usage.input_tokens/output_tokens sur un échantillon de 50 docs réels (PAS une estimation tiktoken — c'est le tokenizer OpenAI, il sous-compte Claude de 15-20% ; utilise client.messages.count_tokens(model="claude-opus-4-8", ...)). Le taux d'erreur résiduel n'est PAS 0 : strict élimine les erreurs de forme (< 0.1%) mais laisse les erreurs métier (hallucination bien formée) — chiffre-les avec un eval sur dataset annoté, pas au doigt mouillé. Point mort = (coût humain actuel − coût LLM − coût correction des erreurs métier résiduelles) ; si le client traite 1M docs/mois, fais aussi le calcul Batches vs temps-réel. La compétence testée : savoir quels chiffres tu mesures vs lesquels tu supposes, et le dire à voix haute.

Exercice 8 — L'énum à 300 valeurs qui tue ta latence

Objectif : on te demande de classifier un ticket support dans l'une de 300 catégories produit. Le réflexe naïf (Literal[...] à 300 valeurs en strict) dégrade latence ET qualité. Conçois la bonne architecture et prouve le gain. Indice/Solution : un enum à 300 valeurs gonfle le FSM (l'automate doit encoder chaque branche) et fait "chercher" le modèle. Deux fixes de staff : (a) hiérarchise — extraction en 2 passes (catégorie macro parmi ~10, puis sous-catégorie parmi ~30) pour réduire l'espace par appel ; ou (b) extraction libre + post-validation : laisse le modèle sortir un label string libre, puis mappe-le contre ta liste via embeddings (cosine top-1) ou fuzzy match. Mesure les deux contre le Literal à 300 sur 100 tickets : latence p50/p99, accuracy, et coût. Sache défendre quand tu prends (a) [taxonomie stable, hiérarchie naturelle] vs (b) [liste qui bouge souvent, tolérance au quasi-match].

🎤 En entretien

Questions seniors que ce sujet appelle, avec la réponse d'une ligne.

"Le mode strict garantit-il que les données sont correctes ?" → Non. Le strict masque les logits pour garantir la forme (conformité au schema au décodage) ; une hallucination bien formée passe. La justesse vient de la validation métier (Pydantic ranges/cross-field) + un champ confiance + flag humain.
"Pourquoi un schema généré dynamiquement par requête dégrade ta latence p99 ?" → Le provider compile le schema en grammaire/FSM et la cache ~24h ; un schema neuf à chaque requête = recompilation systématique + préfixe de prompt différent = cache toujours froid. Schemas figés → cache chaud.
"Tu as une énum de 300 catégories à extraire. Comment tu fais ?" → Pas en enum strict : ça gonfle le FSM, augmente la latence et dégrade la qualité. Extraction libre du label + post-validation/mapping (fuzzy ou embedding) contre la liste, ou hiérarchise (catégorie → sous-catégorie) pour réduire l'espace par appel.
"Sur claude-opus-4-8, tu mets temperature=0 pour du déterministe sur ton extraction ?" → Non : temperature/top_p/top_k sont supprimés (HTTP 400), tout comme budget_tokens. Le déterminisme se pilote par effort: "low" + prompt serré ; et temperature=0 n'a jamais garanti des sorties identiques bit-à-bit de toute façon.
"messages.parse() ou tool_choice forcé — comment tu choisis côté Anthropic ?" → messages.parse(output_config={"format": Schema}) quand il y a un schema garanti par appel (extraction, classif) : l'API force la forme, le SDK valide Pydantic, tu lis parsed_output. tool_choice quand le modèle doit router parmi N schemas (un tool par branche, {"type": "any"}), ou quand tu veux le résultat dans un tool_use block pour ton agent loop — là tu re-valides Pydantic toi-même.
"1 million de factures à extraire ce week-end. Temps-réel ou batch, et combien ça coûte ?" → Batch : non-latence-sensible ⇒ Batches API à 50% du prix, async (< 1h en général, max 24h, 100k req/batch), structured outputs supportés sans surcoût. Coût = (in_tok*5 + out_tok*25)/1e6 * 0.5 sur Opus 4.8, ou bascule sur claude-haiku-4-5 (1 $/5 $) si le schema est simple. Je mesure les vrais usage sur un échantillon avant de chiffrer — pas d'estimation tiktoken (mauvais tokenizer pour Claude), j'utilise count_tokens.
"Ton cache_read_input_tokens est à 0 alors que tu réutilises le même schema. Pourquoi ?" → Invalidateur silencieux dans le préfixe (tools → system → messages) : un datetime.now() ou un UUID dans le system prompt, une sérialisation JSON non triée, ou le schema regénéré dynamiquement par requête (grammaire recompilée + préfixe différent). Le cache est un prefix-match : un seul octet qui change en amont invalide tout l'aval. Fix : figer le préfixe, sérialiser le schema déterministe, mettre le cache_control sur le bloc stable.

Liens

OpenAI structured outputs : https://platform.openai.com/docs/guides/structured-outputs
Anthropic structured outputs (output_config.format, messages.parse()) : https://platform.claude.com/docs/en/build-with-claude/structured-outputs
Anthropic tool use (routing N-schemas, tool_choice) : https://platform.claude.com/docs/en/agents-and-tools/tool-use/overview
Mistral JSON mode : https://docs.mistral.ai/capabilities/structured-output/
Pydantic v2 : https://docs.pydantic.dev/latest/
Zod : https://zod.dev/
Instructor : https://python.useinstructor.com/
Outlines : https://dottxt-ai.github.io/outlines/
Guidance : https://github.com/guidance-ai/guidance
lm-format-enforcer : https://github.com/noamgat/lm-format-enforcer
Vercel AI SDK structured outputs : https://sdk.vercel.ai/docs/ai-sdk-core/generating-structured-data
LlamaIndex Pydantic programs : https://docs.llamaindex.ai/en/stable/module_guides/querying/structured_outputs/

Structured outputs 2026 ​

Mental model ​

Le mental model du décodeur (ce qui te distingue d'un junior) ​

Code minimal ​

Comment le "strict" fonctionne vraiment (mental model staff) ​

Production : async, caching, observabilité ​

Le budget de latence (comment un staff raisonne sur le p99) ​

Failure modes (ce qui casse en prod, pas en démo) ​

Cas d'usage concrets ​

RH — Parsing CV → Pydantic schema, ATS éditeur recrutement IT Paris (40 consultants) ​

E-commerce — Enrichissement catalogue produit auto (marketplace mode 25 000 SKU) ​

FinTech — KYC complet : selfie + pièce d'identité → profil client (néobanque pro B2B) ​

Exemple end-to-end ​

Patterns courants ​

1. Pydantic Discriminated Unions pour outputs polymorphes ​

2. Retry avec correction ("re-ask") ​

3. Escape hatch : champ "unable_to_extract" ​

4. Streaming structured outputs (partial parse) ​

5. Outlines / lm-format-enforcer pour modèles open self-hosted ​

6. Instructor (cross-provider Pydantic-first) ​

Versions & écosystème 2026 ​

Pitfalls ​

Pricing / ROI client ​

Testing / Eval ​

1. Tests unitaires Pydantic ​

2. Eval sur dataset annoté ​

3. Schema diffing en CI ​

Quand utiliser / éviter ​

🏋️ Exercices ​

Exercice 1 — Le pipeline de base (échauffement) ​

Exercice 2 — Casse le strict, puis répare-le ​

Exercice 3 — Fais chauffer le cache (défends ta latence p99) ​

Exercice 4 — Production-grade sous rate limit ​

Exercice 5 — Discriminated union + routing (architecture) ​

Exercice 6 — Le schema-drift qui casse la prod (défends ton CI) ​

Exercice 7 — Défends ton chiffre devant le CFO (build vs buy) ​

Exercice 8 — L'énum à 300 valeurs qui tue ta latence ​

🎤 En entretien ​

Liens ​

Structured outputs 2026

Mental model

Le mental model du décodeur (ce qui te distingue d'un junior)

Code minimal

Comment le "strict" fonctionne vraiment (mental model staff)

Production : async, caching, observabilité

Le budget de latence (comment un staff raisonne sur le p99)

Failure modes (ce qui casse en prod, pas en démo)

Cas d'usage concrets

RH — Parsing CV → Pydantic schema, ATS éditeur recrutement IT Paris (40 consultants)

E-commerce — Enrichissement catalogue produit auto (marketplace mode 25 000 SKU)

FinTech — KYC complet : selfie + pièce d'identité → profil client (néobanque pro B2B)

Exemple end-to-end

Patterns courants

1. Pydantic Discriminated Unions pour outputs polymorphes

2. Retry avec correction ("re-ask")

3. Escape hatch : champ "unable_to_extract"

4. Streaming structured outputs (partial parse)

5. Outlines / lm-format-enforcer pour modèles open self-hosted

6. Instructor (cross-provider Pydantic-first)

Versions & écosystème 2026

Pitfalls

Pricing / ROI client

Testing / Eval

1. Tests unitaires Pydantic

2. Eval sur dataset annoté

3. Schema diffing en CI

Quand utiliser / éviter

🏋️ Exercices

Exercice 1 — Le pipeline de base (échauffement)

Exercice 2 — Casse le strict, puis répare-le

Exercice 3 — Fais chauffer le cache (défends ta latence p99)

Exercice 4 — Production-grade sous rate limit

Exercice 5 — Discriminated union + routing (architecture)

Exercice 6 — Le schema-drift qui casse la prod (défends ton CI)

Exercice 7 — Défends ton chiffre devant le CFO (build vs buy)

Exercice 8 — L'énum à 300 valeurs qui tue ta latence

🎤 En entretien

Liens