일시중단 및 재개

추가 데이터를 수집하거나, API 콜백을 기다리거나, 비용이 많이 드는 작업을 제한하거나, 사람 개입형 입력을 요청하기 위해 어느 단계에서든 Workflow를 일시 중지할 수 있습니다. Workflow가 일시 중지되면 현재 실행 상태가 스냅샷으로 저장됩니다. 나중에 특정 단계 ID에서 Workflow를 재개하여 해당 스냅샷에 캡처된 정확한 상태를 복원할 수 있습니다. 스냅샷은 구성된 스토리지 Provider에 저장되며 배포 및 애플리케이션 재시작 후에도 유지됩니다.

suspend()를 사용하여 Workflow 일시 중지

특정 단계에서 Workflow 실행을 일시 중지하려면 suspend()를 사용하세요. resumeData의 값을 사용해 단계의 execute 블록에 일시 중지 조건을 정의할 수 있습니다.

• 조건이 충족되지 않으면 Workflow가 일시 중지되고 반환됩니다.suspend().
• 조건이 충족되면 Workflow는 단계의 나머지 논리로 계속됩니다.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    userEmail: z.string(),
  }),
  outputSchema: z.object({
    output: z.string(),
  }),
  resumeSchema: z.object({
    approved: z.boolean(),
  }),
  execute: async ({ inputData, resumeData, suspend }) => {
    const { userEmail } = inputData
    const { approved } = resumeData ?? {}

    if (!approved) {
      return await suspend({})
    }

    return {
      output: `Email sent to ${userEmail}`,
    }
  },
})

export const testWorkflow = createWorkflow({
  id: 'test-workflow',
  inputSchema: z.object({
    userEmail: z.string(),
  }),
  outputSchema: z.object({
    output: z.string(),
  }),
})
  .then(step1)
  .commit()

다음을 사용하여 Workflow 다시 시작resume()

일시 중지된 Workflow를 중단된 단계부터 다시 시작하려면 resume()을 사용하세요. 단계의 resumeSchema와 일치하는 resumeData를 전달해 일시 중지 조건을 충족하고 실행을 계속하세요.

import { step1 } from './workflows/test-workflow'

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun()

await run.start({
  inputData: {
    userEmail: 'alex@example.com',
  },
})

const handleResume = async () => {
  const result = await run.resume({
    step: step1,
    resumeData: { approved: true },
  })
}

step 객체를 전달하면 resumeData에 완전한 타입 안전성이 제공됩니다. 또는 ID가 사용자 입력이나 데이터베이스에서 제공되는 경우 더 유연하게 단계 ID를 전달할 수 있습니다.

const result = await run.resume({
  step: 'step-1',
  resumeData: { approved: true },
})

한 단계만 일시 중지된 경우 단계 인수를 완전히 생략할 수 있으며 Mastra는 Workflow에서 마지막 일시 중지된 단계를 재개합니다.

runId만으로 재개할 때는 먼저 createRun()으로 실행 인스턴스를 만드세요.

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun({ runId: '123' })

const stream = run.resume({
  resumeData: { approved: true },
})

HTTP 엔드포인트, 이벤트 핸들러, 사람의 입력에 대한 응답 또는 타이머 등 애플리케이션 어디에서나 resume()을 호출할 수 있습니다.

const midnight = new Date()
midnight.setUTCHours(24, 0, 0, 0)

setTimeout(async () => {
  await run.resume({
    step: 'step-1',
    resumeData: { approved: true },
  })
}, midnight.getTime() - Date.now())

다음을 사용하여 일시중단 데이터에 액세스suspendData

단계가 일시 중지되면 나중에 해당 단계가 재개될 때 suspend()에 제공했던 데이터에 접근할 수 있습니다. 이 데이터에 접근하려면 단계의 execute 함수에서 suspendData 매개변수를 사용하세요.

src/mastra/workflows/user-approval.ts

