跳到主要内容

数据截至 (上游 commit 7cdd13a1beec)

最精妙处:control-flow-excluded 如何撑起分支、循环与竞速

30 秒导读: Rivet 是个纯数据流引擎——节点靠边相连,谁的输入齐了谁就跑(见 02 章)。可现实里的图需要 if、循环、短路、竞速这些控制流。Rivet 的绝招是:不额外建一张控制流图,而是发明一种特殊数据值 control-flow-excluded(下称"排除值"),让它像毒药一样沿着边往下游传染。一个节点如果输入里带着这个"毒",它自己就"不执行"、并把毒继续传下去;只有极少数节点(if/ifElse/coalesce/graphOutput/raceInputs/loopController)能"解毒"。分支、循环、竞速全建立在这一个机制上。

1. 先建直觉:为什么数据流引擎需要"排除值"​

1.1 数据流引擎的天生短板​

先回顾上一章的心脏:GraphProcessor 是拉取式数据流——它不按顺序执行语句,而是看"哪个节点的输入都齐了",齐了就调度它跑。

这套模型天生没有 if。传统程序里 if (cond) { A } else { B } 意味着"要么执行 A,要么执行 B,不能都执行"。但在数据流图里,A 和 B 都是节点,它们的输入一旦齐了就都会跑——引擎不知道该"跳过"谁。

1.2 Rivet 的答案:用一个"值"表示"这条路不该走"​

Rivet 没有为控制流单独造一套机制,而是把"该不该执行"编码进数据本身。它定义了一种特殊数据类型:

真源码 packages/core/src/model/DataValue.ts:116:

export type ControlFlowExcludedDataValue = DataValueDef<'control-flow-excluded', undefined | 'loop-not-broken'>;

这就是"排除值"。它是一个合法的 DataValue,可以像 string、number 一样在边上流动。它的语义是一句话:

"这条边上流的不是数据,而是一张『此路不通』的通行禁令。"

1.3 一句话直觉/类比​

把它想成电路里的一段断线,或者更贴切——沿着水管流下去的一滴毒药:

  • 普通节点喝到毒(输入里有排除值)→ 它自己也"死"(不执行),并把毒灌进自己所有的输出,继续往下游流。
  • 只有装了"解毒器"的特殊节点(if、coalesce…)能拦下这滴毒,把它转成一个真正的决策。

整章讲的就是这一个隐喻怎么被代码精确实现。

2. 顶层全景:两个方法 + 一份白名单​

整个控制流机制的核心只有三样东西,全在 GraphProcessor.ts 里:

部件干什么在哪
#excludedDueToControlFlow每个节点执行前的关卡:判断"这个节点该不该被排除"GraphProcessor.ts:1685
#markAsExcluded把一个节点判死:所有输出置为排除值,向下游扩散GraphProcessor.ts:1752
nodesAllowedToConsumeExcludedValue解毒白名单:6 种能"吃掉"排除值的节点GraphProcessor.ts:1722

它们的关系:

                 节点即将执行
                       │
                       ▼
        ┌──────────────────────────────┐
        │  #excludedDueToControlFlow    │  ← 关卡:该排除吗?
        └──────────────────────────────┘
           │ 是                    │ 否
           ▼                       ▼
   ┌───────────────┐        ┌─────────────┐
   │ #markAsExcluded│       │  正常 process │
   │ 所有输出=排除值 │       │  产出真数据   │
   └───────┬───────┘        └─────────────┘
           │  毒往下游流
           ▼
      下游节点的输入里出现排除值
      → 下游再次撞上这道关卡(递归传染)

怎么读这张图: 每个节点执行前都要过 #excludedDueToControlFlow 这道关卡;被判排除的节点由 #markAsExcluded 把所有输出染成排除值,于是毒流到下游,下游节点再撞上同一道关卡——排除就这样沿着边一层层传染下去,不需要任何独立的控制流图。

3. 关卡:#excludedDueToControlFlow 的三条排除规则​

