暂停与恢复

🌐 Suspend & Resume

工作流可以在任何步骤暂停，以收集额外的数据、等待 API 回调、限制高成本操作，或请求人机交互输入。当工作流被挂起时，其当前的执行状态将被保存为快照。你可以稍后从特定步骤 ID恢复工作流，恢复快照中捕获的精确状态。快照会存储在你配置的存储提供商中，并可跨部署和应用重启使用。

🌐 Workflows can be paused at any step to collect additional data, wait for API callbacks, throttle costly operations, or request human-in-the-loop input. When a workflow is suspended, its current execution state is saved as a snapshot. You can later resume the workflow from a specific step ID, restoring the exact state captured in that snapshot. Snapshots are stored in your configured storage provider and across deployments and application restarts.

使用 suspend() 暂停工作流

🌐 Pausing a workflow with suspend()

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

🌐 Use suspend() to pause workflow execution at a specific step. You can define a suspend condition in the step’s execute block using values from resumeData.

• 如果条件不满足，工作流将暂停并返回 suspend()。
• 如果条件满足，工作流将继续执行该步骤中的剩余逻辑。

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() 重新启动工作流

🌐 Restarting a workflow with resume()

使用 resume() 从工作流暂停的步骤重新启动工作流。传递与步骤的 resumeSchema 匹配的 resumeData 以满足暂停条件并继续执行。

🌐 Use resume() to restart a suspended workflow from the step where it paused. Pass resumeData matching the step's resumeSchema to satisfy the suspend condition and continue execution.

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，以获得更大的灵活性。

🌐 Passing the step object provides full type-safety for resumeData. Alternatively, you can pass a step ID for more flexibility when the ID comes from user input or a database.

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

如果只有一步被挂起，你可以完全省略 step 参数，Mastra 将会恢复工作流中上一次被挂起的步骤。

🌐 If only one step is suspended, you can omit the step argument entirely and Mastra will resume the last suspended step in the workflow.

在仅使用 runId 继续时，首先使用 createRun() 创建一个运行实例。

🌐 When resuming with only a runId, create a run instance first using createRun().

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

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

你可以在应用的任何地方调用 resume()，包括 HTTP 端点、事件处理程序、响应人工输入或定时器时。

🌐 You can call resume() from anywhere in your application, including HTTP endpoints, event handlers, in response to human input, or timers.

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 访问挂起数据

🌐 Accessing suspend data with suspendData

当一个步骤被暂停时，你可能希望在步骤稍后恢复时访问提供给 suspend() 的数据。使用步骤执行函数中的 suspendData 参数来访问这些数据。

🌐 When a step is suspended, you may want to access the data that was provided to suspend() when the step is later resumed. Use the suspendData parameter in your step's execute function to access this data.

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() 函数的确切数据。这使你能够保持对工作流为何被挂起的上下文理解，并在恢复过程中使用该信息。

🌐 The suspendData parameter is automatically populated when a step is resumed and contains the exact data that was passed to the suspend() function during the original suspension. This allows you to maintain context about why the workflow was suspended and use that information during the resume process.

识别暂停执行

🌐 Identifying suspended executions

当工作流被暂停时，它会从暂停的步骤重新开始。你可以检查工作流的 status 以确认它已暂停，并使用 suspended 来识别暂停的步骤或嵌套工作流。

🌐 When a workflow is suspended, it restarts from the step where it paused. You can check the workflow's status to confirm it's suspended, and use suspended to identify the paused step or nested 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 数组包含运行中任何被挂起的工作流和步骤的 ID。调用 resume() 时可以将这些 ID 传递给 step 参数，以定位并恢复被挂起的执行路径。

🌐 The suspended array contains the IDs of any suspended workflows and steps from the run. These can be passed to the step parameter when calling resume() to target and resume the suspended execution path.

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

休眠

🌐 Sleep

睡眠方法可用于在工作流级别暂停执行，这会将状态设置为 waiting。相比之下，suspend() 在特定步骤内暂停执行，并将状态设置为 suspended。

🌐 Sleep methods can be used to pause execution at the workflow level, which sets the status to waiting. By comparison, suspend() pauses execution within a specific step and sets the status to suspended.

可用方法：

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

相关

🌐 Related

• 控制流
• 人类在环
• 快照
• 时间旅行