数据截至 2026-08-10(上游 commit 35efe178d76b)

Workflow 引擎:声明式多步编排与暂停/恢复

30 秒导读: Agent(第 1 章)是"让模型自由发挥、边想边做手脚";Workflow 反过来—— 用代码提前把步骤写死成一条流水线,该走哪步、走几步、并行还是串行,全由你声明,不靠模型临场决定。 换来的是确定性和一个杀手锏:每步跑完都存一份快照,于是流程能在任意步暂停、等人审批, 之后跨进程恢复接着跑,甚至回到历史某一步重放。

本章讲这套引擎怎么搭、怎么转、怎么在中断后原地复活。不重复 Agent 一次生成的内部细节(见 01-agent-runtime.md),只讲"步骤如何把 agent 当积木嵌进来、状态如何持久化以支持恢复"。

---

1. 这是什么(零基础也能懂)

一句话定义

Workflow 是一条用链式代码声明出来的多步流水线:你把每一步(跑个函数、调个 agent、并行几件事、 按条件分支、循环、睡一会儿……)一节一节接起来,引擎负责按顺序驱动、在步与步之间传数据、 并在需要时把整条流水线冻结成快照。

解决什么问题 / 给谁用

假设你要做一个"用户入职"自动化:拉用户资料 → 生成欢迎语(要用大模型)→ 停下来等管理员点批准 → 发通知。这里的难点不是任何单步,而是那个"停下来等人"——审批可能几分钟也可能三天,你不能让一个 进程干等着。Workflow 让你把"停下来"写成一行 suspend(),引擎把现场存进数据库、进程可以退出; 等审批来了,再用一行 resume() 从那一步接着跑。

它给这几类人用:

• 要把 LLM 调用和普通业务逻辑混编成可靠管线的工程师。
• 需要 human-in-the-loop(人工审批、补充输入)的流程。
• 需要崩溃可恢复、可回放调试的生产级编排。

它能做什么(功能一览)

能力	对应原语	白话
跑一段函数	andThen	最基本的一步
调一个 agent	andAgent	把第 1 章的 Agent 当一步嵌进来
条件执行	andWhen	满足条件才跑这步
多路分支	andBranch	所有命中条件的分支都跑
并行等全部	andAll	几件事同时做,等齐
并行取最快	andRace	谁先完成用谁
遍历数组	andForEach	对每个元素跑一步(可限并发)
循环	andDoWhile / andDoUntil	反复跑直到条件
取数据塑形	andMap	从 input/步骤/上下文拼出新对象
定时等待	andSleep / andSleepUntil	睡一段/睡到某时刻
护栏校验	andGuardrail	校验/清洗数据
旁路观察	andTap	看一眼数据但不改它
嵌套子流程	andWorkflow	把另一条 workflow 当一步

用起来什么样

一段最小的真实风格代码,直观感受"链式声明":

// 示意,非源码:一条把 agent 嵌进流水线的最小 workflow
const workflow = createWorkflowChain({
  id: "user-processing",
  input: z.object({ userId: z.string() }),        // 入口数据形状
  result: z.object({ content: z.string() }),      // 出口数据形状
})
  .andThen({                                       // 第 1 步:拉资料
    id: "fetch-user",
    execute: async ({ data }) => {
      const userInfo = await fetchUserInfo(data.userId);
      return { ...data, userInfo };                // 返回值 = 下一步的 data
    },
  })
  .andAgent(                                        // 第 2 步:让 agent 写欢迎语
    ({ data }) => `给 ${data.userInfo.name} 写一句欢迎语`,
    agent,
    { schema: z.object({ content: z.string() }) },
  );

const { result } = await workflow.run({ userId: "123" });

三个关键直觉,先记住:

1. 一步的返回值,就是下一步的 data——数据顺着链子往下淌。
2. 步骤是"声明"出来的,不是模型选的;流程结构在写代码时就定死了。
3. input / result 用 zod schema 卡住两端形状,类型顺着链子自动推导。

一句话类比

