Workflows

Workflows 让你用清晰、结构化的步骤定义复杂任务序列，而不是依赖单个 Agent 推理整个流程。你可以控制任务拆分方式，以及数据的流转和时序。Workflows 默认使用内置执行引擎，也可以通过 Workflow Runner（例如 Inngest）使用托管基础设施。

致 AI agents： 运行 npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' 可直接启动此 workflow，无需编写临时脚本。该命令需要运行中的 Mastra 服务端；可用 npx mastra dev 启动本地服务端，或通过 --url 传入可访问的服务端基础 URL。构造不同的输入前，先运行 npx mastra api workflow run start --schema。安装 Mastra skill（npx skills add mastra-ai/skills --skill mastra）可获取完整的 API CLI 发现、目标定位、schema、身份验证和错误处理指南。

何时使用 workflows

当任务在执行前就已明确定义，且包含多个步骤并具有特定执行顺序时，使用 workflows。它能让你精细控制步骤之间的数据流转与转换，以及每个阶段调用哪些原语。

核心原则

Mastra workflows 按以下原则运行：

• 使用 createStep 定义步骤，指定输入/输出 schema 和业务逻辑。
• 使用 createWorkflow 组合步骤，定义执行流程。
• 运行 workflows 以执行整个序列，内置支持挂起、恢复和结果流式传输。

创建 workflow 步骤

步骤是 workflows 的构建块。使用 createStep() 创建步骤，并通过 inputSchema 和 outputSchema 定义它接受和返回的数据。两个 schema 都可以使用 Standard JSON Schema（Zod、Valibot、ArkType 等）定义。

execute 函数定义步骤的具体行为。可用它调用代码库中的函数、外部 API、Agents 或工具。

Zod

src/mastra/workflows/test-workflow.ts

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

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

    return {
      formatted: message.toUpperCase(),
    }
  },
})

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

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

    return {
      formatted: message.toUpperCase(),
    }
  },
})

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

完整配置选项参见 Step。

使用 Agents 和工具

Workflow 步骤也可以调用已注册的 Agents，或直接导入并执行工具。更多信息参见使用工具页面。

创建 workflow

使用 createWorkflow() 创建 workflow，并通过 inputSchema 和 outputSchema 定义它接受和返回的数据。两个 schema 都可以使用 Standard JSON Schema（Zod、Valibot、ArkType 等）定义。用 .then() 添加步骤，并用 .commit() 完成 workflow。

Zod

src/mastra/workflows/test-workflow.ts

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

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

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    output: z.string()
  })
})
  .then(step1)
  .commit();

Valibot

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import * as v from "valibot";
import { toStandardJsonSchema } from "@valibot/to-json-schema";

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

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: toStandardJsonSchema(v.object({
    message: v.string()
  })),
  outputSchema: toStandardJsonSchema(v.object({
    output: v.string()
  }))
})
  .then(step1)
  .commit();

ArkType

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { type } from "arktype";

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

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: type({
    message: "string"
  }),
  outputSchema: type({
    output: "string"
  })
})
  .then(step1)
  .commit();

完整配置选项参见 Workflow 类。

理解控制流

Workflows 可通过多种方法组合。所选方法决定每个步骤的 schema 应如何组织。更多信息参见控制流页面。

Studio

打开 Studio，在 Workflows 标签页中选择一个 workflow。

• 图视图：中间面板可视化 workflow 的步骤和执行流程。
• 输入表单：右侧边栏根据 workflow 的 inputSchema 生成表单。填写后即可开始运行。
• 实时状态：执行期间，图表会实时更新每个步骤的状态。侧边栏显示 workflow 的输入、输出、状态和日志。
• 时间旅行：运行完成后，可重放单个步骤以检查或重试。

Workflow 状态

Workflow 状态让你无需通过每个步骤的 inputSchema 和 outputSchema 传递，也能在各步骤间共享值。可用状态跟踪进度、累积结果，或在整个 workflow 中共享配置。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  stateSchema: z.object({ counter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // Read from state
    console.log(state.counter)

    // Update state for subsequent steps
    setState({ ...state, counter: state.counter + 1 })

    return { formatted: inputData.message.toUpperCase() }
  },
})

关于状态 schema、初始状态、跨挂起/恢复的持久化以及嵌套 workflows 的完整文档，参见 Workflow 状态。

将 workflows 用作步骤

将 workflow 用作步骤，可在更大的组合中复用其逻辑。输入和输出遵循核心原则中描述的相同 schema 规则。

src/mastra/workflows/test-workflow.ts

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