这节讲那道"该不该执行"的关卡到底怎么判。它在节点真正 process 之前被调用(见 #processNormalNode 的 GraphProcessor.ts:1418、#processSplitRunNode 的 GraphProcessor.ts:1294),返回 true 就意味着这个节点不跑了。

它按顺序检查三种情况,命中任一就排除。

3.1 规则一:节点被禁用(disabled)​

真源码 GraphProcessor.ts:1691:

if (node.disabled) {
  this.#visitedNodes.add(node.id);
  this.#markAsExcluded(node, processId, inputValues, 'disabled');
  return true;
}

用户在编辑器里把节点"禁用",它就直接被判死——注意它同样调 #markAsExcluded,所以禁用一个节点会让下游全部跟着被排除。这本身就是控制流机制的一个免费副产品。

3.2 规则二:条件节点的 if 端口为假(isConditional)​

任何节点都可以打开 isConditional 开关,引擎会给它加一个隐藏的 if 输入端口(NodeImpl.ts:84 在 getInputDefinitions 里 ports.push(IF_PORT))。

真源码 GraphProcessor.ts:1700:

if (node.isConditional && typeOfExclusion === undefined) {
  const ifValue = coerceTypeOptional(inputValues[IF_PORT.id], 'boolean');
  if (ifValue === false) {
    this.#visitedNodes.add(node.id);
    this.#markAsExcluded(node, processId, inputValues, 'if port is false');
    return true;
  }
}

这让任意节点都能变成条件节点:if 端口喂 false,这个节点连同它的下游整片就被排除。这是"分支"最轻量的一种形态,不需要专门的 If 节点。

3.3 规则三:输入里带着排除值(传染的核心)​

这是最关键的一条——毒的传染就发生在这里。

真源码 GraphProcessor.ts:1711:

const controlFlowExcludedValues = inputsWithValues.filter(
  ([, value]) =>
    value &&
    getScalarTypeOf(value.type) === 'control-flow-excluded' &&
    (!typeOfExclusion || value.value === typeOfExclusion),
);
const inputIsExcludedValue = inputsWithValues.length > 0 && controlFlowExcludedValues.length > 0;

只要任何一个输入是排除值,inputIsExcludedValue 就为真。接着看它是不是白名单里的"解毒"节点(下一节讲),如果不是——

真源码 GraphProcessor.ts:1734:

if (inputIsExcludedValue && !allowedToConsumedExcludedValue) {
  // ...
  this.#markAsExcluded(node, processId, inputValues, 'input is excluded value');
  return true;
}

判死,并把毒继续往下游灌。 一个普通节点只要沾到一点排除值,自己就被排除——这正是"毒药沿边传染"的字面实现。

4. 判死:#markAsExcluded 如何把毒灌满输出​

这节讲"判死"具体做了什么。逻辑很短但很关键。

真源码 GraphProcessor.ts:1752:

#markAsExcluded(node: ChartNode, processId: ProcessId, inputValues: Inputs, reason: string) {
  const outputs: Outputs = {};
  for (const output of this.#definitions[node.id]!.outputs) {
    outputs[output.id] = { type: 'control-flow-excluded', value: undefined };
  }
  // Prevent infinite loop, a control-flow-excluded to loop controller shouldn't set the break port
  if (node.type === 'loopController') {
    outputs['break' as PortId] = { type: 'control-flow-excluded', value: 'loop-not-broken' };
  }
  this.#nodeResults.set(node.id, outputs);
  // ... emit 'nodeExcluded'
}

关键动作:把这个节点的每一个输出端口都填上排除值,然后写进 #nodeResults。下游节点去取输入时(#getInputValuesForNode,GraphProcessor.ts:1775),取到的就是这些排除值——于是它们撞上 §3.3 的关卡,继续被排除。传染链条就此闭合。

注意 loopController 那个特例:它被排除时,break 端口不填普通的 undefined 排除值,而是填一个带标签 'loop-not-broken' 的排除值。这个特殊标签是防死锁的关键,§7 详解。

5. 解毒者:6 种能"吃掉"排除值的节点​

上面反复提到"白名单"。这节讲这份名单,以及为什么只有它们特殊。

真源码 GraphProcessor.ts:1722:

const nodesAllowedToConsumeExcludedValue: BuiltInNodeType[] = [
  'if',
  'ifElse',
  'coalesce',
  'graphOutput',
  'raceInputs',
  'loopController',
];
const allowedToConsumedExcludedValue =
  nodesAllowedToConsumeExcludedValue.includes(node.type as BuiltInNodeType) && !isWaitingForLoop;

