Advanced Patterns

Updated November 3, 2026: The sub-workflow previously returned execution.output, which is missing from mistralai SDK 3.7.0. The result is now read from result, using wait_for_result=True to wait for completion.

Beyond Linear Workflows

So far, we’ve looked at workflows that execute activities sequentially. In practice, real-world use cases require more sophisticated patterns: chaining workflows, running activities in parallel, or making complex conditional decisions. These four patterns—chaining, continuation, parallelism, and fan-out/fan-in—cover nearly every architecture you’ll need to build.

Chaining Workflows

A workflow can trigger another workflow. This capability serves two distinct goals: breaking down complex logic into reusable sub-workflows, and bypassing the 51,200-event limit per execution, since each child workflow gets its own budget. The mechanism relies on an activity, the only place where the SDK can legitimately make a network call.

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

@workflows.activity()
async def launch_sub_workflow(workflow_name: str, input_data: dict) -> dict:
    """Launches a child workflow and waits for its result."""
    client = Mistral()
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier=workflow_name,
        input=input_data,
        wait_for_result=True,  # wait for the child workflow result
    )
    return execution.result

@workflows.workflow.define(name="main_pipeline")
class MainPipeline:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        # Step 1: Data collection (dedicated sub-workflow)
        data = await launch_sub_workflow(
            "data_collection",
            {"sources": params["sources"]}
        )

        # Step 2: Analysis (dedicated sub-workflow)
        analysis = await launch_sub_workflow(
            "content_analysis",
            {"data": data}
        )

        # Step 3: Report generation (dedicated sub-workflow)
        report = await launch_sub_workflow(
            "report_generation",
            {"analysis": analysis, "format": params["format"]}
        )

        return report

The benefit goes beyond simple readability: data_collection becomes a standalone workflow, testable in isolation and reusable by other pipelines, with its own timeouts and retry policies.

The same mechanism, applied to a workflow that calls itself, creates the continuation pattern. Instead of processing one hundred thousand items in a single execution—which the event cap forbids—the workflow processes a batch and delegates the rest to a new instance.

@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

        # Retrieve the current batch
        batch = await retrieve_batch(offset, batch_size)

        if not batch["items"]:
            # All batches processed
            return {"status": "completed", "total_processed": offset}

        # Process the current batch
        await process_batch(batch["items"])

        # Launch the next batch via a new workflow
        next_batch = await launch_sub_workflow(
            "batch_processing",
            {"offset": offset + batch_size}
        )

        return next_batch

Each new execution starts with a clean journal. The number of processable items becomes unlimited, with the stopping condition simply being an empty batch.

Parallelizing with asyncio.gather

When multiple activities are independent of one another, there’s no reason to run them in sequence. asyncio.gather submits them to the platform simultaneously, and the total time approaches that of the slowest activity rather than the sum of all of them.

import asyncio
import mistralai.workflows as workflows

@workflows.activity()
async def analyze_source(url: str) -> dict:
    """Analyzes a data source."""
    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:
        # Launch all analyses in parallel
        results = await asyncio.gather(
            *[analyze_source(url) for url in urls]
        )

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

One caveat applies, however: by default, gather propagates the first exception encountered and aborts the entire set. If you’re analyzing forty external sources, just one being offline causes you to lose the other thirty-nine. The return_exceptions=True parameter changes this behavior: errors are returned as ordinary values, leaving you to sort them out.

@workflows.workflow.define(name="fault_tolerant_analysis")
class FaultTolerantAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, urls: list) -> dict:
        # return_exceptions=True prevents one error from blocking the others
        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)
        }

This workflow returns a partial result accompanied by a success rate, allowing downstream processes to decide whether the quality is sufficient or if a retry is needed.

Making Decisions in the Workflow

Complex branching is not only allowed but often the main point of using workflows, provided the conditions derive from activity results. The content moderation case illustrates this cascading logic well: automatic decision when the model is highly confident, a second opinion when confidence is moderate, and human escalation in case of disagreement or low confidence.

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

        if analysis["confidence_score"] > 0.95:
            # High confidence: automatic decision
            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:
            # Moderate confidence: second analysis with a different model
            second_analysis = await analyze_content_model_b(content["text"])

            if analysis["category"] == second_analysis["category"]:
                # Both models agree
                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:
                # Disagreement: human escalation
                await escalate_moderation(content["id"], analysis, second_analysis)
                return {"decision": "pending", "method": "human_escalation"}

        else:
            # Low confidence: direct escalation
            await escalate_moderation(content["id"], analysis, None)
            return {"decision": "pending", "method": "human_escalation"}

Notice the method field returned at every exit point: it lets you measure retrospectively what proportion of content was processed without human intervention, and thus adjust the thresholds accordingly.

Fan-out / Fan-in

The final pattern combines the previous ones: distribute work in parallel, then aggregate the results at a single point. Here, a document is translated simultaneously into three languages, each translation is analyzed in parallel, and everything converges into a summary.

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

        # Fan-out: analyze each translation in parallel
        analyses = await asyncio.gather(
            *[analyze_sentiment(trad["translated_text"]) for trad in translations]
        )

        # Fan-in: aggregate results
        summary = await aggregate_analyses(list(analyses))

        return summary

The two waves of parallelism are successive because the second depends on the results of the first. In strict sequence, this workflow would require seven chained calls; here, it costs only three orchestration steps.

In summary, break down complex workflows into reusable sub-workflows, parallelize whenever activities are independent, enable return_exceptions=True when partial failures are tolerable, keep continuation in reserve for very long processing jobs, and always ensure your branching relies on activity results rather than non-deterministic values.

Key Takeaways

• A workflow can trigger other workflows via an activity that calls the SDK
• The continuation pattern allows processing an unlimited number of items in successive batches
• asyncio.gather lets you run activities in parallel to reduce total time
• Complex conditional branching is supported as long as it depends on activity results
• The fan-out / fan-in pattern distributes work and then aggregates the results