心脏:GraphProcessor 拉取式数据流引擎

30 秒导读: Rivet 的图跑起来靠一个类——GraphProcessor。它不像你以为的那样"从入口节点顺着箭头往下推",而是反过来:先找到图里没有任何下游的"输出叶子",再从叶子递归向上拉取每个节点缺的输入,等一个节点的输入都齐了就把它扔进一个无并发上限的队列里跑。跑完的结果写进一张 Map,并通过事件流实时播报给外面的 UI / 调试器。本章讲清这套"拉取式"调度的骨架;分支、循环、竞速这三种"精妙控制流"留给 下一章。

前置词汇(节点、DataValue、NodeConnection、图模型)见 01 章;本章假设你已经知道"节点有输入输出端口、连线把一个节点的输出端口接到另一个节点的输入端口"。

---

1. 这是什么:一个执行图的调度器

一句话:GraphProcessor 是 Rivet 的执行引擎——你给它一个图(NodeGraph)和一批输入,它负责决定谁先跑、谁后跑、谁能并发,把每个节点的 process() 依次调起来,最后把图的输出还给你。

它对外的主入口只有一个方法:

// 示意,非源码 —— 典型调用方式
const processor = new GraphProcessor(project, 'mainGraph');
const outputs = await processor.processGraph(
  context,               // 给节点的能力面(下面详解)
  { userQuestion: ... }, // 喂给 GraphInputNode 的输入
);

真源码里这个方法签名在 packages/core/src/model/GraphProcessor.ts:771 的 processGraph,注释写得很直白:"Runs a graph and returns the outputs from the output nodes of the graph."

它要解决的核心难题

一个可视化图,节点之间是任意接线的有向图。要把它跑起来,你至少要回答三个问题:

难题	朴素做法	Rivet 的选择
谁先跑?	拓扑排序,从入口正序推	从输出叶子反向拉取依赖
能不能并发?	一层一层 barrier 同步	无上限并发,谁就绪谁跑
没跑到的节点怎么办?	全跑一遍	拉不到的天然不跑(隐式剪枝)

下面几节就围绕这三点展开。

---

2. 顶层全景:一次 processGraph 怎么转