对这 6 种节点,输入带排除值不会触发 §3.3 的自动排除——引擎放行,让节点自己的 process 去观察并处理这个排除值。这就是"解毒":它们不被毒传染,而是把毒当成一个可决策的信号。

为什么偏偏是这 6 种?因为它们各自要"看着排除值做决定":

节点为什么必须能吃掉排除值
if要根据"某条边是否被排除"决定走 output 还是 falseOutput
ifElse要在 true/false 两个输入间选,其中一个可能是排除值
coalesce"取第一个非排除的输入"——必须能识别哪些输入是排除值
graphOutput子图的出口:某个输出被排除是正常结果,不能让它把整个子图判死
raceInputs竞速:落败分支就是排除值,要跳过它取第一个有效值
loopController循环控制器:靠排除值区分"还在循环 / 已跳出",详见 §7

5.1 If:排除值就是"假"​

真源码 IfNode.ts:103 的 process:输入 if 端口如果是排除值,直接当"假"处理,走 isFalse 分支——

if (ifValue.type === 'control-flow-excluded') {
  return isFalse;   // output=排除值, falseOutput=真数据
}

它的输出很讲究:走哪条路,另一条就吐排除值。IfNode.ts:157 的正常返回把 output 给真数据、falseOutput 给排除值——于是接在 falseOutput 上的下游整片被传染排除。分支的"另一条腿"就这样被关掉。

5.2 Coalesce:短路取值,自己也可被排除​

CoalesceNode 是"取第一个有效输入"(类似 ??)。它的精妙在注释里说得很清楚——

真源码 CoalesceNode.ts:98:

// This lets the coalesce actually be control-flow-excluded itself, because otherwise
// the input control-flow-excluded are consumed.
if (conditional?.type === 'control-flow-excluded') {
  return { ['output']: { type: 'control-flow-excluded', value: undefined } };
}

它有个专门的 conditional 端口:如果这个端口被排除,coalesce 自己就吐排除值(把毒传下去);否则它扫描 input1..inputN,跳过所有排除值,返回第一个非排除且非空的值(CoalesceNode.ts:114)。若全是排除值,输出也是排除值(CoalesceNode.ts:119)。这让 coalesce 既能"解毒短路",又能在该传染时传染。

5.3 Match:一进多出的多路分支​

MatchNode(MatchNode.ts:124)是"正则多路分支":拿输入字符串去匹配多个 case,命中的那个 case 端口吐真数据,没命中的全吐排除值——

if (match && canMatch) {
  output[`case${i + 1}`] = { type: outputType, value: outputValue };
} else {
  output[`case${i + 1}`] = { type: 'control-flow-excluded', value: undefined };
}

于是每个 case 后面挂的分支,只有命中那一条会活下来,其余全被传染排除。这就是 switch/case 在数据流里的实现——注意 Match 本身不在白名单里,它靠"给未命中端口主动填排除值"来关分支,不需要吃掉输入的排除值。

6. 演示:排除值像毒药一样沿边传染​

下面这段示意代码(非源码),把 §3-§5 的传染逻辑浓缩成最小模型,帮你在脑子里跑一遍:

# 示意,非源码:排除值如何沿边传染
EXCLUDED = ("control-flow-excluded", None)     # 那滴"毒"
SOLVENT = {"if", "ifElse", "coalesce", "graphOutput", "raceInputs", "loopController"}

def run_node(node, inputs):
    # 关卡:任一输入带毒,且自己不是解毒者 → 自己也死,输出全灌毒
    poisoned = any(v == EXCLUDED for v in inputs.values())
    if poisoned and node.type not in SOLVENT:
        return {port: EXCLUDED for port in node.outputs}   # #markAsExcluded

    # 解毒者 / 干净节点:正常算,自己决定哪条腿吐毒
    return node.process(inputs)

# if 节点:真数据走 output,另一条腿主动吐毒
def if_process(inputs):
    if inputs["if"] in (False, EXCLUDED, None, ""):
        return {"output": EXCLUDED, "falseOutput": inputs["value"]}   # 关掉 output 那条腿
    return {"output": inputs["value"], "falseOutput": EXCLUDED}       # 关掉 falseOutput 那条腿