const approvalStep = createStep({
  id: 'user-approval',
  inputSchema: z.object({
    requestId: z.string(),
  }),
  resumeSchema: z.object({
    approved: z.boolean(),
  }),
  suspendSchema: z.object({
    reason: z.string(),
    requestDetails: z.string(),
  }),
  outputSchema: z.object({
    result: z.string(),
  }),
  execute: async ({ inputData, resumeData, suspend, suspendData }) => {
    const { requestId } = inputData
    const { approved } = resumeData ?? {}

    // On first execution, suspend with context
    if (!approved) {
      return await suspend({
        reason: 'User approval required',
        requestDetails: `Request ${requestId} pending review`,
      })
    }

    // On resume, access the original suspend data
    const suspendReason = suspendData?.reason || 'Unknown'
    const details = suspendData?.requestDetails || 'No details'

    return {
      result: `${details} - ${suspendReason} - Decision: ${approved ? 'Approved' : 'Rejected'}`,
    }
  },
})

단계가 재개되면 suspendData 매개변수가 자동으로 채워지며, 최초 일시 중지 시 suspend() 함수에 전달했던 정확한 데이터가 포함됩니다. Workflow가 일시 중지된 이유에 관한 컨텍스트를 유지하고 재개 과정에서 해당 정보를 활용할 수 있습니다.

일시 중지된 실행 식별

Workflow가 일시 중지되면 일시 중지된 단계부터 다시 시작됩니다. Workflow의 status를 확인해 일시 중지 상태인지 검증하고, suspended를 사용해 일시 중지된 단계 또는 중첩 Workflow를 식별할 수 있습니다.

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun()

const result = await run.start({
  inputData: {
    userEmail: 'alex@example.com',
  },
})

if (result.status === 'suspended') {
  console.log(result.suspended[0])
  await run.resume({
    step: result.suspended[0],
    resumeData: { approved: true },
  })
}

예제 출력

suspended 배열에는 해당 실행에서 일시 중지된 모든 Workflow와 단계의 ID가 포함됩니다. resume() 호출 시 이 값을 step 매개변수에 전달하면 일시 중지된 실행 경로를 지정해 재개할 수 있습니다.

['nested-workflow', 'step-1']

일시 중단된 실행 복구

애플리케이션에서 스토리지의 일시 중지된 실행을 복구해야 하면 createWorkflowStateReader()와 함께 workflow.getWorkflowRunById()를 사용하세요. 리더를 사용하면 원시 스냅샷 구조를 읽지 않고도 일시 중지된 단계, 재개 레이블, 단계 페이로드 및 단계 출력에 접근할 수 있습니다.

src/mastra/workflows/recover-run.ts

import { createWorkflowStateReader } from '@mastra/core/workflows'

const workflow = mastra.getWorkflow('testWorkflow')
const state = await workflow.getWorkflowRunById('run-123')

if (state?.status === 'suspended') {
  const reader = createWorkflowStateReader(state)
  const suspendedStep = reader.getSuspendedStep()
  const approvalLabel = reader.getResumeLabel('approve')
  const run = await workflow.createRun({ runId: state.runId })

  await run.resume({
    step: approvalLabel?.stepId ?? suspendedStep?.path,
    resumeData: { approved: true },
    forEachIndex: approvalLabel?.foreachIndex,
  })
}

중첩 Workflow에서는 suspendedStep.path에 재개 경로가 포함됩니다. foreach 일시 중지의 경우 레이블이 특정 반복을 가리키면 일치하는 재개 레이블에 foreachIndex가 포함됩니다.

잠

절전 메서드를 사용하면 Workflow 수준에서 실행을 일시 정지하고 상태를 waiting으로 설정할 수 있습니다. 반면 suspend()는 특정 단계 내에서 실행을 일시 중지하고 상태를 suspended로 설정합니다. 사용 가능한 방법:

• .sleep(): 지정된 밀리초 동안 일시 중지합니다.
• .sleepUntil(): 특정 날짜까지 일시 중지

관련된

• 제어 흐름
• 인간 참여형
• 스냅샷
• 시간여행
• Workflow 상태 리더