Run.timeTravelStream()

.timeTravelStream() 方法会从任意指定步骤开始重新执行 Workflow，并流式传输事件。你可以在时间旅行执行期间接收实时更新，同时完整了解每个步骤的进度。

用法示例

const run = await workflow.createRun()

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

// Process events as they arrive
for await (const event of output.fullStream) {
  console.log(event.type, event.payload)
}

// Get the final result
const result = await output.result

参数

所有参数都与 Run.timeTravel() 相同。有关详细的参数文档，请参阅 timeTravel 参考。

返回值

output:

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

同时包含 Stream 和结果 Promise 的对象

output.fullStream:

ReadableStream<WorkflowStreamEvent>

随着执行推进发出 Workflow 事件的可读 Stream

output.result:

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

解析为最终 Workflow 执行结果的 Promise

output.traceId?:

string

启用 Tracing 时与本次执行关联的 Trace ID

Stream 事件

Stream 会在执行期间发出 Workflow 事件：

• workflow-step-start：步骤开始执行时发出
• workflow-step-finish：步骤成功完成时发出
• workflow-step-error：步骤遇到错误时发出
• workflow-step-suspended：步骤暂停时发出
• 其他事件取决于步骤类型（Agent、Tool 等）

扩展用法示例

在时间旅行期间处理事件

const run = await workflow.createRun()

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

for await (const event of output.fullStream) {
  switch (event.type) {
    case 'workflow-step-start':
      console.log(`Starting step: ${event.payload.stepName}`)
      break
    case 'workflow-step-finish':
      console.log(`Completed step: ${event.payload.stepName}`)
      break
    case 'workflow-step-error':
      console.error(`Error in step: ${event.payload.stepName}`, event.payload.error)
      break
  }
}

const result = await output.result
console.log('Time travel completed:', result)

带上下文的时间旅行 Stream

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

for await (const event of output.fullStream) {
  // Handle events
  console.log(event)
}

const result = await output.result

带嵌套 Workflow 的时间旅行 Stream

const output = run.timeTravelStream({
  step: ['nestedWorkflow', 'step3'],
  inputData: { value: 10 },
  nestedStepsContext: {
    nestedWorkflow: {
      step2: {
        status: 'success',
        payload: { step1Result: 2 },
        output: { step2Result: 3 },
        startedAt: Date.now(),
        endedAt: Date.now(),
      },
    },
  },
})

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

const result = await output.result

注意事项

• 时间旅行执行完成或遇到错误时，Stream 会自动关闭
• 可以在 Workflow 仍在执行时处理 Stream 中的事件
• result Promise 仅在所有步骤完成后才会解析
• Stream 事件采用与常规 Workflow Streaming 相同的格式
• 时间旅行 Streaming 依赖持久化的 Workflow 快照，因此需要配置存储

相关内容

• Run.timeTravel()
• 时间旅行
• Workflow Streaming
• Run.stream()
• Run.resumeStream()