重点看: 毒只需在 if 的某个输出端口生成一次,之后所有下游普通节点都会因为"输入带毒"而自动被 run_node 的第一段判死——一次决策,整片子图静默关闭,引擎全程只是在传数据,从没有"跳转"这个概念。

7. 循环:用排除值区分"还在转"和"已跳出"​

循环是这套机制最烧脑的应用。因为数据流图里的循环就是图里的一个环(cycle),而引擎默认"每个节点只跑一次"。要让环转起来又能停下,LoopControllerNode 和引擎两边配合,核心还是那滴毒——只是这次毒带了标签。

7.1 LoopController 的 continue / break​

循环控制器有个 continue 输入和 break 输出。它的 process(LoopControllerNode.ts:183)决定这一轮是"再转一圈"还是"跳出":

真源码 LoopControllerNode.ts:231:

if (continueValue) {
  output['break' as PortId] = { type: 'control-flow-excluded', value: 'loop-not-broken' };
} else {
  // Break gets an array of all the input values
  output['break' as PortId] = { type: 'any[]', value: inputValues };
}

看这个巧妙的反转:

状态break 端口的值含义
继续循环(continue)排除值 'loop-not-broken'break 这条出口关着,数据继续在环里转
跳出循环(break)真数据(所有输入的数组)break 出口打开,值流向环外的下游

也就是说:"继续循环"时,通往环外的 break 出口反而是被排除的。只有真正 break 时,环外的下游才第一次拿到真数据、开始执行。循环内部的 output1..N 端口则相反——continue 时吐真数据(喂给下一圈),break 时吐排除值(LoopControllerNode.ts:244)。

7.2 'loop-not-broken':防死锁的特殊标签​

为什么排除值要带 'loop-not-broken' 这个标签,而不是普通的 undefined?

因为环外的下游节点这时输入里带着 break 端口的排除值——按 §3.3 它们本该被永久判死。但循环还没跑完,它们其实是"还没轮到",不是"永远不执行"。这两种状态必须区分开,否则下游会被过早排除、循环结果永远流不出去。

引擎用这个标签做区分。真源码 GraphProcessor.ts:1079,在调度每个节点前:

if (this.#excludedDueToControlFlow(node, inputValues, nanoid() as ProcessId, 'loop-not-broken')) {
  this.#emitTraceEvent(`Node ${node.title} is excluded due to control flow`);
  return;   // 只是"这轮先不跑",不写进 nodeResults,不判死
}

配合关卡里的 isWaitingForLoop 判断(GraphProcessor.ts:1720):

const isWaitingForLoop = controlFlowExcludedValues.some((value) => value?.[1]?.value === 'loop-not-broken');

当毒是 'loop-not-broken' 时,节点只是安静地跳过本轮、不被 #markAsExcluded 判死(GraphProcessor.ts:1735 的 if (!isWaitingForLoop) 守卫),等循环真正 break 后再执行。这就是"将不执行"和"尚未执行"的区别——注释在 GraphProcessor.ts:1076 写得明明白白:difference between "will not execute" and "has not executed yet"。

7.3 每轮清空环内节点:让"只跑一次"的节点重复跑​

