Erweiterte Muster

Aktualisiert am 3. November 2026: Das Unter-Workflow, das zuvor execution.output zurückgab, ist im SDK 3.7.0 von mistralai nicht mehr vorhanden. Das Ergebnis wird nun aus result gelesen, wobei wait_for_result=True verwendet wird, um auf die Fertigstellung zu warten.

Jenseits linearer Workflows

Bisher haben wir Workflows betrachtet, die Aktivitäten sequentiell ausführen. In der Praxis erfordern reale Anwendungsfälle jedoch anspruchsvollere Muster: das Verkettung von Workflows, das parallele Ausführen von Aktivitäten oder das Treffen komplexer bedingter Entscheidungen. Diese vier Muster – Verkettung, Fortsetzung, Parallelität und Fan-out/Fan-in – decken nahezu jede Architektur ab, die Sie zum Aufbau benötigen werden.

Verkettung von Workflows

Ein Workflow kann einen anderen Workflow auslösen. Diese Fähigkeit dient zwei unterschiedlichen Zwecken: der Zerlegung komplexer Logik in wiederverwendbare Unter-Workflows und dem Umgehen der Ereignisgrenze von 51.200 Ereignissen pro Ausführung, da jedes Kind-Workflow sein eigenes Budget erhält. Der Mechanismus stützt sich auf eine Aktivität, den einzigen Ort, an dem das SDK rechtmäßig eine Netzwerkanfrage stellen kann.

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

@workflows.activity()
async def launch_sub_workflow(workflow_name: str, input_data: dict) -> dict:
    """Startet ein Kind-Workflow und wartet auf das Ergebnis."""
    client = Mistral()
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier=workflow_name,
        input=input_data,
        wait_for_result=True,  # Warten auf das Ergebnis des Kind-Workflows
    )
    return execution.result

@workflows.workflow.define(name="main_pipeline")
class MainPipeline:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        # Schritt 1: Datenerfassung (dedizierter Unter-Workflow)
        data = await launch_sub_workflow(
            "data_collection",
            {"sources": params["sources"]}
        )

        # Schritt 2: Analyse (dedizierter Unter-Workflow)
        analysis = await launch_sub_workflow(
            "content_analysis",
            {"data": data}
        )

        # Schritt 3: Berichterstellung (dedizierter Unter-Workflow)
        report = await launch_sub_workflow(
            "report_generation",
            {"analysis": analysis, "format": params["format"]}
        )

        return report

Der Vorteil geht über reine Lesbarkeit hinaus: data_collection wird zu einem eigenständigen Workflow, der isoliert getestet und von anderen Pipelines wiederverwendet werden kann, mit eigenen Zeitlimits und Wiederholungsrichtlinien.

Dieselbe Mechanik, angewendet auf einen Workflow, der sich selbst aufruft, erzeugt das Muster der Fortsetzung. Anstatt hunderttausend Elemente in einer einzigen Ausführung zu verarbeiten – was die Ereignisgrenze verbietet – verarbeitet der Workflow eine Charge und delegiert den Rest an eine neue Instanz.

@workflows.workflow.define(name="batch_processing")
class BatchProcessing:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        offset = params.get("offset", 0)
        batch_size = 100

        # Hole die aktuelle Charge
        batch = await retrieve_batch(offset, batch_size)

        if not batch["items"]:
            # Alle Chargen verarbeitet
            return {"status": "completed", "total_processed": offset}

        # Verarbeite die aktuelle Charge
        await process_batch(batch["items"])

        # Starte die nächste Charge über einen neuen Workflow
        next_batch = await launch_sub_workflow(
            "batch_processing",
            {"offset": offset + batch_size}
        )

        return next_batch

Jede neue Ausführung beginnt mit einem sauberen Journal. Die Anzahl der verarbeitbaren Elemente wird unbegrenzt, wobei die Abbruchbedingung lediglich eine leere Charge ist.

Parallelisierung mit asyncio.gather

Wenn mehrere Aktivitäten voneinander unabhängig sind, besteht kein Grund, sie sequentiell auszuführen. asyncio.gather reicht sie dem Plattform gleichzeitig ein, und die Gesamtzeit nähert sich der Dauer der langsamsten Aktivität an, statt die Summe aller zu sein.

import asyncio
import mistralai.workflows as workflows

