Control flow

Workflows 通过将每个任务划分为 steps 来让你控制预定义任务。这些步骤可以顺序或并行运行，也可以遵循条件路径。

每个步骤通过定义的 schema 连接到 workflow 中的下一个步骤，从而保持数据受控且一致。

核心原则

• 第一个步骤的 inputSchema 必须与 workflow 的 inputSchema 匹配。
• 最后一个步骤的 outputSchema 必须与 workflow 的 outputSchema 匹配。
• 每个步骤的 outputSchema 必须与下一个步骤的 inputSchema 匹配。
  • 如果不匹配，请使用输入数据映射将数据转换为所需的结构。

使用 .then() 串联步骤

使用 .then() 按顺序运行步骤，使每个步骤都能访问前一个步骤的结果。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    formatted: z.string(),
  }),
})

const step2 = createStep({
  inputSchema: z.object({
    formatted: z.string(),
  }),
  outputSchema: z.object({
    emphasized: z.string(),
  }),
})

export const testWorkflow = createWorkflow({
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    emphasized: z.string(),
  }),
})
  .then(step1)
  .then(step2)
  .commit()

使用 .parallel() 同时运行步骤

使用 .parallel() 同时运行步骤。所有并行步骤都必须完成后，workflow 才能继续执行下一步。每个步骤的 id 用于定义后续步骤的 inputSchema，并作为 inputData 对象上的键来访问前一步骤的值。随后，后续步骤可以引用或合并并行步骤的输出。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
})

const step2 = createStep({
  id: 'step-2',
})

const step3 = createStep({
  id: 'step-3',
  inputSchema: z.object({
    'step-1': z.object({
      formatted: z.string(),
    }),
    'step-2': z.object({
      emphasized: z.string(),
    }),
  }),
  outputSchema: z.object({
    combined: z.string(),
  }),
  execute: async ({ inputData }) => {
    const { formatted } = inputData['step-1']
    const { emphasized } = inputData['step-2']
    return {
      combined: `${formatted} | ${emphasized}`,
    }
  },
})

export const testWorkflow = createWorkflow({
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    combined: z.string(),
  }),
})
  .parallel([step1, step2])
  .then(step3)
  .commit()

输出结构

当步骤并行运行时，输出是一个对象，每个键是步骤的 id，值是该步骤的输出。你可以独立访问每个并行步骤的结果。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'format-step',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  execute: async ({ inputData }) => ({
    formatted: inputData.message.toUpperCase(),
  }),
})

const step2 = createStep({
  id: 'count-step',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ count: z.number() }),
  execute: async ({ inputData }) => ({
    count: inputData.message.length,
  }),
})

const step3 = createStep({
  id: 'combine-step',
  // The inputSchema must match the structure of parallel outputs
  inputSchema: z.object({
    'format-step': z.object({ formatted: z.string() }),
    'count-step': z.object({ count: z.number() }),
  }),
  outputSchema: z.object({ result: z.string() }),
  execute: async ({ inputData }) => {
    // Access each parallel step's output by its id
    const formatted = inputData['format-step'].formatted
    const count = inputData['count-step'].count
    return {
      result: `${formatted} (${count} characters)`,
    }
  },
})

export const testWorkflow = createWorkflow({
  id: 'parallel-output-example',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ result: z.string() }),
})
  .parallel([step1, step2])
  .then(step3)
  .commit()

// When executed with { message: "hello" }
// The parallel output structure will be:
// {
//   "format-step": { formatted: "HELLO" },
//   "count-step": { count: 5 }
// }

要点：

• 每个并行步骤的输出以其 id 为键
• 所有并行步骤同时执行
• 下一步骤接收一个包含所有并行步骤输出的对象
• 你必须定义后续步骤的 inputSchema 以匹配此结构

处理步骤失败

如果任何并行步骤抛出错误，整个并行块都会失败。要构建即使某些步骤失败也能保持弹性的并行 workflows（例如多个研究 agents 中有一个的 auth token 可能已过期），请在步骤内部使用 try/catch 处理错误：

src/mastra/workflows/test-workflow.ts

