Snapshots

在 Mastra 中，快照是工作流在特定时间点完整执行状态的可序列化表示。快照捕获从上次中断处精确恢复工作流所需的所有信息，包括：

• 工作流中每个步骤的当前状态
• 已完成步骤的输出
• 工作流所经过的执行路径
• 任何挂起的步骤及其元数据
• 每个步骤剩余的 retry 次数
• 恢复执行所需的额外上下文数据

每当工作流挂起时，Mastra 都会自动创建和管理快照，并将其持久化到配置的存储系统。

快照在挂起和恢复中的作用

快照是实现 Mastra 挂起和恢复能力的关键机制。当工作流步骤调用 await suspend() 时：

1. 工作流执行在该精确位置暂停
2. 工作流的当前状态被捕获为快照
3. 快照被持久化到存储
4. 工作流步骤被标记为"suspended"，状态为 'suspended'
5. 之后，当对挂起的步骤调用 resume() 时，快照会被检索
6. 工作流执行从上次中断处精确恢复

该机制提供了一种强大的方式来实现 human-in-the-loop 工作流、处理速率限制、等待外部资源，以及实现可能需要长时间暂停的复杂分支工作流。

快照结构

每个快照包含 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 表中，并通过工作流的 runId 标识。

了解更多：

• libSQL 存储
• Upstash 存储
• PostgreSQL 存储
• OracleDB 存储

保存快照

当工作流挂起时，Mastra 会自动持久化工作流快照，步骤如下：

1. 步骤执行中的 suspend() 函数触发快照流程
2. WorkflowInstance.suspend() 方法记录挂起的机器
3. 调用 persistWorkflowSnapshot() 保存当前状态
4. 快照被序列化并存储在配置数据库的 workflow_snapshots 表中
5. 存储记录包含工作流名称、run ID 和序列化后的快照

检索快照

当工作流恢复时，Mastra 会检索已持久化的快照，步骤如下：

1. 使用特定步骤 ID 调用 resume() 方法
2. 通过 loadWorkflowSnapshot() 从存储加载快照
3. 快照被解析并准备用于恢复
4. 使用快照状态重建工作流执行
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 实例持久化。该存储层由注册到该实例的所有工作流共享。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
• 更多存储提供商

最佳实践

1. 确保可序列化：需要包含在快照中的任何数据都必须可序列化（可转换为 JSON）。
2. 尽量减小快照大小：避免将大型数据对象直接存储在工作流上下文中。取而代之，存储对它们的引用（如 ID），并在需要时检索数据。
3. 谨慎处理恢复上下文：恢复工作流时，仔细考虑要提供什么上下文。它将与现有快照数据合并。
4. 设置适当的监控：对挂起的工作流（尤其是长时间运行的）实施监控，并确保它们被正确恢复。
5. 考虑存储扩展性：对于有许多挂起工作流的应用，确保存储解决方案具有适当的扩展能力。

自定义快照元数据

你可以通过定义 suspendSchema 在挂起工作流时附加自定义元数据。该元数据存储在快照中，并在工作流恢复时可用。

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