第8章 Workflows 工作流

# Workflows（工作流） Workflows 让你以清晰、结构化的步骤来定义复杂的任务序列，而不是依赖单个 agent 推理整个过程。任务的拆分、数据的流转和时机都由你掌控。Workflows 默认使用内置的执行引擎，也可以通过 Inngest 这类 workflow runner 使用托管基础设施。 针对 AI agent：直接运行 npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' 即可启动这个 workflow，无需编写临时脚本。该命令需要 Mastra 服务器处于运行状态：可以用 npx mastra dev 启动本地服务器，或通过 --url 传入可达服务器的 base URL。在构造其他输入之前，先运行 npx mastra api workflow run start --schema 查看输入结构。安装 Mastra 的 skill（npx skills add mastra-ai/skills --skill mastra）可获得完整的 CLI API 使用指引，包括发现、调用、schema、认证和错误处理。 ## 何时使用 workflows 当任务在开始前就有明确定义、包含多个需要按特定顺序执行的步骤时，使用 workflows。它能让你精细控制数据在步骤之间如何流转和转换，以及每个阶段调用哪些原语。 ## 核心原则 Mastra workflows 遵循以下原则： - 用 createStep 定义步骤，声明输入/输出 schema 和业务逻辑。 - 用 createWorkflow 组合步骤，定义执行流程。 - 运行 workflow 执行完整序列，内置对挂起、恢复和流式结果的支持。 ## 创建 workflow 步骤 步骤（Step）是 workflow 的基本构建单元。使用 createStep() 创建步骤，并通过 inputSchema 和 outputSchema 定义它接收和返回的数据。这两种 schema 都可以用 Standard JSON Schema 来定义（Zod、Valibot、ArkType 等）。 execute 函数定义步骤要做什么。可以用它调用你代码库中的函数、外部 API、agent 或 tools。 ZodValibotArkTypesrc/mastra/workflows/test-workflow.tsimport { 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 {
查看 Step 了解完整的配置选项列表。 使用 Agent 和 Tool Workflow 步骤还可以调用已注册的 Agent，或直接导入并执行 Tool。更多信息请参阅 Using Tools 页面。 创建 Workflow 使用 createWorkflow() 创建 workflow，通过 inputSchema 和 outputSchema 定义它接收和返回的数据。你可以使用 Standard JSON Schema（Zod、Valibot、ArkType 等）定义这两种 schema。使用 .then() 添加步骤，并用 .commit() 完成 workflow。 Zod Valibot ArkType 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();
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();
src/mastra/workflows/test-workflow.tsimport { 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 Class 以查看完整的配置选项。 理解控制流 直接链接至：理解控制流 工作流可以通过多种方法组合。选择的方法决定了每个步骤 schema 的结构。更多信息请访问“控制流”页面。 Studio 直接链接至：Studio 打开 Studio，并在“工作流”标签页中选择一个工作流。 - 图形视图：中间面板可视化显示工作流的步骤和执行流程。 - 输入表单：右侧边栏根据工作流的 inputSchema 生成表单。填写后即可启动运行。 - 实时状态：执行期间，图形会实时更新每个步骤的状态。边栏显示工作流的输入、输出、状态和日志。 - 时间旅行：运行完成后，可以重放单个步骤进行检查或重试。 工作流状态 直接链接至：工作流状态 工作流状态允许你在步骤间共享值，而无需通过每个步骤的 inputSchema 和 outputSchema 传递。你可以使用状态来跟踪进度、累积结果，或者在整个工作流中共享配置。 src/mastra/workflows/test-workflow.tsconst 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() }
},
})
如需了解状态模式、初始状态、挂起/恢复后的持久化以及嵌套 workflow 等完整文档，请访问 Workflow State。 将 workflow 作为 step 可以把 workflow 用作 step，从而在更大的组合中复用它的逻辑。输入和输出遵循核心原则中描述的相同 schema 规则。 src/mastra/workflows/test-workflow.tsconst 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.tsimport { 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，即可在整个应用中使用它。注册之后，它可以被 agent 或 tool 调用，并能访问共享资源，例如日志和可观测性功能： src/mastra/index.tsimport { Mastra } from '@mastra/core/mastra'
import { testWorkflow } from './workflows/test-workflow'

export const mastra = new Mastra({
workflows: { testWorkflow },
})
引用 workflow 可以从 agent、tool、Mastra Client 或命令行中运行 workflow。根据你的使用方式，调用 mastra 或 mastraClient 实例上的 .getWorkflow() 获取引用： const testWorkflow = mastra.getWorkflow('testWorkflow')
```

建议使用 mastra.getWorkflow() 而非直接导入，原因有二：一是可以访问 Mastra 实例的配置（如日志、遥测、存储、已注册的 agent 及向量数据库）；二是为 workflow 的输入和输出 schema 提供完整的 TypeScript 类型推断。

使用 workflow 的注册 key（即添加到 Mastra 时使用的 key）调用 getWorkflow()。通过 getWorkflowById() 可以依据其 id 属性获取 workflow，但其类型推断能力较弱。

## 运行 Workflows

Workflows 支持两种运行模式：start 会等待所有步骤执行完毕后再返回结果；stream 则会在执行过程中持续发出事件。你可以根据场景选择：若只需最终结果，使用 start；若需监控进度或在步骤完成时触发动作，使用 stream。

.start().stream()

使用 createRun() 创建 workflow 运行实例，然后调用 .start() 并传入符合 workflow inputSchema 的 inputData。workflow 将执行所有步骤并返回最终结果。

```js const run = await testWorkflow.createRun() const result = await run.start({ inputData: { message: 'Hello world', }, }) if (result.status === 'success') { console.log(result.result) } ```

使用 .createRun() 创建 workflow 运行实例，然后调用 .stream() 并传入符合 workflow inputSchema 的 inputData。迭代 fullStream 以跟踪进度，随后通过 await result 获取最终的 workflow 结果。

```js const run = await testWorkflow.createRun() const stream = run.stream({ inputData: { message: 'Hello world', }, }) for await (const chunk of stream.fullStream) { console.log(chunk) } // 获取最终结果（类型与 run.start() 相同） const result = await stream.result if (result.status === 'success') { console.log(result.result) } ```

``` Workflow 结果类型Direct link to Workflow result type run.start() 和 stream.result 都返回基于 status 属性的可辨识联合类型（discriminated union），status 的取值可以是 success、failed、suspended、tripwire 或 paused。无论状态如何，你都可以安全地访问 result.status、result.input、result.steps，以及可选的 result.state。 此外，根据状态不同，会开放不同的属性： statusUnique propertiesDescriptionsuccessresultWorkflow 的输出数据failederror导致失败的错误信息tripwiretripwire包含原因、retry?、metadata?、processorId?suspendedsuspendPayload, suspended暂停数据以及挂起步骤的路径数组paused(none)仅可用通用属性 若要访问状态特有的属性，需先检查 status： const result = await run.start({ inputData: { message: 'Hello world' } })

if (result.status === 'success') {
console.log(result.result) // 仅在状态为 "success" 时可用
} else if (result.status === 'failed') {
console.log(result.error.message)
} else if (result.status === 'suspended') {
console.log(result.suspendPayload)
}
Workflow 输出Direct link to Workflow output 以下是一个成功的 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!!!"
}
}
流式传输Direct link to Streaming 关于通用 writer API 的用法，请参阅 Streaming。 查看 Workflow 流数据负载Direct link to Inspecting workflow stream payloads 写入流的事件会包含在发出的数据块（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') {
// 如果工作流被挂起，可以用 resumeStream 方法恢复
const resumedStream = await run.resumeStream({
resumeData: { value: 'resume data' },
})

for await (const chunk of resumedStream) {
console.log(chunk)
}
}
恢复被中断的工作流流 如果工作流流因任何原因被关闭或中断，可以使用 resumeStream 方法恢复，它会返回一个新的 ReadableStream，用于继续观察工作流事件。 const newStream = await run.resumeStream()

for await (const chunk of newStream) {
console.log(chunk)
}
在工作流中使用 agent 将 agent 的 textStream 管道传输到工作流步骤的 writer，即可流式输出部分结果，Mastra 会自动把 agent 的用量汇总到工作流运行中。 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,
}
},
})
重启活跃的工作流运行 当工作流运行与服务器断开连接后，可以从最近一次活跃的步骤重新启动。对于断连时可能仍在运行的长任务工作流，这个功能非常有用。重启后会从最近活跃的步骤继续执行。 用 restartAllActiveWorkflowRuns() 重启工作流的所有活跃运行 使用 restartAllActiveWorkflowRuns() 可以一次性重启某个工作流的所有活跃运行，无需手动遍历每个运行并逐个重启。 workflow.restartAllActiveWorkflowRuns()
重启并发执行，每批最多处理 recovery.workflowConcurrency 个任务（默认值为 5）。如果你的运行任务共享不可并行的副作用，请在 Mastra 配置中将其设置为 1。 若需将某个工作流排除在由 Mastra.restartAllActiveWorkflowRuns() 触发的自动启动时重启之外，请在工作流的选项中设置 autoRestartActiveRuns: false。这对于包含不可被批量重启重新驱动副作用的工作流非常有用。持久化智能体工作流会默认自动设置此选项，因为其恢复机制由 opt-in 的 recovery.durableAgents: 'auto' 路径负责管理。 使用 restart() 重启活跃的工作流运行 （直接链接：restarting-an-active-workflow-run-with-restart） 使用 restart() 从最后一个活跃步骤重启活跃的工作流运行。执行将从该步骤恢复，工作流将从那里继续。 const run = await workflow.createRun()

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

const restartedResult = await run.restart()
识别活跃的工作流运行 （直接链接：Identifying active workflow runs） 当工作流处于活跃状态时，其状态为 running（运行中）或 waiting（等待中）。你可以检查工作流状态以确认其活跃，并使用 active 标识活跃的工作流运行。 const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
console.log(activeRuns.runs)
}
注意：在运行本地 mastra 服务器时，当服务器启动时，所有活跃的工作流运行都会自动重启。 使用 RequestContext （直接链接：using-requestcontext） 使用 RequestContext 访问请求特定的值。这允许你根据请求上下文有条件地调整行为。 src/mastra/workflows/test-workflow.tsexport 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 Validation”。 相关文档：直达“相关文档”链接 若希望更深入地了解工作流，请查阅我们的工作流指南，该指南通过一个实用示例逐步讲解核心概念。 工作流状态 控制流 挂起与恢复 错误处理 Workers：在专门的后台进程中运行工作流执行 📹 基于 Mastra 的 Agentic workflows 工作坊