const resilientStep = createStep({
  id: 'researcher',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({
    brief: z.string().nullable(),
    failed: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    try {
      const result = await fetchExternalData(inputData.query)
      return { brief: result, failed: false }
    } catch {
      return { brief: null, failed: true }
    }
  },
})

这样该步骤总能成功返回带类型的结果，下游步骤可以过滤掉失败的结果：

src/mastra/workflows/test-workflow.ts

const writerStep = createStep({
  id: 'writer',
  inputSchema: z.object({
    'researcher-a': z.object({ brief: z.string().nullable(), failed: z.boolean() }),
    'researcher-b': z.object({ brief: z.string().nullable(), failed: z.boolean() }),
  }),
  outputSchema: z.object({ synthesis: z.string() }),
  execute: async ({ inputData }) => {
    const briefs = Object.values(inputData)
      .filter(v => !v.failed && v.brief)
      .map(v => v.brief)
    return { synthesis: briefs.join('; ') }
  },
})

访问选择合适的模式了解何时使用 .parallel() 与 .foreach()。

使用 .branch() 实现条件逻辑

使用 .branch() 根据条件选择要运行的步骤。分支中的所有步骤都需要相同的 inputSchema 和 outputSchema，因为分支需要一致的 schema，这样 workflows 才能沿不同路径执行。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...})

const stepA = createStep({
  inputSchema: z.object({
    value: z.number()
  }),
  outputSchema: z.object({
    result: z.string()
  })
});

const stepB = createStep({
  inputSchema: z.object({
    value: z.number()
  }),
  outputSchema: z.object({
    result: z.string()
  })
});

export const testWorkflow = createWorkflow({
  inputSchema: z.object({
    value: z.number()
  }),
  outputSchema: z.object({
    result: z.string()
  })
})
  .then(step1)
  .branch([
    [async ({ inputData: { value } }) => value > 10, stepA],
    [async ({ inputData: { value } }) => value <= 10, stepB]
  ])
  .commit();

使用 classifier 进行路由

配置好的 classifier 可以在 .branch() 之前生成带类型的答案。根据与每种问题类型匹配的答案字段进行路由：choice、score 或 probability。

src/mastra/workflows/ticket-triage.ts

import { Classifier } from '@mastra/core/classifier'
import { typeSafeAi } from '@ai-sdk/typesafe-ai'
import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const router = new Classifier({
  id: 'ticket-router',
  model: typeSafeAi.decisionModel('jev-latest'),
  questions: {
    route: {
      type: 'choice',
      criteria: {
        billing: 'Billing, invoices, refunds, and payments',
        support: 'Account access and product help',
        other: 'Anything else',
      },
    },
  },
})

export const ticketTriage = createWorkflow({
  id: 'ticket-triage',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.any(),
})
  .map({ message: { initData: true, path: 'message' } })
  .classifier(router)
  .branch([
    [async ({ inputData }) => inputData.answers.route.choice === 'billing', billingStep],
    [async ({ inputData }) => inputData.answers.route.choice === 'support', supportStep],
    [async ({ inputData }) => inputData.answers.route.choice === 'other', fallbackStep],
  ])
  .commit()

choice 字段保留配置的 criteria 键联合类型，因此 inputData.answers.route.choice 的类型为 'billing' | 'support' | 'other' | undefined。Score 答案暴露 score。布尔答案将 P(true) 值作为 probability 暴露。路由时请应用明确的阈值。

当决策模型拒绝回答问题时，其答案为 { type: 'refusal' }，且 choice、score 和 probability 均为 undefined。在上面的示例中，被拒绝的 route 问题不匹配任何条件。添加针对 inputData.answers.route.type === 'refusal' 的条件以显式处理。

输出结构

使用条件分支时，只会根据第一个求值为 true 的条件执行一个分支。输出结构与 .parallel() 类似，结果以执行的步骤 id 为键。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'initial-step',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ value: z.number() }),
  execute: async ({ inputData }) => inputData,
})

const highValueStep = createStep({
  id: 'high-value-step',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ result: z.string() }),
  execute: async ({ inputData }) => ({
    result: `High value: ${inputData.value}`,
  }),
})

const lowValueStep = createStep({
  id: 'low-value-step',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ result: z.string() }),
  execute: async ({ inputData }) => ({
    result: `Low value: ${inputData.value}`,
  }),
})

const finalStep = createStep({
  id: 'final-step',
  // The inputSchema must account for either branch's output
  inputSchema: z.object({
    'high-value-step': z.object({ result: z.string() }).optional(),
    'low-value-step': z.object({ result: z.string() }).optional(),
  }),
  outputSchema: z.object({ message: z.string() }),
  execute: async ({ inputData }) => {
    // Only one branch will have executed
    const result = inputData['high-value-step']?.result || inputData['low-value-step']?.result
    return { message: result }
  },
})

