快照

在 Mastra 中，快照是 Workflow 在特定時間點完整執行狀態的可序列化表示。快照會擷取從中斷處精確繼續 Workflow 所需的所有資訊，包括：

• Workflow 中每個步驟的目前狀態
• 已完成步驟的輸出
• Workflow 採用的執行路徑
• 所有暫停的步驟及其中繼資料
• 每個步驟剩餘的重試次數
• 繼續執行所需的額外情境資料

每當 Workflow 暫停時，Mastra 都會自動建立並管理快照，且將其持久保存至設定的儲存系統。

快照在暫停與繼續中的作用

快照是實現 Mastra 暫停與繼續功能的關鍵機制。Workflow 步驟呼叫 await suspend() 時：

1. Workflow 執行會在該時間點精確暫停
2. Workflow 的目前狀態會擷取為快照
3. 快照會持久保存至儲存空間
4. Workflow 步驟會標記為「已暫停」，狀態為 'suspended'
5. 稍後對暫停的步驟呼叫 resume() 時，系統會擷取快照
6. Workflow 會從中斷處精確地繼續執行

此機制提供強大的方式，可實作人機協作 Workflow、處理速率限制、等待外部資源，以及實作可能需要長時間暫停的複雜分支 Workflow。

快照結構

每個快照都包含 runId、輸入、步驟狀態（success、suspended 等）、所有暫停與繼續承載資料，以及最終輸出。因此，繼續執行時可取得完整情境。

{
  "runId": "34904c14-e79e-4a12-9804-9655d4616c50",
  "status": "success",
  "value": {},
  "context": {
    "input": {
      "value": 100,
      "user": "Michael",
      "requiredApprovers": ["manager", "finance"]
    },
    "approval-step": {
      "payload": {
        "value": 100,
        "user": "Michael",
        "requiredApprovers": ["manager", "finance"]
      },
      "startedAt": 1758027577955,
      "status": "success",
      "suspendPayload": {
        "message": "Workflow suspended",
        "requestedBy": "Michael",
        "approvers": ["manager", "finance"]
      },
      "suspendedAt": 1758027578065,
      "resumePayload": { "confirm": true, "approver": "manager" },
      "resumedAt": 1758027578517,
      "output": { "value": 100, "approved": true },
      "endedAt": 1758027578634
    }
  },
  "activePaths": [],
  "serializedStepGraph": [
    {
      "type": "step",
      "step": {
        "id": "approval-step",
        "description": "Accepts a value, waits for confirmation"
      }
    }
  ],
  "suspendedPaths": {},
  "waitingPaths": {},
  "result": { "value": 100, "approved": true },
  "requestContext": {},
  "timestamp": 1758027578740
}

如何儲存與擷取快照

快照會儲存至設定的儲存系統。預設使用 libSQL，但也可改為設定 Upstash、PostgreSQL 或 OracleDB。每個快照都儲存在 workflow_snapshots 資料表中，並以 Workflow 的 runId 識別。

進一步了解：

• libSQL 儲存空間
• Upstash 儲存空間
• PostgreSQL 儲存空間
• OracleDB 儲存空間

儲存快照

Workflow 暫停時，Mastra 會透過下列步驟自動持久保存 Workflow 快照：

1. 步驟執行中的 suspend() 函式會觸發快照流程
2. WorkflowInstance.suspend() 方法會記錄暫停的狀態機
3. 呼叫 persistWorkflowSnapshot() 儲存目前狀態
4. 快照經過序列化後，會儲存至所設定資料庫的 workflow_snapshots 資料表
5. 儲存記錄包含 Workflow 名稱、執行 ID 與序列化快照

擷取快照

Workflow 繼續時，Mastra 會透過下列步驟擷取持久保存的快照：

1. 使用特定步驟 ID 呼叫 resume() 方法
2. 使用 loadWorkflowSnapshot() 從儲存空間載入快照
3. 解析快照並準備繼續執行
4. 使用快照狀態重新建立 Workflow 執行
5. 繼續暫停的步驟及其後續執行

const storage = mastra.getStorage()
const workflowStore = await storage?.getStore('workflows')

const snapshot = await workflowStore?.loadWorkflowSnapshot({
  runId: '<run-id>',
  workflowName: '<workflow-id>',
})

console.log(snapshot)

快照的儲存選項

快照會使用在 Mastra 類別上設定的 storage 個體持久保存。註冊至該個體的所有 Workflow 會共用此儲存層。Mastra 支援多種儲存選項，以因應不同環境的需求。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { LibSQLStore } from '@mastra/libsql'
import { approvalWorkflow } from './workflows'

export const mastra = new Mastra({
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: ':memory:',
  }),
  workflows: { approvalWorkflow },
})

• libSQL 儲存空間
• PostgreSQL 儲存空間
• OracleDB 儲存空間
• MongoDB 儲存空間
• Upstash 儲存空間
• Cloudflare D1
• DynamoDB
• 更多儲存 Provider

最佳實務

1. 確保可序列化：需要納入快照的任何資料都必須可序列化（可轉換為 JSON）。
2. 縮減快照大小：避免直接在 Workflow 情境中儲存大型資料物件。請改為儲存其參照（例如 ID），並在需要時擷取資料。
3. 謹慎處理繼續情境：繼續 Workflow 時，請仔細考量要提供的情境。該情境會與現有快照資料合併。
4. 設定適當監控：對暫停的 Workflow（尤其是長時間執行者）實作監控，並確保它們能正確繼續。
5. 考量儲存空間擴充：對於有大量暫停 Workflow 的應用程式，請確保儲存解決方案已適當擴充。

自訂快照中繼資料

你可以定義 suspendSchema，在暫停 Workflow 時附加自訂中繼資料。此中繼資料會儲存在快照中，並可在 Workflow 繼續時使用。

src/mastra/workflows/test-workflow.ts

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

const approvalStep = createStep({
  id: 'approval-step',
  description: 'Accepts a value, waits for confirmation',
  inputSchema: z.object({
    value: z.number(),
    user: z.string(),
    requiredApprovers: z.array(z.string()),
  }),
  suspendSchema: z.object({
    message: z.string(),
    requestedBy: z.string(),
    approvers: z.array(z.string()),
  }),
  resumeSchema: z.object({
    confirm: z.boolean(),
    approver: z.string(),
  }),
  outputSchema: z.object({
    value: z.number(),
    approved: z.boolean(),
  }),
  execute: async ({ inputData, resumeData, suspend }) => {
    const { value, user, requiredApprovers } = inputData
    const { confirm } = resumeData ?? {}

    if (!confirm) {
      return await suspend({
        message: 'Workflow suspended',
        requestedBy: user,
        approvers: [...requiredApprovers],
      })
    }

    return {
      value,
      approved: confirm,
    }
  },
})

提供繼續資料

繼續暫停的步驟時，請使用 resumeData 傳入結構化輸入。該輸入必須符合步驟的 resumeSchema。

const workflow = mastra.getWorkflow('approvalWorkflow')

const run = await workflow.createRun()

const result = await run.start({
  inputData: {
    value: 100,
    user: 'Michael',
    requiredApprovers: ['manager', 'finance'],
  },
})

if (result.status === 'suspended') {
  const resumedResult = await run.resume({
    step: 'approval-step',
    resumeData: {
      confirm: true,
      approver: 'manager',
    },
  })
}

相關資源

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