Automated Workflows and Pipelines

Moving to production

Previous lessons covered the basics and individual use cases. This final lesson brings it all together to build automated workflows and validation pipelines ready for production.

Workflow: document processing

A complete pipeline to process incoming documents automatically:

from pydantic import BaseModel
from typing import Literal

# Step 1: Document classification
class ClassificationDocument(BaseModel):
    type: Literal["facture", "contrat", "devis", "courrier", "autre"]
    langue: str
    urgence: Literal["basse", "normale", "haute"]
    departement: Literal["comptabilite", "juridique", "commercial", "direction"]

# Step 2: Extraction by type
class DonneesFacture(BaseModel):
    numero: str
    date_emission: str
    date_echeance: str
    emetteur: str
    montant_ht: float
    montant_ttc: float
    lignes: list[dict]

# Step 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):
    # Classify
    _, classification = client.chat.parse(
        model="grok-4.20-0309-reasoning",
        messages=[{"role": "user", "content": f"Classify this document: {texte}"}],
        response_format=ClassificationDocument
    )

    # Extract by type
    if classification.type == "facture":
        _, donnees = client.chat.parse(
            model="grok-4.20-0309-reasoning",
            messages=[{"role": "user", "content": f"Extract data from this invoice: {texte}"}],
            response_format=DonneesFacture
        )

    # Validate
    _, validation = client.chat.parse(
        model="grok-4.20-0309-reasoning",
        messages=[{
            "role": "user",
            "content": f"Validate this extracted data: {donnees.model_dump_json()}"
        }],
        response_format=ResultatValidation
    )

    return classification, donnees, validation

Data validation pipeline

Use structured outputs to validate and normalize incoming data:

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]

This pipeline takes contacts in various formats and normalizes them into a uniform structure.

Multi-step pipeline architecture

Here is the skeleton of a robust production pipeline:

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

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

    # Each step uses a dedicated schema
    etapes = [
        ("extraction", SchemaExtraction, "Extract data from this text"),
        ("enrichissement", SchemaEnrichi, "Enrich this data"),
        ("validation", SchemaValidation, "Validate this data"),
        ("formatage", SchemaFinal, "Format the final result"),
    ]

    donnees_courantes = texte
    for nom, schema, instruction in etapes:
        try:
            _, parsed = client.chat.parse(
                model="grok-4.20-0309-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

Production best practices

Idempotence

Design your pipelines to be rerun without side effects. The same document processed twice must produce the same result.

Structured logging

Log every call with the schema used, token count, and response time.

Retry with 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-0309-reasoning",
                messages=messages,
                response_format=schema
            )
        except Exception as e:
            if tentative < max_retries - 1:
                time.sleep(2 ** tentative)
            else:
                raise

Monitoring

Track these metrics for each pipeline:

• Success rate per step
• Average processing time
• Number of tokens consumed
• Fallback frequency

Course summary

Over these 12 lessons, you learned to:

• Understand structured outputs and their guarantees
• Master supported types and their limits
• Use parse() and response_format according to your needs
• Define schemas with Pydantic (Python) and Zod (JavaScript)
• Combine structured outputs with web search and code execution
• Build extraction, classification, and validation pipelines

Structured outputs transform LLMs from conversational tools into reliable software components integrable into any architecture.

Key takeaways

• Multi-step workflows use a dedicated schema per step
• Cross-validation compares extraction results to reference sources
• In production: idempotence, structured logging, retry with backoff, monitoring
• Structured outputs are the foundation of any reliable AI pipeline
• Combine schemas (structure) + prompts (content) + tools (data) for optimal results

Test your knowledge

Guaranteed JSON, schemas, pipelines: structured output, Grok version.

1. What does structured output by schema guarantee?

Answer: Output strictly compliant with the declared JSON schema—types, fields, structure—usable by code without defensive parsing: the format is constrained at generation.

2. What does the parse() method add compared to manual response_format?

Answer: The fully automated loop: the Pydantic model (or Zod in JS) declares the schema, and the response comes back already validated and typed as an object—zero plumbing.

3. Why know unsupported types?

Answer: Because a schema using unsupported constructions fails or degrades: you design your models within the accepted subset and validate on output.

4. What does combining structured outputs + web search or code execution allow?

Answer: Pipelines where search or computation feed a structured output: fresh or calculated information arrives directly in your typed fields.

5. Give two typical use cases in production.

Answer: Entity extraction and classification (documents, tickets, content) connected to your systems—automated workflows where AI fills structures, not paragraphs.

Declared schema, constrained generation, output validation: structured output is the interface between Grok and your software—treat it as such.