export const testWorkflow = createWorkflow({
  id: 'branch-output-example',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ message: z.string() }),
})
  .then(step1)
  .branch([
    [async ({ inputData }) => inputData.value > 10, highValueStep],
    [async ({ inputData }) => inputData.value <= 10, lowValueStep],
  ])
  .then(finalStep)
  .commit()

// When executed with { value: 15 }
// Only the high-value-step executes, output structure:
// {
//   "high-value-step": { result: "High value: 15" }
// }

// When executed with { value: 5 }
// Only the low-value-step executes, output structure:
// {
//   "low-value-step": { result: "Low value: 5" }
// }

要点：

• 只会根据条件求值顺序执行一个分支
• 输出以执行的步骤 id 为键
• 后续步骤应处理所有可能的分支输出
• 当下一步骤需要处理多个可能分支时，在 inputSchema 中使用可选字段
• 条件按定义顺序求值

输入数据映射

使用 .then()、.parallel() 或 .branch() 时，有时需要转换前一步骤的输出以匹配下一步骤的输入。在这些情况下，你可以使用 .map() 访问 inputData 并转换它，为下一步骤创建合适的数据结构。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});
const step2 = createStep({...});

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .map(async ({ inputData }) => {
    const { foo } = inputData;
    return {
      bar: `new ${foo}`,
    };
  })
  .then(step2)
  .commit();

.map() 方法为更复杂的映射场景提供了额外的辅助函数。

可用的辅助函数：

• getStepResult()：访问特定步骤的完整输出
• getInitData<any>()：访问 workflow 的初始输入数据
• mapVariable()：使用声明式对象语法提取和重命名字段

Parallel 和 Branch 输出

.parallel() 和 .branch() 都返回以子步骤 id 为键的对象。使用 .map() 重塑该输出以供下一个操作使用。此示例将两个并行结果映射为 .branch() 所需的共享输入：

src/mastra/workflows/order-routing-workflow.ts

import { createStep, createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const orderSchema = z.object({ amount: z.number(), inventory: z.number() })
const routingSchema = z.object({ riskScore: z.number(), inStock: z.boolean() })
const decisionSchema = z.object({ action: z.enum(['review', 'approve']) })

const scoreRisk = createStep({
  id: 'score-risk',
  inputSchema: orderSchema,
  outputSchema: z.object({ riskScore: z.number() }),
  execute: async ({ inputData }) => ({
    riskScore: inputData.amount >= 1_000 ? 80 : 20,
  }),
})

const checkInventory = createStep({
  id: 'check-inventory',
  inputSchema: orderSchema,
  outputSchema: z.object({ inStock: z.boolean() }),
  execute: async ({ inputData }) => ({ inStock: inputData.inventory > 0 }),
})

const manualReview = createStep({
  id: 'manual-review',
  inputSchema: routingSchema,
  outputSchema: decisionSchema,
  execute: async () => ({ action: 'review' as const }),
})

const autoApprove = createStep({
  id: 'auto-approve',
  inputSchema: routingSchema,
  outputSchema: decisionSchema,
  execute: async () => ({ action: 'approve' as const }),
})

export const orderRoutingWorkflow = createWorkflow({
  id: 'order-routing',
  inputSchema: orderSchema,
  outputSchema: z.object({
    'manual-review': decisionSchema.optional(),
    'auto-approve': decisionSchema.optional(),
  }),
})
  .parallel([scoreRisk, checkInventory])
  .map(async ({ inputData }) => ({
    riskScore: inputData['score-risk'].riskScore,
    inStock: inputData['check-inventory'].inStock,
  }))
  .branch([
    [async ({ inputData }) => inputData.riskScore >= 70 || !inputData.inStock, manualReview],
    [async ({ inputData }) => inputData.riskScore < 70 && inputData.inStock, autoApprove],
  ])
  .commit()

在 .map() 之前，并行输出的结构如下：

{
  'score-risk': { riskScore: 80 },
  'check-inventory': { inStock: true },
}

映射后的 { riskScore, inStock } 对象会传递给每个分支条件和分支步骤，因此它们的 inputSchema 值必须匹配 routingSchema。放在 .branch() 之后的 .map() 工作方式相同。使用 inputData['manual-review'] 或 getStepResult(manualReview) 读取分支输出。

循环步骤

Workflows 支持不同的循环方法，让你重复执行步骤直到或当条件满足时，或者遍历数组。循环可以与其他控制方法（如 .then()）组合使用。

使用 .dountil() 循环

使用 .dountil() 重复运行步骤，直到条件变为 true。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});

