스냅샷

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 스토리지
• 업스태시 스토리지
• 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 스토리지
• 몽고DB 스토리지
• 업스태시 스토리지
• 클라우드플레어 D1
• DynamoDB
• 더 많은 스토리지 Provider

모범 사례

1. 직렬화 가능성 보장: 스냅샷에 포함되어야 하는 모든 데이터는 직렬화 가능해야 합니다(JSON으로 변환 가능).
2. 스냅샷 크기 최소화: Workflow 컨텍스트에 직접 대규모 데이터 개체를 저장하지 마십시오. 대신, 이에 대한 참조(예: ID)를 저장하고 필요할 때 데이터를 검색하세요.
3. 이력서 컨텍스트를 신중하게 처리하세요.: Workflow를 재개할 때 제공할 컨텍스트를 신중하게 고려하세요. 이는 기존 스냅샷 데이터와 병합됩니다.
4. 적절한 모니터링 설정: 일시 중단된 Workflow, 특히 장기 실행 Workflow에 대한 모니터링을 구현하고 제대로 재개되는지 확인합니다.
5. 스토리지 확장 고려: 일시 중단된 Workflow가 많은 애플리케이션의 경우 스토리지 솔루션의 크기가 적절하게 조정되었는지 확인하세요.

커스텀 스냅샷 메타데이터

Workflow를 일시 중지할 때 suspendSchema를 정의해 사용자 정의 메타데이터를 첨부할 수 있습니다. 이 메타데이터는 스냅샷에 저장되며 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',
    },
  })
}

관련된

• 제어 흐름
• 일시중단 및 재개
• 시간여행
• 인간 참여형