把 Workflow 想成 Unix 管道 a | b | c,只不过每个环节可以是"调大模型"这种异步重活, 而且整条管道能在任意接口处按下暂停键、把半成品连同现场一起塞进抽屉,过几天再拉出来接着流。

---

2. 顶层全景(它大概怎么转)

三层结构

引擎从"你怎么写"到"它怎么跑"分三层。先看这张图(从上到下是从你的代码到底层执行):

  你写的链式代码
  ┌─────────────────────────────────────────────┐
  │  createWorkflowChain(cfg).andThen().andAgent()│  ← ① 构建层
  │  每个 .andX() 造一个 step 对象、push 进数组     │     chain.ts
  └───────────────────────┬─────────────────────┘
                          │ .run() / .stream()
                          ▼
  ┌─────────────────────────────────────────────┐
  │  createWorkflow(cfg, ...steps) → executeInternal│ ← ② 执行层
  │  for 每个 step:  存输入 → 跑 → 存输出 → 存快照   │     core.ts
  │  中间随时能被 AbortSignal 打断 = 暂停/取消       │
  └──────┬───────────────────────┬──────────────┘
         │ 存/读现场               │ 登记活跃执行
         ▼                        ▼
  ┌──────────────┐        ┌──────────────────┐
  │  Memory V2    │        │ WorkflowRegistry │   ← ③ 支撑层
  │ 存 workflow   │        │ 单例·活跃执行表   │     registry.ts
  │ 执行状态+快照  │        │ 全局暂停/恢复入口 │
  └──────────────┘        └──────────────────┘

部件一句话职责

部件	干什么	在哪个文件
WorkflowChain	链式构建器,收集步骤,末端转成可运行 workflow	workflow/chain.ts:111
andThen 等原语	每个造一个 {type, execute} 步骤对象	workflow/steps/and-*.ts
createWorkflow	把 config + 步骤数组组装成 workflow 对象	workflow/core.ts:908
executeInternal(内联在 createWorkflow 里)	真正的执行循环:驱动每步、处理暂停/恢复/重试	workflow/core.ts:1782
WorkflowStateManager	一次执行的内存态(data/status/suspension…)	workflow/internal/state.ts:107
WorkflowSuspendController	一个带"暂停/取消"语义的 AbortController 包装	workflow/suspend-controller.ts:25
Memory V2(updateWorkflowState 等)	把执行状态和快照落库,支持跨进程恢复	(记忆子系统,见 03-memory.md)
WorkflowRegistry	全局单例:登记 workflow、追踪活跃执行、统一恢复入口	workflow/registry.ts:42
WorkflowStreamController / ...Writer	把执行过程做成事件流,供 UI 实时观察	workflow/stream.ts:9

主线走一遍(高层)

一次 workflow.run(input) 大致这样流动:

run(input)
  └─ executeInternal(core.ts:1782 起)
       ├─ stateManager.start(input)          建立内存态,status=running
       ├─ 建 executionContext(stepData 表、stream writer、trace)
       └─ for (index, step) of steps:        ← 一步一步来
            ├─ if index < startStepIndex: continue   (恢复时跳过已完成步)
            ├─ 若 AbortSignal 已 aborted → 暂停或取消,存快照后返回
            ├─ stepData.set(step.id, {input, status:"running"})
            ├─ result = await step.execute(stepContext)   ← 真正干活
            ├─ stateManager.update({ data: result })      结果成为下一步输入
            ├─ stepData.set(step.id, {output, status:"success"})
            └─ persistRunningCheckpoint(index)            存"跑到这了"的快照
       └─ 全部跑完 → stateManager.finish() → 返回 {status:"completed", result}

关键点:for 循环本身就是引擎(core.ts:1782),没有魔法调度器;整条链子是一个普通的 顺序 for,而"跳步恢复""暂停""重试"都是在这个 for 里加判断实现的。这份朴素恰恰是它可靠、 可预测的原因。

---

3. 核心原理(逐个机制,由浅入深)

3.1 链式构建器:.andX() 到底做了什么