const step2 = createStep({
  execute: async ({ inputData }) => {
    const { number } = inputData;
    return {
      number: number + 1
    };
  }
});

export const testWorkflow = createWorkflow({})
  .then(step1)
  .dountil(step2, async ({ inputData: { number } }) => number > 10)
  .commit();

使用 .dowhile() 循环

使用 .dowhile() 在条件保持为 true 时重复运行步骤。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});

const step2 = createStep({
  execute: async ({ inputData }) => {
    const { number } = inputData;
    return {
      number: number + 1
    };
  }
});

export const testWorkflow = createWorkflow({})
  .then(step1)
  .dowhile(step2, async ({ inputData: { number } }) => number < 10)
  .commit();

在迭代之间传递数据

.dountil() 或 .dowhile() 的每次迭代都会将上一次迭代的输出作为其 inputData。当启用输入校验（默认）时，该输出会在循环体运行前根据循环体的 inputSchema 进行校验。对于 z.object() 之类的对象 schema，schema 未声明的键会被移除。允许未知键的 schema（例如 .passthrough()）会保留它们。

当循环体是嵌套 workflow 时，输出会先根据嵌套 workflow 的 inputSchema 校验，然后根据其第一个步骤的 inputSchema 校验。只有当嵌套 workflow 返回某个值并且两个 schema 都声明了它时，该值才能在下次迭代中使用。这包括嵌套 workflow 内部产生的值，例如从挂起并恢复的步骤返回的反馈。

要在迭代之间传递数据，请在循环体的 inputSchema 中声明它，并从循环体的最后一个步骤返回它：

src/mastra/workflows/test-workflow.ts

const loopSchema = z.object({
  draft: z.string(),
  feedback: z.string().optional(),
  attempts: z.number(),
})

const reviewStep = createStep({
  id: 'review',
  inputSchema: loopSchema,
  outputSchema: loopSchema,
  execute: async ({ inputData }) => {
    // ...
    return { draft, feedback, attempts: inputData.attempts + 1 }
  },
})

对于不属于步骤输入的值（例如计数器或审计数据），请改用 workflow state。

使用 .foreach() 循环

使用 .foreach() 为数组中的每个项运行同一步骤。输入必须是 array 类型，以便循环遍历其值，并对每个值应用步骤逻辑。关于何时使用 .foreach() 与其他方法的指导，请参阅选择合适的模式。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  inputSchema: z.string(),
  outputSchema: z.string(),
  execute: async ({ inputData }) => {
    return inputData.toUpperCase();
  }
});

const step2 = createStep({...});

export const testWorkflow = createWorkflow({
  inputSchema: z.array(z.string()),
  outputSchema: z.array(z.string())
})
  .foreach(step1)
  .then(step2)
  .commit();

输出结构

.foreach() 方法始终返回一个数组，包含每次迭代的输出。输出的顺序与输入的顺序一致。

src/mastra/workflows/test-workflow.ts

const addTenStep = createStep({
  id: 'add-ten',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ value: z.number() }),
  execute: async ({ inputData }) => ({
    value: inputData.value + 10,
  }),
})

export const testWorkflow = createWorkflow({
  id: 'foreach-output-example',
  inputSchema: z.array(z.object({ value: z.number() })),
  outputSchema: z.array(z.object({ value: z.number() })),
})
  .foreach(addTenStep)
  .commit()

// When executed with [{ value: 1 }, { value: 22 }, { value: 333 }]
// Output: [{ value: 11 }, { value: 32 }, { value: 343 }]

并发限制

使用 concurrency 控制处理的数组项数量。默认值为 1，即按顺序运行步骤。增大该值可让 .foreach() 同时处理多个项。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...})

export const testWorkflow = createWorkflow({...})
  .foreach(step1, { concurrency: 4 })
  .commit();

在 .foreach() 之后聚合结果

由于 .foreach() 输出数组，你可以使用 .then() 或 .map() 来聚合或转换结果。.foreach() 之后的步骤会将整个数组作为其输入。

src/mastra/workflows/test-workflow.ts