先看大盘。processGraph 的生命周期可以画成一条流水线:

 processGraph(context, inputs)
        │
        ▼
 ① #initProcessState()      重置所有 per-process 状态(见 :736)
        │                    建 p-queue(concurrency: Infinity)
        ▼
 ② #preprocessGraph()       建节点实例、连线查找表、输入/输出定义(见 :295)
        │
        ▼
 ③ emit 'start' / 'graphStart'   Emittery 播报开始
        │
        ▼
 ④ 挑 startNodes            = 没有任何下游的"输出叶子"(见 :834)
        │
        ▼
 ⑤ 对每个 startNode:
        │   queue.add(() => #fetchNodeDataAndProcessNode(node))   反向拉取(见 :840)
        ▼
 ⑥ await queue.onIdle()     等整张依赖图跑干净(见 :847)
        │
        ▼
 ⑦ 聚合 erroredNodes → AggregateError?   有错就抛(见 :860)
        │
        ▼
 ⑧ emit 'graphFinish'/'done'/'finish',return graphOutputs

部件一句话职责

部件(方法/字段)	干什么	位置
processGraph	主入口,编排整条生命周期	GraphProcessor.ts:771
#initProcessState	每次运行前重置 per-process 状态、建队列	GraphProcessor.ts:736
#preprocessGraph	建节点实例、连线查找表、SCC	GraphProcessor.ts:295
#fetchNodeDataAndProcessNode	反向递归拉取一个节点的上游	GraphProcessor.ts:952
#processNodeIfAllInputsAvailable	就绪判定,满足才真正跑	GraphProcessor.ts:1019
#processNode → #processNormalNode	调节点的 process(),写结果	GraphProcessor.ts:1262 / 1415
#processNodeWithInputData	组装 InternalProcessContext 并 await instance.process	GraphProcessor.ts:1482
#inputNodesTo / #outputNodesFrom	按连线算上游/下游	GraphProcessor.ts:1797 / 1817
#getInputValuesForNode	从 #nodeResults 取上游输出,拼成本节点输入	GraphProcessor.ts:1775
#nodeResults	存每个节点算出来的输出	GraphProcessor.ts:227
#emitter(Emittery)	对外播报所有事件	GraphProcessor.ts:187

---

3. 核心原理一:为什么是"反向拉取"而不是"拓扑正序"

这是整个引擎最反直觉、也最关键的设计。理解了它,后面全通。

3.1 直觉:从"我要什么"倒推,而不是从"我有什么"顺推

想象你要做一道菜(输出),菜谱说需要"炒好的料"(中间节点),炒料又需要"切好的菜"(上游)。

• 正序(拓扑推):从冰箱里的原料开始,挨个处理,一路往前推到成品。问题是——你可能切了一堆根本不会用到的菜。
• 反向拉取(Rivet):从"我要这道菜"出发,问它需要什么 → 需要炒料 → 炒料需要切菜 → 于是去切菜。只有被最终产物依赖到的东西才会被碰。

Rivet 选后者。好处是:图里那些"没连到任何输出"的死枝,天然不会被执行——剪枝是免费的副产品,不需要单独一遍分析。

3.2 起点:挑"输出叶子"当 startNodes

反向拉取要从叶子开始。Rivet 定义的"叶子"是:没有任何下游节点的节点。

真源码:

// GraphProcessor.ts:834
const startNodes = this.runToNodeIds
  ? this.#graph.nodes.filter((node) => this.runToNodeIds?.includes(node.id))
  : this.#graph.nodes.filter((node) => this.#outputNodesFrom(node).nodes.length === 0);

#outputNodesFrom(node).nodes.length === 0 就是"这个节点没有下游"——它是数据流的终点(通常是 GraphOutputNode、副作用节点等)。这些叶子被塞进队列,作为反向拉取的发起点(GraphProcessor.ts:834-845)。

注意 runToNodeIds:如果调用方指定了"只跑到某几个节点",startNodes 就换成那几个目标节点——但拉取逻辑不变,依旧是从目标往上游拉。这是调试器"运行到此节点"功能的底座。

3.3 递归:#fetchNodeDataAndProcessNode 往上游拉

拿到一个节点后,#fetchNodeDataAndProcessNode 做的事是:先确保上游都被拉取,再尝试处理自己。

它的骨架逻辑(GraphProcessor.ts:952):

1. 去重:已经在处理 / 已经排队 / 已有结果 / 已出错的,直接返回(:953-959)。
2. 算上游:const inputNodes = this.#inputNodesTo(node)(:961)。
3. 就绪预检:上游有谁出错了就放弃(:964-968);再检查"必需输入是否都有连线"(:971-975)。
4. 标记已排队,然后把每个上游节点也 addAll 进队列递归拉取(:1002-1013):

// GraphProcessor.ts:1005
this.#processingQueue.addAll(
  inputNodes.map((inputNode) => {
    return async () => {
      await this.#fetchNodeDataAndProcessNode(inputNode);   // 递归向上
    };
  }),
);

await this.#processNodeIfAllInputsAvailable(node);          // 再试着跑自己

关键点:上游的拉取是丢进队列异步跑的,不是同步 await 一个个来。所以整张依赖图会像水漫开一样并发地向上游铺开,谁的依赖先齐,谁就先被真正执行。

3.4 一张图看懂拉取方向

以"A→B→D、A→C→D"这个菱形图为例(D 是输出叶子):

   拉取发起                真正执行(依赖齐了才跑)
   ─────────►             ◄─────────────

   D (叶子, startNode)      D 最后跑
   │ 需要 B、C
   ├─► B  需要 A            B、C 谁先拿到 A 谁先跑(并发)
   └─► C  需要 A            
        └─► A  无上游        A 最先跑(没有依赖,立即就绪)

从 D 发起,递归拉到 A;A 没有上游,立即就绪先跑;A 跑完 B、C 同时就绪并发跑;B、C 都跑完 D 才就绪。执行的实际顺序,是拉取递归"触底反弹"回来的顺序。

---

4. 核心原理二:就绪判定——一个节点凭什么可以跑

拉取只是"排队意向";真正决定"现在能不能跑"的是 #processNodeIfAllInputsAvailable(GraphProcessor.ts:1019)。它是一串短路 return 的守卫(guard)——任何一条不满足就退出,等下次被上游完成时再被唤醒重试。

4.1 就绪的三道关

抛开循环/竞速/控制流那几条特殊分支(留给 03 章),普通节点要跑,核心是三个条件:

关卡	判定	源码
① 必需输入都有连线	每个 required 输入端口都能找到一条连到它的连线	:1064-1067
② 上游全部 visited	每个上游节点都已在 #visitedNodes 里(即已产出结果)	:1088-1108
③ 上游没有 errored	任一上游在 #erroredNodes 里就放弃	:1055-1060

第 ① 关的代码很能说明"就绪"的定义——只看必需(required)输入,可选输入没接线不影响:

// GraphProcessor.ts:1064
const inputsReady = this.#definitions[node.id]!.inputs.every((input) => {
  const connectionToInput = connections?.find(
    (conn) => conn.inputId === input.id && conn.inputNodeId === node.id,
  );
  return connectionToInput || !input.required;   // 有连线,或本就不是必需
});

第 ② 关是"数据是否真的到位"——光有连线不够,连线另一头的上游节点必须已经跑完(在 #visitedNodes 中),否则本节点要"等这个上游"(:1105 waitingForInputNode)。

4.2 为什么是"重试式"而不是"计数式"

值得留意:Rivet 没有用"入度计数器,减到 0 就跑"那套经典调度。它每次上游完成,都会把下游重新 addAll 进队列再跑一遍 #processNodeIfAllInputsAvailable(GraphProcessor.ts:1243-1249),让下游自己重新检查一遍所有守卫:

// GraphProcessor.ts:1243 —— 一个节点跑完后,唤醒它的所有下游重试
this.#processingQueue.addAll(
  outputNodes.nodes.map((outputNode) => async () => {
    await this.#processNodeIfAllInputsAvailable(outputNode);
  }),
);

这种"幂等重试 + 短路守卫"的写法,好处是不用维护精确的计数状态——尤其是循环会反复清空/重跑节点(见 03 章),计数器很难维护正确,而"每次重新问一遍所有条件"永远不会算错。代价是同一个节点的就绪检查可能被触发多次,但都被开头的去重守卫(:1036-1050)挡掉了。

---

5. 核心原理三:依赖图怎么组织——连线查找表

反向拉取要频繁问三个问题:"谁是我的上游""谁是我的下游""我每个输入端口的值从哪来"。为了 O(1)ish 地回答,#preprocessGraph 预先建了连线查找表 #connections。

5.1 预处理:把连线按节点索引

#preprocessGraph(GraphProcessor.ts:295)遍历所有连线,把同一条连线同时挂到它两端的节点名下:

// GraphProcessor.ts:323
this.#connections[conn.inputNodeId] ??= [];
this.#connections[conn.outputNodeId] ??= [];
this.#connections[conn.inputNodeId]!.push(conn);
this.#connections[conn.outputNodeId]!.push(conn);

于是 #connections[nodeId] 就是"所有碰到这个节点的连线"。再靠连线里的 inputNodeId / outputNodeId 区分方向。

术语提醒(易混):在一条 NodeConnection 里,outputNodeId 是数据的来源(上游),inputNodeId 是数据的去处(下游)。命名是站在"连线接到某节点的 input 端口 / output 端口"的角度。

5.2 三个访问器

方法	问题	实现要点	源码
#inputNodesTo(node)	谁是我的上游?	取 inputNodeId === node.id 的连线,拿它们的 outputNodeId	:1797
#outputNodesFrom(node)	谁是我的下游?	取 outputNodeId === node.id 的连线,拿它们的 inputNodeId,并 uniqBy 去重	:1817
#getInputValuesForNode(node)	我每个输入端口的值?	对每个输入定义,找连它的连线,去 #nodeResults 取上游那个输出端口的值	:1775

#getInputValuesForNode 是"依赖图"与"结果存储"的接缝——它把上游算出来的 #nodeResults 按端口重新组装成本节点的输入:

// GraphProcessor.ts:1782
const connection = connections.find((conn) => conn.inputId === input.id && conn.inputNodeId === node.id);
if (connection) {
  const outputNode = this.#nodeInstances[connection.outputNodeId]!.chartNode;
  const outputNodeOutputs = this.#nodeResults.get(outputNode.id);
  const outputResult = outputNodeOutputs?.[connection.outputId];   // 取上游那个具体端口
  values[input.id] = outputResult;
}

---

6. 核心原理四:并发引擎——concurrency 为 Infinity 的 p-queue

Rivet 的并发模型简单到有点极端:没有并发上限。

#initProcessState 里建队列(GraphProcessor.ts:748):

this.#processingQueue = new PQueue({ concurrency: Infinity });