@workflows.activity()
async def analyze_source(url: str) -> dict:
    """Analysiert eine Datenquelle."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"url": url, "size": len(response.text)}

@workflows.workflow.define(name="parallel_analysis")
class ParallelAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, urls: list) -> dict:
        # Starte alle Analysen parallel
        results = await asyncio.gather(
            *[analyze_source(url) for url in urls]
        )

        return {
            "source_count": len(results),
            "results": list(results)
        }

Es gilt jedoch eine Einschränkung: Standardmäßig propagiert gather die erste Ausnahme, die auftritt, und bricht den gesamten Satz ab. Wenn Sie vierzig externe Quellen analysieren, führt bereits eine einzige Offline-Quelle dazu, dass Sie die anderen dreißig neun verlieren. Der Parameter return_exceptions=True ändert dieses Verhalten: Fehler werden als gewöhnliche Werte zurückgegeben, sodass Sie diese selbst sortieren können.

@workflows.workflow.define(name="fault_tolerant_analysis")
class FaultTolerantAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, urls: list) -> dict:
        # return_exceptions=True verhindert, dass ein Fehler die anderen blockiert
        results = await asyncio.gather(
            *[analyze_source(url) for url in urls],
            return_exceptions=True
        )

        successes = []
        errors = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                errors.append({"url": urls[i], "error": str(result)})
            else:
                successes.append(result)

        return {
            "successes": successes,
            "errors": errors,
            "success_rate": len(successes) / len(urls)
        }

Dieser Workflow gibt ein teilweises Ergebnis zusammen mit einer Erfolgsquote zurück, sodass nachgelagerte Prozesse entscheiden können, ob die Qualität ausreicht oder ob ein Neustart erforderlich ist.

Entscheidungsfindung im Workflow

Komplexe Verzweigungen sind nicht nur erlaubt, sondern oft der Hauptgrund für den Einsatz von Workflows, sofern die Bedingungen sich aus Aktivitätsergebnissen ableiten. Der Fall der Inhaltsmoderation veranschaulicht diese kaskadierende Logik gut: automatische Entscheidung, wenn das Modell sehr sicher ist, zweite Meinung bei mäßiger Sicherheit und menschliche Eskalation bei Diskrepanz oder geringer Sicherheit.

@workflows.workflow.define(name="content_moderation")
class ContentModeration:
    @workflows.workflow.entrypoint
    async def run(self, content: dict) -> dict:
        # Schritt 1: Automatische Analyse
        analysis = await auto_analyze_content(content["text"])

        if analysis["confidence_score"] > 0.95:
            # Hohe Sicherheit: automatische Entscheidung
            if analysis["category"] == "acceptable":
                await publish_content(content["id"])
                return {"decision": "published", "method": "automatic"}
            else:
                await reject_content(content["id"], analysis["reason"])
                return {"decision": "rejected", "method": "automatic"}

        elif analysis["confidence_score"] > 0.7:
            # Mäßige Sicherheit: zweite Analyse mit einem anderen Modell
            second_analysis = await analyze_content_model_b(content["text"])

            if analysis["category"] == second_analysis["category"]:
                # Beide Modelle stimmen überein
                if analysis["category"] == "acceptable":
                    await publish_content(content["id"])
                    return {"decision": "published", "method": "ai_consensus"}
                else:
                    await reject_content(content["id"], analysis["reason"])
                    return {"decision": "rejected", "method": "ai_consensus"}
            else:
                # Diskrepanz: menschliche Eskalation
                await escalate_moderation(content["id"], analysis, second_analysis)
                return {"decision": "pending", "method": "human_escalation"}

        else:
            # Geringe Sicherheit: direkte Eskalation
            await escalate_moderation(content["id"], analysis, None)
            return {"decision": "pending", "method": "human_escalation"}

Beachten Sie das Feld method, das an jedem Austrittspunkt zurückgegeben wird: Es ermöglicht es Ihnen, retrospektiv zu messen, welcher Anteil des Inhalts ohne menschliches Eingreifen verarbeitet wurde, und die Schwellenwerte entsprechend anzupassen.

Fan-out / Fan-in

Das letzte Muster kombiniert die vorherigen: Verteilen Sie die Arbeit parallel und aggregieren Sie die Ergebnisse an einem einzigen Punkt. Hier wird ein Dokument gleichzeitig in drei Sprachen übersetzt, jede Übersetzung wird parallel analysiert und alles fließt in eine Zusammenfassung zusammen.

@workflows.workflow.define(name="multi_language_analysis")
class MultiLanguageAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, document: dict) -> dict:
        # Fan-out: Übersetzen parallel in 3 Sprachen
        translations = await asyncio.gather(
            translate(document["text"], "en"),
            translate(document["text"], "de"),
            translate(document["text"], "es")
        )

        # Fan-out: Analysiere jede Übersetzung parallel
        analyses = await asyncio.gather(
            *[analyze_sentiment(trad["translated_text"]) for trad in translations]
        )

        # Fan-in: Ergebnisse aggregieren
        summary = await aggregate_analyses(list(analyses))

        return summary

Die zwei Wellen der Parallelität sind aufeinanderfolgend, da die zweite von den Ergebnissen der ersten abhängt. In strikter Reihenfolge würde dieser Workflow sieben verkettete Aufrufe erfordern; hier kostet er nur drei Orchestrierungsschritte.

Zusammenfassend: Zerlegen Sie komplexe Workflows in wiederverwendbare Unter-Workflows, parallelisieren Sie, wann immer Aktivitäten unabhängig sind, aktivieren Sie return_exceptions=True, wenn partielle Fehler tolerierbar sind, halten Sie die Fortsetzung für sehr lange Verarbeitungsjobs bereit und stellen Sie immer sicher, dass Ihre Verzweigungen sich auf Aktivitätsergebnisse stützen und nicht auf nicht-deterministische Werte.

Kernaussagen

• Ein Workflow kann andere Workflows über eine Aktivität auslösen, die das SDK aufruft
• Das Muster der Fortsetzung ermöglicht die Verarbeitung einer unbegrenzten Anzahl von Elementen in aufeinanderfolgenden Chargen
• asyncio.gather ermöglicht es Ihnen, Aktivitäten parallel auszuführen, um die Gesamtzeit zu reduzieren
• Komplexe bedingte Verzweigungen werden unterstützt, solange sie sich auf Aktivitätsergebnisse stützen
• Das Muster Fan-out / Fan-in verteilt die Arbeit und aggregiert dann die Ergebnisse