快照

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

• Workflow 中每個步驟的目前狀態
• 已完成步驟的輸出
• Workflow 採用的執行路徑
• 任何已暫停的步驟及其 metadata
• 每個步驟剩餘的重試次數
• 恢復執行所需的其他上下文資料

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

快照在暫停及恢復中的作用

快照是 Mastra 實現暫停及恢復功能的關鍵機制。當 Workflow 步驟呼叫 await suspend() 時：

1. Workflow 會在該確切位置暫停執行
2. Workflow 的目前狀態會擷取為快照
3. 快照會持久儲存至儲存空間
4. Workflow 步驟會標記為「已暫停」，狀態為 'suspended'
5. 之後，當已暫停的步驟呼叫 resume() 時，系統會擷取快照
6. Workflow 會從中斷的確切位置恢復執行

此機制提供強大的方式來實現 human-in-the-loop Workflow、處理速率限制、等候外部資源，以及實現可能需要長時間暫停的複雜分支 Workflow。

快照結構

每個快照都包含 runId、輸入、步驟狀態（success、suspended 等）、任何暫停及恢復 payload，以及最終輸出。因此，恢復執行時可以取得完整的上下文。

{
  "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 名稱、run 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)

快照的儲存選項

快照會使用 storage 實例持久儲存，而該實例是在 Mastra 類別上設定。此儲存層由註冊至該實例的所有 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（尤其是長時間運行的 Workflow）實施監察，並確保它們能夠正常恢復。
5. 考慮儲存空間擴展能力：如果應用程式有大量已暫停的 Workflow，請確保儲存方案已適當擴展。

自訂快照 metadata

你可以定義 suspendSchema，在暫停 Workflow 時附加自訂 metadata。這些 metadata 會儲存在快照中，並在 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',
    },
  })
}

相關內容

• 控制流程
• 暫停及恢復
• 時間回溯
• Human-in-the-loop