含义:任何时刻,所有已就绪的节点全部并行执行,队列不做节流。这跟前面"反向拉取会像水漫一样并发铺开"是配套的——引擎不关心"同时跑几个",只关心"依赖对不对"。真正的限速交给节点自己(比如 LLM 节点内部的速率控制)。

队列的两个用途

#processingQueue 同时承担了"任务调度"和"完成信号":

• 调度:add / addAll 把"拉取上游""重试下游"当成任务丢进去(:842、:1005、:1243)。
• 完成信号:await this.#processingQueue.onIdle()(:847)——当队列彻底空了、没有任何在跑的任务时,就代表整张可达依赖图都跑完了,processGraph 才继续往下走收尾。

        p-queue (concurrency: Infinity)
   ┌──────────────────────────────────────┐
   │  [拉取A] [拉取B] [跑A] [重试D] ...      │  ← 就绪即入队,无上限并发
   └──────────────────────────────────────┘
                     │
              onIdle() 兑现  ← 队列空 = 全图跑完

---

7. 从"就绪"到"结果":#processNode 这条链

一个节点通过了所有就绪守卫后,真正的执行走这条调用链:

#processNodeIfAllInputsAvailable
        │  currentlyProcessing.add(node)   标记在跑(:1115)
        ▼
#processNode(node)                          (:1262)
        │  再查一次 abort / 上游是否 errored
        ▼
