挂起与恢复

在任意步骤暂停 workflow，以收集额外数据、等待 API 回调、限制高成本操作的频率，或请求人类参与（human-in-the-loop）的输入。挂起会将当前执行状态保存为快照。之后，可从特定步骤 ID恢复，还原精确捕获的状态。快照存储在你配置的存储提供商中，并可跨部署和应用重启持久保留。

使用 suspend() 暂停 workflow

使用 suspend() 在特定步骤暂停 workflow 执行。你可以在步骤的 execute 块中使用 resumeData 的值定义挂起条件。

• 如果条件不满足，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()

使用 resume() 重启 workflow

使用 resume() 从暂停的步骤重启已挂起的 workflow。传入符合步骤 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 },
})

如果只有一个步骤被挂起，可以完全省略 step 参数，Mastra 会恢复 workflow 中最后一个被挂起的步骤。

仅使用 runId 恢复时，请先通过 createRun() 创建运行实例。

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

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

你可以在应用中的任何位置调用 resume()，例如 HTTP 端点或事件处理器。定时器和响应人工输入的代码也可以恢复 workflow。

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 数组包含本次运行中所有被挂起的 workflows 和步骤的 ID。调用 resume() 时，可将这些 ID 传给 step 参数，以定位并恢复被挂起的执行路径。

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

恢复已挂起的运行

当应用需要从存储中恢复已挂起的运行时，可结合 workflow.getWorkflowRunById() 与 createWorkflowStateReader() 使用。该 reader 会暴露被挂起的步骤、恢复标签、步骤 payload 和步骤输出，而无需读取原始快照结构。

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,
  })
}

对于嵌套 workflows，suspendedStep.path 包含恢复路径。对于 foreach 挂起，当标签指向特定迭代时，匹配的恢复标签会包含 foreachIndex。

Sleep

Sleep 方法可用于在 workflow 级别暂停执行，此时状态为 waiting。相比之下，suspend() 在特定步骤内暂停执行，此时状态为 suspended。

可用方法：

• .sleep()：暂停指定的毫秒数
• .sleepUntil()：暂停到指定日期

相关内容

• 控制流
• 人类参与（Human-in-the-loop）
• 快照
• 时间旅行
• Workflow 状态读取器