const processItemStep = createStep({
  id: 'process-item',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ processed: z.number() }),
  execute: async ({ inputData }) => ({
    processed: inputData.value * 2,
  }),
})

const aggregateStep = createStep({
  id: 'aggregate',
  // Input is an array of outputs from foreach
  inputSchema: z.array(z.object({ processed: z.number() })),
  outputSchema: z.object({ total: z.number() }),
  execute: async ({ inputData }) => ({
    // Sum all processed values
    total: inputData.reduce((sum, item) => sum + item.processed, 0),
  }),
})

export const testWorkflow = createWorkflow({
  id: 'foreach-aggregate-example',
  inputSchema: z.array(z.object({ value: z.number() })),
  outputSchema: z.object({ total: z.number() }),
})
  .foreach(processItemStep)
  .then(aggregateStep) // Receives the full array from foreach
  .commit()

// When executed with [{ value: 1 }, { value: 2 }, { value: 3 }]
// After foreach: [{ processed: 2 }, { processed: 4 }, { processed: 6 }]
// After aggregate: { total: 12 }

你也可以使用 .map() 转换数组输出：

src/mastra/workflows/test-workflow.ts

export const testWorkflow = createWorkflow({...})
  .foreach(processItemStep)
  .map(async ({ inputData }) => ({
    // Transform the array into a different structure
    values: inputData.map(item => item.processed),
    count: inputData.length
  }))
  .then(nextStep)
  .commit();

串联多个 .foreach() 调用

当串联 .foreach() 调用时，每个调用都会操作上一步骤的数组输出。当数组中的每个项需要按顺序经过多个步骤转换时，这很有用。

src/mastra/workflows/test-workflow.ts

const chunkStep = createStep({
  id: 'chunk',
  // Takes a document, returns an array of chunks
  inputSchema: z.object({ content: z.string() }),
  outputSchema: z.array(z.object({ chunk: z.string() })),
  execute: async ({ inputData }) => {
    // Split document into chunks
    const chunks = inputData.content.match(/.{1,100}/g) || []
    return chunks.map(chunk => ({ chunk }))
  },
})

const embedStep = createStep({
  id: 'embed',
  // Takes a single chunk, returns embedding
  inputSchema: z.object({ chunk: z.string() }),
  outputSchema: z.object({ embedding: z.array(z.number()) }),
  execute: async ({ inputData }) => ({
    embedding: [/* vector embedding */],
  }),
})

// For a single document that produces multiple chunks:
export const singleDocWorkflow = createWorkflow({
  id: 'single-doc-rag',
  inputSchema: z.object({ content: z.string() }),
  outputSchema: z.array(z.object({ embedding: z.array(z.number()) })),
})
  .then(chunkStep) // Returns array of chunks
  .foreach(embedStep) // Process each chunk -> array of embeddings
  .commit()

对于多个文档且每个文档产生多个分块的情况，你有以下选择：

选项 1：在单个步骤中处理所有文档并控制批处理

src/mastra/workflows/test-workflow.ts

const downloadAndChunkStep = createStep({
  id: "download-and-chunk",
  inputSchema: z.array(z.string()),  // Array of URLs
  outputSchema: z.array(z.object({ chunk: z.string(), source: z.string() })),
  execute: async ({ inputData: urls }) => {
    // Control batching/parallelization within the step
    const allChunks = [];
    for (const url of urls) {
      const content = await fetch(url).then(r => r.text());
      const chunks = content.match(/.{1,100}/g) || [];
      allChunks.push(...chunks.map(chunk => ({ chunk, source: url })));
    }
    return allChunks;
  }
});

export const multiDocWorkflow = createWorkflow({...})
  .then(downloadAndChunkStep)  // Returns flat array of all chunks
  .foreach(embedStep, { concurrency: 10 })  // Embed each chunk in parallel
  .commit();

选项 2：使用 foreach 处理文档并聚合分块，再用 then/foreach 处理嵌入

src/mastra/workflows/test-workflow.ts

const downloadStep = createStep({
  id: 'download',
  inputSchema: z.string(), // Single URL
  outputSchema: z.object({ content: z.string(), source: z.string() }),
  execute: async ({ inputData: url }) => ({
    content: await fetch(url).then(r => r.text()),
    source: url,
  }),
})

