> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# Workflow 类

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

## 使用示例

```typescript
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

可以使用任何支持 [Standard JSON Schema](https://standardschema.dev/json-schema) 的库定义 workflow 的 `inputSchema` 和 `outputSchema`，包括 [Zod](https://zod.dev/)、[Valibot](https://valibot.dev/) 和 [ArkType](https://arktype.io/) 等库。

**Zod**:

```typescript
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**:

```typescript
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**:

```typescript
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();
```

## 构造函数参数

**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 指南。

**schedule.id** (`string`): 调度的稳定标识符。传入调度数组时必需；传入单个调度对象时默认为 workflow ID。

**schedule.cron** (`string`): 由 5、6 或 7 个部分组成的 cron 表达式。在构造 workflow 时验证。

**schedule.timezone** (`string`): IANA 时区，例如 "America/New\_York"。默认为主机的本地时区。请在生产环境中明确设置此项，避免触发时间依赖服务器 locale。

**schedule.inputData** (`TInput`): 每次触发时作为 workflow 输入传递的 payload。

**schedule.initialState** (`TState`): run 的初始状态。

**schedule.requestContext** (`Record<string, unknown>`): 附加到 run 的请求上下文。

**schedule.metadata** (`Record<string, unknown>`): 与调度记录一同持久化的任意 metadata。

**options** (`WorkflowOptions`): workflow 的可选配置

**options.tracingPolicy** (`TracingPolicy`): workflow 的可选 tracing 策略

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

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

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

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

**options.onFinish.status** (`WorkflowRunStatus`): workflow 状态：'success'、'failed'、'suspended' 或 'tripwire'

**options.onFinish.result** (`any`): workflow 输出（status 为 'success' 时）

**options.onFinish.error** (`SerializedError`): 错误详情（status 为 'failed' 时）

**options.onFinish.steps** (`Record<string, StepResult>`): 包含状态和输出的各步骤结果

**options.onFinish.tripwire** (`StepTripwireInfo`): tripwire 信息（status 为 'tripwire' 时）

**options.onFinish.runId** (`string`): 此 workflow run 的唯一标识符

**options.onFinish.workflowId** (`string`): workflow 的标识符

**options.onFinish.resourceId** (`string`): 可选资源标识符（如果在创建 run 时提供）

**options.onFinish.getInitData** (`() => any`): 返回传给 workflow 的初始输入数据的函数

**options.onFinish.mastra** (`Mastra`): Mastra 实例（如果 workflow 已在 Mastra 中注册）

**options.onFinish.requestContext** (`RequestContext`): 请求范围的上下文数据

**options.onFinish.logger** (`IMastraLogger`): workflow 的 logger 实例

**options.onFinish.state** (`Record<string, any>`): workflow 的当前状态对象

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

**options.onError.status** (`'failed' | 'tripwire'`): workflow 状态（'failed' 或 'tripwire'）

**options.onError.error** (`SerializedError`): 错误详情

**options.onError.steps** (`Record<string, StepResult>`): 包含状态和输出的各步骤结果

**options.onError.tripwire** (`StepTripwireInfo`): tripwire 信息（status 为 'tripwire' 时）

**options.onError.runId** (`string`): 此 workflow run 的唯一标识符

**options.onError.workflowId** (`string`): workflow 的标识符

**options.onError.resourceId** (`string`): 可选资源标识符（如果在创建 run 时提供）

**options.onError.getInitData** (`() => any`): 返回传给 workflow 的初始输入数据的函数

**options.onError.mastra** (`Mastra`): Mastra 实例（如果 workflow 已在 Mastra 中注册）

**options.onError.requestContext** (`RequestContext`): 请求范围的上下文数据

**options.onError.logger** (`IMastraLogger`): workflow 的 logger 实例

**options.onError.state** (`Record<string, any>`): workflow 的当前状态对象

## 使用初始状态运行

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

```typescript
const run = await workflow.createRun()

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

`initialState` 对象应与 workflow 的 `stateSchema` 所定义的结构匹配。更多详情请参阅 [Workflow 状态](https://mastra.zisheng.pro/docs/workflows/workflow-state)。

## Workflow 状态

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

**success** (`string`): 所有步骤均已成功执行完毕，并产生有效的结果输出

**failed** (`string`): workflow 执行过程中遇到错误，并提供错误详情

**suspended** (`string`): workflow 执行已暂停并等待恢复，同时提供已暂停步骤的信息

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

### 处理 tripwire 状态

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

```typescript
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 有意停止了执行（例如为了内容审核）。

## 相关内容

- [Step 类](https://mastra.zisheng.pro/reference/workflows/step)
- [Workflow 状态](https://mastra.zisheng.pro/docs/workflows/workflow-state)
- [控制流](https://mastra.zisheng.pro/docs/workflows/control-flow)