Workflow 概覽

Workflow 可讓你透過清楚且結構化的步驟定義複雜的任務序列，而不必依賴單一 Agent 的推理。你可以完整控制任務如何拆分、資料如何在任務之間流動，以及何時執行哪些內容。Workflow 預設使用內建執行引擎，也可以部署至 Inngest 等 Workflow runner，使用代管基礎架構。

給 AI Agent： 執行 npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' 可直接啟動此 Workflow，無須撰寫暫用指令碼。此命令需要執行中的 Mastra 伺服器；請使用 npx mastra dev 啟動本機伺服器，或透過 --url 傳入可連線伺服器的基底 URL。建立不同的輸入前，請先執行 npx mastra api workflow run start --schema。使用 npx skills add mastra-ai/skills --skill mastra 安裝 Mastra Skill，即可取得完整的 API CLI 探索、目標指定、結構描述、驗證及錯誤處理指引。

適合使用 Workflow 的時機

若任務可預先明確定義，且包含多個具特定執行順序的步驟，請使用 Workflow。你可以精細控制資料如何在步驟之間流動與轉換，以及每個階段會呼叫哪些基本元件。

核心原則

Mastra Workflow 依循以下原則運作：

• 使用 createStep 定義步驟，並指定輸入／輸出結構描述與商業邏輯。
• 使用 createWorkflow 組合步驟，以定義執行流程。
• 執行 Workflow 以執行整個序列，並透過內建功能支援暫停、繼續與串流結果。

建立 Workflow 步驟

步驟是 Workflow 的組成元件。使用 createStep() 建立步驟，並以 inputSchema 和 outputSchema 定義它接受與傳回的資料。這兩種結構描述都可以使用 Standard JSON Schema（例如 Zod、Valibot、ArkType）定義。

execute 函式會定義步驟的行為。你可以用它呼叫程式碼庫中的函式、外部 API、Agent 或 Tool。

Zod

src/mastra/workflows/test-workflow.ts

import { 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 {
      formatted: message.toUpperCase(),
    }
  },
})

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

