Workflows automatisés et pipelines — Sorties structurées JSON

Passer à la production

Les leçons précédentes ont couvert les bases et les cas d’usage individuels. Cette dernière leçon rassemble tout pour construire des workflows automatisés et des pipelines de validation prêts pour la production.

Workflow : traitement de documents

Un pipeline complet pour traiter des documents entrants automatiquement :

from pydantic import BaseModel
from typing import Literal

# Étape 1 : Classification du document
class ClassificationDocument(BaseModel):
    type: Literal["facture", "contrat", "devis", "courrier", "autre"]
    langue: str
    urgence: Literal["basse", "normale", "haute"]
    departement: Literal["comptabilite", "juridique", "commercial", "direction"]

# Étape 2 : Extraction selon le type
class DonneesFacture(BaseModel):
    numero: str
    date_emission: str
    date_echeance: str
    emetteur: str
    montant_ht: float
    montant_ttc: float
    lignes: list[dict]

# Étape 3 : Validation
class ResultatValidation(BaseModel):
    document_valide: bool
    champs_manquants: list[str]
    anomalies: list[str]
    score_confiance: float
    action_recommandee: str

async def traiter_document(texte: str):
    # Classifier
    _, classification = client.chat.parse(
        model="grok-4.20-reasoning",
        messages=[{"role": "user", "content": f"Classifie ce document : {texte}"}],
        response_format=ClassificationDocument
    )

    # Extraire selon le type
    if classification.type == "facture":
        _, donnees = client.chat.parse(
            model="grok-4.20-reasoning",
            messages=[{"role": "user", "content": f"Extrais les données de cette facture : {texte}"}],
            response_format=DonneesFacture
        )

    # Valider
    _, validation = client.chat.parse(
        model="grok-4.20-reasoning",
        messages=[{
            "role": "user",
            "content": f"Valide ces données extraites : {donnees.model_dump_json()}"
        }],
        response_format=ResultatValidation
    )

    return classification, donnees, validation

Pipeline de validation de données

Utilisez les structured outputs pour valider et normaliser des données entrantes :

class AdresseNormalisee(BaseModel):
    ligne_1: str
    ligne_2: str | None
    code_postal: str
    ville: str
    pays: str
    format_valide: bool

class ContactNormalise(BaseModel):
    prenom: str
    nom: str
    email_valide: bool
    telephone_normalise: str | None
    adresse: AdresseNormalisee

class ResultatNormalisation(BaseModel):
    contacts: list[ContactNormalise]
    nombre_traites: int
    nombre_erreurs: int
    erreurs: list[str]

Ce pipeline prend des contacts dans des formats variés et les normalise en une structure uniforme.

Architecture de pipeline multi-étapes

Voici le squelette d’un pipeline robuste en production :

class PipelineResult(BaseModel):
    etape: str
    succes: bool
    donnees: dict
    erreurs: list[str]
    duree_ms: int

def pipeline_complet(texte: str) -> list[PipelineResult]:
    resultats = []

    # Chaque étape utilise un schéma dédié
    etapes = [
        ("extraction", SchemaExtraction, "Extrais les données de ce texte"),
        ("enrichissement", SchemaEnrichi, "Enrichis ces données"),
        ("validation", SchemaValidation, "Valide ces données"),
        ("formatage", SchemaFinal, "Formate le résultat final"),
    ]

    donnees_courantes = texte
    for nom, schema, instruction in etapes:
        try:
            _, parsed = client.chat.parse(
                model="grok-4.20-reasoning",
                messages=[{
                    "role": "user",
                    "content": f"{instruction} : {donnees_courantes}"
                }],
                response_format=schema
            )
            resultats.append(PipelineResult(
                etape=nom, succes=True,
                donnees=parsed.model_dump(),
                erreurs=[], duree_ms=0
            ))
            donnees_courantes = parsed.model_dump_json()
        except Exception as e:
            resultats.append(PipelineResult(
                etape=nom, succes=False,
                donnees={}, erreurs=[str(e)], duree_ms=0
            ))
            break

    return resultats

Bonnes pratiques pour la production

Idempotence

Concevez vos pipelines pour qu’ils puissent être relancés sans effets de bord. Un même document traité deux fois doit produire le même résultat.

Logging structuré

Chaque appel doit être loggé avec le schéma utilisé, le nombre de tokens, et le temps de réponse.

Retry avec backoff

import time

def appel_avec_retry(messages, schema, max_retries=3):
    for tentative in range(max_retries):
        try:
            return client.chat.parse(
                model="grok-4.20-reasoning",
                messages=messages,
                response_format=schema
            )
        except Exception as e:
            if tentative < max_retries - 1:
                time.sleep(2 ** tentative)
            else:
                raise

Monitoring

Surveillez ces métriques pour chaque pipeline :

Taux de succès par étape
Temps de traitement moyen
Nombre de tokens consommés
Fréquence des fallbacks

Récapitulatif du cours

Au fil de ces 12 leçons, vous avez appris à :

Comprendre les sorties structurées et leurs garanties
Maîtriser les types supportés et leurs limites
Utiliser parse() et response_format selon vos besoins
Définir des schémas avec Pydantic (Python) et Zod (JavaScript)
Combiner structured outputs avec web search et code execution
Construire des pipelines d’extraction, de classification et de validation

Les sorties structurées transforment les LLM d’outils conversationnels en composants logiciels fiables intégrables dans n’importe quelle architecture.

Points clés à retenir

Les workflows multi-étapes utilisent un schéma dédié par étape
La validation croisée compare les résultats d’extraction à des sources de référence
En production : idempotence, logging structuré, retry avec backoff, monitoring
Les structured outputs sont la base de tout pipeline d’IA fiable
Combinez schémas (structure) + prompts (contenu) + outils (données) pour des résultats optimaux