数据截至 (上游 commit 7a975c596eca)
环、循环与条件路由 — 这张图为什么不是 DAG
30 秒导读: 前一章(调度引擎)讲的是"按拓扑序把顶点一个个跑完"。这章讲 Langflow 在这套 DAG 调度之上多加的两件事——允许图里有环(节点可以回跳到上游重跑)和允许运行期剪枝 (一个 If-Else 判完之后,没选中的那半张图整个不跑)。这两件事各自都会打破"所有前驱跑完才能跑我" 这条 DAG 铁律,所以引擎里为它们准备了单独的判据和单独的状态机。
1. 这节讲什么:普通 DAG 引擎缺的两样东西
一个普通的 DAG(有向无环图)工作流引擎有两条硬规矩:
- 不许有环——有环就没法拓扑排序,调度器不知道谁先谁后。
- 顶点全跑——图上画了的节点,只要前驱跑完了就得跑。
画布上的用户偏偏想要这两件被禁止的事:
| 用户想要的 | 画布上长什么样 | 打破了哪条规矩 |
|---|---|---|
| "答案不满意就回去重问一 次" | 下游节点连回上游节点,形成回边 | 不许有环 |
| "条件为真走这半边,为假走那半边" | 一个 If-Else 分出两条支路,只该跑一条 | 顶点全跑 |
| "对列表里每一项都跑一遍这段流程" | Loop 的 item 输出接一段子流程再接回来 | 两条都破 |
Langflow 的做法不是"把环拆掉再当 DAG 跑",而是保留环、给环上的顶点一套单独的放行规则; 不是"跑完再丢弃没用的结果",而是在调度阶段就把不该跑的分支标记成不可运行。
一句话直觉: 把 DAG 引擎想成"红绿灯按拓扑序依次放行"。Langflow 干了两件事——给环上的路口 换了一套放行逻辑(否则环上每个路口都在等对面先走,谁也走不了),再给分支路口装了可以临时封路的路障。
2. 顶层全景:一次带环带分支的运行
先看一张最小的、同时有环和有分支的流:一个 If-Else 判断结果够不够好,不够好就把消息拼一拼再回去重判, 够好了才往输出走。
怎么读这张图: 实线是普通数据流,↰ 那条是回边(构成环);虚线框里是被剪掉的分支。从左往右是主流向。
┌──────────────┐
ChatInput ───────▶ │ If-Else │ ──true──▶ TextOutput ──▶ ChatOutput
▲ │ (路由顶点) │
│ └──────┬───────┘
│ │ false
│ ▼
│ ┌─────────────┐
└──── ↰ ───────│ Concatenate │ ← 这条回边让 ChatInput / Concatenate /
回边 └─────────────┘ If-Else 三个顶点同处一个环
每一轮判断结束时,引擎要同时做两件事:
- 剪掉没选中的那半边(true 走了就把 false 那支封掉,反之亦然);
- 决定环要不要再转一圈(false 分支通了,就意味着要回到 ChatInput 重跑)。
这两件事在代码里由两套完全独立的机制完成。下面这张表是本章的骨架,后面每一节展开其中一 格:
| 关切 | 引擎里的机制 | 核心符号 | 生命周期 |
|---|---|---|---|
| 谁在环上 | 强连通分量检测 | find_cycle_vertices | 建图时算一次并缓存 |
| 环上顶点何时能跑 | 两套前驱判据 | are_all_predecessors_fulfilled | 每次询问时现算 |
| 环怎么停下来 | 每顶点产出计数 + 上限 | should_continue / yielded_counts | 整个 run 期间累加 |
| 分支封路(服务于环) | ACTIVE / INACTIVE 顶点状态 | mark_branch | 每步结束即重置 |
| 分支封路(服务于路由) | 条件排除集合 | conditionally_excluded_vertices | 由源顶点持有,直到它重新判定 |
最容易踩的坑就在最后两行:Langflow 里并存着两套剪枝状态机,它们看起来在做同一件事, 但生命周期完全相反。第 5 节专门讲这个。
3. 环:怎么识别、怎么标记
3.1 谁在环上 —— 强连通分量
判断"图里有没有环"和"具体哪些顶点在环上"是两个问题,Langflow 的工具箱里四个函数各 管一摊:
| 函数 | 回答什么问题 | 算法 | 位置 |
|---|---|---|---|
has_cycle | 有没有环(布尔) | DFS + 递归栈 | utils.py:408 |
find_cycle_edge | 第一条造成环的边 | DFS,命中即返回 | utils.py:444 |
find_all_cycle_edges | 所有造成环的边 | DFS,全部收集 | utils.py:481 |
find_cycle_vertices | 哪些顶点在环上 | networkx 强连通分量 | utils.py:524 |
真正被运行期依赖的是最后一个。它不用 DFS,而是借 networkx 求强连通分量(SCC,一组互相都能到达的
顶点),分量里顶点多于一个、或者有自环,就整组算作环上顶点
(src/lfx/src/lfx/graph/graph/utils.py:531-533,find_cycle_vertices):
for component in nx.strongly_connected_components(graph):
if len(component) > 1 or graph.has_edge(tuple(component)[0], tuple(component)[0]):
cycle_vertices.update(component)
为什么用 SCC 而不是 DFS? DFS 找到的是"回边"(哪条边造成了环),但运行期真正需要的是"这个顶点是不是 在环里"——因为放行判据是按顶点问的。SCC 天然给出的就是顶点集合,而且不依赖从哪个入口开始遍历。
Graph 上三个属性把这些包成缓存:
| 属性 | 内容 | 位置 |
|---|---|---|
Graph.cycle_vertices | 环上顶点集合(懒算 + 缓存) | base.py:2204-2209 |
Graph.is_cyclic | 就是 bool(self.cycle_vertices) | base.py:645-653 |
Graph.cycles | 造成环的边列表,需要 _start 才算 | base.py:2193-2202 |
注意 cycles 和 cycle_vertices 用的不是同一套算法:前者调 find_all_cycle_edges,必须有起点顶点
self._start,没有就返回空列表;后者不需要起点。所以从 JSON 载入的流(没有显式 start/end)上
graph.cycles 恒为空,但 graph.cycle_vertices 照常工作——运行期依赖的是后者。
缓存会在 add_nodes_and_edges 里被两次清空(base.py:276-277、base.py:282-284),因为图结构变了
环也就变了。
3.2 环上的边换一个类:CycleEdge
建边的时候,只要两端有任意一端落在 cycle_vertices 里,这条边就不是普通 Edge 而是 CycleEdge
(src/lfx/src/lfx/graph/graph/base.py:2612-2615,Graph.build_edge):
if any(v in self.cycle_vertices for v in [source.id, target.id]):
new_edge: CycleEdge | Edge = CycleEdge(source, target, edge)
else:
new_edge = Edge(source, target, edge)
两个类的差别很小但很关键:
Edge | CycleEdge | |
|---|---|---|
is_cycle | False(edge/base.py:52) | True(edge/base.py:291) |
| 给两端顶点打标 | 无 | source.has_cycle_edges = True,target 同样(edge/base.py:292-293) |
| 额外能力 | 无 | honor() / get_result_from_source()(edge/base.py:295-332) |
CycleEdge.honor 做的事是"履约":把已经建好的源顶点的结果塞进目标顶点的 params,并置 is_fulfilled
(src/lfx/src/lfx/graph/edge/base.py:295-318)。它明确拒绝在这里触发构建——源没 built 就直接抛错,
注释写得很直白:这条路径必须是只读的。
is_cycle 这个标志还有一个下游用处:当某个前驱始终没建成时,ComponentVertex._get_result 会沿着
cycle 边去取模板默认值,而不是报错(见第 6 节)。
3.3 环上的顶点强制关缓存
环的意义就是"同一个顶点跑第二遍要得到新结果",所以建图时会把环上所有顶点的输出缓存关掉
(src/lfx/src/lfx/graph/graph/base.py:1886-1892,Graph._set_cache_to_vertices_in_cycle):
cycle_vertices = set(find_cycle_vertices(edges))
for vertex in self.vertices:
if vertex.id in cycle_vertices:
vertex.apply_on_outputs(lambda output_object: setattr(output_object, "cache", False))
它在 _build_graph 里被调用(base.py:1487),紧接着的循环把环上顶点登记进
run_manager.cycle_vertices(base.py:1489-1491),prepare() 里还会再登记一次首层里的环上顶点
(base.py:2301-2304)。调度器判断"这个顶点在不在环上"读的是 run_manager.cycle_vertices 这份副本,
不是 Graph.cycle_vertices——两者由上面这些调用保持同步。
顺带一提,冻结(frozen)机制也给环开了口子:Loop 类顶点即使被冻结也必须重建
(base.py:1717-1718,is_loop_component = vertex.display_name == "Loop" or vertex.is_loop)。
3.4 拓扑排序怎么给环找一个入口
环上每个顶点的入度都 ≥ 1,标准的 Kahn 算法开局就找不到"入度为 0"的顶点,队列是空的。
layered_topological_sort 对这种情况有专门分支(src/lfx/src/lfx/graph/graph/utils.py:568-585):
is_cyclic 且所有顶点入度 > 0
│
├── 有 start_id ──────────▶ 队列 = [start_id],并把它的入度强行改成 0
│
└── 没有 start_id ────────▶ find_start_component_id 找 webhook/chat 类输入顶点
找不到就随便挑一个顶点当入口
排序阶段还允许环上顶点在层里重复出现最多两次——常量 MAX_CYCLE_APPEARANCES = 2
(utils.py:11),判据在 utils.py:645 和 utils.py:670-673。这只是给运行期铺个路,真正决定跑多少圈的
是运行期的判据和止损计数,不是这份静态分层。分层本身的细节见 调度引擎。
3.5 止损:环靠什么停下来
Langflow 不做静态的循环次数分析,它靠一个朴素的运行期计数器兜底。
async_start 每 yield 一个结果就给该顶点的计数加一(src/lfx/src/lfx/graph/graph/base.py:496-506):
yielded_counts: dict[str, int] = defaultdict(int)
while should_continue(yielded_counts, max_iterations):
result = await self.astep(...)
yield result
if isinstance(result, Finish):
return
if hasattr(result, "vertex"):
yielded_counts[result.vertex.id] += 1
should_continue 的判据是"产出次数最多的那个顶点还没超过上限"
(src/lfx/src/lfx/graph/graph/utils.py:518-521):
def should_continue(yielded_counts: dict[str, int], max_iterations: int | None) -> bool:
if max_iterations is None:
return True
return max(yielded_counts.values(), default=0) <= max_iterations
三个容易搞错的细节:
-
正常结束走的是
return,不是循环条件。 队列空了astep返回Finish,生成器直接 return (base.py:401-402)。只有真的转超了才会走到循环外面那句raise ValueError("Max iterations reached")(base.py:420-421)。 -
计数是"每顶点"的,不是"总步数"。 一条长直线流跑 50 个不同顶点,每个计数都是 1,不会触发上限。
-
同步
start()强制要求max_iterations,异步async_start()不要求。 只有Graph.start里有这道前置检查(base.py:464-466):if self.is_cyclic and max_iterations is None:msg = "You must specify a max_iterations if the graph is cyclic"raise ValueError(msg)也就是说,直接调
async_start且不传max_iterations的环流,理论上会无限转——should_continue在max_iterations is None时恒为True。真实的兜底通常来自组件层自己的max_iterations(见 7.1 的 If-Else)。
回归测试直接锁住了这个行为:src/lfx/tests/unit/graph/graph/test_cycles.py:114-115 用
max_iterations=2 跑一个必然转不完的环,断言抛出 Max iterations reached。
4. 环上顶点的放行判据(本章最硬的一段)
4.1 问题:DAG 判据在环上必然死锁
DAG 调度器的放行条件是"我的所有前驱都跑完了"。把它套到环上:
A ──▶ B ──▶ C
▲ │
└──── 回边 ───┘
A 等 C 跑完 C 等 B 跑完 B 等 A 跑完 → 谁也跑不了
所以环上顶点必须有一套不同的判据。这套判据全部集中在
RunnableVerticesManager.are_all_predecessors_fulfilled
(src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py:67-100)。
4.2 三种情形,三条判据
先看总表,再看代码:
| 顶点情形 | 还有待清前驱时的判据 | 直觉 |
|---|---|---|
| 不在环上 | 直接 False,老老实实等 | 普通 DAG 行为 |
在环上 · 首轮(没进过 ran_at_least_once) | is_loop 且 待清前驱全部落在环内 | 只对循环组件开这个口子,且只允许它无视"环内互等" |
| 在环上 · 已跑过至少一次 | 严格等到待清前驱列表清空 | 第二圈起回到 DAG 的严格语义 |
源码(runnable_vertices_manager.py:82-100,注释已略):
pending = self.run_predecessors.get(vertex_id, [])
if not pending:
return True # 前驱清空 → 任何顶点都放行
if vertex_id in self.cycle_vertices:
pending_set = set(pending)
running_predecessors = pending_set & self.vertices_being_run
if vertex_id in self.ran_at_least_once:
return not (pending_set or running_predecessors)
return is_loop and pending_set <= self.cycle_vertices
return False
关于 ran_at_least_once 那一支的一个精确读法: 走到那行时 pending 必然非空(空的已经在上面返回
True 了),所以 pending_set 恒为真值,not (pending_set or ...) 恒为 False。
换句话说,一个跑过一遍的环上顶点,唯一的再次放行途径是让它的 run_predecessors 列表真的被清空,
走的是最上面那个早返回。running_predecessors 这个中间变量对结果没有影响。
那"清空"由谁做?两条路:
- 前驱正常跑完 →
remove_vertex_from_runnables→remove_from_predecessors(runnable_vertices_manager.py:102-107),把自己从各个后继的待清列表里划掉; - 前驱被标成 INACTIVE →
Graph.mark_vertex也会调同一个方法(base.py:982-987)。 这就是剪枝和环的接合点:封掉一条支路,等于替下游把这个前驱"划掉",下游因此能跑。
命名陷阱:
remove_from_predecessors(v)读起来像"从 v 的前驱里删",实际是"把 v 从它后继们的 待清前驱列表里删"——因为run_map存的是前驱 → [后继...](runnable_vertices_manager.py:109-116)。
4.3 is_loop 是从哪来的
首轮判据里的 is_loop 是顶点级属性,不是"这个顶点在环上"的意思。它的定义只有三行
(src/lfx/src/lfx/graph/vertex/base.py:126-131,Vertex.is_loop):
@property
def is_loop(self) -> bool:
if self._is_loop is None:
self._is_loop = any(output.get("allows_loop", False) for output in self.outputs)
return self._is_loop
追溯链条:
组件的 Output(..., allows_loop=True) ← 组件作者声明
│ Output 字段定义: template/field/base.py:206
▼
顶点的 outputs 字典里带 allows_loop
│
▼
Vertex.is_loop == True ← vertex/base.py:124
│
▼
Graph.is_vertex_runnable 把它传下去 ← base.py:2436
│
▼
are_all_predecessors_fulfilled 首轮放行 ← runnable_vertices_manager.py:99
全仓只有一个组件设了 allows_loop=True:Loop 组件的 item 输出
(src/lfx/src/lfx/components/flow_controls/loop.py:36-45)。所以首轮那条特权规则实际上是给 Loop
量身定做的:Loop 的 item 输出接出去一段子流程又接回它自己的 item 输入,它的待清 前驱全在环内,
按 DAG 规矩永远起不来,只能靠这条特权先跑起第一圈。
4.4 关键细节:今天的 Loop 其实不靠外层环转圈
读到这里很容易以为"Loop 就是靠上面那条特权在外层图里转 N 圈"。不是。
LoopComponent._iterate 走的是子图路线:它先用图遍历圈出 loop body 的顶点集合,再对数据列表里的
每一项开一个隔离子图跑一遍,把结果聚合起来(src/lfx/src/lfx/base/flow_controls/loop_utils.py:231-256,
execute_loop_body → graph.create_subgraph,后者定义在 base.py:2516)。
外层图这边,Loop 组件做的是把 item 分支封掉
(src/lfx/src/lfx/components/flow_controls/loop.py:252,item_output 的第一行 self.stop("item")),
免得外层调度器又把 loop body 跑一遍。
所以现状是:
| 图的静态形态 | 运行期实际行为 | |
|---|---|---|
| Loop + body | 确实成环,body 顶点都进 cycle_vertices | 迭代在子图里发生,外层 item 分支被剪掉 |
| If-Else 回边 | 成环 | 真的在外层图里一圈圈转 |
is_loop 特权和 CycleEdge、关缓存这些机制仍然对 Loop 生效(图的形态没变),但"转圈"这件事的重心
已经从外层调度器挪到了子图执行器。这是读代码时最 容易误判的一处。
5. 条件剪枝:两套并存的状态机
5.1 先看差异表
Langflow 里有两套"把分支关掉"的机制,它们同时存在、同时被 If-Else 调用。搞混这两套是理解本章 的最大障碍,所以先摆表:
| ACTIVE / INACTIVE | conditionally_excluded_vertices | |
|---|---|---|
| 存在哪 | 每个 Vertex.state(vertex/base.py:41-46) | Graph 上的一个集合(base.py:119) |
| 谁设置 | Graph.mark_branch / mark_vertex | Graph.exclude_branch(es)_conditionally |
| 谁清除 | 每次 astep 结尾自动重置(base.py:1635-1636) | 只有当初那个源顶点重新判定时(base.py:1082-1083) |
| 被谁读 | Vertex.is_active() → run_manager.is_vertex_runnable | Graph.is_vertex_runnable 的第一道检查(base.py:2433) |
| 服务于 | 环:让分支下一圈能重新活过来 | 分支路由:让没选中的那半边永久不跑 |
| 副作用 | 会调 remove_from_predecessors,替下游清前驱 | 无副作用, 纯查表 |
为什么必须两套? 因为二者的需求直接冲突:
环的需求: 这一步封掉的分支,下一圈必须能重新打开
↓ 否则环只能转一圈
→ 必须"每步重置"
路由的需求: 判完 false 之后,true 那半边整个 run 都不该再跑
↓ 否则重置一次它就复活了
→ 必须"持久保留"
一套状态机做不到既每步重置又持久保留,所以就并存了两套。SmartRouterComponent 的文档字符串把这个
bug 讲得很明白(src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py:267-270):
INACTIVE 会在调度轮次之间被重置,少了持久排除,一个"复活"的未选中分支只要在下游某个汇合节点重新会师,
就会被捡起来执行。
5.2 第一套:ACTIVE / INACTIVE
状态定义只有三个值(src/lfx/src/lfx/graph/vertex/base.py:41-46,VertexStates):ACTIVE、
INACTIVE、ERROR。Vertex.is_active() 就是 state == ACTIVE(vertex/base.py:158-159)。
入口是 Graph.mark_branch(base.py:1050-1073),组件层通过 Component.stop(output_name) 调到它
(src/lfx/src/lfx/custom/custom_component/custom_component.py:141-154;对应的 start() 是反向操作,
标 ACTIVE)。整条链路:
组件调 self.stop("false_result")
│
▼
custom_component.py:151 graph.mark_branch(vertex_id, output_name, "INACTIVE")
│
├─ (a) 算出"受保护顶点" _get_vertices_reachable_from_other_outputs base.py:1015
│
├─ (b) 沿该 output 递归下钻 _mark_branch → mark_vertex → vertex.set_state base.py:1021 / 982
│
└─ (c) 重建被访问顶点的前驱表并交给 run_manager base.py:1062-1073
(a) 受保护顶点——为什么需要它。 如果 If-Else 的 true 支和 false 支在下游汇合到同一个节点(比如一个
合并组件),沿 false 支下钻会一路把那个汇合节点也标成 INACTIVE,选中的 true 支就白跑了。
_get_vertices_reachable_from_other_outputs 先算出"从其他输出也能到达的顶点集合"
(base.py:1015-1019),_mark_branch 遇到集合里的顶点就跳过(base.py:1033-1034):
if state == VertexStates.INACTIVE and vertex_id in (protected_vertices or set()):
return visited
还有一层保险在 Vertex.set_state(vertex/base.py:149-156):只有入度 ≤ 1 的顶点被标 INACTIVE 时
才会登记进 graph.inactivated_vertices,注释直说了理由——入度 > 1 说明它是个汇合点。
副作用是:汇合点即使被标了 INACTIVE 也进不了 inactivated_vertices,因而不会被 reset_inactivated_vertices
恢复成 ACTIVE,这正是 (a) 那层保护必须存在的原因 (inferred)。
(b) 递归下钻。 _mark_branch(base.py:1021-1048)有个容易看漏的细节:第一次调用不标记源顶点自己——
visited is None 这一支只建集合、不 mark_vertex。也就是说 stop("false_result") 封的是 false 那条支路
上的下游顶点,路由器本身仍是 ACTIVE。往下钻时按 output_name 过滤边(base.py:1043-1046),只走
该输出对应的那些子节点。
(c) 重建前驱表。 标完之后重算前驱映射并只保留刚访问过的顶点;如果源顶点在环上,还要再滤一道
(base.py:1064-1069):
if vertex_id in self.cycle_vertices:
new_predecessor_map = {
k: [dep for dep in v if dep in self.cycle_vertices and dep in self.run_manager.ran_at_least_once]
for k, v in new_predecessor_map.items()
}
白话:环上重新洗牌依赖时,只保留"同在环内、且已经跑过至少一次"的依赖,环外的依赖和还没跑过的 环内依赖统统丢掉。这样下一圈才不会被那些不会再产生新值的旧依赖卡住。
重置在哪。 astep 每处理完一个顶点,把下一批可运行顶点塞进队列之后,立刻清空两组标记
(src/lfx/src/lfx/graph/graph/base.py:2004-2006):
self.extend_run_queue(next_runnable_vertices)
self.reset_inactivated_vertices()
self.reset_activated_vertices()
reset_inactivated_vertices(base.py:970-975)把 inactivated_vertices 里每个顶点标回 ACTIVE。
所以准确说法是"每一步结束就重置",不是"每一轮"。INACTIVE 的有效期只覆盖当前这一步里
get_next_runnable_vertices 的计算,用完即弃。
另外 sort_vertices 开头还有一次全局重置:self.mark_all_vertices("ACTIVE")(base.py:2376,
方法定义在 base.py:977-980),每次 prepare() 都会执行。
状态顶点的反向操作。 activate_state_vertices(name, caller)(base.py:563-611)是这套机制的
"唤醒"方向:Notify/Listen 这类 StateVertex 按 context_key 匹配上以后,把自己、所有后继、以及这些
后继的前驱统统从 INACTIVE 拉回 ACTIVE(base.py:595-596),再把新算出的前驱表并进 run_manager。
被唤醒的顶点 id 存在 self.activated_vertices 里,get_next_runnable_vertices 在源顶点是状态顶点时
会把它们追加进下一批(base.py:1927-1928),然后同样在 astep 结尾被
reset_activated_vertices(base.py:613-615)清空。
5.3 第二套:conditionally_excluded_vertices
两个字段(src/lfx/src/lfx/graph/graph/base.py:166-168):
self.conditionally_excluded_vertices: set = set() # 被排除的顶点
self.conditional_exclusion_sources: dict[str, set[str]] = {} # 源顶点 → 它排除掉的那批
第二个字段是这套机制的关键:排除是记在源顶点名下的,谁排的谁负责撤。
_replace_conditional_exclusions(base.py:1075-1086)是唯一的写入口,永远是"先撤销该源上次排的,
再记这次的":
if vertex_id in self.conditional_exclusion_sources:
self.conditionally_excluded_vertices -= self.conditional_exclusion_sources.pop(vertex_id)
self.conditionally_excluded_vertices.update(excluded)
if excluded:
self.conditional_exclusion_sources[vertex_id] = excluded
这正是"持久,但可重新判定"的实现:环转到第二圈,同一个 If-Else 再次执行,它上一圈排掉的分支会被 它自己先撤销,再按新的条件重排。别的顶点动不了它。
对外的两个方法分工如下:
| 方法 | 用途 | 位置 |
|---|---|---|
exclude_branch_conditionally(vertex_id, output_name=None) | 排除一个输出分支;output_name 为空则排掉全部下游 | base.py:1088-1113 |
exclude_branches_conditionally(vertex_id, output_names) | 一次排除多个输出分支,累积到同一个源键下 | base.py:1115-1138 |
前者在传了 output_name 时直接委托给后者(base.py:1105-1110),注释讲了理由:"保留共享下游节点"
的逻辑只想写一遍。多路路由器(Smart Router)必须用后者,因为它要一次排掉 N-1 条分支,
用单支版本会互相清掉。
两个私有工具:
_collect_branch_vertices(vertex_id, output_names=None)(base.py:996-1013)——从源顶点出发做 DFS,只在第一跳按output_names过滤边,之后一路收全部后代;visited里预置了源顶点自己, 所以源不会被排除,也天然能在环上终止。_get_vertices_reachable_from_other_outputs(vertex_id, output_names)(base.py:1015-1019)—— 算出"从这个源的其他输出也能到达的顶点",在exclude_branches_conditionally里从排除集里减掉 (base.py:1137)。这就是"两支汇合的合并节点不该被排除"的保障,和 5.2 里 ACTIVE/INACTIVE 的protected_vertices是同一个思路的两次实现。
读取只有一处,而且在最前面(src/lfx/src/lfx/graph/graph/base.py:2808-2815,Graph.is_vertex_runnable):
if vertex_id in self.conditionally_excluded_vertices:
return False
is_active = self.get_vertex(vertex_id).is_active()
is_loop = self.get_vertex(vertex_id).is_loop
return self.run_manager.is_vertex_runnable(vertex_id, is_active=is_active, is_loop=is_loop)
条件排除是第一道闸,直接短路,连 run_manager 都不问。
前端那边,两套状态在发 SSE 事件时被合并成一个列表给 UI 打灰(base.py:2040):
inactivated = list(self.inactivated_vertices.union(self.conditionally_excluded_vertices))
6. 被剪掉的上游,下游怎么拿值
剪枝制造了一个新问题:下游节点的某个输入连着一个永远不会构建的上游。如果照常去拉结果, 会撞上 "has not been built yet" 的错误;如果放着不管,下游也跑不起来。Langflow 分两种情形处理。
6.1 单值输入:返回该输入的模板默认值
ComponentVertex._get_result(src/lfx/src/lfx/graph/vertex/vertex_types.py:93)在顶点没 built 时,
第一件事就是查条件排除集(vertex_types.py:108-109):
if self.id in self.graph.conditionally_excluded_vertices and target_handle_name:
return requester.get_value_from_template_dict(target_handle_name)
requester 是来拉值的下游顶 点,target_handle_name 是它正在读的那个输入名。返回的是下游自己模板里
这个输入的默认值——比如一个空 Message。注释里点明了两件事:这条路径上 target_handle_name 一定有值;
以及这里绝不会触发被排除顶点的构建,因为 is_vertex_runnable 早就把它拦下了。
紧跟着的是环的兜底路径(vertex_types.py:111-121):如果不是条件排除,就看有没有 is_cycle 的边,
有的话同样退回默认值(目标参数恰好是 requester 的某个输出名时退成 None)。两条兜底都没命中才抛
"has not been built yet"。
6.2 列表输入:直接跳过,不补默认值
如果下游的这个输入是列表(多个上游汇进同一个 handle),补默认值就错了——会在真实分支的结果旁边
塞进一个空元素。所以 _build_list_of_vertices_and_update_params 选择跳过
(src/lfx/src/lfx/graph/vertex/base.py:683-685):
if not vertex.built and vertex.id in self.graph.conditionally_excluded_vertices:
continue
result = await vertex.get_result(self, target_handle_name=key)
两种处理并排看:
| 输入形态 | 被排除的上游贡献什么 | 位置 |
|---|---|---|
| 单值 handle | 该输入的模板默认值 | vertex_types.py:108-109 |
| 列表 handle | 什么都不贡献(跳过这一项) | vertex/base.py:683-684 |
| 单值 + 环边(非条件排除) | cycle 边的默认值 / None | vertex_types.py:111-121 |
7. 真实使用方:谁在调这些接口
7.1 If-Else(ConditionalRouterComponent)
文件在 src/lfx/src/lfx/components/flow_controls/conditional_router.py
(lfx.components.logic.conditional_router 是向后兼容别名,见 components/logic/__init__.py)。
它有两个输出 true_result / false_result(conditional_router.py:87-90)。
核心方法是 iterate_and_stop_once(route_to_stop)(conditional_router.py:131-166),它是本章两套机制
唯一一处并排调用的地方:
# 1. stop() → ACTIVE/INACTIVE,服务于环(每步重置)
self.stop(route_to_stop)
# 2. 条件排除,服务于路由(持久保留)
self.graph.exclude_branch_conditionally(self._id, output_name=route_to_stop)
组件自己也有一个环的止损,而且比 max_iterations 更聪明。计数存在图 context 里
(conditional_router.py:141,键是 f"{self._id}_iteration"),达到上限时走一条特殊路径
(conditional_router.py:146-159):
第 N 次判定,且这次要封的正好是 default_route
│
├─ 撤销本路由器此前所有的条件排除(直接操作 conditional_exclusion_sources)
│
├─ 把 route_to_stop 翻转成另一条(封掉非默认路)
│
├─ 只调 stop(),不再加条件排除
│
└─ 提前 return → 默认路通了,环从这条路走出去
翻译成人话:转够了就强行让默认分支通过,用"放行出口"的方式打破环,而不是抛异常。
true_response 里对应有个 force_output 判断(conditional_router.py:204-213),
达到上限时直接产出,并且不再去封另一支。
测试 test_conditional_router_max_iterations(src/lfx/tests/unit/graph/graph/test_cycles.py:293-335)
断言的正是这条路径:图级 max_iterations=20,组件级 max_iterations=5,最后 context 里的迭代计数停 在 5——
组件级止损先生效,图级止损只是兜底。
7.2 Loop(LoopComponent)
src/lfx/src/lfx/components/flow_controls/loop.py。和本章相关的三点:
item输出带allows_loop=True(loop.py:36-45),这是全仓唯一一处,决定了Vertex.is_loop;item_output第一行self.stop("item")(loop.py:252)——只用 ACTIVE/INACTIVE 那套, 不用条件排除,因为它每次构建都要重新封;- 真正的迭代在
_iterate→execute_loop_body里靠子图完成(见 4.4)。
值得注意:item_output 还会在 done 输出没有下游消费者时主动触发 _iterate
(loop.py:253-254),而 _iterate 本身用 ctx 里的 _iterated 标志做幂等
(loop.py:201-205),因为两个输出可能在同一次顶点构建里都被调用。
7.3 Smart Router(多路)
src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py:259-282,_deactivate_branches。
它把 N 条分支里没选中的那些一次性排掉:
for name in output_names:
self.stop(name)
...
self._excluded_outputs.update(output_names)
self._vertex.graph.exclude_branches_conditionally(self._id, sorted(self._excluded_outputs))
注意它用的是累积集合 _excluded_outputs 再整体提交——因为 process_case 每个连着的输出都会跑一次,
如果每次都用单支版本,后一次会把前一次的排除清掉。_pre_run_setup(llm_conditional_router.py:249-257)
在每次构建前把这个集合清空,保证每轮重新判定。
8. 巧妙之处(值得带走的)
-
用强连通分量而不是 DFS 回边来定义"环上顶点"。 运行期的判据是按顶点问的,SCC 直接给顶点集合, 而且与遍历入口无关(
utils.py:524-535,find_cycle_vertices)。 -
环上顶点的"首轮特权"用一个组件级开关控制,而不是全局放开。
is_loop and pending_set <= self.cycle_vertices(runnable_vertices_manager.py:99)—— 只有声明了allows_loop的组件能靠这条起步,把"打破 DAG 铁律"的权限缩到最小面。 -
剪枝顺手替下游清依赖。
mark_vertex标 INACTIVE 时调remove_from_predecessors(base.py:986-987),下游因此不必等一个永远不会来的前驱。剪枝和调度是同一个动作的两面。 -
排除记在源顶点名下,让"持久"和"可重判"共存。
conditional_exclusion_sources这个源 → 被排除集的映射(base.py:1075-1086), 使得同一个路由器在环的下一圈能干净地撤销自己上一圈的决定,而不影响别人。 -
同一个"保护汇合节点"的思路做了两遍。 ACTIVE/INACTIVE 侧是
protected_vertices(base.py:1051-1055),条件排除侧是减去_get_vertices_reachable_from_other_outputs(base.py:1137)。两套状态机各自实现,说明这个坑踩得够深。 -
组件级止损比图级止损更优雅。 图级
max_iterations到点抛异常;If-Else 到点是放行默认分支 (conditional_router.py:146-159),用户拿到的是一个正常结束的流,而不是一个报错。
9. 边界与局限(诚实说)
-
async_start不强制max_iterations。 前置检查只在同步Graph.start里(base.py:464-466)。 纯图 API 直接跑一个没有组件级止损的环,should_continue在max_iterations is None时恒真 (utils.py:519-520),会一直转。 -
止损粒度是"单顶点产出次数",不是"环转了几圈"。 一圈里跑 5 个顶点,计数各加 1;
max_iterations=10在这里意味着大约 10 圈,但如果某个顶点一圈被跑两次,含义就变了。 代码里没有"圈数"这个概念。 -
Graph.cycles在没有_start时恒为空(base.py:2196-2197),而从 JSON 载入的流通常没有_start。想知道"哪条边造成了环"就拿不到——只能拿到cycle_vertices。 -
条件排除没有全局清空入口。 全仓写入
conditionally_excluded_vertices的地方只有_replace_conditional_exclusions(base.py:1082-1086)和 If-Else 的破环路径 (conditional_router.py:148-151)。它不随prepare()/sort_vertices重置——生产环境靠 "每个请求Graph.from_payload建一张新图"来保证干净,复用同一个 Graph 对象连跑两次时这层状态会带过去 (inferred)。 -
_collect_branch_vertices在环上会连带排除源顶点的上游。 它只保证不重复访问、不含源顶点本身 (base.py:996-1013),分支若绕回源的祖先,那些祖先会一并进排除集。对"封掉回边"的场景这是想要的, 对别的拓扑可能过宽。 -
are_all_predecessors_fulfilled里的running_predecessors是无效计算(见 4.2 的分析)。 不影响正确性,但会误导读者以为"正在运行的前驱"被单独考虑了。 -
拓扑分层里的
MAX_CYCLE_APPEARANCES = 2是个硬编码常量(utils.py:11),不可配置; 它只影响静态分层结果,不影响运行期实际转多少圈。
10. 与本组其它章的关系
| 想知道什么 | 去哪章 |
|---|---|
allows_loop、Output 这些声明写在组件类的哪里 | 01 组件模型 |
cycle_vertices 是在建图流程的哪一步算出来的 | 02 从 JSON 到图 |
astep / _run_queue / 分层排序的完整逻辑 | 03 调度引擎 |
| 被剪掉的顶点怎么在前端打灰(SSE 事件) | 05 运行时与事件流 |
11. 代码地图(导航索引)
| 主题 | 文件路径(相对克隆根) | 符号名 |
|---|---|---|
| 环上顶点检测(SCC) | src/lfx/src/lfx/graph/graph/utils.py | find_cycle_vertices |
| 有没有环 / 回边 | src/lfx/src/lfx/graph/graph/utils.py | has_cycle、find_cycle_edge、find_all_cycle_edges |
| 迭代止损判据 | src/lfx/src/lfx/graph/graph/utils.py | should_continue、MAX_CYCLE_APPEARANCES |
| 有环时的拓扑入口选择 | src/lfx/src/lfx/graph/graph/utils.py | layered_topological_sort、find_start_component_id |
| 图级环属性 | src/lfx/src/lfx/graph/graph/base.py | Graph.is_cyclic、Graph.cycles、Graph.cycle_vertices |
| 环上顶点关缓存 | src/lfx/src/lfx/graph/graph/base.py | Graph._set_cache_to_vertices_in_cycle |
| 环边 vs 普通边的分派 | src/lfx/src/lfx/graph/graph/base.py | Graph.build_edge、Graph._build_edges |
| 迭代计数与上限 | src/lfx/src/lfx/graph/graph/base.py | Graph.async_start(yielded_counts)、Graph.start |
| ACTIVE/INACTIVE 剪枝 | src/lfx/src/lfx/graph/graph/base.py | mark_branch、_mark_branch、mark_vertex、mark_all_vertices |
| ACTIVE/INACTIVE 重置 | src/lfx/src/lfx/graph/graph/base.py | reset_inactivated_vertices、reset_activated_vertices |
| 状态顶点唤醒 | src/lfx/src/lfx/graph/graph/base.py | activate_state_vertices |
| 条件排除写入 | src/lfx/src/lfx/graph/graph/base.py | exclude_branch_conditionally、exclude_branches_conditionally、_replace_conditional_exclusions |
| 分支顶点收集 / 汇合点保护 | src/lfx/src/lfx/graph/graph/base.py | _collect_branch_vertices、_get_vertices_reachable_from_other_outputs |
| 可运行判定(条件排除第一道闸) | src/lfx/src/lfx/graph/graph/base.py | Graph.is_vertex_runnable |
| 环上顶点的两套前驱判据 | src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py | are_all_predecessors_fulfilled、ran_at_least_once、cycle_vertices |
| 前驱清除(命名易误读) | src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py | remove_from_predecessors、build_run_map |
| 环边类型与履约 | src/lfx/src/lfx/graph/edge/base.py | CycleEdge、CycleEdge.honor、Edge.is_cycle |
| 顶点状态与 is_loop | src/lfx/src/lfx/graph/vertex/base.py | VertexStates、Vertex.is_loop、Vertex.set_state、Vertex.is_active |
| 列表输入跳过被剪前驱 | src/lfx/src/lfx/graph/vertex/base.py | _build_list_of_vertices_and_update_params |
| 被剪前驱返模板默认值 | src/lfx/src/lfx/graph/vertex/vertex_types.py | ComponentVertex._get_result |
| 组件层封路入口 | src/lfx/src/lfx/custom/custom_component/custom_component.py | stop、start |
| If-Else(两套机制并用) | src/lfx/src/lfx/components/flow_controls/conditional_router.py | ConditionalRouterComponent.iterate_and_stop_once |
| Loop(子图迭代) | src/lfx/src/lfx/components/flow_controls/loop.py | LoopComponent.item_output、LoopComponent._iterate |
| Loop 子图执 行器 | src/lfx/src/lfx/base/flow_controls/loop_utils.py | execute_loop_body、get_loop_body_vertices |
| 多路路由(累积排除) | src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py | SmartRouterComponent._deactivate_branches |
| 环行为的回归测试 | src/lfx/tests/unit/graph/graph/test_cycles.py | test_cycle_in_graph_max_iterations、test_conditional_router_max_iterations |