跳至主要內容

Workflow 概覽

Workflow 讓你透過清晰且結構化的步驟定義複雜的任務序列,無需依賴單一 Agent 的推理。你可以完全控制如何拆分任務、數據如何在各任務之間流動,以及何時執行哪些任務。Workflow 預設使用內置執行引擎運行,也可以部署至 Inngest 等 Workflow runner,使用託管基礎設施。

何時使用 Workflow
何時使用 Workflow 的直接連結

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

核心原則
核心原則 的直接連結

Mastra Workflow 按照以下原則運作:

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

建立 Workflow 步驟
建立 Workflow 步驟 的直接連結

步驟是 Workflow 的構成單元。使用 createStep() 建立步驟,並以 inputSchema 和 outputSchema 定義其接受及傳回的數據。兩個 schema 均可使用 Standard JSON Schema(Zod、Valibot、ArkType 等)定義。

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

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

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

使用 Agent 及 Tool
使用 Agent 及 Tool 的直接連結

Workflow 步驟亦可呼叫已註冊的 Agent,或直接匯入並執行 Tool。詳情請參閱使用 Tool 頁面。

建立 Workflow
建立 Workflow 的直接連結

使用 createWorkflow() 建立 Workflow,並以 inputSchema 和 outputSchema 定義其接受及傳回的數據。兩個 schema 均可使用 Standard JSON Schema(Zod、Valibot、ArkType 等)定義。使用 .then() 加入步驟,再以 .commit() 完成 Workflow。

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

如需完整的設定選項清單,請參閱 Workflow Class。

了解控制流程
了解控制流程 的直接連結

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

Studio
Studio 的直接連結

開啟 Studio,然後在 Workflows 分頁選擇一個 Workflow。

  • 圖形檢視:中央面板會以圖像呈現 Workflow 的步驟及執行流程。
  • 輸入表格:右側邊欄會根據 Workflow 的 inputSchema 產生表格。填寫後即可開始運行。
  • 即時狀態:執行期間,圖形會即時更新每個步驟的狀態。側邊欄會顯示 Workflow 的輸入、輸出、狀態及日誌。
  • 時間回溯:運行完成後,可重播個別步驟,以檢查或重試。

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 用作步驟 的直接連結

將 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 的直接連結

如要重用 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
註冊 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
參照 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 的直接連結

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

使用 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)
}

Workflow 結果類型
Workflow 結果類型 的直接連結

run.start() 和 stream.result 都會根據 status 屬性傳回可辨識聯合類型,而該屬性的值可以是 success、failed、suspended、tripwire 或 paused。無論狀態為何,你都可以安全地存取 result.status、result.input、result.steps,以及選擇性提供的 result.state。

此外,可用的不同屬性取決於狀態:

狀態專屬屬性說明
successresultWorkflow 的輸出數據
failederror導致失敗的錯誤
tripwiretripwire包含 reason、retry?、metadata?、processorId?
suspendedsuspendPayload, 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 輸出 的直接連結

以下是成功 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
檢查 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 串流 的直接連結

如果 Workflow 串流因任何原因關閉或中斷,你可以使用 resumeStream 方法恢復串流。此方法會傳回新的 ReadableStream,讓你觀察 Workflow 事件。

const newStream = await run.resumeStream()

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

使用 Agent 的 Workflow
使用 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 可能仍在運行。重新啟動 Workflow 運行會從最後一個進行中的步驟恢復執行,並由該處繼續。

使用 restartAllActiveWorkflowRuns() 重新啟動某個 Workflow 的所有進行中運行
restarting-all-active-workflow-runs-of-a-workflow-with-restartallactiveworkflowruns 的直接連結

使用 restartAllActiveWorkflowRuns() 重新啟動某個 Workflow 的所有進行中運行。這樣便無需手動逐一循環處理每次運行並重新啟動。

workflow.restartAllActiveWorkflowRuns()

使用 restart() 重新啟動進行中的 Workflow 運行
restarting-an-active-workflow-run-with-restart 的直接連結

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

const run = await workflow.createRun()

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

const restartedResult = await run.restart()

識別進行中的 Workflow 運行
識別進行中的 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
using-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 指南,當中會透過實際範例逐步講解核心概念。