它要解决的小问题: 怎么让 a().andThen().andAgent() 这种写法既好写、又能让 TypeScript 一路推导出每步的数据类型。

思路: WorkflowChain 内部就攒一个数组 this.steps;每个 andX 造一个步骤对象、push 进去、 return this。类型的"魔法"全在方法重载的泛型签名里,运行时其实平平无奇。

看 andThen 的运行时实现,只有三行实质逻辑:

// workflow/chain.ts:429
andThen(config: any): any {
  const step = andThen(config) as WorkflowStep<...>;
  this.steps.push(step);
  return this;   // 返回自己,好接着 .andX()
}

链子的类型参数 CURRENT_DATA 随每次 andX 变化:andThen<NEW_DATA> 让返回的链变成 WorkflowChain<..., NEW_DATA, ...>(chain.ts:406),这样下一步的 data 类型就是上一步的返回值。 这就是"一步的返回值是下一步的 data"在类型层面的实现。

到末端,.run() / .toWorkflow() 才真正把攒好的数组喂给 createWorkflow:

// workflow/chain.ts:949  toWorkflow()
return createWorkflow(this.config, ...this.steps);  // 变长参数展开步骤数组

一个坑(见 chain.ts:993 注释): chain.run/stream/timeTravel/restart 每次调用都重新 createWorkflow(...) 造一个新实例。做一次性执行没问题;但若要 timeTravel/restart 这类 需要"找回之前那次执行状态"的操作,官方建议先 const wf = chain.toWorkflow() 复用同一个实例, 并配持久化 memory,否则新实例找不到旧执行。

3.2 步骤原语:统一形状 + 各自的执行语义

它要解决的小问题: 十几种步骤(函数、agent、并行、循环……)怎么被同一个执行循环一视同仁地驱动。

思路: 所有 andX 工厂都返回同一种形状的对象——{ type, id, name, ...(可选 schema), execute }。 执行循环只认 step.execute(context);每种步骤的"个性"全封装在自己的 execute 里。defaultStepConfig (internal/utils.ts:56)负责补齐 name/purpose 默认值。

下面是所有原语的执行语义速查(每格都指向真实工厂函数):

原语	type	执行语义(一句话)	源码
andThen	func	跑 execute,返回值成为新 data	and-then.ts:31
andAgent	agent	调 agent.generateText,输出经可选 map 塑形	and-agent.ts:69
andWhen	conditional-when	condition 真才跑内嵌步,否则原样透传 data	and-when.ts:33
andTap	tap	跑副作用,吞掉异常,永远返回原 data	and-tap.ts:32
andGuardrail	guardrail	对 data 跑输入/输出护栏校验	and-guardrail.ts:16
andAll	parallel-all	Promise.allSettled 跑全部,任一失败抛首个错	and-all.ts:59
andRace	parallel-race	并发跑,取最先完成者	and-race.ts
andBranch	branch	所有 condition 命中的分支都跑,结果成数组	and-branch.ts:10
andForEach	foreach	对每个元素跑一步,concurrency 控并发	and-foreach.ts:11
andDoWhile/andDoUntil	loop	反复跑到条件满足	and-loop.ts:146
andMap	map	从 value/data/input/step/context/fn 拼新对象	and-map.ts:79
andSleep/andSleepUntil	sleep/sleep-until	睡一段/睡到某时刻(可被信号打断)	and-sleep.ts:17
andWorkflow	workflow	把另一条 workflow 当一步跑(实验性)	and-workflow.ts:36

两个值得记的细节:

andForEach 不是傻串行,也不是无限并发,而是一个固定大小的 worker 池:concurrency=1 走串行快路, 否则起 min(concurrency, 元素数) 个 worker 抢着取下一个下标做:

// workflow/and-foreach.ts:109  worker 池抢下标
const workers = Array.from({ length: Math.min(maxConcurrency, itemList.length) },
  async () => {
    while (nextIndex < itemList.length) {
      const index = nextIndex; nextIndex += 1;      // 抢一个下标
      results[index] = await runItem(itemList[index], index);
    }
  });
await Promise.all(workers);

