跳到主要内容

Run.resumeStream()

.resumeStream() 方法使用新数据恢复已暂停的 Workflow 运行，使你可以从特定步骤继续执行并观察事件 Stream。

用法示例
用法示例的直接链接

const run = await workflow.createRun()

const stream = run.stream({
  inputData: {
    value: 'initial data',
  },
})

const result = await stream.result

if (result!.status === 'suspended') {
  const resumedStream = await run.resumeStream({
    resumeData: {
      value: 'resume data',
    },
  })
}

参数
参数的直接链接

resumeData?:

z.infer<TInput>
与 Workflow 输入 schema 匹配的输入数据

requestContext?:

RequestContext
Workflow 执行期间使用的 Request Context 数据

step?:

Step<string, any, any, any, any, TEngineType>
从中恢复执行的步骤

forEachIndex?:

number
以已暂停的 .foreach() 步骤的特定迭代为目标。传入要恢复的迭代的从零开始的索引；其他迭代将保持暂停状态。省略此参数可使用相同的 resumeData 恢复该步骤所有已暂停的迭代。

tracingOptions?:

TracingOptions
Tracing 配置选项。

metadata?:

Record<string, any>
要添加到根 Trace span 的元数据。适合添加用户 ID、会话 ID 或功能标志等自定义属性。

requestContextKeys?:

string[]
要作为此 Trace 的元数据提取的其他 RequestContext 键。支持使用点号表示法访问嵌套值（例如 'user.id'）。

traceId?:

string
本次执行使用的 Trace ID（1–32 个十六进制字符）。如果提供，该 Trace 将成为指定 Trace 的一部分。

parentSpanId?:

string
本次执行使用的父 span ID（1–16 个十六进制字符）。如果提供，将创建根 span 作为该 span 的子 span。

tags?:

string[]
要应用到此 Trace 的标签。用于对 Trace 进行分类和筛选的字符串标签。

返回值
返回值的直接链接

stream:

MastraWorkflowStream<ChunkType>
扩展 ReadableStream<ChunkType> 并添加 Workflow 专用属性的自定义 Stream

stream.status:

Promise<RunStatus>
解析为当前 Workflow 运行状态的 Promise

stream.result:

Promise<WorkflowResult<TState, TOutput, TSteps>>
解析为最终 Workflow 结果的 Promise

stream.usage:

Promise<{ inputTokens: number; outputTokens: number; totalTokens: number, reasoningTokens?: number, cacheInputTokens?: number }>
解析为 token 用量统计信息的 Promise

Stream 事件
Stream 事件的直接链接

Stream 会在 Workflow 执行期间发出不同类型的事件。每个事件都有一个 type 字段，以及一个包含相关数据的 payload：

  • workflow-start：Workflow 开始执行
  • workflow-step-start：步骤开始执行
  • workflow-step-output：步骤的自定义输出
  • workflow-step-result：步骤完成并产生结果
  • workflow-finish：Workflow 执行完成，并附带用量统计信息。对于成功的运行，payload.finalWorkflowResult 包含 Workflow 的最终结果
On this page