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 探索、目標指定、schema、驗證及錯誤處理指引。

何時使用 Workflow

Workflow 適合用於事先已有明確定義、包含多個步驟並有特定執行次序的任務。你可以細緻控制數據如何在步驟之間流動及轉換，以及每個階段會呼叫哪些基本操作。

核心原則

Mastra Workflow 按照以下原則運作：

• 使用 createStep 定義步驟，並指定輸入／輸出 schema 及業務邏輯。
• 使用 createWorkflow 組合步驟，以定義執行流程。
• 運行 Workflow 以執行整個序列，並內置支援暫停、恢復及串流結果。

建立 Workflow 步驟

步驟是 Workflow 的構成單元。使用 createStep() 建立步驟，並以 inputSchema 和 outputSchema 定義其接受及傳回的數據。兩個 schema 均可使用 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 定義其接受及傳回的數據。兩個 schema 均可使用 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 Class。

了解控制流程

Workflow 可透過多種不同方法組合。你選擇的方法會決定每個步驟的 schema 應如何組織。詳情請參閱控制流程頁面。

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

如需有關狀態 schema、初始狀態、暫停／恢復時的持久化，以及巢狀 Workflow 的完整文件，請參閱 Workflow 狀態。

將 Workflow 用作步驟

將 Workflow 用作步驟，即可在較大型的組合中重用其邏輯。輸入及輸出遵循核心原則所述的相同 schema 規則。

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。每個複製項目均獨立運行，並會在日誌及可觀測性工具中顯示為不同的 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 即可呼叫此 Workflow，而它亦可存取日誌及可觀測性功能等共用資源：

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。呼叫 .getWorkflow()，並視乎你的設定在 mastra 或 mastraClient 實例上取得參照：

const testWorkflow = mastra.getWorkflow('testWorkflow')

資訊

建議使用 mastra.getWorkflow() 而非直接匯入，原因有二：

1. 它可讓你存取 Mastra 實例設定（logger、telemetry、storage、已註冊的 Agent 及 vector store）
2. 它為 Workflow 的輸入及輸出 schema 提供完整的 TypeScript 類型推斷

請以 Workflow 的註冊 key（將其加入 Mastra 時使用的 key）呼叫 getWorkflow()。雖然你可以使用 getWorkflowById()，按 Workflow 的 id 屬性擷取 Workflow，但它無法提供同等程度的類型推斷。

運行 Workflow

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

.start()

使用 createRun() 建立 Workflow 運行實例，然後呼叫 .start() 並傳入 inputData；該數據須符合 Workflow 的 inputSchema。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 運行實例，然後呼叫 .stream() 並傳入 inputData；該數據須符合 Workflow 的 inputSchema。逐一處理 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 串流 payload

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

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 存取特定請求的值。你可以由此根據請求的 context，有條件地調整行為。

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。

提示

如需類型安全的請求 context schema 驗證，請參閱 Schema Validation。

相關內容

如要深入了解 Workflow，請參閱我們的 Workflow 指南，當中會透過實際範例逐步講解核心概念。

• Workflow 狀態
• 控制流程
• 暫停及恢復
• 錯誤處理
• Worker：在專用背景程序中執行 Workflow
• 📹 Mastra Agentic Workflow 工作坊