引擎默认每个节点跑一次(#visitedNodes 记录已跑)。循环要让环内节点跑很多次,靠的是每转一圈就把环内节点从"已访问"里删掉。

真源码 GraphProcessor.ts:1156(loopController 处理完之后):

if (!didBreak) {
  for (const loopNodeId of attachedData.loopInfo?.nodes ?? []) {
    const cycleNode = this.#nodesById[loopNodeId]!;
    this.#visitedNodes.delete(cycleNode.id);       // 清"已访问"
    this.#currentlyProcessing.delete(cycleNode.id);
    this.#remainingNodes.add(cycleNode.id);        // 重新排队
    this.#nodeResults.delete(cycleNode.id);        // 清缓存结果
  }
}

didBreak 的判断(GraphProcessor.ts:1151)正是看 break 端口是不是那个 'loop-not-broken' 排除值——没 break 就把整个环重置,让下一圈重新算。哪些节点算"环内"?靠 loopInfo.nodes 这个集合,节点在处理时把自己加进去(GraphProcessor.ts:1123)。

7.4 谁跟谁在同一个环:Tarjan SCC​

引擎怎么知道两个节点在同一个循环里?启动时用 Tarjan 强连通分量算法(#tarjanSCC,GraphProcessor.ts:1851)把图里所有的环预先算出来。之后一个查询函数即可判断:

真源码 GraphProcessor.ts:1901:

#nodesAreInSameCycle(a: NodeId, b: NodeId) {
  return this.#scc.find((cycle) => cycle.find((node) => node.id === a) && cycle.find((node) => node.id === b));
}

这个判断用在调度逻辑里(GraphProcessor.ts:1094):loopController 第一圈时,允许同环的输入节点还没处理完就先跑控制器(否则环会互相死等),但第二圈起就必须正常等待。这是打破"环内互相依赖"死锁的关键一手。

7.5 maxIterations 保护​

循环必须能防跑飞。控制器有 maxIterations(默认 100)和到达上限时的动作 atMaxIterationsAction:

真源码 LoopControllerNode.ts:208:

if (iterationCount >= (this.data.maxIterations ?? 100) && this.data.atMaxIterationsAction !== 'break') {
  throw new Error(`Loop controller exceeded max iterations of ${this.data.maxIterations ?? 100}`);
}

超限要么抛错中止整个图,要么(若配置为 'break')强制把 continueValue 置假、干净跳出(LoopControllerNode.ts:227)。另外引擎在 GraphProcessor.ts:1195 还会检测并拒绝嵌套循环(Nested loops are not supported)。

8. 竞速与并行:让最快的赢,中止落败者​

8.1 raceInputs:第一个到达的胜出​

RaceInputsNode 取"多个输入里第一个完成的",其余取消。它的 process 极简(RaceInputsNode.ts:94)——因为竞速的重活全在引擎里,注释直说 GraphProcessor handles most of the racing/aborting logic for us:

const value = Object.entries(inputs).find(
  ([key, value]) => key.startsWith('input') && value !== undefined && value.type !== 'control-flow-excluded',
);

它只是挑出第一个"非排除值"的输入。真正的"标记赢家、中止输家"由引擎完成。

8.2 引擎侧:打标记 + abort 落败分支​

竞速靠附着数据 AttachedNodeData.races 跟踪。当一个节点是 raceInputs(或已被打上 race 标记)时,引擎给它上游每个输入节点贴上 raceIds 标签(GraphProcessor.ts:985):

inputNodeAttachedData.races = { propagate: false, raceIds: [...raceIds], completed: false };

一旦 raceInputs 跑完(有赢家),引擎立刻中止所有同 race 的分支——真源码 GraphProcessor.ts:1171:

if (node.type === 'raceInputs') {
  const allNodesForRace = [...this.#attachedNodeData.entries()].filter(([, { races }]) =>
    races?.raceIds.includes(`race-${node.id}` as RaceId));
  for (const [nodeId] of allNodesForRace) {
    for (const [key, abortController] of this.#nodeAbortControllers.entries()) {
      if (key.startsWith(nodeId)) {
        abortController.abort();      // 中止还没跑完的落败分支
      }
    }
    // Mark every attached data as completed for the race
    // ... races.completed = true
  }
}

每个节点执行时都注册了自己的 AbortController(GraphProcessor.ts:1494),所以这里能精确 abort 掉正在跑(比如正在等 LLM 返回)的落败分支。已经跑完的不受影响,还没开始的则因为 races.completed 被标记为 true 而在调度时直接跳过(GraphProcessor.ts:1127)。

8.3 Split run:一个节点扇出成数组并行​

这是另一种并行:node.isSplitRun 让一个节点把数组输入拆开、对每个元素并行跑一遍。调度入口 GraphProcessor.ts:1282:

if (node.isSplitRun) {
  await this.#processSplitRunNode(node, processId);
}

#processSplitRunNode(GraphProcessor.ts:1291)先算扇出数量(取各数组输入的最大长度,封顶 splitRunMax 默认 10,GraphProcessor.ts:1298),然后:

  • 默认并行:Promise.all 同时跑所有分片(GraphProcessor.ts:1354);
  • 若 isSplitSequential 则串行逐个跑(GraphProcessor.ts:1320)。

最后把 N 个结果聚合回数组(GraphProcessor.ts:1397,把 Record<Port, DataValue[]> 合成 Record<Port, ArrayDataValue>)。注意它开头也过了一遍 #excludedDueToControlFlow(GraphProcessor.ts:1294)——排除值机制对 split run 同样生效。

9. 三种控制流,同一个机制​

回头看,分支、循环、竞速这三件看似不同的事,底层都是同一滴毒:

控制流怎么用排除值实现关键代码
分支(if / ifElse / match)没走的那条腿吐排除值,下游整片被传染关闭IfNode.ts:157、MatchNode.ts:148
短路(coalesce)跳过排除值输入,取第一个有效值;全排除则自己也排除CoalesceNode.ts:114
循环(loopController)用带 'loop-not-broken' 标签的排除值区分"还在转/已跳出",每轮清空环内节点LoopControllerNode.ts:231、GraphProcessor.ts:1156
竞速(raceInputs)落败分支是排除值;赢家出现后 abort 其余分支RaceInputsNode.ts:96、GraphProcessor.ts:1176

这就是本章的精华: Rivet 没有为控制流建第二套系统。它只是发明了一种"表示此路不通"的数据值,让它沿着已有的数据边传染,再指定极少数节点能拦截它——分支、循环、竞速全部涌现自这一个设计。这是"用数据表达控制"的一个漂亮范例。

10. 边界与坑​

  • 不支持嵌套循环。 引擎显式拒绝(GraphProcessor.ts:1195 抛 Nested loops are not supported)。
  • 循环里 loopInfo.nodes 是"可能在环里跑的所有节点",清空时会全部重置(GraphProcessor.ts:1158 注释:clear every node that _potentially_ could run in the loop)。
  • loopController 判死时必须走 break 特例(GraphProcessor.ts:1759),否则一个被排除的控制器会死循环——这是注释里点名的防护。
  • split run 封顶 10(splitRunMax ?? 10,GraphProcessor.ts:1300),超出的数组元素被截断。
  • didBreak 那行用了 ?? 而非 ||(GraphProcessor.ts:1153),逻辑略绕,读源码时留意其真值语义。

11. 横向对比与延伸​

本章是 Rivet 系列的"最精妙处",建议连着这几章一起读:

对比同类:多数可视化工作流引擎(如 n8n、传统 BPMN 引擎)会显式建控制流——单独的分支节点、循环块、专门的执行栈。Rivet 反其道而行:控制流是数据流的一种特例,靠一个哨兵值涌现,引擎主体始终只做一件事——传数据、调度输入齐了的节点。

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

主题文件路径符号名
排除值类型定义packages/core/src/model/DataValue.tsControlFlowExcludedDataValue(含 'loop-not-broken')
排除关卡(三条规则)packages/core/src/model/GraphProcessor.ts#excludedDueToControlFlow
解毒白名单packages/core/src/model/GraphProcessor.tsnodesAllowedToConsumeExcludedValue
判死并灌毒下游packages/core/src/model/GraphProcessor.ts#markAsExcluded
循环重置环内节点packages/core/src/model/GraphProcessor.ts#processNodeIfAllInputsAvailable(loopController 段,didBreak / #visitedNodes.delete)
竞速中止落败分支packages/core/src/model/GraphProcessor.ts#nodeAbortControllers(raceInputs 段的 abort())
split run 扇出packages/core/src/model/GraphProcessor.ts#processSplitRunNode
同环判定 / SCCpackages/core/src/model/GraphProcessor.ts#nodesAreInSameCycle、#tarjanSCC
附着数据结构packages/core/src/model/GraphProcessor.tsAttachedNodeData、LoopInfo、RaceId
If 分支packages/core/src/model/nodes/IfNode.tsIfNodeImpl.process
IfElse 分支packages/core/src/model/nodes/IfElseNode.tsIfElseNodeImpl.process
Coalesce 短路packages/core/src/model/nodes/CoalesceNode.tsCoalesceNodeImpl.process
Match 多路分支packages/core/src/model/nodes/MatchNode.tsMatchNodeImpl.process
循环控制器packages/core/src/model/nodes/LoopControllerNode.tsLoopControllerNodeImpl.process(continue/break/maxIterations)
竞速节点packages/core/src/model/nodes/RaceInputsNode.tsRaceInputsNodeImpl.process