数据截至 (上游 commit 7cdd13a1beec)
心脏: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):
- 去重:已经在处理 / 已经排队 / 已有结果 / 已出错的,直接返回(
:953-959)。 - 算上游:
const inputNodes = this.#inputNodesTo(node)(:961)。 - 就绪预检:上游有谁出错了就放弃(
:964-968);再检查"必需输入是否都有连线"(:971-975)。 - 标记已排队,然后把每个上游节点也
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)时:
this.#abortController.abort()—— 触发信号(:453);- 记录
#abortSuccessfully/#abortError(区分"正常早退"和"报错中止"); - 发
graphAbort(以及根处理器发abort)事件; 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 如何撑起分支、循环与竞速