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

# Workflow 概覽

Workflow 可讓你透過清楚且結構化的步驟定義複雜的任務序列，而不必依賴單一 Agent 的推理。你可以完整控制任務如何拆分、資料如何在任務之間流動，以及何時執行哪些內容。Workflow 預設使用內建執行引擎，也可以部署至 Inngest 等 [Workflow runner](https://mastra.zisheng.pro/zh-TW/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 探索、目標指定、結構描述、驗證及錯誤處理指引。

## 適合使用 Workflow 的時機

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

## 核心原則

Mastra Workflow 依循以下原則運作：

- 使用 [`createStep`](https://mastra.zisheng.pro/zh-TW/reference/workflows/step) 定義**步驟**，並指定輸入／輸出結構描述與商業邏輯。
- 使用 [`createWorkflow`](https://mastra.zisheng.pro/zh-TW/reference/workflows/workflow) 組合**步驟**，以定義執行流程。
- 執行 **Workflow** 以執行整個序列，並透過內建功能支援暫停、繼續與串流結果。

## 建立 Workflow 步驟

步驟是 Workflow 的組成元件。使用 `createStep()` 建立步驟，並以 `inputSchema` 和 `outputSchema` 定義它接受與傳回的資料。這兩種結構描述都可以使用 [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-TW/reference/workflows/step)。

### 使用 Agent 與 Tool

Workflow 步驟也可以呼叫已註冊的 Agent，或直接匯入並執行 Tool。如需更多資訊，請參閱[使用 Tool](https://mastra.zisheng.pro/zh-TW/docs/agents/using-tools)。

## 建立 Workflow

使用 `createWorkflow()` 建立 Workflow，並以 `inputSchema` 和 `outputSchema` 定義它接受與傳回的資料。這兩種結構描述都可以使用 [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 類別](https://mastra.zisheng.pro/zh-TW/reference/workflows/workflow)。

### 瞭解控制流程

Workflow 可以使用多種不同方法組合。你選擇的方法會決定每個步驟的結構描述應如何安排。如需更多資訊，請參閱[控制流程](https://mastra.zisheng.pro/zh-TW/docs/workflows/control-flow)。

## Studio

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

- **圖形檢視**：中央面板會以視覺化方式呈現 Workflow 的步驟與執行流程。
- **輸入表單**：右側邊欄會依據 Workflow 的 `inputSchema` 產生表單。填寫表單後即可開始執行。
- **即時狀態**：執行期間，圖形會即時更新每個步驟的狀態。側邊欄會顯示 Workflow 的輸入、輸出、狀態與記錄。
- [**時間回溯**](https://mastra.zisheng.pro/zh-TW/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() }
  },
})
```

如需狀態結構描述、初始狀態、暫停／繼續期間的持久化，以及巢狀 Workflow 的完整說明，請參閱 [Workflow 狀態](https://mastra.zisheng.pro/zh-TW/docs/workflows/workflow-state)。

## 將 Workflow 作為步驟

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

```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。每個複本都會獨立執行，並在記錄與可觀測性 Tool 中顯示為不同的 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 可以呼叫它，而它也能存取記錄與可觀測性功能等共用資源：

```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。請依據你的設定，在 `mastra` 或 `mastraClient` 執行個體上呼叫 `.getWorkflow()` 以取得參照：

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

```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 執行個體，再以符合 Workflow `inputSchema` 的 `inputData` 呼叫 `.stream()`。逐一處理 `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-TW/guides/concepts/streaming)。

### 檢查 Workflow 串流承載資料

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

```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-TW/docs/server/request-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-TW/docs/server/request-context)。

> **提示:** 如需具型別安全性的要求情境結構描述驗證，請參閱[結構描述驗證](https://mastra.zisheng.pro/zh-TW/docs/server/request-context)。

## 相關內容

如要深入瞭解 Workflow，請參閱 [Workflow 指南](https://mastra.zisheng.pro/zh-TW/guides/guide/ai-recruiter)，其中會透過實際範例逐步說明核心概念。

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