andBranch 和 andWhen/andAll/andForEach/andLoop 一样,跑内嵌子步时把 workflowContext 置空(and-branch.ts:59 的 subState = { ...state, workflowContext: undefined }),这样子步不会重复 发一遍事件——外层容器步已经代表它们发过了。这是"套娃步骤不重复上报"的统一做法。

3.3 step 如何把 agent 嵌进来(andAgent)

它要解决的小问题: 让一个完整的 Agent(带记忆、工具、子代理)在流水线里当"一步",还要把用量、 上下文、追踪 span 都串起来,但不重造 agent 的生成逻辑。

思路: andAgent 的 execute 里就是调一次 agent.generateText,把 workflow 的现场信息 "喂"进去、把结果"接"出来。它自己不管 token 怎么生成——那是第 1 章的事。

三个"接线"动作看这段真实源码:

// workflow/steps/and-agent.ts:137  在 workflow 上下文里调 agent
const result = await agent.generateText(finalTask, {
  ...restConfig,
  context: restConfig.context ?? state.context,             // 继承 workflow 上下文
  conversationId: restConfig.conversationId ?? state.conversationId,
  userId: restConfig.userId ?? state.userId,
  parentSpan: state.workflowContext?.currentStepSpan,        // 挂到当前步的 span 下
  output,                                                    // 由 schema 决定结构化输出
});

接线三件事:

1. 继承身份/上下文:conversationId、userId、context 若步骤没显式给,就沿用 workflow 现场。
2. span 挂载:把 agent 的 span 挂到 currentStepSpan 下,于是可观测性里 agent 调用是这一步的子节点 (可观测性细节见 06-observability-guardrails-voltops.md)。
3. 用量累加:每次 agent 调用后把 token 用量加进 state.usage(and-agent.ts:150),整条 workflow 的总用量因此可汇总。

输出结构由 config.schema 决定:是 zod schema 就包成 Output.object({schema}),是 AI SDK 的 Output 规格就直接用(and-agent.ts:95);还能传第四个参数 map 把 agent 原始输出塑形/合并进现有 data。

task 和 schema 都可以是函数,在执行时拿到 context 再动态算(and-agent.ts:93-94)—— 于是提示词能引用前面步骤产出的数据。

3.4 执行引擎:一个 for 循环撑起一切

它要解决的小问题: 顺序驱动、步间传数据、存现场、还要能被打断——怎么在一个循环里全办了。

思路: 引擎核心是 executeInternal 里对步骤数组的一个 for...of(core.ts:1782)。每一轮做四件事: 判断要不要跳过/暂停 → 存输入 → 跑 → 存输出+快照。状态由 WorkflowStateManager 托管。

先看状态机。WorkflowStateManager(internal/state.ts:117)持有一次执行的全部内存态,状态只在 这几个值间流转:

        start()
  pending ──────► running ──┬─ finish() ─► completed   (终态)
                            ├─ fail()   ─► failed       (终态)
                            ├─ suspend()─► suspended    (可 resume 回 running)
                            └─ cancel() ─► cancelled    (终态)

一个关键护栏:到了 completed/failed 就禁止再改状态——任何 update/suspend/cancel 都会先跑 assertCanMutate,终态就抛错:

// workflow/internal/state.ts:277
function assertCanMutate(value): asserts value is RunningWorkflowState {
  if (!hasState(value) || value.status === "completed" || value.status === "failed") {
    throw new Error("Cannot mutate state after workflow has finished");
  }
}

注意 suspended/cancelled 不在禁止之列(只挡 completed/failed),因为暂停后还要被恢复继续改。

再看每步怎么拿到上下文。执行循环用 createStepExecutionContext(internal/utils.ts:72)给 step.execute 装配一个对象,步骤代码里 async ({ data, suspend, getStepData, ... }) 解构的就是它:

字段	给步骤的能力
data	上一步的输出
state	只读现场:executionId、input、usage、signal……
getStepData(id) / getStepResult(id)	回看任意已完成步的输入/输出
suspend(reason, suspendData)	主动暂停这条 workflow(见 3.5)
bail(result) / abort()	提前成功收尾 / 中止
resumeData	恢复时注入的外部输入(仅被恢复的那一步能拿到)
workflowState / setWorkflowState	读写跨步共享的可变状态
writer	往执行事件流里写自定义事件(见 3.7)

其中 setWorkflowState 的闭包(core.ts:2309)把新状态同时写进 stateManager、executionContext 和 stepContext 三处,保证快照存的是最新的共享状态——这点对恢复正确性很重要。

3.5 暂停与恢复:本章的心脏

这是 Workflow 相对"裸 for 循环"最大的价值。先建立直觉,再看两条触发路径,最后看快照与恢复。

直觉:暂停 = 把"跑到哪 + 现场数据"冻进抽屉

暂停不是"挂起线程"。它是:把当前进度(第几步)和现场(当前 data、共享状态、各步产物)打包成一个 checkpoint 存进数据库,然后让这次执行以 suspended 状态干净地返回。进程可以退出。恢复时另起一次 执行,从数据库读回 checkpoint,跳过已完成的步,从暂停那步接着跑。

  run() ───► step0 ✔ ─► step1 ✔ ─► step2 ⏸ suspend!
                                        │  存 checkpoint:
                                        │   • resumeStepIndex = 2
                                        │   • stepExecutionState = 当前 data
                                        │   • workflowState / 各步产物 / usage
                                        ▼
                              status = "suspended"  ← 干净返回,进程可退出
                    ……(几分钟/几天后,拿到审批输入)……
  resume(input) ─► 读回 checkpoint ─► 新一次 executeInternal:
                     for 循环里 index<2 的步直接 continue 跳过
                     到 step2:把 input 作为 resumeData 注入,接着跑 ─► step3 ✔ ─► done

两条暂停触发路径

路径 A — 步骤内主动暂停(human-in-the-loop 常用): 步骤代码调 await suspend("等审批")。 这个 suspend 就是 suspendFn(core.ts:2110):它把 suspendData 暂存进 context、触发控制器、 然后抛一个特殊错误 WORKFLOW_SUSPENDED:

// workflow/core.ts:2110  步骤级 suspend
const suspendFn = async (reason?, suspendData?): Promise<never> => {
  if (suspendData !== undefined) executionContext.context.set("suspendData", suspendData);
  if (options?.suspendController) options.suspendController.suspend(reason ?? "Step requested suspension");
  throw new Error("WORKFLOW_SUSPENDED");   // 用异常把控制权弹回执行循环
};

路径 B — 外部信号暂停(如服务器要关机): 外部持有一个 WorkflowSuspendController,调它的 suspend(reason)。这本质是 abort 一个 AbortSignal:

// workflow/suspend-controller.ts:43
suspend: (reason?) => {
  if (!suspended && !cancelled) {
    suspensionReason = reason; suspended = true;
    triggerAbort("suspended");     // abortController.abort({type:"suspended", reason})
  }
},

执行循环在每步开始前检查这个信号(core.ts:1928),已 abort 就地暂停;步骤执行中途也会被 executeWithSignalCheck(core.ts:2324 调用、:3657 定义)轮询打断——suspensionMode:"immediate" 时 50ms 一查、否则 500ms(core.ts:2328)。像 andSleep 这类等待用 waitWithSignal(steps/signal.ts:26) 监听 abort,能被立刻唤醒。

两条路径最终汇流到同一个 catch:执行循环捕获到 WORKFLOW_SUSPENDED 错误(core.ts:2419), 调 handleStepSuspension(core.ts:2129)。信号如何被翻译成这个错误?看 steps/signal.ts:1—— getAbortError 按 abort 原因的 type 决定抛 WORKFLOW_CANCELLED 还是 WORKFLOW_SUSPENDED。 于是"暂停 vs 取消"用同一套 AbortSignal 机制、靠 reason 区分。

checkpoint:冻的到底是什么

handleStepSuspension 造 checkpoint 并交给 stateManager.suspend(reason, checkpoint, index):