const childWorkflow = createWorkflow({
  id: "child-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(step1)
  .then(step2)
  .commit();

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(childWorkflow)
  .commit();

克隆 workflow

若想复用 workflow 的逻辑，但以新 ID 单独跟踪，可使用 cloneWorkflow() 克隆。每个克隆独立运行，并在日志和可观测性工具中显示为不同的 workflow。

src/mastra/workflows/test-workflow.ts

import { cloneWorkflow } from "@mastra/core/workflows";

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

const parentWorkflow = createWorkflow({...})
const clonedWorkflow = cloneWorkflow(parentWorkflow, { id: "cloned-workflow" });

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .then(clonedWorkflow)
  .commit();

注册 workflow

在 Mastra 实例中注册 workflow，使其在整个应用中可用。注册后，Agents 或工具可以调用它，并可使用日志和可观测性等共享资源：

src/mastra/index.ts

import { Mastra } from '@mastra/core/mastra'
import { testWorkflow } from './workflows/test-workflow'

export const mastra = new Mastra({
  workflows: { testWorkflow },
})

引用 workflow

你可以从 Agents、工具、Mastra Client 或命令行运行 workflows。根据你的设置，在 mastra 或 mastraClient 实例上调用 .getWorkflow() 获取引用：

const testWorkflow = mastra.getWorkflow('testWorkflow')

备注

相比直接导入，推荐使用 mastra.getWorkflow()，原因有二：

1. 它可以访问 Mastra 实例配置（logger、telemetry、storage、已注册的 agents 和 vector stores）
2. 它为 workflow 的输入和输出 schema 提供完整的 TypeScript 类型推断

使用 workflow 的注册键（添加到 Mastra 时使用的键）调用 getWorkflow()。getWorkflowById() 按其 id 属性检索 workflows，但不提供同等程度的类型推断。

运行 workflows

Workflows 有两种运行模式：start 会等待所有步骤完成后返回，而 stream 在执行期间发出事件。根据用例选择：只需最终结果时用 start，需要监控进度或在步骤完成时触发操作时用 stream。

.start()

使用 createRun() 创建工作流运行实例，然后用与 workflow inputSchema 匹配的 inputData 调用 .start()。Workflow 会执行所有步骤并返回最终结果。

const run = await testWorkflow.createRun()

const result = await run.start({
  inputData: {
    message: 'Hello world',
  },
})

if (result.status === 'success') {
  console.log(result.result)
}

.stream()

使用 .createRun() 创建工作流运行实例，然后用与 workflow inputSchema 匹配的 inputData 调用 .stream()。遍历 fullStream 跟踪进度，再 await result 获取最终的 workflow 结果。

const run = await testWorkflow.createRun()

const stream = run.stream({
  inputData: {
    message: 'Hello world',
  },
})

for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get the final result (same type as run.start())
const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Workflow 结果类型

run.start() 和 stream.result 都返回基于 status 属性的可辨识联合，status 可为 success、failed、suspended、tripwire 或 paused。无论状态如何，都可以安全访问 result.status、result.input、result.steps，以及可选的 result.state。

此外，不同状态还提供不同属性：

状态	特有属性	说明
success	result	Workflow 的输出数据
failed	error	导致失败的错误
tripwire	tripwire	包含 reason、retry?、metadata?、processorId?
suspended	suspendPayload、suspended	挂起数据以及被挂起步骤路径的数组
paused	（无）	仅提供通用属性

要访问状态特有的属性，请先检查 status：

const result = await run.start({ inputData: { message: 'Hello world' } })

if (result.status === 'success') {
  console.log(result.result) // Only available when status is "success"
} else if (result.status === 'failed') {
  console.log(result.error.message)
} else if (result.status === 'suspended') {
  console.log(result.suspendPayload)
}

Workflow 输出

以下是一个成功 workflow 结果的示例，展示了 input、steps 和 result 属性：

{
  "status": "success",
  "steps": {
    "step-1": {
      "status": "success",
      "payload": {
        "message": "Hello world"
      },
      "output": {
        "formatted": "HELLO WORLD"
      }
    },
    "step-2": {
      "status": "success",
      "payload": {
        "formatted": "HELLO WORLD"
      },
      "output": {
        "emphasized": "HELLO WORLD!!!"
      }
    }
  },
  "input": {
    "message": "Hello world"
  },
  "result": {
    "emphasized": "HELLO WORLD!!!"
  }
}

流式传输

关于通用 writer API 的用法，参见流式传输。

检查 workflow 流负载

写入流的事件会包含在发出的 chunks 中。检查这些 chunks 可访问自定义字段，例如事件类型、中间值或步骤特定数据。

const testWorkflow = mastra.getWorkflow('testWorkflow')

const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}

if (result!.status === 'suspended') {
  // if the workflow is suspended, we can resume it with the resumeStream method
  const resumedStream = await run.resumeStream({
    resumeData: { value: 'resume data' },
  })

  for await (const chunk of resumedStream) {
    console.log(chunk)
  }
}

恢复中断的 workflow 流

如果 workflow 流因任何原因关闭或中断，可使用 resumeStream 方法恢复。它会返回一个新的 ReadableStream，用于观察 workflow 事件。

const newStream = await run.resumeStream()

for await (const chunk of newStream) {
  console.log(chunk)
}

在 workflow 中使用 Agent

将 Agent 的 textStream 管道到 workflow 步骤的 writer。这样会流式传输部分输出，Mastra 会自动将 Agent 的用量聚合到 workflow 运行中。

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

export const testStep = createStep({
  execute: async ({ inputData, mastra, writer }) => {
    const { city } = inputData

    const testAgent = mastra?.getAgent('testAgent')
    const stream = await testAgent?.stream(`What is the weather in ${city}?`)

    await stream!.textStream.pipeTo(writer!)

    return {
      value: await stream!.text,
    }
  },
})

重启活跃的 workflow 运行

当 workflow 运行与服务端失去连接时，可以从最后一个活跃步骤重启。这对于服务端断连时可能仍在运行的长时 workflows 很有用。重启 workflow 运行会从最后一个活跃步骤恢复执行，并从那里继续。

使用 restartAllActiveWorkflowRuns() 重启某个 workflow 的所有活跃运行

使用 restartAllActiveWorkflowRuns() 重启某个 workflow 的所有活跃运行。这样无需手动遍历每个运行并逐一重启。

workflow.restartAllActiveWorkflowRuns()

运行会并发重启，最多同时执行 recovery.workflowConcurrency 个（默认 5）。如果多个运行共享不能并行执行的副作用，请在 Mastra 配置中将其设为 1。

若要排除某个 workflow，使其不受 Mastra.restartAllActiveWorkflowRuns() 触发的启动时自动重启影响，可在 workflow 的 options 中设置 autoRestartActiveRuns: false。这适用于带有副作用、不能被一刀切重启再次驱动的 workflows。持久化 Agent workflows 会自动设置此项，因为其恢复由需显式启用的 recovery.durableAgents: 'auto' 路径负责。

运行执行期间，Mastra 会持有该运行的存储租约并在后台续租。restart() 拒绝接管租约尚未过期的运行：它会抛出 WORKFLOW_RESTART_RUN_OWNED 错误，扫描流程也会跳过该运行。租约是建议性的，不会隔离写入。如果进程停止续租但仍继续执行步骤（例如事件循环阻塞时间超过租约 TTL），恢复流程可能会再次执行该步骤。启动时扫描 mastra.restartAllActiveWorkflowRuns() 会在租约可能已过期后重新检查被跳过的运行，因此进程崩溃的运行仍可恢复。workflow.restartAllActiveWorkflowRuns() 不会重新检查，直接跳过该运行；如需恢复该运行，请在租约过期后调用。不支持运行所有权的存储适配器保持原有行为，会重启所有活跃运行。

使用 restart() 重启活跃的 workflow 运行

使用 restart() 从最后一个活跃步骤重启活跃的 workflow 运行。它会从最后一个活跃步骤恢复执行，workflow 将从那里继续。

const run = await workflow.createRun()

const result = await run.start({ inputData: { value: 'initial data' } })

const restartedResult = await run.restart()

识别活跃的 workflow 运行

当 workflow 运行处于活跃状态时，其 status 为 running 或 waiting。你可以检查 workflow 的 status 确认其是否活跃，并使用 active 识别活跃的 workflow 运行。

const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
  console.log(activeRuns.runs)
}

备注

运行本地 mastra 服务端时，所有活跃的 workflow 运行会在服务端启动时自动重启。

使用 RequestContext

使用 RequestContext 访问请求特定的值。这样可以根据请求上下文有条件地调整行为。

src/mastra/workflows/test-workflow.ts

export type UserTier = {
  'user-tier': 'enterprise' | 'pro'
}

const step1 = createStep({
  execute: async ({ requestContext }) => {
    const userTier = requestContext.get('user-tier') as UserTier['user-tier']

    const maxResults = userTier === 'enterprise' ? 1000 : 50

    return { maxResults }
  },
})

更多信息参见 Request Context。

提示

如需类型安全的请求上下文 schema 校验，参见 Schema 校验。

相关内容

想进一步了解 workflows，请参阅我们的 Workflow 指南，其中通过一个实际示例讲解了核心概念。

• Workflow 状态
• 控制流
• 挂起与恢复
• 错误处理
• Workers：在专用后台进程中执行 workflow
• 📹 使用 Mastra 的 Agentic workflows 研讨会