錯誤處理

Mastra Workflow 支援透過執行後的結果狀態檢查、針對暫時性失敗的重試原則，以及用於集中記錄錯誤或發出警示的生命週期回呼來處理錯誤。

處理 Workflow 結果

執行 Workflow 時，結果物件會包含狀態與發生的所有錯誤。

檢查結果狀態

src/run-workflow.ts

import { mastra } from './mastra'

const workflow = mastra.getWorkflow('myWorkflow')
const run = await workflow.createRun()
const result = await run.start({ inputData: { value: 'test' } })

switch (result.status) {
  case 'success':
    console.log('Workflow completed:', result.result)
    break
  case 'failed':
    console.error('Workflow failed:', result.error)
    break
  case 'suspended':
    console.log('Workflow suspended, waiting for resume')
    break
}

結果物件結構

結果物件包含：

• status - Workflow 狀態：'success'、'failed'、'suspended' 或 'tripwire'
• result - Workflow 輸出（狀態為 'success' 時）
• error - 錯誤詳細資料（狀態為 'failed' 時）
• steps - 個別步驟的結果，包含其狀態與輸出

存取步驟結果

你可以檢查個別步驟結果，以了解失敗發生的位置：

src/run-workflow.ts

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

if (result.status === 'failed') {
  // Find which step failed
  for (const [stepId, stepResult] of Object.entries(result.steps)) {
    if (stepResult.status === 'failed') {
      console.error(`Step ${stepId} failed:`, stepResult.error)
    }
  }
}

生命週期回呼

若需要在不等待結果的情況下處理 Workflow 完成事件，例如背景工作、啟動後不等待的 Workflow 或集中式記錄，可使用生命週期回呼。

onFinish

Workflow 以任何狀態（成功、失敗、暫停或 tripwire）完成時呼叫：

src/mastra/workflows/order-workflow.ts

import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const orderWorkflow = createWorkflow({
  id: 'order-processing',
  inputSchema: z.object({ orderId: z.string() }),
  outputSchema: z.object({ orderId: z.string(), status: z.string() }),
  options: {
    onFinish: async result => {
      if (result.status === 'success') {
        await db.updateOrderStatus(result.result.orderId, result.status)
      }
      await analytics.track('workflow_completed', {
        workflowId: 'order-processing',
        status: result.status,
      })
    },
  },
})

onFinish 回呼會收到：

• status - Workflow 狀態
• result - Workflow 輸出（成功時）
• error - 錯誤詳細資料（失敗時）
• steps - 個別步驟結果
• tripwire - Tripwire 資訊（狀態為 'tripwire' 時）
• runId - 此次 Workflow 執行的唯一識別碼
• workflowId - Workflow 的識別碼
• resourceId - 選用的資源識別碼（建立執行時提供）
• getInitData<any>() - 回傳初始輸入資料的函式
• mastra - Mastra 個體（若 Workflow 已註冊至 Mastra）
• requestContext - 要求範圍的情境資料
• logger - Workflow 的 logger 個體
• state - Workflow 的目前狀態物件

onError

僅在 Workflow 失敗（狀態為 'failed' 或 'tripwire'）時呼叫：

src/mastra/workflows/payment-workflow.ts

import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const paymentWorkflow = createWorkflow({
  id: 'payment-processing',
  inputSchema: z.object({ amount: z.number() }),
  outputSchema: z.object({ transactionId: z.string() }),
  options: {
    onError: async errorInfo => {
      await alertService.notify({
        channel: 'payments-alerts',
        message: `Payment workflow failed: ${errorInfo.error?.message}`,
      })
      await errorTracker.capture(errorInfo.error)
    },
  },
})

onError 回呼會收到：

• status - 'failed' 或 'tripwire'
• error - 錯誤詳細資料
• steps - 個別步驟結果
• tripwire - Tripwire 資訊（狀態為 'tripwire' 時）
• runId - 此次 Workflow 執行的唯一識別碼
• workflowId - Workflow 的識別碼
• resourceId - 選用的資源識別碼（建立執行時提供）
• getInitData<any>() - 回傳初始輸入資料的函式
• mastra - Mastra 個體（若 Workflow 已註冊至 Mastra）
• requestContext - 要求範圍的情境資料
• logger - Workflow 的 logger 個體
• state - Workflow 的目前狀態物件