#processNormalNode(node, processId)         (:1415)  ← 普通节点走这条
        │  (isSplitRun 的节点走 #processSplitRunNode 并行分裂, :1291)
        ▼
#processNodeWithInputData(...)              (:1482)
        │  组装 InternalProcessContext
        ▼
  instance.process(inputValues, context)    (:1674)  ← 真正调节点逻辑
        │
        ▼
  #nodeResults.set(node.id, outputValues)   (:1437)  写结果
  #visitedNodes.add(node.id)                        标记完成
  emit 'nodeFinish'                                  播报

#processNormalNode 的核心就三步:算输入 → 调 process → 写结果(GraphProcessor.ts:1415-1447):

// GraphProcessor.ts:1415(节选骨架)
const inputValues = this.#getInputValuesForNode(node);
if (this.#excludedDueToControlFlow(node, inputValues, processId)) return;  // 控制流剪枝,详见 03 章
this.#emitter.emit('nodeStart', { node, inputs: inputValues, processId });
const outputValues = await this.#processNodeWithInputData(node, inputValues, 0, processId, ...);
this.#nodeResults.set(node.id, outputValues);
this.#visitedNodes.add(node.id);
this.#emitter.emit('nodeFinish', { node, outputs: outputValues, processId });

#processNodeWithInputData 里那句 await instance.process(inputValues, context)(GraphProcessor.ts:1674)才是引擎与具体节点的边界——引擎负责调度和喂数据,process() 里是每种节点自己的活(拼 prompt、调 API、跑代码……)。各类节点的 process() 见 04 章。

#processSplitRunNode(:1291)是同一个节点对数组输入并行分裂多跑几份再聚合的变体(isSplitRun),用 Promise.all 把 N 份并行跑完再合成一个数组结果(:1354、:1397)。属于同一条链上的旁支。

---

8. 对外的窗口:Emittery 事件流与 events()

引擎在内部跑得飞快且并发,外面的 UI / 调试器 / 测试怎么知道进度?靠 Emittery(一个类型安全的事件发射器)。

8.1 ProcessEvents:一张"会发生什么"的清单

所有能被订阅的事件都在 ProcessEvents 类型里声明(GraphProcessor.ts:52)。挑几个关键的:

事件	何时发	携带
start	根图开始	project、startGraph、inputs
graphStart / graphFinish	每个图/子图开始、结束	graph、inputs/outputs
nodeStart	某节点开始跑	node、inputs、processId
nodeFinish	某节点跑完	node、outputs、processId
nodeError	某节点抛错	node、error、processId
nodeExcluded	某节点被控制流排除(不执行)	node、reason、inputs、outputs
partialOutput	节点流式产出中间结果(如 LLM 逐 token)	node、outputs、index
done	根图成功完成	results
abort	被中止	successful、error
finish	无论成败都最后发一次	—

发事件的地方散落在执行链里:节点完成发 nodeFinish(:1443)、被剪枝发 nodeExcluded(:1766)、出错发 nodeError(:1452)。

8.2 events():把事件流变成 async 迭代器

订阅方最常用的是 events() 异步生成器——for await 就能顺序拿到所有事件,直到 finish(GraphProcessor.ts:496):

// GraphProcessor.ts:496
async *events(): AsyncGenerator<ProcessEvent> {
  for await (const [event, data] of this.#emitter.anyEvent()) {
    yield { type: event, ...(data as any) };
    if (event === 'finish') break;   // finish 一到,迭代器收尾
  }
}

用起来像这样:

// 示意,非源码 —— 一边跑图一边消费事件
const run = processor.processGraph(context, inputs);
for await (const ev of processor.events()) {
  if (ev.type === 'nodeStart')  console.log('开始', ev.node.title);
  if (ev.type === 'partialOutput') render(ev.outputs);   // 实时流式渲染
}
await run;

这套事件流也是录制/回放的基础:replayRecording(GraphProcessor.ts:545)把录下来的事件序列用 ts-pattern 逐条 match 再重新 emit 一遍,就能"重演"一次历史执行,不必真的再调 API。回放细节见 05 章。

---

9. 节点的能力面:InternalProcessContext

节点的 process(inputs, context) 里那个 context,就是引擎递给节点的能力面——节点想跟外界或引擎打交道,全靠它。类型定义在 packages/core/src/model/ProcessContext.ts:66 的 InternalProcessContext,实例在 #processNodeWithInputData 里逐字段组装(GraphProcessor.ts:1509)。

9.1 能力清单

能力(字段/方法)	节点能干什么	定义 / 装配
createSubProcessor	新建子处理器跑子图(子图节点靠它)	ProcessContext.ts:118 / 装配 GraphProcessor.ts:1567
raiseEvent	向全树广播一个用户事件	:104 / :1529
waitEvent	挂起等待某个用户事件到来	:106 / :1521
getGlobal / setGlobal	读写跨图共享的全局变量	:124-127 / :1554-1559
waitForGlobal	等某个全局变量被设置	:129 / :1560
abortGraph	从节点内部中止整张图	:135 / :1636
executionCache	全 execution 共享的缓存 Map	:112 / :1515
onPartialOutputs	汇报流式中间结果(触发 partialOutput)	:115 / :1534
signal	本节点的 AbortSignal,用来响应中止	:77 / :1552
requestUserInput	向用户要一段输入(UserInput 节点)	:141 / :1641
trace	打 trace 日志	:132 / :1633

9.2 子处理器:子图怎么复用同一套机器

createSubProcessor(GraphProcessor.ts:1567)是子图执行的关键:它 new 一个新的 GraphProcessor,但共享父的关键状态——#executionCache、#externalFunctions、#contextValues、#globals 都直接指过去,并把自己标成 #isSubProcessor = true:

// GraphProcessor.ts:1571(节选)
const processor = new GraphProcessor(project ?? this.#project, subGraphId, this.#registry);
processor.#isSubProcessor = true;
processor.#executionCache = this.#executionCache;   // 共享缓存
processor.#globals = this.#globals;                 // 共享全局
processor.#parent = this;

然后把子处理器的一堆事件(nodeStart/nodeFinish/partialOutput…)转发到父的 emitter(:1584-1594),这样外面订阅根处理器,就能连子图里发生的一切也一并收到。父被 abort 时子也跟着 abort(:1623)。

#isSubProcessor 这个标志决定了收尾时的差别:只有根处理器才发 start / done / error / finish 这类"整体级"事件(:803、:905),子图只发 graphStart/graphFinish,避免每个子图都宣布一次"全局完成"。

---

10. 中止与错误聚合

10.1 abort:一个 AbortController 统管全局

引擎持有一个 #abortController(GraphProcessor.ts:228),在 #initProcessState 里新建(:758)。调 abort()(GraphProcessor.ts:448)时:

1. this.#abortController.abort() —— 触发信号(:453);
2. 记录 #abortSuccessfully / #abortError(区分"正常早退"和"报错中止");
3. 发 graphAbort(以及根处理器发 abort)事件;
4. await this.#processingQueue.onIdle() 等在跑的任务收干净。

每个节点执行时又会派生一个子 AbortController(#processNodeWithInputData :1490),挂在全局信号下——全局一 abort,所有节点的 signal 一起触发(:1495),节点内部的 await 就能感知并抛出。竞速(race)也复用这套机制来"掐掉输的分支",细节见 03 章。

10.2 错误聚合:erroredNodes → AggregateError

节点抛错不会当场炸掉整张图——#nodeErrored(GraphProcessor.ts:1449)只是把错误记进 #erroredNodes 这张 Map,并发 nodeError 事件。图继续跑其他不依赖它的分支。

等 onIdle() 之后,processGraph 才统一清算(GraphProcessor.ts:860):

// GraphProcessor.ts:860(节选)
const erroredNodes = [...this.#erroredNodes.entries()].filter(([nodeId]) => {
  const d = this.#getAttachedDataTo(nodeId);
  return d.races == null || d.races.completed === false;   // 竞速中"输掉"的错误不算数
});
if (erroredNodes.length && !this.#abortSuccessfully) {
  // 一个错误 → 直接包一层 Error(cause);多个 → AggregateError
  error = erroredNodes.length === 1
    ? new Error(message, { cause: nodeError })
    : new AggregateError(erroredNodes.map(([, e]) => e), message);
  await this.#emitter.emit('graphError', { graph: this.#graph, error });
  throw error;
}

两个巧妙处:

• 竞速容错:属于"已完成的竞速"里输掉那条分支的错误会被过滤掉(:862)——分支被主动 abort 而报错,不该算图失败。
• 一 vs 多:单个错误用 Error 带 cause 保留原始堆栈;多个错误用 AggregateError 打包,一次性抛出所有失败节点(:874-879)。

---

11. 边界与坑

• 纯拉取 ⇒ 死枝不跑:没有连到任何输出叶子的节点永远不会执行。这是特性不是 bug,但初学者容易困惑"我这个节点怎么没跑"——多半是它没有下游。
• 无并发上限:concurrency: Infinity 意味着引擎本身不保护你——一张宽图可能同时发起几十个 LLM 调用。限流得靠节点自己或外部。
• 就绪判定只认 required 输入:可选输入没接线不阻塞执行(:1067),节点 process() 里要自己处理 undefined。
• #nodeResults 会被循环清空:循环控制器每轮会 delete 掉环内节点的结果再重跑(:1164),所以"节点跑一次"这个直觉对循环里的节点不成立——见 03 章。
• 状态是 per-process 的:所有 #-前缀的 per-process 字段在 #initProcessState(:736)里整体重置,processGraph 不能对同一实例并发调用(:782 会抛 "Cannot process graph while already processing")。

---

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

主题	文件	符号
主入口/生命周期编排	packages/core/src/model/GraphProcessor.ts	processGraph
per-process 状态重置 + 建队列	packages/core/src/model/GraphProcessor.ts	#initProcessState
预处理:实例/连线表/SCC	packages/core/src/model/GraphProcessor.ts	#preprocessGraph
挑输出叶子当 startNodes	packages/core/src/model/GraphProcessor.ts	processGraph(startNodes 段)
反向递归拉取上游	packages/core/src/model/GraphProcessor.ts	#fetchNodeDataAndProcessNode
就绪判定(守卫链)	packages/core/src/model/GraphProcessor.ts	#processNodeIfAllInputsAvailable
调用节点 process 并写结果	packages/core/src/model/GraphProcessor.ts	#processNode / #processNormalNode
组装 context + await process	packages/core/src/model/GraphProcessor.ts	#processNodeWithInputData
分裂并行运行	packages/core/src/model/GraphProcessor.ts	#processSplitRunNode
上游/下游/输入值	packages/core/src/model/GraphProcessor.ts	#inputNodesTo / #outputNodesFrom / #getInputValuesForNode
结果存储	packages/core/src/model/GraphProcessor.ts	#nodeResults
事件类型清单	packages/core/src/model/GraphProcessor.ts	ProcessEvents
事件异步迭代器	packages/core/src/model/GraphProcessor.ts	events
录制回放	packages/core/src/model/GraphProcessor.ts	replayRecording
子处理器	packages/core/src/model/GraphProcessor.ts	createSubProcessor(在 #processNodeWithInputData 内)
中止	packages/core/src/model/GraphProcessor.ts	abort / #newAbortController
错误聚合	packages/core/src/model/GraphProcessor.ts	#nodeErrored(+ processGraph 清算段)
控制流剪枝(详见 03)	packages/core/src/model/GraphProcessor.ts	#excludedDueToControlFlow / #markAsExcluded
节点能力面(类型)	packages/core/src/model/ProcessContext.ts	InternalProcessContext

---

上一章 01 · 词汇表:节点、数据值与图模型 · 下一章 03 · control-flow-excluded 如何撑起分支、循环与竞速