暫停及恢復

Workflow 可以在任何步驟暫停，以收集額外資料、等待 API 回呼、限制高成本操作的頻率，或要求人工介入輸入。Workflow 暫停時，其目前的執行狀態會儲存為快照。之後，你可以從特定步驟 ID 恢復 workflow，還原該快照擷取的確切狀態。快照會儲存在你已設定的儲存 Provider 中，並會在部署及應用程式重新啟動後保留。

使用 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 暫停的步驟重新啟動。傳入 resumeData；其內容須符合步驟的 resumeSchema，以符合暫停條件並繼續執行。

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() 建立 run 實例。

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

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

你可以在應用程式的任何位置呼叫 resume()，包括 HTTP 端點、事件處理器、回應人工輸入時，或計時器中。

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 陣列包含 run 中所有已暫停 workflow 及步驟的 ID。這些 ID 可以傳給 step 參數，以便在呼叫 resume() 時指定並恢復已暫停的執行路徑。

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

復原已暫停的 run

當應用程式需要從儲存空間復原已暫停的 run 時，請同時使用 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,
  })
}

對於巢狀 workflow，suspendedStep.path 包含恢復路徑。對於 foreach 暫停，如果恢復標籤指向特定迭代，相符的標籤會包含 foreachIndex。

Sleep

Sleep 方法可用於在 workflow 層級暫停執行，並將狀態設為 waiting。相比之下，suspend() 會在特定步驟內暫停執行，並將狀態設為 suspended。

可用方法：

• .sleep()：暫停指定的毫秒數
• .sleepUntil()：暫停至特定日期

相關內容

• 控制流程
• 人工介入
• 快照
• 時間回溯
• Workflow state reader