const chunkDocStep = createStep({
  id: 'chunk-doc',
  inputSchema: z.object({ content: z.string(), source: z.string() }),
  outputSchema: z.array(z.object({ chunk: z.string(), source: z.string() })),
  execute: async ({ inputData }) => {
    const chunks = inputData.content.match(/.{1,100}/g) || []
    return chunks.map(chunk => ({ chunk, source: inputData.source }))
  },
})

export const multiDocWorkflow = createWorkflow({
  id: 'multi-doc-rag',
  inputSchema: z.array(z.string()), // Array of URLs
  outputSchema: z.array(z.object({ embedding: z.array(z.number()) })),
})
  .foreach(downloadStep, { concurrency: 5 }) // Download docs in parallel
  .foreach(chunkDocStep) // Chunk each doc -> array of chunk arrays
  .map(async ({ inputData }) => {
    // Flatten nested arrays: [[chunks], [chunks]] -> [chunks]
    return inputData.flat()
  })
  .foreach(embedStep, { concurrency: 10 }) // Embed all chunks
  .commit()

关于串联 .foreach() 的要点：

• 每个 .foreach() 都操作上一步骤产生的数组
• 如果 .foreach() 内的步骤返回数组，输出将变成数组的数组
• 需要时使用 .map() 搭配 .flat() 展平嵌套数组
• 对于复杂的 RAG 管线，选项 1（在单个步骤中处理批处理）通常能提供更好的控制

foreach 中的嵌套 workflows

.foreach() 之后的步骤只有在所有迭代完成后才会执行。如果你需要为每个项运行多个顺序操作，请使用嵌套 workflow，而不是串联多个 .foreach() 调用。这样可以将每个项的所有操作保持在一起，并使数据流更清晰。

src/mastra/workflows/test-workflow.ts

// Define a workflow that processes a single document
const processDocumentWorkflow = createWorkflow({
  id: 'process-document',
  inputSchema: z.object({ url: z.string() }),
  outputSchema: z.object({
    embeddings: z.array(z.array(z.number())),
    metadata: z.object({ url: z.string(), chunkCount: z.number() }),
  }),
})
  .then(downloadStep) // Download the document
  .then(chunkStep) // Split into chunks
  .then(embedChunksStep) // Embed all chunks for this document
  .then(formatResultStep) // Format the final output
  .commit()

// Use the nested workflow inside foreach
export const batchProcessWorkflow = createWorkflow({
  id: 'batch-process-documents',
  inputSchema: z.array(z.object({ url: z.string() })),
  outputSchema: z.array(
    z.object({
      embeddings: z.array(z.array(z.number())),
      metadata: z.object({ url: z.string(), chunkCount: z.number() }),
    }),
  ),
})
  .foreach(processDocumentWorkflow, { concurrency: 3 })
  .commit()

// Each document goes through all 4 steps before the next document starts (with concurrency: 1)
// With concurrency: 3, up to 3 documents process their full pipelines in parallel

为什么使用嵌套 workflows：

• 更好的并行性：使用 concurrency: N 时，多个项会同时运行各自的完整管线。串联的 .foreach().foreach() 会让所有项先完成第 1 步、等待，然后全部完成第 2 步；而嵌套 workflows 让每个项独立推进
• 一个项的所有步骤会在收集结果之前一起完成
• 比创建嵌套数组的多个 .foreach() 调用更清晰
• 每个嵌套 workflow 执行都是独立的，拥有自己的数据流
• 更容易单独测试和复用每个项的逻辑

工作方式：

1. 父 workflow 将每个数组项传递给嵌套 workflow 的一个实例
2. 每个嵌套 workflow 为该条目运行其完整的步骤序列
3. 当 concurrency > 1 时，多个嵌套 workflows 并行执行
4. 嵌套 workflow 的最终输出成为结果数组中的一个元素
5. 所有嵌套 workflows 完成后，父 workflow 的下一步骤接收完整数组

标记 control-flow 条目

每个 control-flow 方法都接受一个可选的最后一个参数，包含稳定的 id、description 和可 JSON 序列化的 metadata。这些字段标识并描述图条目本身，与其中的步骤分开。它们会在序列化和存储后保留，并出现在 serializedStepGraph 中。它们对执行没有影响。

workflow
  .parallel([validateStep, enrichStep], {
    id: 'independent-enrichment',
    description: 'Run independent enrichment tasks concurrently',
    metadata: { title: 'Independent enrichment' },
  })
  .sleep(5000, { id: 'wait-before-retry', description: 'Pause before retrying' })

