錯誤處理

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 中斷。

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

重試

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

在 Workflow 層級使用 retryConfig

你可以在 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()，以成功結果提前結束。這會將所提供的 payload 作為步驟輸出傳回，並結束 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)
}

相關內容

• 控制流程
• 暫停及恢復
• 時間回溯
• 人在迴路