Erstellen eines Workflows

Die Rolle des Workflows

Der Workflow bildet die Orchestrierungsschicht Ihrer Anwendung. Er definiert, was geschehen muss, in welcher Reihenfolge und unter welchen Bedingungen – ohne die eigentliche Arbeit auszuführen. Konkret koordiniert er Aktivitäten, wartet auf externe Ereignisse, verwaltet Timer und bedingte Verzweigungen, berührt jedoch niemals das Netzwerk oder die Festplatte.

Sie können dies in einem Satz zusammenfassen, den Sie sich bei jeder Codeprüfung merken sollten: Aktivitäten erledigen die Arbeit, der Workflow dirigiert. Wenn Sie httpx oder open() innerhalb einer Workflow-Klasse sehen, muss etwas verschoben werden.

Workflow-Struktur

Ein Workflow ist eine Python-Klasse, die mit @workflows.workflow.define() dekoriert ist und eine Einstiegsmethode enthält, die mit @workflows.workflow.entrypoint dekoriert ist:

import mistralai.workflows as workflows

@workflows.workflow.define(name="mon_premier_workflow")
class MonPremierWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        """Einstiegspunkt des Workflows."""
        resultat = await saluer(nom)
        return resultat

Der Dekorierer @workflows.workflow.define(name="...") registriert die Klasse als Workflow. Der empfangene name ist der eindeutige Identifikator, den Sie zum Auslösen einer Ausführung verwenden – dieser String, nicht der Name der Python-Klasse, ist das, was der Aufrufer sieht. Der Dekorierer @workflows.workflow.entrypoint kennzeichnet die Methode, die beim Start aufgerufen wird. Der erste Parameter, self, bleibt wie bei jedem Python-Objekt die Instanz der Klasse, und jeder Aufruf einer Aktivität verwendet await.

Vollständiges Beispiel: Verarbeitungs-Pipeline

Hier ist ein realistischer Workflow, der mehrere Aktivitäten orchestriert, um ein Dokument zu verarbeiten. Beachten Sie die Arbeitsteilung: Die drei Funktionen oben enthalten den gesamten Code, der die Außenwelt berührt, während die Klasse unten lediglich await-Aufrufe verknüpft und ein Ergebnis zusammenfügt.

import mistralai.workflows as workflows
from mistralai.client import Mistral

# --- Aktivitäten ---

@workflows.activity()
async def extraire_texte(url_document: str) -> dict:
    """Herunterladen und Extrahieren von Text aus einem Dokument."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url_document)
        texte = response.text
    return {"texte": texte, "longueur": len(texte)}

@workflows.activity()
async def analyser_avec_mistral(texte: str) -> dict:
    """Text mit Mistral analysieren, um Schlüsselpunkte zu extrahieren."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Extrahieren Sie die 5 Schlüsselpunkte aus diesem Text."},
            {"role": "user", "content": texte}
        ]
    )
    return {"analyse": response.choices[0].message.content}