// workflow/core.ts:2143  暂停时打包现场
const suspensionMetadata = stateManager.suspend(reason, {
  stepExecutionState: stateManager.state.data,           // 当前步的输入数据 = 恢复起点
  completedStepsData: steps.slice(0, index).map(...),    // 已完成步的产物摘要
  workflowState: stateManager.state.workflowState,       // 跨步共享状态
  stepData: serializeStepDataSnapshot(),                 // 各步 input/output/status 快照
  usage: stateManager.state.usage,                       // 累计用量
}, index);                                               // index = 暂停在第几步

这份 WorkflowSuspensionMetadata 的结构见 types.ts:17,suspendedStepIndex(:23)记着"从第几步恢复", checkpoint(:29)装现场。随后 saveSuspensionState(core.ts:940)把它 updateWorkflowState 到 memory,状态置 suspended。至此现场完全落库,内存里那次执行可以消失。

恢复:跳步 + 注入 resumeData

用户拿到 suspended 结果后调 result.resume(input)。它绕道 WorkflowRegistry.resumeSuspendedWorkflow (registry.ts:152):校验状态确实是 suspended(:175),从 metadata 取 checkpoint,拼出 resumeFrom 再调 workflow.run(input, resumeOptions):

// workflow/registry.ts:209  恢复选项
resumeFrom: {
  executionId,
  checkpoint: suspensionMetadata.checkpoint,
  resumeStepIndex: suspensionMetadata.stepIndex,   // 从暂停那步起
  lastEventSequence: suspensionMetadata.lastEventSequence,
}

回到 executeInternal,恢复分支(core.ts:1431)干三件事:把 startStepIndex 设成 resumeFrom.resumeStepIndex;用 checkpoint.stepExecutionState 覆盖当前 data、恢复 workflowState / usage / 各步 stepData;把外部输入存进 resumeInputData。

于是 for 循环里两处配合完成"跳步续跑":

// workflow/core.ts:1784  已完成的步,直接跳过
if (index < startStepIndex) { continue; }
...
// workflow/core.ts:2274  只有"被暂停的那一步"能拿到 resumeData
const isResumingThisStep =
  options?.resumeFrom && index === startStepIndex && resumeInputData !== undefined;
// → createStepExecutionContext(..., isResumingThisStep ? resumeInputData : undefined, ...)

这就是完整闭环:resumeStepIndex 决定从哪跳过、stepExecutionState 供回起点数据、resumeData 把外部输入(如审批意见)只喂给暂停那一步。resumeStepIndex 还能被 resume(input, { stepId }) 覆盖到别的步(registry.ts:218),甚至倒回重跑。

3.6 崩溃恢复 vs 时间旅行:另外两种"重来"

暂停/恢复是主动的。引擎还有两种"重来",别混淆:

机制	触发场景	从哪读	源状态要求
suspend / resume	主动暂停等外部	suspension.checkpoint	suspended
restart(崩溃恢复)	进程崩了,一次 running 没跑完	metadata 里的 restart checkpoint	running(core.ts:2729)
timeTravel(回放调试)	想回到历史某步重放	历史执行的快照/事件	非 running(core.ts:2837)

崩溃恢复靠"跑着也存快照"。 每步成功后 persistRunningCheckpoint(core.ts:1483)会把进度写进 memory——但状态仍是 running,checkpoint 塞在 metadata 的 VOLTAGENT_RESTART_CHECKPOINT_KEY (core.ts:1519)下。存的频率由 checkpointInterval 节流(core.ts:1488),可用 disableCheckpointing 关掉。之后 restart(executionId)(core.ts 的 restartExecution,读 checkpoint 在 :2735、拼 resumeFrom 在 :2765)就能从崩溃点接着跑;restartAllActive 批量恢复所有 running 的执行。