可视化编辑器和审阅工具可以按 id 索引布局和选择状态，并将 metadata.title 渲染为显示标签。对于 .foreach()，请在与 concurrency 相同的 options 对象中传入这些字段。对于 .map()、.sleep() 和 .sleepUntil()，提供的 id 会替换生成的条目 id。

选择合适的模式

使用本节作为选择合适控制流方法的参考。

快速参考

方法	用途	输入	输出	并发
.then(step)	顺序处理	T	U	不适用（一次一个）
.parallel([a, b])	对相同输入执行不同操作	T	{ a: U, b: V }	全部同时运行
.foreach(step)	对每个数组项执行相同操作	T[]	U[]	可配置（默认：1）
.branch([...])	条件路径选择	T	{ selectedStep: U }	只运行一个分支

.parallel() 与 .foreach()

当你有一个输入需要不同处理时，使用 .parallel()：

// Same user data processed differently in parallel
workflow.parallel([validateStep, enrichStep, scoreStep]).then(combineResultsStep)

当你有许多输入需要相同处理时，使用 .foreach()：

// Multiple URLs each processed the same way
workflow.foreach(downloadStep, { concurrency: 5 }).then(aggregateStep)

何时使用嵌套 workflows

在 .foreach() 内部——当每个数组项需要多个顺序步骤时：

// Each document goes through a full pipeline
const processDocWorkflow = createWorkflow({...})
  .then(downloadStep)
  .then(parseStep)
  .then(embedStep)
  .commit();

workflow.foreach(processDocWorkflow, { concurrency: 3 })

单个 .foreach() 可以保持结果扁平。串联 .foreach().foreach() 会创建嵌套数组。

在 .parallel() 内部——当并行分支需要自己的多步骤管线时：

const pipelineA = createWorkflow({...}).then(step1).then(step2).commit();
const pipelineB = createWorkflow({...}).then(step3).then(step4).commit();

workflow.parallel([pipelineA, pipelineB])

串联模式

模式	发生什么	常见用例
.then().then()	顺序步骤	简单管线
.parallel().then()	并行运行，然后合并	Fan-out/fan-in
.foreach().then()	处理所有项，然后聚合	Map-reduce
.foreach().foreach()	创建数组的数组	避免——改用嵌套 workflow 或 .map() 搭配 .flat()
.foreach(workflow)	每个项的完整管线	每个数组项的多步骤处理

同步：下一步骤何时运行？

.parallel() 和 .foreach() 都是同步点。workflow 中的下一步骤只有在所有并行分支或所有数组迭代完成后才会执行。

workflow
  .parallel([stepA, stepB, stepC]) // All 3 run simultaneously
  .then(combineStep) // Waits for ALL 3 to finish before running
  .commit()

workflow
  .foreach(processStep, { concurrency: 5 }) // Up to 5 items process at once
  .then(aggregateStep) // Waits for ALL items to finish before running
  .commit()

这意味着：

• .parallel() 将所有分支输出收集到一个对象，然后传递给下一步骤
• .foreach() 将所有迭代输出收集到一个数组，然后传递给下一步骤
• 结果无法在其完成时"流式"传递给下一步骤

并发行为

方法	行为
.then()	顺序——一次一个步骤
.parallel()	所有分支同时运行（无限制选项）
.foreach()	通过 { concurrency: N } 控制——默认为 1（顺序）
.foreach() 中的嵌套 workflow	遵循父级的并发设置

**性能提示：**对于 .foreach() 中的 I/O 密集型操作，请提高并发数以并行处理各项：

// Process up to 10 items simultaneously
workflow.foreach(fetchDataStep, { concurrency: 10 })

循环管理

循环条件可以通过不同方式实现，取决于你希望循环如何结束。

常见模式会检查 inputData 中返回的值，并设置最大迭代次数。它们也可以在达到限制时中止执行。

中止循环

使用 iterationCount 限制循环运行次数。如果计数超过阈值，请抛出错误以使该步骤失败并停止 workflow。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});

export const testWorkflow = createWorkflow({...})
  .dountil(step1, async ({ inputData: { userResponse, iterationCount } }) => {
    if (iterationCount >= 10) {
      throw new Error("Maximum iterations reached");
    }
    return userResponse === "yes";
  })
  .commit();

相关内容

• 挂起与恢复
• Human-in-the-loop