@workflows.activity()
async def sauvegarder_resultat(analyse: dict) -> dict:
    """Das Analyseergebnis speichern."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/rapports",
            json=analyse
        )
        return {"id_rapport": response.json()["id"]}

# --- Workflow ---

@workflows.workflow.define(name="pipeline_analyse_document")
class PipelineAnalyseDocument:
    @workflows.workflow.entrypoint
    async def run(self, url_document: str) -> dict:
        """Orchestrierung von Extraktion, Analyse und Speichern."""
        # Schritt 1: Text extrahieren
        extraction = await extraire_texte(url_document)

        # Schritt 2: Mit Mistral analysieren
        analyse = await analyser_avec_mistral(extraction["texte"])

        # Schritt 3: Speichern
        sauvegarde = await sauvegarder_resultat(analyse)

        return {
            "url_source": url_document,
            "longueur_texte": extraction["longueur"],
            "id_rapport": sauvegarde["id_rapport"]
        }

Jedes await im Workflow stellt einen Persistenzpunkt dar. Es lohnt sich, dies mental am obigen Beispiel zu überprüfen: Wenn der Prozess zwischen Schritt 2 und Schritt 3 fehlschlägt, wird das Dokument nicht erneut heruntergeladen und das Modell wird nicht erneut aufgerufen – es fallen also keine zusätzlichen Token-Gebühren an. Der Workflow setzt die Ausführung direkt beim Speicherschritt fort.

Workflow-Dauerhaftigkeit

Workflows überleben Neustarts, Infrastrukturausfälle und vorübergehende Fehler, da die Plattform jedes Ereignis aufzeichnet und den Workflow-Zustand während der Wiedergabe rekonstruiert. Diese Rekonstruktion ist nur unter einer Bedingung zuverlässig, die bereits erwähnt wurde: Der Workflow-Code muss für dieselben Eingaben immer dieselbe Sequenz von Operationen erzeugen. Dies ist die Determinismus-Einschränkung. Wir werden die Details in Lektion 8 behandeln.

Workflow mit bedingter Verzweigung

Ein Workflow ist nicht auf eine lineare Sequenz beschränkt. Er kann bedingte Logik enthalten, wie in diesem Beispiel zur Bestellverarbeitung, bei dem der Pfad vom Lagerbestand abhängt:

@workflows.workflow.define(name="traitement_commande")
class OrderProcessing:
    @workflows.workflow.entrypoint
    async def run(self, order: dict) -> dict:
        # Lagerbestand prüfen
        stock = await check_stock(order["produit_id"])

        if stock["disponible"]:
            # Standardpfad
            payment = await process_payment(order["montant"])
            shipment = await prepare_shipment(order)
            await send_confirmation(order["email"], shipment)
            return {"status": "confirmée", "tracking": shipment["tracking"]}
        else:
            # Alternativer Pfad
            await notify_stockout(order["email"], order["produit_id"])
            return {"status": "rupture_stock"}

if/else-Verzweigungen werden vollständig unterstützt. Die einzige Einschränkung betrifft das Material, das für die Entscheidung verwendet wird: Das Ergebnis der Verzweigung muss vom Rückgabewert einer Aktivität abhängen – ein Wert, der im Protokoll aufgezeichnet und identisch wiedergegeben wird – und nicht von einem direkten Aufruf eines externen Dienstes aus dem Workflow. Hier fragt verifier_stock den Lagerbestand innerhalb einer Aktivität ab, und die if-Anweisung liest lediglich ein Wörterbuch, das bereits persistiert wurde.

Eingaben mit Pydantic

Ein Workflow, der sechs Positionalparameter annimmt, wird schnell unlesbar. Für strukturierte Eingaben sollten Sie Pydantic-Modelle verwenden, die zudem die Typvalidierung am Start der Ausführung bereitstellen:

from pydantic import BaseModel

class ReportParams(BaseModel):
    report_type: str
    include_details: bool = False
    start_date: str
    end_date: str

@workflows.workflow.define(name="generation_rapport")
class ReportGeneration:
    @workflows.workflow.entrypoint
    async def run(self, params: ReportParams) -> dict:
        data = await collect_data(params.report_type, params.start_date, params.end_date)
        
        if params.include_details:
            data = await enrich_data(data)
        
        report = await generate_pdf_report(data)
        return {"url_rapport": report["url"]}

Ein Punkt der Vorsicht beim Auslösen dieses Workflows: Wenn Sie ein BaseModel verwenden, werden die Felder zu Top-Level-Schlüsseln in der JSON-Eingabe, die an den Workflow gesendet wird. Sie senden {"type_rapport": ..., "date_debut": ...}, nicht ein Objekt, das unter einem params-Schlüssel verschachtelt ist.

Best Practices

Vier Gewohnheiten machen langfristig einen Unterschied. Erstens: Halten Sie sich an eine Verantwortung pro Workflow. Das Mischen von Abrechnungen und Follow-ups zu Verkäufen in derselben Klasse macht es unmöglich, das eine ohne das andere wiederzugeben. Zweitens: Achten Sie auf die Benennung, da der Workflow-name ein öffentlicher Identifikator ist, der von anderen Systemen aufgerufen wird – "traitement_commande" ist wartbar, "workflow_1" nicht. Drittens: Führen Sie keine I/O-Operationen im Workflow ein: Jeder Netzwerkaufruf, jede Datei-Zugriff oder jede Datenbankabfrage gehört in eine Aktivität. Viertens: Halten Sie den Workflow-Code einfach, indem Sie komplexe Logik – Parsing, Transformationen, schwere Berechnungen – in Aktivitäten auslagern, wo sie isoliert getestet werden können, ohne Determinismus-Einschränkungen.

Kernaussagen

• Ein Workflow ist eine Klasse, die mit @workflows.workflow.define() dekoriert ist und Aktivitäten orchestriert.
• Die Methode @workflows.workflow.entrypoint ist der Einstiegspunkt, der beim Start aufgerufen wird.
• Jedes await einer Aktivität ist ein Persistenzpunkt (automatische Fortsetzung bei Fehlern).
• Bedingte Verzweigungen und Schleifen werden unterstützt.
• Verwenden Sie Pydantic-Modelle für strukturierte Eingabeparameter.
• Der Workflow darf keine direkten I/O-Operationen enthalten.