const step1 = createStep({
  id: 'step-1',
  inputSchema: toStandardJsonSchema(
    v.object({
      message: v.string(),
    }),
  ),
  outputSchema: toStandardJsonSchema(
    v.object({
      formatted: v.string(),
    }),
  ),
  execute: async ({ inputData }) => {
    const { message } = inputData

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

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

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

如需完整的設定選項清單，請參閱 Step。

使用 Agent 與 Tool

Workflow 步驟也可以呼叫已註冊的 Agent，或直接匯入並執行 Tool。如需更多資訊，請參閱使用 Tool。

建立 Workflow

使用 createWorkflow() 建立 Workflow，並以 inputSchema 和 outputSchema 定義它接受與傳回的資料。這兩種結構描述都可以使用 Standard JSON Schema（例如 Zod、Valibot、ArkType）定義。使用 .then() 新增步驟，再以 .commit() 完成 Workflow。

Zod

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

Valibot

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

ArkType

src/mastra/workflows/test-workflow.ts

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

如需完整的設定選項清單，請參閱 Workflow 類別。

瞭解控制流程

Workflow 可以使用多種不同方法組合。你選擇的方法會決定每個步驟的結構描述應如何安排。如需更多資訊，請參閱控制流程。

Studio

開啟 Studio，然後從 Workflows 分頁選取一個 Workflow。

• 圖形檢視：中央面板會以視覺化方式呈現 Workflow 的步驟與執行流程。
• 輸入表單：右側邊欄會依據 Workflow 的 inputSchema 產生表單。填寫表單後即可開始執行。
• 即時狀態：執行期間，圖形會即時更新每個步驟的狀態。側邊欄會顯示 Workflow 的輸入、輸出、狀態與記錄。
• 時間回溯：執行完成後，可重新執行個別步驟，以便檢查或重試。

Workflow 狀態

Workflow 狀態可讓你在步驟之間共用值，而不必透過每個步驟的 inputSchema 與 outputSchema 傳遞。你可以使用狀態追蹤進度、累積結果，或在整個 Workflow 中共用設定。

src/mastra/workflows/test-workflow.ts

const 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 狀態。

將 Workflow 作為步驟

將 Workflow 作為步驟，即可在更大型的組合中重複使用其邏輯。輸入與輸出遵循核心原則所述的相同結構描述規則。

src/mastra/workflows/test-workflow.ts

const 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。每個複本都會獨立執行，並在記錄與可觀測性 Tool 中顯示為不同的 Workflow。

src/mastra/workflows/test-workflow.ts

import { 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.ts

import { 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()，而非直接匯入，原因有二：

1. 它可以存取 Mastra 執行個體設定（記錄器、遙測、儲存空間、已註冊的 Agent 與向量儲存區）
2. 它可針對 Workflow 輸入與輸出結構描述提供完整的 TypeScript 型別推斷

請將 Workflow 的註冊鍵（將其加入 Mastra 時使用的鍵）傳給 getWorkflow()。雖然可以使用 getWorkflowById()，依 Workflow 的 id 屬性取得 Workflow，但它無法提供相同程度的型別推斷。

執行 Workflow

Workflow 可使用兩種模式執行：start 會等所有步驟完成後才傳回，而 stream 會在執行期間發出事件。請依使用情境選擇合適的方法：只需要最終結果時使用 start；想監控進度或在步驟完成時觸發動作，則使用 stream。

.start()

使用 createRun() 建立 Workflow 執行個體，再以符合 Workflow inputSchema 的 inputData 呼叫 .start()。Workflow 會執行所有步驟並傳回最終結果。

const run = await testWorkflow.createRun()

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

if (result.status === 'success') {
  console.log(result.result)
}

.stream()

使用 .createRun() 建立 Workflow 執行個體，再以符合 Workflow inputSchema 的 inputData 呼叫 .stream()。逐一處理 fullStream 以追蹤進度，然後等候 result 取得 Workflow 的最終結果。

const run = await testWorkflow.createRun()

const stream = run.stream({
  inputData: {
    message: 'Hello world',
  },
})

for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get the final result (same type as run.start())
const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Workflow 結果型別

run.start() 與 stream.result 都會根據 status 屬性傳回可辨識聯集；其值可以是 success、failed、suspended、tripwire 或 paused。無論狀態為何，你都可以安全地存取 result.status、result.input、result.steps，以及選擇性的 result.state。

此外，可用的屬性會依狀態而異：

狀態	特有屬性	說明
success	result	Workflow 的輸出資料
failed	error	導致失敗的錯誤
tripwire	tripwire	包含 reason、retry?、metadata?、processorId?
suspended	suspendPayload、suspended	暫停資料與已暫停步驟路徑的陣列
paused	（無）	僅提供共用屬性

若要存取特定狀態的屬性，請先檢查 status：

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

if (result.status === 'success') {
  console.log(result.result) // Only available when status is "success"
} else if (result.status === 'failed') {
  console.log(result.error.message)
} else if (result.status === 'suspended') {
  console.log(result.suspendPayload)
}

Workflow 輸出

以下是成功 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!!!"
  }
}

串流

如需一般 writer API 的用法，請參閱串流。

檢查 Workflow 串流承載資料

寫入串流的事件會包含在發出的區塊中。檢查這些區塊，即可存取事件類型、中間值或步驟特定資料等自訂欄位。

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') {
  // if the workflow is suspended, we can resume it with the resumeStream method
  const resumedStream = await run.resumeStream({
    resumeData: { value: 'resume data' },
  })

  for await (const chunk of resumedStream) {
    console.log(chunk)
  }
}

繼續中斷的 Workflow 串流

若 Workflow 串流因任何原因關閉或中斷，可以使用 resumeStream 方法繼續。此方法會傳回新的 ReadableStream，供你觀察 Workflow 事件。

const newStream = await run.resumeStream()

for await (const chunk of newStream) {
  console.log(chunk)
}

使用 Agent 的 Workflow

將 Agent 的 textStream 以管線傳送至 Workflow 步驟的 writer。這會串流部分輸出，而 Mastra 會自動將 Agent 的用量彙總至 Workflow 執行。

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,
    }
  },
})

重新啟動作用中的 Workflow 執行

當 Workflow 執行與伺服器中斷連線時，可以從最後一個作用中步驟重新啟動。這對長時間執行的 Workflow 很有幫助，因為伺服器中斷連線時，Workflow 可能仍在執行。重新啟動 Workflow 執行會從最後一個作用中步驟恢復，並由該處繼續。

使用 restartAllActiveWorkflowRuns() 重新啟動 Workflow 的所有作用中執行

使用 restartAllActiveWorkflowRuns() 重新啟動某個 Workflow 的所有作用中執行。如此便不必手動逐一走訪並重新啟動每次執行。

workflow.restartAllActiveWorkflowRuns()

使用 restart() 重新啟動作用中的 Workflow 執行

使用 restart()，從最後一個作用中步驟重新啟動作用中的 Workflow 執行。執行會從該步驟恢復，Workflow 也會由該處繼續。

const run = await workflow.createRun()

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

const restartedResult = await run.restart()

識別作用中的 Workflow 執行

作用中的 Workflow 執行，其 status 會是 running 或 waiting。你可以檢查 Workflow 的 status 以確認它是否作用中，並使用 active 識別作用中的 Workflow 執行。

const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
  console.log(activeRuns.runs)
}

備註

執行本機 mastra 伺服器時，伺服器啟動後會自動重新啟動所有作用中的 Workflow 執行。

使用 RequestContext

使用 RequestContext 存取要求特定的值。如此即可依要求的情境，有條件地調整行為。

src/mastra/workflows/test-workflow.ts

export 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。

提示

如需具型別安全性的要求情境結構描述驗證，請參閱結構描述驗證。

相關內容

如要深入瞭解 Workflow，請參閱 Workflow 指南，其中會透過實際範例逐步說明核心概念。

• Workflow 狀態
• 控制流程
• 暫停與繼續
• 錯誤處理
• Worker：在專用的背景處理程序中執行 Workflow
• 📹 使用 Mastra 建立 Agentic Workflow 工作坊