Run.timeTravel()

.timeTravel() 方法从任意指定步骤开始重新执行 workflow，可使用已存储的快照数据，也可使用你提供的自定义上下文。它可用于调试失败的 workflow、使用不同输入测试单个步骤，还能在不重新运行整个 workflow 的情况下从错误中恢复。

使用示例

const run = await workflow.createRun()

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

参数

step:

Step<string, any, TInputSchema, any, any, any, TEngineType> | [...Step<string, any, any, any, any, any, TEngineType>[], Step<string, any, TInputSchema, any, any, any, TEngineType>] | string | string[]

开始执行的目标步骤。可以是 Step 实例、Step 数组（用于嵌套 workflow）、步骤 ID 字符串或步骤 ID 字符串数组。嵌套 workflow 步骤请使用点表示法或数组（例如 'nestedWorkflow.step3' 或 ['nestedWorkflow', 'step3']）

inputData?:

z.infer<TInputSchema>

目标步骤的输入数据。必须与该步骤的输入 schema 匹配。若未提供，则使用 workflow 快照中的数据

resumeData?:

any

如果 workflow 之前已暂停，要提供的恢复数据

initialState?:

z.infer<TState>

为 workflow run 设置的初始状态。用于在执行前设置 workflow 级状态

context?:

TimeTravelContext<any, any, any, any>

包含目标步骤之前各步骤结果的执行上下文。每个键都是步骤 ID，其值为 StepResult 对象，包含 status、payload、output、startedAt、endedAt、suspendPayload 和 resumePayload

nestedStepsContext?:

Record<string, TimeTravelContext<any, any, any, any>>

嵌套 workflow 步骤的上下文。以嵌套 workflow ID 为键，每个值包含该嵌套 workflow 的步骤结果

requestContext?:

RequestContext

Time Travel 执行期间使用的 Request Context 数据

outputWriter?:

(chunk: TOutput) => Promise<void>

用于在生成输出 chunk 时进行处理的可选异步函数

tracingContext?:

TracingContext

用于创建子 span 和添加 metadata 的 Tracing 上下文。使用 Mastra 的 tracing 系统时会自动注入。

currentSpan?:

Span

用于创建子 span 和添加 metadata 的当前 span。可在执行期间用它创建自定义子 span 或更新 span 属性。

tracingOptions?:

TracingOptions

Tracing 配置选项。

metadata?:

Record<string, any>

要添加到根 trace span 的 metadata。可用于添加用户 ID、会话 ID 或 feature flag 等自定义属性。

requestContextKeys?:

string[]

要提取为此 trace 的 metadata 的其他 RequestContext 键。支持使用点表示法访问嵌套值（例如 'user.id'）。

traceId?:

string

此次执行使用的 trace ID（1-32 个十六进制字符）。提供后，此 trace 将归入指定的 trace。

parentSpanId?:

string

此次执行使用的父 span ID（1-16 个十六进制字符）。提供后，根 span 将创建为该 span 的子 span。

tags?:

string[]

应用于此 trace 的标签，即用于对 trace 进行分类和筛选的字符串标签。

outputOptions?:

OutputOptions

输出配置选项。

includeState?:

boolean

是否在结果中包含 workflow run 状态。

includeResumeLabels?:

boolean

是否在结果中包含恢复标签。

返回值

result:

Promise<WorkflowResult<TState, TInput, TOutput, TSteps>>

解析为 workflow 执行结果的 promise，其中包含步骤输出和状态

traceId?:

string

启用 Tracing 时与此次执行关联的 trace ID。可用它关联日志并调试执行流程。

spanId?:

string

启用 Tracing 时与此次执行关联的根 span ID。可用于 span 级别的查找和关联。

扩展使用示例

使用自定义上下文进行 Time Travel

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

Time Travel 到嵌套 workflow 步骤

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

使用初始状态进行 Time Travel

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

使用嵌套 workflow 上下文进行 Time Travel

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

注意事项

• Time Travel 依赖已持久化的 workflow 快照，因此需要配置存储
• 重新执行 workflow 时，workflow 会从存储中加载现有快照（如果有）
• 目标步骤之前的步骤结果会从快照或提供的上下文中重建
• 使用提供或重建的输入数据，从指定步骤开始执行
• workflow 会从该处继续执行直至完成
• 如果提供自定义上下文或起始步骤的输入数据，也可以对尚未运行过的 workflow 使用 Time Travel。

相关内容

• Time Travel
• Workflows 概览
• Workflow.createRun()
• 快照
• 暂停与恢复