時間旅行

時間旅行讓你可從任何指定步驟開始重新執行 Workflow，並使用已儲存的快照資料或你提供的自訂 context。

此功能適合用於為失敗的 Workflow 除錯、以不同輸入測試個別步驟，或在無需重新執行整個 Workflow 的情況下從錯誤中復原。你亦可使用時間旅行執行尚未運行的 Workflow，並從任何指定步驟開始。

時間旅行的運作方式

當你在 Workflow run 上呼叫 timeTravel() 時：

1. Workflow 會從儲存空間載入現有快照（如有）
2. 目標步驟之前的步驟結果會從快照或所提供的 context 重建
3. 執行會從指定步驟開始，並使用所提供或重建的輸入資料
4. Workflow 會從該位置繼續執行直至完成

時間旅行依賴持久保存的 Workflow 快照，因此必須配置儲存空間。

基本用法

使用 run.timeTravel() 從指定步驟重新執行 Workflow：

import { mastra } from './mastra'

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

const result = await run.timeTravel({
  step: 'step2',
  inputData: { previousStepResult: 'custom value' },
})

指定目標步驟

你可使用步驟參照或步驟 ID 指定目標步驟：

使用步驟參照

const result = await run.timeTravel({
  step: step2,
  inputData: { value: 10 },
})

使用步驟 ID

const result = await run.timeTravel({
  step: 'step2',
  inputData: { value: 10 },
})

巢狀 Workflow 步驟

對於巢狀 Workflow 內的步驟，請使用點號標記法、步驟 ID 陣列或步驟參照陣列：

// Using dot notation
const result = await run.timeTravel({
  step: 'nestedWorkflow.step3',
  inputData: { value: 10 },
})

// Using array of step IDs
const result = await run.timeTravel({
  step: ['nestedWorkflow', 'step3'],
  inputData: { value: 10 },
})

// Using array of step references
const result = await run.timeTravel({
  step: [nestedWorkflow, step3],
  inputData: { value: 10 },
})

提供執行 context

進行時間旅行時，你可提供 context 以指定先前步驟的狀態：

const result = await run.timeTravel({
  step: 'step2',
  context: {
    step1: {
      status: 'success',
      payload: { value: 0 },
      output: { step1Result: 2 },
      startedAt: Date.now(),
      endedAt: Date.now(),
    },
  },
})

context 物件包含以步驟 ID 為 key 的步驟結果。每個步驟結果包括：

• status：步驟的執行狀態（success、failed、suspended）
• payload：傳遞至步驟的輸入資料
• output：步驟的輸出資料（適用於成功的步驟）
• startedAt：步驟開始時的時間戳記
• endedAt：步驟結束時的時間戳記（適用於已完成的步驟）
• suspendPayload：傳遞至 suspend() 的資料（適用於已暫停的步驟）
• resumePayload：傳遞至 resume() 的資料（適用於已恢復的步驟）

重新執行失敗的 Workflow

時間旅行特別適合用於為失敗的 Workflow 執行除錯，以及從中復原：

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

// Initial run fails at step2
const failedResult = await run.start({
  inputData: { value: 1 },
})

if (failedResult.status === 'failed') {
  // Re-run from step2 with corrected input
  const recoveredResult = await run.timeTravel({
    step: 'step2',
    inputData: { step1Result: 5 }, // Provide corrected input
  })
}

對已暫停的 Workflow 使用時間旅行

你可透過時間旅行，從較早的步驟恢復已暫停的 Workflow：

const run = await workflow.createRun()

// Start workflow - suspends at promptAgent step
const initialResult = await run.start({
  inputData: { input: 'test' },
})

if (initialResult.status === 'suspended') {
  // Time travel back to an earlier step with resume data
  const result = await run.timeTravel({
    step: 'getUserInput',
    resumeData: {
      userInput: 'corrected input',
    },
  })
}

串流傳送時間旅行結果

使用 timeTravelStream() 在時間旅行執行期間接收串流事件：

const run = await workflow.createRun()

const stream = run.timeTravelStream({
  step: 'step2',
  inputData: { value: 10 },
})

for await (const event of stream.fullStream) {
  console.log(event.type, event.payload)
}

const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

對初始狀態使用時間旅行

進行時間旅行時，你可提供初始狀態，以設定 Workflow 層級的狀態：

const result = await run.timeTravel({
  step: 'step2',
  inputData: { value: 10 },
  initialState: {
    counter: 5,
    metadata: { source: 'time-travel' },
  },
})

錯誤處理

在特定情況下，時間旅行會拋出錯誤：

正在運行的 Workflow

你無法對目前正在運行的 Workflow 進行時間旅行：

try {
  await run.timeTravel({ step: 'step2' })
} catch (error) {
  // "This workflow run is still running, cannot time travel"
}

無效的步驟 ID

如果 Workflow 中不存在目標步驟，時間旅行會拋出錯誤：

try {
  await run.timeTravel({ step: 'nonExistentStep' })
} catch (error) {
  // "Time travel target step not found in execution graph: 'nonExistentStep'. Verify the step id/path."
}

無效的輸入資料

啟用 validateInputs 後，時間旅行會根據步驟的 schema 驗證輸入資料：

try {
  await run.timeTravel({
    step: 'step2',
    inputData: { invalidField: 'value' },
  })
} catch (error) {
  // "Invalid inputData: \n- step1Result: Required"
}

巢狀 Workflow 的 context

當時間旅行進入巢狀 Workflow 時，你可為父層及巢狀 Workflow 步驟提供 context：

const result = await run.timeTravel({
  step: 'nestedWorkflow.step3',
  context: {
    step1: {
      status: 'success',
      payload: { value: 0 },
      output: { step1Result: 2 },
      startedAt: Date.now(),
      endedAt: Date.now(),
    },
    nestedWorkflow: {
      status: 'running',
      payload: { step1Result: 2 },
      startedAt: Date.now(),
    },
  },
  nestedStepsContext: {
    nestedWorkflow: {
      step2: {
        status: 'success',
        payload: { step1Result: 2 },
        output: { step2Result: 3 },
        startedAt: Date.now(),
        endedAt: Date.now(),
      },
    },
  },
})

使用案例

為失敗的步驟除錯

使用相同或修改後的輸入重新執行失敗的步驟，以診斷問題：

const result = await run.timeTravel({
  step: failedStepId,
  context: originalContext, // Use context from the failed run
})

在新的 Workflow run 上測試步驟邏輯

在新的 Workflow run 上使用指定輸入測試個別步驟，適合用於無需從頭開始執行 Workflow 即可測試步驟邏輯的情況。

const result = await run.timeTravel({
  step: 'processData',
  inputData: { testData: 'specific test case' },
})

從暫時性故障中復原

重新執行因暫時性問題（網絡錯誤、速率限制）而失敗的步驟：

// After fixing the external service issue
const result = await run.timeTravel({
  step: 'callExternalApi',
  inputData: savedInputData,
})

相關內容

• 快照
• 暫停和恢復
• 錯誤處理
• 控制流程