Chapter 8 · Branching and Parallel Execution

Chapter goals: Master .branch() conditional branching and .parallel() parallel fan-out, and learn to aggregate multi-path results at the convergence point.

8.1 Beyond Linearity: Three Control Flow Primitives

Real business flows are rarely a straight line. Mastra Workflow provides three control flow primitives:

.then()      Linear serial:  A → B → C
.branch()    Conditional:    A → [condition1→B / condition2→C]
.parallel()  Parallel fan-out: A → (B ∥ C) → merge

8.2 .branch(): Conditional Routing

Scenario: customer support tickets are first triaged by urgency, then routed to different handling steps.

// src/mastra/workflows/ticket-workflow.ts (excerpt)
import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';

// Step 1: classify and tag
const classifyStep = createStep({
  id: 'classify',
  inputSchema: z.object({ text: z.string() }),
  outputSchema: z.object({
    text: z.string(),
    urgent: z.boolean(), // Classification result serves as routing basis
  }),
  execute: async ({ inputData }) => {
    const urgent = /宕机|无法支付|数据丢失/.test(inputData.text);
    return { text: inputData.text, urgent };
  },
});

const urgentHandler = createStep({
  id: 'urgent-handler',
  inputSchema: z.object({ text: z.string(), urgent: z.boolean() }),
  outputSchema: z.object({ handled: z.boolean(), channel: z.string() }),
  execute: async () => ({ handled: true, channel: 'pager-oncall' }),
});

const normalQueue = createStep({
  id: 'normal-queue',
  inputSchema: z.object({ text: z.string(), urgent: z.boolean() }),
  outputSchema: z.object({ handled: z.boolean(), channel: z.string() }),
  execute: async () => ({ handled: true, channel: 'normal-queue' }),
});

// Assemble branches: each branch is a [filter, target step] tuple
export const ticketWorkflow = createWorkflow({
  id: 'ticket',
  inputSchema: z.object({ text: z.string() }),
  outputSchema: z.object({ handled: z.boolean(), channel: z.string() }),
})
  .then(classifyStep)
  .branch([
    // Branch 1: take urgent handler when urgent is true
    [async ({ inputData }) => inputData.urgent === true, urgentHandler],
    // Branch 2: otherwise go to normal queue
    [async ({ inputData }) => inputData.urgent === false, normalQueue],
  ])
  .commit();

Each branch's condition function receives the context and returns a boolean; the matching branch executes, non-matching ones are skipped.

8.3 .parallel(): Parallel Fan-out

Scenario: before content publishing, both "moderation check" and "SEO scoring" are needed — they're independent, so running in parallel halves the elapsed time:

// Two parallel steps: same input, independent outputs
export const publishCheck = createWorkflow({
  id: 'publish-check',
  inputSchema: z.object({ article: z.string() }),
  outputSchema: z.object({
    review: z.object({ pass: z.boolean() }),
    seo: z.object({ score: z.number() }),
  }),
})
  .parallel([
    // Branch 1: moderation check
    moderationStep,
    // Branch 2: SEO scoring
    seoScoreStep,
  ])
  .commit();

The parallel block's output is aggregated by key: each branch's output hangs under its own step id for downstream to read — the downstream step's inputSchema must declare this nested structure.

8.4 Aggregating After Convergence

// Convergence step: consumes the parallel-aggregated output
const mergeStep = createStep({
  id: 'merge',
  inputSchema: z.object({
    // Key names = each branch step's id
    moderation: z.object({ pass: z.boolean() }),
    seo: z.object({ score: z.number() }),
  }),
  outputSchema: z.object({ ok: z.boolean() }),
  execute: async ({ inputData }) => {
    const ok =
      inputData.moderation.pass && inputData.seo.score >= 60;
    return { ok }; // Only allow publish if both checks pass
  },
});

// Full pipeline: parallel → converge → decide
export const fullPipeline = createWorkflow({
  id: 'full-pipeline',
  inputSchema: z.object({ article: z.string() }),
  outputSchema: z.object({ ok: z.boolean() }),
})
  .parallel([moderationStep, seoScoreStep])
  .then(mergeStep)
  .commit();

Performance tip

Put mutually independent IO-heavy steps (multi-API calls, multiple model requests) into parallel; total elapsed time ≈ the slowest branch, not the sum of all.

8.5 Chapter Summary

• Three control flows: .then() serial, .branch() conditional routing, .parallel() parallel fan-out;
• Each branch of branch is a [conditionFn, step] tuple; conditions judge based on upstream inputData;
• parallel output aggregates by branch step id into an object; downstream schema must declare the nested shape;
• Parallelize IO steps with no dependencies to显著 compress end-to-end latency.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the structure of each element in .branch()?

2. What is the optimal organization for two independent external API call steps?

3. After .parallel([aStep, bStep]), how should the downstream step's inputSchema be declared?

4. What happens if none of the branch condition functions match?

🛠️ Hands-On Exercises

1. Extend ticketWorkflow into three branches: urgent / normal / spam (add a spam-filtering condition).
2. Add a third branch readabilityStep to fullPipeline's parallel; update mergeStep's schema and decision logic.
3. Trigger urgent=true and urgent=false tickets in Studio; observe which paths light up in the Graph view.

Next: Chapter 9 · Suspend & Resume: Human-in-the-Loop Collaboration