同時使用兩個回呼

你可以同時使用兩個回呼：

src/mastra/workflows/pipeline-workflow.ts

import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const pipelineWorkflow = createWorkflow({
  id: 'data-pipeline',
  inputSchema: z.object({ source: z.string() }),
  outputSchema: z.object({ recordsProcessed: z.number() }),
  options: {
    onFinish: async result => {
      // Always log completion
      await logger.info('Pipeline completed', { status: result.status })
    },
    onError: async errorInfo => {
      // Alert on failures
      await pagerDuty.alert('Data pipeline failed', errorInfo.error)
    },
  },
})

回呼中的錯誤處理

回呼內擲回的錯誤會被攔截並記錄，不會影響 Workflow 結果或導致 Workflow 失敗。因此，回呼問題不會中斷正式環境中的 Workflow。

options: {
  onFinish: async (result) => {
    // If this throws, it's logged but the workflow result is unchanged
    await externalService.notify(result);
  },
}

重試

Mastra 為因暫時性錯誤而失敗的 Workflow 或步驟提供重試機制，例如步驟與可能暫時無法使用的外部服務或資源互動時。

使用 retryConfig 設定 Workflow 層級重試

你可以在 Workflow 層級設定重試，此設定會套用至 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({
  retryConfig: {
    attempts: 5,
    delay: 2000
  }
})
  .then(step1)
  .commit();

使用 retries 設定步驟層級重試

你可以使用 retries 屬性設定個別步驟的重試。這會覆寫該特定步驟的 Workflow 層級重試設定：

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import { z } from 'zod'

const step1 = createStep({
  execute: async () => {
    const response = await fetch('example-url')

    if (!response.ok) {
      throw new Error('Error')
    }

    return {
      value: '',
    }
  },
  retries: 3,
})

條件分支

你可以使用條件邏輯，根據先前步驟成功或失敗的結果建立替代 Workflow 路徑：

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { z } from "zod";

const step1 = createStep({
  execute: async () => {
    try {
      const response = await fetch('example-url');

      if (!response.ok) {
        throw new Error('error');
      }

      return {
        status: "ok"
      };
    } catch (error) {
      return {
        status: "error"
      };
    }
  }
});

const step2 = createStep({...});
const fallback = createStep({...});

export const testWorkflow = createWorkflow({})
  .then(step1)
  .branch([
    [async ({ inputData: { status } }) => status === "ok", step2],
    [async ({ inputData: { status } }) => status === "error", fallback]
  ])
  .commit();

檢查先前步驟結果

使用 getStepResult() 檢查先前步驟的結果。

src/mastra/workflows/test-workflow.ts

import { createStep } from "@mastra/core/workflows";
import { z } from "zod";

const step1 = createStep({...});

const step2 = createStep({
  execute: async ({ getStepResult }) => {
    const step1Result = getStepResult(step1);

    return {
      value: ""
    };
  }
});

使用 bail() 提前結束

在步驟中使用 bail()，以成功結果提前結束。這會將提供的承載資料作為步驟輸出回傳，並結束 Workflow 執行。

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { z } from "zod";

const step1 = createStep({
  id: 'step1',
  execute: async ({ bail }) => {
    return bail({ result: 'bailed' });
  },
  inputSchema: z.object({ value: z.string() }),
  outputSchema: z.object({ result: z.string() }),
});

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .commit();

使用 Error() 提前結束

在步驟中使用 throw new Error()，以錯誤結束執行。

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { z } from "zod";

const step1 = createStep({
  id: 'step1',
  execute: async () => {
    throw new Error('error');
  },
  inputSchema: z.object({ value: z.string() }),
  outputSchema: z.object({ result: z.string() }),
});

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .commit();

## 使用 stream() 監控錯誤

你可以使用 stream 監控 Workflow 錯誤：

src/test-workflow.ts

import { mastra } from '../src/mastra'

const workflow = mastra.getWorkflow('testWorkflow')

const run = await workflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream.stream) {
  console.log(chunk.payload.output.stats)
}

相關資源

• 控制流程
• 暫停與繼續
• 時間旅行
• 人機協作