Time travel

时间旅行允许你从任意特定步骤重新执行工作流，可使用已存储的快照数据或你提供的自定义上下文。

这适用于调试失败的工作流，或使用不同输入测试单个步骤，也适用于在不重新运行整个工作流的情况下从错误中恢复。你还可以使用时间旅行来执行一个尚未运行过的工作流，从任意特定步骤开始。

时间旅行的工作原理

当你在工作流运行上调用 timeTravel() 时：

1. 工作流从存储加载现有快照（如果可用）
2. 目标步骤之前的步骤结果会从快照或提供的上下文重建
3. 执行从指定步骤开始，使用提供的或重建的输入数据
4. 工作流从该点继续执行至完成

如果自运行记录以来工作流定义已发生变化（例如步骤被重命名），或记录的运行从未到达目标之前的某个步骤，时间旅行会在执行任何内容之前失败并给出描述性错误，已存储的快照保持不变。

时间旅行需要配置存储，因为它依赖已持久化的工作流快照。

基本用法

使用 run.timeTravel() 从特定步骤重新执行工作流：

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

嵌套工作流步骤

对于嵌套工作流内部的步骤，可使用点号表示法、步骤 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 },
})

提供执行上下文

你可以提供上下文来指定时间旅行时先前步骤的状态：

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

上下文对象包含以步骤 ID 为键的步骤结果。每个步骤结果包括：

• status：步骤的执行状态。常见为 success、failed 或 suspended，但接受任何工作流运行状态：running、success、failed、tripwire、suspended、waiting、pending、canceled、bailed、paused 或 skipped
• payload：传递给步骤的输入数据
• output：步骤的输出数据（对于成功的步骤）
• startedAt：步骤开始的时间戳
• endedAt：步骤结束的时间戳（对于已完成的步骤）
• suspendPayload：传递给 suspend() 的数据（对于挂起的步骤）
• resumePayload：传递给 resume() 的数据（对于已恢复的步骤）

重新运行失败的工作流

时间旅行对于调试和从失败的工作流执行中恢复特别有用：

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

对挂起工作流进行时间旅行

你可以通过时间旅行从较早的步骤恢复挂起的工作流：

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

带初始状态的时间旅行

你可以在时间旅行时提供初始状态以设置工作流级状态：

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

错误处理

时间旅行会在特定情况下抛出错误。在执行任何内容之前，时间旅行会原子地认领该运行，因此一次只有一个调用继续。并发调用会抛出 WORKFLOW_TIME_TRAVEL_ALREADY_CLAIMED，如果运行已在执行则抛出 WORKFLOW_TIME_TRAVEL_RUN_RUNNING。参见并发时间旅行调用。

运行中的工作流

你不能对当前正在运行的工作流进行时间旅行：

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

无效的步骤 ID

如果目标步骤在当前工作流定义中不存在，时间旅行会抛出错误：

try {
  await run.timeTravel({ step: 'nonExistentStep' })
} catch (error) {
  // "Cannot time travel to step 'nonExistentStep': the step does not exist in the current execution graph. ..."
}

错误还会列出该运行快照中记录的最多 10 个步骤 ID（以及剩余 ID 的数量），并确认已存储的快照未被修改。如果该运行没有记录的步骤结果，则错误内容为 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"
}

嵌套工作流上下文

对嵌套工作流进行时间旅行时，你可以同时为父工作流和嵌套工作流步骤提供上下文：

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

在新的工作流运行上测试步骤逻辑

在新的工作流运行上使用特定输入测试单个步骤，适用于无需从头开始执行工作流即可测试步骤逻辑。

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

相关内容

• 快照
• 挂起与恢复
• 错误处理
• 控制流