时间旅行 = 用历史快照伪造一次"从第 N 步开始"的执行。 prepareTimeTravelExecution(core.ts:2818) 先按 stepId 找到目标步下标(:2843),再为它之前的每一步重建产物——优先用 checkpoint 里的 stepData,没有就回退去翻历史 step-complete 事件(:2888);然后算出目标步的输入 replayStepInput (:2936,允许 inputData 覆盖),新建一个 executionId,把这些伪造的历史塞进 resumeFrom.checkpoint (:3000)。跑起来就复用 3.5 的"跳步"机制,只不过跳过的步的产物是从历史"喂"进去的,不是真跑出来的。 适合"改一下第 3 步的输入,只重跑 3 及之后"这种调试。

3.7 事件流:让长流程可被实时观察

它要解决的小问题: 一条要跑几十秒、还会调大模型逐字输出的 workflow,前端怎么实时看到进度。

思路: workflow.stream() 走同一个执行引擎,只是塞进一个 WorkflowStreamController (stream.ts:9)。执行循环每到一步就给它配一个 WorkflowStreamWriterImpl(stream.ts:219),步骤里 context.writer.write(...) 就能往流里发事件;控制器用一个事件队列 + EventTarget 把事件做成 getStream() 异步迭代器(stream.ts:53)和 watch() 回调(stream.ts:120)。

对比 run():非流式执行时装的是 NoOpWorkflowStreamWriter(stream.ts:194),write 直接丢弃—— 执行循环靠 streamController 存不存在来二选一(core.ts:2059)。同一套步骤代码,流式与否只换一个 writer。

最实用的是 pipeFrom(stream.ts:254):把一个 agent 的 fullStream 逐块转成 workflow 事件并入总流—— 它有个大 switch(stream.ts:322)把 AI SDK 的 text-delta/tool-call/tool-result/finish 等 part 一一映射成 WorkflowStreamEvent。于是"workflow 第几步里那个 agent 正逐字吐字、正在调哪个工具"都能透传到前端。

---

4. 巧妙之处(可借鉴的技术)

• 用异常做控制流,把"暂停"弹穿任意深的调用栈。 suspend() 抛 WORKFLOW_SUSPENDED (core.ts:2126),不管步骤内部套了多少层 async,都能一路冒泡回执行循环的统一 catch(core.ts:2419)。 暂停与取消复用同一 AbortSignal,靠 reason.type 区分(steps/signal.ts:1)——一套机制两种语义。

• "跳步"是恢复、崩溃恢复、时间旅行的公共底座。 三种"重来"最后都归约成同一件事:设置 resumeStepIndex + 塞一个 checkpoint,再让 for 循环 if (index < startStepIndex) continue (core.ts:1784)。一个朴素判断复用出三种高级能力。

• 状态终态锁。 assertCanMutate(state.ts:276)只在内存态层面拦 completed/failed 的再修改, 从根上杜绝"已完成的执行被误改结果"。而 suspended 特意放行,因为它天然要被续写。

• 容器步统一"清空 workflowContext"防重复上报。 andAll/andBranch/andForEach/andLoop/andWhen 跑子步前都 subState = { ...state, workflowContext: undefined }(如 and-all.ts:129),让事件只由外层 容器发一次,避免并行/嵌套时事件爆炸。

• 一个 writer 抽象吃掉"流式/非流式"分叉。 NoOpWorkflowStreamWriter(stream.ts:194)让步骤代码 无需关心当前是不是流式——写事件就写,不流式就被静默丢弃。

---

5. 边界与局限(诚实)

• andWorkflow 是实验性的、且不接可观测性。 源码注释明说 "EXPERIMENTAL … doesn't directly hook into or support the Observability"(and-workflow.ts:7);它跑子 workflow 时只透传少量 state (and-workflow.ts:46),trace/事件不会像原生步那样接入。嵌套编排要谨慎。

• startAsync 不支持 resumeFrom。 后台启动的执行不能用来恢复暂停态,源码直接抛错并提示改用 run/stream(core.ts:3075、:3080)。

• checkpoint 依赖数据可序列化。 恢复靠把 data/workflowState/各步产物落库再读回;含类实例、闭包、 不可序列化对象的 data 无法正确还原(引擎存的是它们的序列化快照,见 serializeStepDataSnapshot core.ts:1462)。

