跳到主要内容

Workflow 类

Workflow 类用于为包含条件分支和数据验证的复杂操作序列创建状态机。

使用示例
使用示例的直接链接

src/mastra/workflows/test-workflow.ts
import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

export const workflow = createWorkflow({
  id: 'test-workflow',
  inputSchema: z.object({
    value: z.string(),
  }),
  outputSchema: z.object({
    value: z.string(),
  }),
})

定义 schema
定义 schema的直接链接

可以使用任何支持 Standard JSON Schema 的库定义 workflow 的 inputSchema 和 outputSchema，包括 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();

构造函数参数
构造函数参数的直接链接

id:

string
workflow 的唯一标识符

inputSchema:

StandardJSONSchemaV1
定义 workflow 输入结构的 Standard JSON Schema

outputSchema:

StandardJSONSchemaV1
定义 workflow 输出结构的 Standard JSON Schema

stateSchema?:

StandardJSONSchemaV1
用于 workflow 状态的可选 Standard JSON Schema。使用 Mastra 的状态系统时会自动注入。若未指定，类型为 'any'。

requestContextSchema?:

StandardJSONSchemaV1
用于验证请求上下文值的 Standard JSON Schema。提供后，会在 run.start() 开始时验证上下文；验证失败则抛出错误。

schedule?:

WorkflowScheduleConfig | WorkflowScheduleConfig[]
workflow 的可选 cron 调度。接受单个配置或配置数组，以按多个周期触发。设置此项会自动将 workflow 提升为使用事件驱动的执行引擎。用法请参阅计划 workflow 指南。
WorkflowScheduleConfig

id?:

string
调度的稳定标识符。传入调度数组时必需；传入单个调度对象时默认为 workflow ID。

cron:

string
由 5、6 或 7 个部分组成的 cron 表达式。在构造 workflow 时验证。

timezone?:

string
IANA 时区，例如 "America/New_York"。默认为主机的本地时区。请在生产环境中明确设置此项，避免触发时间依赖服务器 locale。

inputData?:

TInput
每次触发时作为 workflow 输入传递的 payload。

initialState?:

TState
run 的初始状态。

requestContext?:

Record<string, unknown>
附加到 run 的请求上下文。

metadata?:

Record<string, unknown>
与调度记录一同持久化的任意 metadata。

options?:

WorkflowOptions
workflow 的可选配置
WorkflowOptions

tracingPolicy?:

TracingPolicy
workflow 的可选 tracing 策略

validateInputs?:

boolean
用于确定是否验证 workflow 输入的可选标志。它还会应用 workflow/步骤输入或恢复数据中 zodSchemas 的默认值。如果在 start/resume 时输入或恢复数据验证失败，workflow 不会启动或恢复，而会抛出错误。如果执行步骤时输入数据验证失败，该步骤会失败，导致 workflow 失败并返回错误。

shouldPersistSnapshot?:

(params: { stepResults: Record<string, StepResult<any, any, any, any>>; workflowStatus: WorkflowRunStatus }) => boolean
用于确定是否持久化 workflow 快照的可选标志

pruneSnapshot?:

(params: { snapshot: WorkflowRunState; workflowStatus: WorkflowRunStatus }) => WorkflowRunState
在持久化 workflow 快照前立即转换快照的可选 hook。必须返回可 JSON 序列化的数据，并保留 workflow 恢复所需的一切内容（已暂停步骤的 suspendPayloads、suspendedPaths、executionPath 等）。agent run 在内部使用它以保持快照精简；用户 workflow 默认持久化完整快照。

onFinish?:

(result: WorkflowFinishCallbackResult) => void | Promise<void>
workflow 以任意状态（success、failed、suspended、tripwire）完成时调用的回调。接收 workflow 结果，包括状态、输出、错误和步骤结果。此回调中抛出的错误会被捕获并记录，而不会向外传播。
WorkflowFinishCallbackResult

status:

WorkflowRunStatus
workflow 状态：'success'、'failed'、'suspended' 或 'tripwire'

result?:

any
workflow 输出（status 为 'success' 时）

error?:

SerializedError
错误详情（status 为 'failed' 时）

steps:

Record<string, StepResult>
包含状态和输出的各步骤结果

tripwire?:

StepTripwireInfo
tripwire 信息（status 为 'tripwire' 时）

runId:

string
此 workflow run 的唯一标识符

workflowId:

string
workflow 的标识符

resourceId?:

string
可选资源标识符（如果在创建 run 时提供）

getInitData:

() => any
返回传给 workflow 的初始输入数据的函数

mastra?:

Mastra
Mastra 实例（如果 workflow 已在 Mastra 中注册）

requestContext:

RequestContext
请求范围的上下文数据

logger:

IMastraLogger
workflow 的 logger 实例

state:

Record<string, any>
workflow 的当前状态对象

onError?:

(errorInfo: WorkflowErrorCallbackInfo) => void | Promise<void>
仅在 workflow 失败（failed 或 tripwire 状态）时调用的回调。接收错误详情和步骤结果。此回调中抛出的错误会被捕获并记录，而不会向外传播。
WorkflowErrorCallbackInfo

status:

'failed' | 'tripwire'
workflow 状态（'failed' 或 'tripwire'）

error?:

SerializedError
错误详情

steps:

Record<string, StepResult>
包含状态和输出的各步骤结果

tripwire?:

StepTripwireInfo
tripwire 信息（status 为 'tripwire' 时）

runId:

string
此 workflow run 的唯一标识符

workflowId:

string
workflow 的标识符

resourceId?:

string
可选资源标识符（如果在创建 run 时提供）

getInitData:

() => any
返回传给 workflow 的初始输入数据的函数

mastra?:

Mastra
Mastra 实例（如果 workflow 已在 Mastra 中注册）

requestContext:

RequestContext
请求范围的上下文数据

logger:

IMastraLogger
workflow 的 logger 实例

state:

Record<string, any>
workflow 的当前状态对象

使用初始状态运行
使用初始状态运行的直接链接

启动 workflow run 时，可以传入 initialState，为 workflow 状态设置初始值：

const run = await workflow.createRun()

const result = await run.start({
  inputData: { value: 'hello' },
  initialState: {
    counter: 0,
    items: [],
  },
})

initialState 对象应与 workflow 的 stateSchema 所定义的结构匹配。更多详情请参阅 Workflow 状态。

Workflow 状态
Workflow 状态的直接链接

workflow 的 status 表示其当前执行状态。可能的值如下：

success:

string
所有步骤均已成功执行完毕，并产生有效的结果输出

failed:

string
workflow 执行过程中遇到错误，并提供错误详情

suspended:

string
workflow 执行已暂停并等待恢复，同时提供已暂停步骤的信息

tripwire:

string
workflow 已被 processor tripwire 终止。当 workflow 中的 agent 步骤触发 tripwire 时（例如内容被 guardrail 阻止），就会发生这种情况。结果中包含 tripwire 信息。

处理 tripwire 状态
处理 tripwire 状态的直接链接

当 workflow 包含触发 tripwire 的 agent 步骤时，workflow 会返回 status: 'tripwire' 并包含 tripwire 详情：

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

if (result.status === 'tripwire') {
  console.log('Workflow terminated by tripwire:', result.tripwire?.reason)
  console.log('Processor ID:', result.tripwire?.processorId)
  console.log('Retry requested:', result.tripwire?.retry)
}

这与表示意外错误的 status: 'failed' 不同。tripwire 状态表示 processor 有意停止了执行（例如为了内容审核）。

On this page