> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-HK/llms.txt

# Workflow 概覽

Workflow 讓你透過清晰且結構化的步驟定義複雜的任務序列，無需依賴單一 Agent 的推理。你可以完全控制如何拆分任務、數據如何在各任務之間流動，以及何時執行哪些任務。Workflow 預設使用內置執行引擎運行，也可以部署至 Inngest 等 [Workflow runner](https://mastra.zisheng.pro/zh-HK/docs/deployment/workflow-runners)，使用託管基礎設施。

**給 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`](https://mastra.zisheng.pro/zh-HK/reference/workflows/step) 定義**步驟**，並指定輸入／輸出 schema 及業務邏輯。
- 使用 [`createWorkflow`](https://mastra.zisheng.pro/zh-HK/reference/workflows/workflow) 組合**步驟**，以定義執行流程。
- 運行 **Workflow** 以執行整個序列，並內置支援暫停、恢復及串流結果。

## 建立 Workflow 步驟

步驟是 Workflow 的構成單元。使用 `createStep()` 建立步驟，並以 `inputSchema` 和 `outputSchema` 定義其接受及傳回的數據。兩個 schema 均可使用 [Standard JSON Schema](https://standardschema.dev/json-schema)（[Zod](https://zod.dev/)、[Valibot](https://valibot.dev/)、[ArkType](https://arktype.io/) 等）定義。

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

**Zod**:

```typescript
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**:

```typescript
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**:

```typescript
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`](https://mastra.zisheng.pro/zh-HK/reference/workflows/step)。

### 使用 Agent 及 Tool

Workflow 步驟亦可呼叫已註冊的 Agent，或直接匯入並執行 Tool。詳情請參閱[使用 Tool](https://mastra.zisheng.pro/zh-HK/docs/agents/using-tools) 頁面。

## 建立 Workflow

使用 `createWorkflow()` 建立 Workflow，並以 `inputSchema` 和 `outputSchema` 定義其接受及傳回的數據。兩個 schema 均可使用 [Standard JSON Schema](https://standardschema.dev/json-schema)（[Zod](https://zod.dev/)、[Valibot](https://valibot.dev/)、[ArkType](https://arktype.io/) 等）定義。使用 `.then()` 加入步驟，再以 `.commit()` 完成 Workflow。

**Zod**:

```typescript
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**:

```typescript
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**:

```typescript
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](https://mastra.zisheng.pro/zh-HK/reference/workflows/workflow)。

### 了解控制流程

Workflow 可透過多種不同方法組合。你選擇的方法會決定每個步驟的 schema 應如何組織。詳情請參閱[控制流程](https://mastra.zisheng.pro/zh-HK/docs/workflows/control-flow)頁面。

## Studio

開啟 [Studio](https://mastra.zisheng.pro/zh-HK/docs/studio/overview)，然後在 **Workflows** 分頁選擇一個 Workflow。

- **圖形檢視**：中央面板會以圖像呈現 Workflow 的步驟及執行流程。
- **輸入表格**：右側邊欄會根據 Workflow 的 `inputSchema` 產生表格。填寫後即可開始運行。
- **即時狀態**：執行期間，圖形會即時更新每個步驟的狀態。側邊欄會顯示 Workflow 的輸入、輸出、狀態及日誌。
- [**時間回溯**](https://mastra.zisheng.pro/zh-HK/docs/workflows/time-travel)：運行完成後，可重播個別步驟，以檢查或重試。

## Workflow 狀態

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

```typescript
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 狀態](https://mastra.zisheng.pro/zh-HK/docs/workflows/workflow-state)。

## 將 Workflow 用作步驟

將 Workflow 用作步驟，即可在較大型的組合中重用其邏輯。輸入及輸出遵循[核心原則](https://mastra.zisheng.pro/zh-HK/docs/workflows/control-flow)所述的相同 schema 規則。

```typescript
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。

```typescript
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，而它亦可存取日誌及可觀測性功能等共用資源：

```typescript
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` 實例上取得參照：

```typescript
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 會執行所有步驟，並傳回最終結果。

```typescript
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 結果。

```typescript
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`：

```typescript
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` 屬性：

```json
{
  "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 的用法，請參閱[串流](https://mastra.zisheng.pro/zh-HK/guides/concepts/streaming)。

### 檢查 Workflow 串流 payload

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

```typescript
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 事件。

```typescript
const newStream = await run.resumeStream()

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

### 使用 Agent 的 Workflow

將 Agent 的 `textStream` 以管道傳送至 Workflow 步驟的 `writer`。這會串流部分輸出，而 Mastra 會自動將 Agent 的用量彙集至該次 Workflow 運行。

```typescript
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 的所有進行中運行。這樣便無需手動逐一循環處理每次運行並重新啟動。

```typescript
workflow.restartAllActiveWorkflowRuns()
```

### 使用 `restart()` 重新啟動進行中的 Workflow 運行

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

```typescript
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 運行。

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

> **備註:** 運行本機 mastra 伺服器時，伺服器啟動後會自動重新啟動所有進行中的 Workflow 運行。

## 使用 `RequestContext`

使用 [RequestContext](https://mastra.zisheng.pro/zh-HK/docs/server/request-context) 存取特定請求的值。你可以由此根據請求的 context，有條件地調整行為。

```typescript
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](https://mastra.zisheng.pro/zh-HK/docs/server/request-context)。

> **提示:** 如需類型安全的請求 context schema 驗證，請參閱 [Schema Validation](https://mastra.zisheng.pro/zh-HK/docs/server/request-context)。

## 相關內容

如要深入了解 Workflow，請參閱我們的 [Workflow 指南](https://mastra.zisheng.pro/zh-HK/guides/guide/ai-recruiter)，當中會透過實際範例逐步講解核心概念。

- [Workflow 狀態](https://mastra.zisheng.pro/zh-HK/docs/workflows/workflow-state)
- [控制流程](https://mastra.zisheng.pro/zh-HK/docs/workflows/control-flow)
- [暫停及恢復](https://mastra.zisheng.pro/zh-HK/docs/workflows/suspend-and-resume)
- [錯誤處理](https://mastra.zisheng.pro/zh-HK/docs/workflows/error-handling)
- [Worker](https://mastra.zisheng.pro/zh-HK/docs/deployment/workers)：在專用背景程序中執行 Workflow
- 📹 [Mastra Agentic Workflow 工作坊](https://www.youtube.com/watch?v=HGt8pVPpX9g)