跳至主要內容

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
用於驗證請求 context 值的 Standard JSON Schema。提供此項後，context 會在 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"。預設為主機的本地時區。在生產環境中應明確設定，避免觸發時間取決於伺服器地區設定。

inputData?:

TInput
每次觸發時，作為 Workflow 輸入傳入的 payload。

initialState?:

TState
執行的初始狀態。

requestContext?:

Record<string, unknown>
附加至執行的請求 context。

metadata?:

Record<string, unknown>
與排程記錄一同持久保存的任意 metadata。

options?:

WorkflowOptions
Workflow 的可選選項
WorkflowOptions

tracingPolicy?:

TracingPolicy
Workflow 的可選 tracing policy

validateInputs?:

boolean
用於決定是否驗證 Workflow 輸入的可選旗標。此設定亦會將 zodSchemas 的預設值套用至 Workflow／step 的輸入／恢復資料。如在開始／恢復時，輸入／恢復資料驗證失敗，Workflow 不會開始／恢復，而會拋出錯誤。如執行 step 時輸入資料驗證失敗，該 step 會失敗，導致 Workflow 失敗並傳回錯誤。

shouldPersistSnapshot?:

(params: { stepResults: Record<string, StepResult<any, any, any, any>>; workflowStatus: WorkflowRunStatus }) => boolean
用於決定是否持久保存 Workflow snapshot 的可選旗標

pruneSnapshot?:

(params: { snapshot: WorkflowRunState; workflowStatus: WorkflowRunStatus }) => WorkflowRunState
在持久保存 Workflow snapshot 前，立即轉換該 snapshot 的可選 hook。必須傳回可序列化為 JSON 的資料，並保留 Workflow 恢復時所需的一切（已暫停 step 的 suspendPayloads、suspendedPaths、executionPath 等）。Agent 執行會在內部使用此項，令 snapshot 保持精簡；使用者 Workflow 預設會持久保存完整 snapshot。

onFinish?:

(result: WorkflowFinishCallbackResult) => void | Promise<void>
Workflow 以任何狀態（success、failed、suspended、tripwire）完成時調用的 callback。會接收包括狀態、輸出、錯誤及 step 結果在內的 Workflow 結果。此 callback 拋出的錯誤會被捕捉並記錄，不會向外傳播。
WorkflowFinishCallbackResult

status:

WorkflowRunStatus
Workflow 狀態：'success'、'failed'、'suspended' 或 'tripwire'

result?:

any
Workflow 輸出（狀態為 'success' 時）

error?:

SerializedError
錯誤詳細資料（狀態為 'failed' 時）

steps:

Record<string, StepResult>
各個 step 的結果，包括其狀態及輸出

tripwire?:

StepTripwireInfo
Tripwire 資料（狀態為 'tripwire' 時）

runId:

string
此 Workflow 執行的唯一識別符

workflowId:

string
Workflow 的識別符

resourceId?:

string
可選資源識別符（如建立執行時有提供）

getInitData:

() => any
傳回提供予 Workflow 的初始輸入資料的函式

mastra?:

Mastra
Mastra 實例（如 Workflow 已註冊至 Mastra）

requestContext:

RequestContext
以請求為範圍的 context 資料

logger:

IMastraLogger
Workflow 的 logger 實例

state:

Record<string, any>
Workflow 目前的狀態物件

onError?:

(errorInfo: WorkflowErrorCallbackInfo) => void | Promise<void>
僅在 Workflow 失敗（failed 或 tripwire 狀態）時調用的 callback。會接收錯誤詳細資料及 step 結果。此 callback 拋出的錯誤會被捕捉並記錄，不會向外傳播。
WorkflowErrorCallbackInfo

status:

'failed' | 'tripwire'
Workflow 狀態（'failed' 或 'tripwire'）

error?:

SerializedError
錯誤詳細資料

steps:

Record<string, StepResult>
各個 step 的結果，包括其狀態及輸出

tripwire?:

StepTripwireInfo
Tripwire 資料（狀態為 'tripwire' 時）

runId:

string
此 Workflow 執行的唯一識別符

workflowId:

string
Workflow 的識別符

resourceId?:

string
可選資源識別符（如建立執行時有提供）

getInitData:

() => any
傳回提供予 Workflow 的初始輸入資料的函式

mastra?:

Mastra
Mastra 實例（如 Workflow 已註冊至 Mastra）

requestContext:

RequestContext
以請求為範圍的 context 資料

logger:

IMastraLogger
Workflow 的 logger 實例

state:

Record<string, any>
Workflow 目前的狀態物件

使用初始狀態執行
使用初始狀態執行 的直接連結

開始執行 Workflow 時，你可以傳入 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
所有 step 均已成功完成執行，並產生有效的結果輸出

failed:

string
Workflow 執行期間遇到錯誤，並有錯誤詳細資料可供查閱

suspended:

string
Workflow 執行已暫停並等待恢復，並有已暫停 step 的資料可供查閱

tripwire:

string
Workflow 已由 processor tripwire 終止。當 Workflow 內的 Agent step 觸發 tripwire（例如內容被 guardrail 封鎖）時便會發生。結果中會提供 tripwire 資料。

處理 tripwire 狀態
處理 tripwire 狀態 的直接連結

當 Workflow 包含觸發 tripwire 的 Agent step，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