暫停與繼續

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。傳入符合該步驟 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 端點、事件處理常式、回應人工輸入時，或計時器中。

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() 時，可將這些 ID 傳給 step 參數，以指定並繼續暫停的執行路徑。

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

復原暫停的執行

當應用程式需要從儲存空間復原暫停的執行時，請搭配使用 workflow.getWorkflowRunById() 與 createWorkflowStateReader()。此讀取器無須讀取原始快照結構，即可取得暫停的步驟、繼續標籤、步驟承載資料與步驟輸出。

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 狀態讀取器