• 时间旅行需要历史快照存在。 若目标步之前某步既无 checkpoint stepData、又无可回退的 step-complete 事件,prepareTimeTravelExecution 会因"missing historical snapshots"直接抛错(core.ts:2920); 用 in-memory 且没存事件的执行没法回放。

• chain.run/timeTravel/restart 每次重建实例。 对需要找回旧执行的操作,临时链式实例可能找不到状态, 官方要求配持久化 memory 或复用 toWorkflow() 实例(chain.ts:993 注释)。

• 恢复的正确性系于 memory 适配器。 跨进程恢复要求 memory 真持久化;默认回退是 InMemoryStorageAdapter(core.ts:936),进程一退状态就没了——生产必须显式配持久后端。

---

6. 横向对比(同 shelf 兄弟)

VoltAgent 的 Workflow 走的是"代码声明 + 类型推导链"路线,和其它 agent 框架的编排取舍不同:

• 对比"让模型自己规划下一步"(ReAct/planner 式): VoltAgent Workflow 刻意不让模型决定流程, 结构写死在链子里,换确定性与可恢复性;需要模型自由发挥时用 Agent(见 01-agent-runtime.md) 或 04-subagents-supervisor.md 的 Supervisor 模式。

• 对比图/DAG 式编排框架: 它不是显式建 DAG,而是线性链 + 容器步(andAll/andBranch/andForEach) 在链上"就地展开"并行与分支;简单直观,但表达任意 DAG 依赖不如专用图引擎灵活。

• 暂停/恢复对标 durable execution(如 Temporal 一类): 都靠 checkpoint 实现跨进程续跑,但 VoltAgent 是库内自持的轻量实现(memory 落库 + 跳步 for 循环),而非独立的工作流服务/事件溯源引擎。

---

7. 代码地图(导航索引)

主题	文件路径	关键符号
链式构建器	packages/core/src/workflow/chain.ts	WorkflowChain、createWorkflowChain、andThen/andAgent(方法)
组装 workflow	packages/core/src/workflow/core.ts:908	createWorkflow
执行主循环	packages/core/src/workflow/core.ts:1782	executeInternal(内联 for...of)
步骤内 suspend	packages/core/src/workflow/core.ts:2110	suspendFn
暂停落库	packages/core/src/workflow/core.ts:2129 / :940	handleStepSuspension、saveSuspensionState
恢复跳步	packages/core/src/workflow/core.ts:1784 / :2274	startStepIndex、isResumingThisStep
崩溃恢复 checkpoint	packages/core/src/workflow/core.ts:1483	persistRunningCheckpoint、VOLTAGENT_RESTART_CHECKPOINT_KEY
时间旅行准备	packages/core/src/workflow/core.ts:2818	prepareTimeTravelExecution
恢复结果封装	packages/core/src/workflow/core.ts:3581	createWorkflowExecutionResult、resumeFn
状态机	packages/core/src/workflow/internal/state.ts:117	WorkflowStateManager、suspend、assertCanMutate
步骤上下文装配	packages/core/src/workflow/internal/utils.ts:72	createStepExecutionContext
暂停控制器	packages/core/src/workflow/suspend-controller.ts:25	createSuspendController
abort→错误映射	packages/core/src/workflow/steps/signal.ts	getAbortError、throwIfAborted、waitWithSignal
agent 步	packages/core/src/workflow/steps/and-agent.ts:69	andAgent
并行/分支/循环/遍历	packages/core/src/workflow/steps/and-{all,branch,loop,foreach}.ts	andAll/andBranch/andDoWhile/andForEach
事件流	packages/core/src/workflow/stream.ts	WorkflowStreamController、WorkflowStreamWriterImpl、pipeFrom、NoOpWorkflowStreamWriter
全局注册表	packages/core/src/workflow/registry.ts:42	WorkflowRegistry、resumeSuspendedWorkflow、activeExecutions
关键类型	packages/core/src/workflow/types.ts	WorkflowSuspensionMetadata:17、WorkflowStateStore:288、WorkflowTimeTravelOptions:247