Automatisierte Workflows und Pipelines

Übergang in die Produktion

Die vorherigen Lektionen haben die Grundlagen und einzelne Anwendungsfälle behandelt. Diese abschließende Lektion fasst alles zusammen, um automatisierte Workflows und Validierungspipelines zu erstellen, die für die Produktion bereit sind.

Workflow: Dokumentenverarbeitung

Ein vollständiger Pipeline-Prozess zur automatischen Verarbeitung eingehender Dokumente:

from pydantic import BaseModel
from typing import Literal

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

# Schritt 2: Extraktion nach Typ
class DonneesFacture(BaseModel):
    numero: str
    date_emission: str
    date_echeance: str
    emetteur: str
    montant_ht: float
    montant_ttc: float
    lignes: list[dict]

# Schritt 3: Validierung
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):
    # Klassifizieren
    _, classification = client.chat.parse(
        model="grok-4.20-0309-reasoning",
        messages=[{"role": "user", "content": f"Klassifizieren Sie dieses Dokument: {texte}"}],
        response_format=ClassificationDocument
    )

    # Extrahieren nach Typ
    if classification.type == "facture":
        _, donnees = client.chat.parse(
            model="grok-4.20-0309-reasoning",
            messages=[{"role": "user", "content": f"Extrahieren Sie Daten aus dieser Rechnung: {texte}"}],
            response_format=DonneesFacture
        )

    # Validieren
    _, validation = client.chat.parse(
        model="grok-4.20-0309-reasoning",
        messages=[{
            "role": "user",
            "content": f"Validieren Sie diese extrahierten Daten: {donnees.model_dump_json()}"
        }],
        response_format=ResultatValidation
    )

    return classification, donnees, validation

Datenvalidierungspipeline

Verwenden Sie strukturierte Ausgaben, um eingehende Daten zu validieren und zu normalisieren:

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]

Diese Pipeline nimmt Kontakte in verschiedenen Formaten entgegen und normalisiert sie in eine einheitliche Struktur.

Architektur mehrstufiger Pipelines

Hier ist das Gerüst einer robusten Produktionspipeline:

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

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

    # Jeder Schritt verwendet ein dediziertes Schema
    etapes = [
        ("extraction", SchemaExtraction, "Extrahieren Sie Daten aus diesem Text"),
        ("enrichissement", SchemaEnrichi, "Erweitern Sie diese Daten"),
        ("validation", SchemaValidation, "Validieren Sie diese Daten"),
        ("formatage", SchemaFinal, "Formatieren Sie das Endergebnis"),
    ]

    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

Best Practices für die Produktion

Idempotenz

Gestalten Sie Ihre Pipelines so, dass sie ohne Nebenwirkungen neu ausgeführt werden können. Das gleiche Dokument, das zweimal verarbeitet wird, muss das gleiche Ergebnis liefern.

Strukturierte Protokollierung

Protokollieren Sie jeden Aufruf mit dem verwendeten Schema, der Token-Anzahl und der Antwortzeit.

Wiederholung mit 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

Verfolgen Sie für jede Pipeline folgende Metriken:

• Erfolgsquote pro Schritt
• Durchschnittliche Verarbeitungszeit
• Anzahl der verbrauchten Tokens
• Häufigkeit von Fallbacks

Kurszusammenfassung

In diesen 12 Lektionen haben Sie gelernt:

• Strukturierte Ausgaben und ihre Garantien zu verstehen
• Unterstützte Typen und deren Grenzen zu meistern
• parse() und response_format entsprechend Ihren Bedürfnissen zu verwenden
• Schemata mit Pydantic (Python) und Zod (JavaScript) zu definieren
• Strukturierte Ausgaben mit Websuche und Codeausführung zu kombinieren
• Extraktions-, Klassifizierungs- und Validierungspipelines zu erstellen

Strukturierte Ausgaben transformieren LLMs von Gesprächswerkzeugen in zuverlässige Softwarekomponenten, die in jede Architektur integriert werden können.

Wichtige Erkenntnisse

• Mehrstufige Workflows verwenden ein dediziertes Schema pro Schritt
• Kreuzvalidierung vergleicht Extraktionsergebnisse mit Referenzquellen
• In der Produktion: Idempotenz, strukturierte Protokollierung, Wiederholung mit Backoff, Monitoring
• Strukturierte Ausgaben bilden das Fundament jeder zuverlässigen KI-Pipeline
• Kombinieren Sie Schemata (Struktur) + Prompts (Inhalt) + Tools (Daten) für optimale Ergebnisse

Testen Sie Ihr Wissen

Garantiertes JSON, Schemata, Pipelines: strukturierte Ausgabe, Grok-Version.

1. Was garantiert eine strukturierte Ausgabe nach Schema?

Antwort: Eine Ausgabe, die strikt mit dem deklarierten JSON-Schema übereinstimmt – Typen, Felder, Struktur – und die vom Code ohne defensives Parsing verwendet werden kann: Das Format ist bei der Generierung eingeschränkt.

2. Was fügt die parse()-Methode im Vergleich zu manuellem response_format hinzu?

Antwort: Der vollautomatische Zyklus: Das Pydantic-Modell (oder Zod in JS) deklariert das Schema, und die Antwort kommt bereits validiert und als Objekt mit Typ zurück – ohne zusätzliche Verkabelung.

3. Warum ist es wichtig, nicht unterstützte Typen zu kennen?

Antwort: Weil ein Schema, das nicht unterstützte Konstruktionen verwendet, fehlschlägt oder sich verschlechtert: Sie entwerfen Ihre Modelle innerhalb des akzeptierten Teils und validieren die Ausgabe.

4. Was ermöglicht die Kombination von strukturierten Ausgaben + Websuche oder Codeausführung?

Antwort: Pipelines, bei denen Suche oder Berechnung eine strukturierte Ausgabe speisen: Frische oder berechnete Informationen gelangen direkt in Ihre typisierten Felder.

5. Nennen Sie zwei typische Anwendungsfälle in der Produktion.

Antwort: Entitätsextraktion und Klassifizierung (Dokumente, Tickets, Inhalte), die mit Ihren Systemen verbunden sind – automatisierte Workflows, bei denen KI Strukturen füllt, keine Absätze.

Deklaratives Schema, eingeschränkte Generierung, Ausgabevalidierung: Die strukturierte Ausgabe ist die Schnittstelle zwischen Grok und Ihrer Software – behandeln Sie sie als solche.