Workflow 狀態

Workflow 狀態讓你毋須透過每個步驟的 inputSchema 和 outputSchema 傳遞值，也能在步驟之間共享值。這適合用於追蹤進度或累積結果，亦可在整個 Workflow 之間共享設定。

狀態與步驟輸入／輸出的分別

了解狀態與步驟輸入／輸出之間的分別十分重要：

• 步驟輸入／輸出：資料會依次在步驟之間流動。每個步驟都會以 inputData 接收上一步的輸出，並傳回輸出供下一步使用。
• 狀態：所有步驟都可透過 state 和 setState 讀取及更新的共享儲存空間。狀態會在整個 Workflow 執行期間持續存在，包括暫停／恢復週期。

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ workflowInput: z.string() }),
  outputSchema: z.object({ step1Output: z.string() }),
  stateSchema: z.object({ sharedCounter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // inputData comes from workflow input or previous step's output
    console.log(inputData.workflowInput)

    // state is the shared workflow state
    console.log(state.sharedCounter)

    // Update state for subsequent steps
    await setState({ sharedCounter: state.sharedCounter + 1 })

    // Return output that flows to next step's inputData
    return { step1Output: 'processed' }
  },
})

定義狀態 schema

在 Workflow 和個別步驟上定義 stateSchema。Workflow 的 stateSchema 是包含所有可能狀態值的主 schema，而每個步驟只宣告其所需的子集：

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  stateSchema: z.object({
    processedItems: z.array(z.string()),
  }),
  execute: async ({ inputData, state, setState }) => {
    const { message } = inputData
    const { processedItems } = state

    await setState({
      processedItems: [...processedItems, 'item-1', 'item-2'],
    })

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

const step2 = createStep({
  stateSchema: z.object({
    metadata: z.object({
      processedBy: z.string(),
    }),
  }),
  execute: async ({ inputData, state }) => {
    const { formatted } = inputData
    const { metadata } = state

    return {
      emphasized: `${formatted}!! ${metadata.processedBy}`,
    }
  },
})

export const testWorkflow = createWorkflow({
  stateSchema: z.object({
    processedItems: z.array(z.string()),
    metadata: z.object({
      processedBy: z.string(),
    }),
  }),
})
  .then(step1)
  .then(step2)
  .commit()

設定初始狀態

啟動 Workflow 執行時傳入 initialState，以設定初始值：

const run = await workflow.createRun()

const result = await run.start({
  inputData: { message: 'Hello' },
  initialState: {
    processedItems: [],
    metadata: { processedBy: 'system' },
  },
})

initialState 物件的結構應與 Workflow 的 stateSchema 所定義的結構一致。

狀態在暫停／恢復期間持續存在

狀態會自動在暫停及恢復週期之間持續存在。當 Workflow 暫停後再恢復，暫停前作出的所有狀態更新都會保留：

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({}),
  outputSchema: z.object({}),
  stateSchema: z.object({ count: z.number(), items: z.array(z.string()) }),
  resumeSchema: z.object({ proceed: z.boolean() }),
  execute: async ({ state, setState, suspend, resumeData }) => {
    if (!resumeData) {
      // First run: update state and suspend
      await setState({ count: state.count + 1, items: [...state.items, 'item-1'] })
      await suspend({})
      return {}
    }
    // After resume: state changes are preserved (count: 1, items: ["item-1"])
    return {}
  },
})

嵌套 Workflow 中的狀態

使用嵌套 Workflow 時，狀態會由父 Workflow 傳遞至子 Workflow。父 Workflow 在呼叫嵌套 Workflow 前所作的變更，可供嵌套 Workflow 內的步驟讀取：

src/mastra/workflows/test-workflow.ts

const nestedStep = createStep({
  id: 'nested-step',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
  execute: async ({ state }) => {
    // Receives state modified by parent workflow
    return { result: `Received: ${state.sharedValue}` }
  },
})

const nestedWorkflow = createWorkflow({
  id: 'nested-workflow',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
})
  .then(nestedStep)
  .commit()

const parentStep = createStep({
  id: 'parent-step',
  inputSchema: z.object({}),
  outputSchema: z.object({}),
  stateSchema: z.object({ sharedValue: z.string() }),
  execute: async ({ state, setState }) => {
    // Modify state before nested workflow runs
    await setState({ sharedValue: 'modified-by-parent' })
    return {}
  },
})

const parentWorkflow = createWorkflow({
  id: 'parent-workflow',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
})
  .then(parentStep)
  .then(nestedWorkflow)
  .commit()

相關內容

• Workflows 概覽
• 暫停及恢復
• Step Class
• Workflow Class