循环变量不再被报成"未定义":错的是分析器,不是模板

模板⑦「批量处理(循环)」是唯一一个装好就红着的内置模板——新用户点开第一眼
看到「1 个错误」,运行按钮还是灰的:

    {{ vars.item }} 没有任何节点产出,是不是想写 {{ vars.items }}?

**但模板是对的。** 循环节点配的 item_var 就叫 item,control.py:142 每轮往
vars 里注入 vars.item 和 vars.item_index。错的是静态分析器:产出侧只登记了
input.*、vars.<assign_to>、nodes.<id>,从来没登记过循环注入的那两个。
kind="loop" 在 _KIND_ORDER 里一直占着位置,只是没人产出过——漏实现,
不是设计如此。

顺着修下来一共三处误报,都是同一类:

**一、循环变量没登记。** 补上 vars.<item_var> 和 vars.<item_var>_index,
只在 foreach 模式(while 不遍历列表,自然没有"当前项")。

**二、顺序判定被回边搞乱。** 补上之后错误降成警告,但换了一句同样不对的话:
"{{ vars.item }} 由「逐条循环」产出,但那一步在这之后才跑"。循环体末尾有一条
回到循环节点的边,于是循环节点的拓扑深度被推高、反而排在自己的体后面。
改成先算出每个循环的"体"(从 body 出口走可达闭包,遇到循环节点自己就停),
体内引用不套先后判定。体**外面**引用它确实取不到值,那条留着,
并换成说得清的措辞:"只在循环体内部有值"。

**三、表达式字段里的裸引用没扫。** transform 的 expression、branch 的
cases[].condition、skip_if 写的是裸 vars.x,没有 {{ }},而扫描只认模板语法。
于是模板⑦里 `(vars.collected or '') + str(vars.one)` 一个引用都认不出来,
vars.one 被报成"产出了但没有任何地方引用"——它明明在用。

顺带把"没人用"这条收窄到**作者声明过的**变量:循环的 item_index 和入口字段的
vars.<name> 别名都是系统自动提供的,报它们等于噪音(内置变量没有 produced_by,
本来就不在这条路上)。而真正未引用的 assign_to 照报不误——⑤⑥⑧ 还各有一两条。

误报是 linter 最坏的失败模式:它会让人把整个校验一起无视掉。这条理由本来就
写在这个文件的注释里,只是自己没做到。

现在 8 个内置模板全部零错误。验证:后端 284 passed(test_variables.py 新增
8 条,含"体外引用仍然要警告"和"while 模式没有当前项"两条反向断言);
三个前端 check 不回归。
This commit is contained in:
ThearchyHelios
2026-09-20 21:30:42 +08:00
parent 00f8dd78a9
commit 5641d0631f
2 changed files with 248 additions and 2 deletions
+114 -1
View File
@@ -104,6 +104,59 @@ def _depths(spec: GraphSpec) -> dict[str, int]:
return depth
#: 按 AST 白名单求值的字段:里面是裸表达式,不带 {{ }}。
#: cases 是数组,单独处理。
_EXPRESSION_FIELDS = ("expression", "condition", "skip_if")
#: 裸引用:vars.x / input.x / nodes.x[.y…]。只认这三个根,
#: 免得把 str(...) 里的普通标识符也当成变量引用
_BARE_REF_RE = re.compile(r"\b(?:vars|input|nodes)\.[A-Za-z_][\w.]*")
def _iter_expressions(cfg: dict[str, Any]) -> list[tuple[str, str]]:
"""节点配置里那些按表达式求值的字段。"""
out: list[tuple[str, str]] = []
for key in _EXPRESSION_FIELDS:
value = cfg.get(key)
if isinstance(value, str) and value.strip():
out.append((key, value))
for i, case in enumerate(cfg.get("cases") or []):
cond = (case or {}).get("condition")
if isinstance(cond, str) and cond.strip():
out.append((f"cases[{i}].condition", cond))
return out
def _loop_bodies(spec: GraphSpec) -> dict[str, set[str]]:
"""每个循环节点的"体"里有哪些节点。
循环变量(item / item_index)只在体内部有值,而拓扑深度在这里完全不可靠:
循环体末尾有一条回到循环节点的边,于是循环节点的深度被回边推高,反而排在
自己的体后面。模板⑦就因此被报成"{{ vars.item }} 由「逐条循环」产出,
但那一步在这之后才跑"——而那是循环最正常的写法。
从 body 出口走可达闭包,遇到循环节点自己就停(那就是回边)。
"""
bodies: dict[str, set[str]] = {}
outgoing: dict[str, list[tuple[str | None, str]]] = {}
for e in spec.edges:
outgoing.setdefault(e.source, []).append((e.sourceHandle, e.target))
for node in spec.nodes:
if node.type != "loop":
continue
seen: set[str] = set()
stack = [t for h, t in outgoing.get(node.id, []) if h == "body"]
while stack:
cur = stack.pop()
if cur in seen or cur == node.id:
continue
seen.add(cur)
stack.extend(t for _, t in outgoing.get(cur, []))
bodies[node.id] = seen
return bodies
def _iter_strings(value: Any, prefix: str = "") -> list[tuple[str, str]]:
"""把配置里所有字符串摊平成 (字段路径, 文本)。"""
out: list[tuple[str, str]] = []
@@ -172,6 +225,7 @@ def _resolve_fallback(
def analyze(spec: GraphSpec) -> VariableReport:
"""静态分析一张图的变量。不需要跑,也不需要有运行记录。"""
depth = _depths(spec)
bodies = _loop_bodies(spec)
report = VariableReport()
index: dict[str, Variable] = {}
@@ -207,6 +261,28 @@ def analyze(spec: GraphSpec) -> VariableReport:
description=desc,
)
# 循环节点每轮往 vars 里注入当前项和它的序号(control.py:142)。
# 产出侧漏了这一条,于是模板⑦「批量处理」里完全正确的 {{ vars.item }}
# 被报成"没有任何节点产出"——那是唯一一个装好就红着的内置模板,
# 新用户点开第一眼看到的就是个报错,而运行按钮还是灰的。
#
# 误报是 linter 最坏的失败模式:它会让人把整个校验一起无视掉。
# kind="loop" 在 _KIND_ORDER 里一直有位置,只是从来没人产出过。
if node.type == "loop" and str(cfg.get("mode") or "foreach") == "foreach":
# while 模式不遍历列表,自然也没有"当前项"
item_var = str(cfg.get("item_var") or "item").strip()
if item_var:
ensure(
f"vars.{item_var}", kind="loop", label=f"「{label}」每轮的当前项",
produced_by=node.id, produced_by_label=label, order=order,
description="只在循环体内部有值",
)
ensure(
f"vars.{item_var}_index", kind="loop", label=f"「{label}」每轮的序号",
produced_by=node.id, produced_by_label=label, order=order,
description="从 0 开始,只在循环体内部有值",
)
var_name = str(cfg.get("assign_to") or "").strip()
if var_name:
ensure(
@@ -238,6 +314,21 @@ def analyze(spec: GraphSpec) -> VariableReport:
cfg = node.data.config or {}
label = _label_of(node)
order = depth.get(node.id, 0)
# 表达式字段里写的是**裸**引用(vars.one、input.n),没有 {{ }}。
# 只扫模板语法的话,模板⑦里 collect 那个
# `(vars.collected or '') + str(vars.one)` 就一个引用都认不出来,
# 于是 vars.one 被报成"产出了但没有任何地方引用"——而它明明在用。
for field, expr_text in _iter_expressions(cfg):
for m in _BARE_REF_RE.finditer(expr_text):
path, root = _root_of(m.group(0))
if not path:
continue
target = index.get(path)
if target is not None:
target.refs.append(VarRef(
node_id=node.id, node_label=label, field=field, expr=m.group(0),
))
for field, text in _iter_strings(cfg):
for m in _TEMPLATE_RE.finditer(text):
expr = m.group(1).strip()
@@ -253,6 +344,19 @@ def analyze(spec: GraphSpec) -> VariableReport:
target = _resolve_fallback(path, index, input_is_closed=input_is_closed)
if target is not None:
target.refs.append(ref)
# 循环变量单独判:它只在循环体内部有值,而循环体的拓扑
# 深度因为那条回边天然比循环节点小,套用下面的先后判定
# 必然误报(模板⑦就是这么红的)
if target.kind == "loop" and target.produced_by:
if node.id not in bodies.get(target.produced_by, set()):
report.issues.append(VarIssue(
level="warning", node_id=node.id, path=path,
message=f"{{{{ {path} }}}} 是「{target.produced_by_label}」"
f"每轮的循环变量,只在循环体内部有值;"
f"这一步不在循环体里,取到的会是空值",
))
continue
# 变量存在,但产出它的节点排在引用者之后——跑到这儿时它还是空的
if (
target.produced_by
@@ -293,7 +397,16 @@ def analyze(spec: GraphSpec) -> VariableReport:
for var in index.values():
if var.kind in ("builtin", "node"):
continue # nodes.<id> 是自动存在的,没人引用很正常
if not var.refs and var.produced_by:
# 只对**作者声明**的变量报"没人用"。系统自动提供的不算:
# - 循环变量(item / item_index):循环本来就两个都注入,
# 用不用是作者的事,提醒他"你没用 item_index"纯属噪音
# - vars.<入口字段> 这个别名:真正的 input.<字段> 用着呢,
# 拿别名去报未引用只会让人以为哪里写错了
# 内置变量(last_message 等)没有 produced_by,本来就不在这条路上。
auto_provided = var.kind == "loop" or (
var.kind == "input" and var.path.startswith("vars.")
)
if not var.refs and var.produced_by and not auto_provided:
report.issues.append(VarIssue(
level="info", node_id=var.produced_by, path=var.path,
message=f"{{{{ {var.path} }}}} 产出了但没有任何地方引用",
+134 -1
View File
@@ -18,7 +18,10 @@ def _graph(nodes, edges=()):
"data": {"label": n.get("label", ""), "config": n.get("config", {})}}
for n in nodes
],
"edges": [{"source": s, "target": t} for s, t in edges],
"edges": [
{"source": e[0], "target": e[1], **({"sourceHandle": e[2]} if len(e) > 2 else {})}
for e in edges
],
})
@@ -208,3 +211,133 @@ def test_cycle_does_not_hang() -> None:
[("a", "b"), ("b", "a")],
)
assert analyze(g).variables # 跑得完就行
# --------------------------------------------------------------------------
# 循环变量
# --------------------------------------------------------------------------
#
# 模板⑦「批量处理(循环)」装好就红着:{{ vars.item }} 被报成"没有任何节点
# 产出",而那是循环最正常的写法。它是唯一一个内置模板开箱就报错的,
# 新用户点开第一眼看到的就是红字,运行按钮还是灰的。
#
# 根因不在模板,在这里:产出侧从来没登记过循环注入的变量
# (control.py:142 每轮写 vars.<item_var> 和 vars.<item_var>_index)。
# kind="loop" 在 _KIND_ORDER 里一直有位置,只是没人产出过。
def _loop_graph(*, item_var="item", body_ref="{{ vars.item }}", extra_node=None):
"""input → loop →(body) handle → 回到 loop;loop →(done) out。
末尾那条回边是关键:它让 loop 的拓扑深度反而大于自己的循环体。
"""
nodes = [
{"id": "start", "type": "input", "config": {"fields": [{"name": "items"}]}},
{"id": "each", "type": "loop", "label": "逐条循环",
"config": {"mode": "foreach", "items": "{{ input.items }}", "item_var": item_var}},
{"id": "handle", "type": "llm", "label": "处理单条", "config": {"prompt": body_ref}},
{"id": "out", "type": "output", "config": {}},
]
if extra_node:
nodes.append(extra_node)
edges = [("start", "each"), ("each", "handle", "body"),
("handle", "each"), ("each", "out", "done")]
if extra_node:
edges.append(("out", extra_node["id"]))
return _graph(nodes, edges)
def test_loop_item_is_a_known_variable() -> None:
"""循环体里引用 {{ vars.item }} 不该被报成未定义。"""
report = analyze(_loop_graph())
paths = {v.path for v in report.variables}
assert "vars.item" in paths
assert "vars.item_index" in paths
errors = [i for i in report.issues if i.level == "error"]
assert not errors, [i.message for i in errors]
def test_loop_item_follows_the_configured_name() -> None:
report = analyze(_loop_graph(item_var="row", body_ref="{{ vars.row }}"))
paths = {v.path for v in report.variables}
assert {"vars.row", "vars.row_index"} <= paths
assert not [i for i in report.issues if i.level == "error"]
def test_no_false_ordering_warning_inside_the_loop_body() -> None:
"""循环体的拓扑深度因为回边天然小于循环节点,套用"先后"判定必然误报。
以前报的是"{{ vars.item }} 由「逐条循环」产出,但那一步在这之后才跑"——
而那正是循环最正常的写法。
"""
report = analyze(_loop_graph())
assert not [i for i in report.issues if "在这之后才跑" in i.message], \
[i.message for i in report.issues]
def test_loop_variable_outside_the_body_still_warns() -> None:
"""但在循环体**外面**引用它确实取不到值,这条要留着。"""
outside = {"id": "after", "type": "llm", "label": "循环之后",
"config": {"prompt": "{{ vars.item }}"}}
report = analyze(_loop_graph(extra_node=outside))
warns = [i for i in report.issues if i.node_id == "after"]
assert warns, [i.message for i in report.issues]
assert "只在循环体内部有值" in warns[0].message
def test_while_loop_has_no_current_item() -> None:
"""while 模式不遍历列表,自然没有"当前项"(control.py 只在 foreach 分支注入)。"""
report = analyze(_graph([
{"id": "lp", "type": "loop",
"config": {"mode": "while", "condition": "vars.done != True", "item_var": "item"}},
]))
assert "vars.item" not in {v.path for v in report.variables}
# --------------------------------------------------------------------------
# 表达式字段里的裸引用
# --------------------------------------------------------------------------
def test_bare_refs_in_expression_fields_count_as_usage() -> None:
"""表达式字段写的是裸 vars.x,没有 {{ }}——只扫模板语法会认不出来。
模板⑦里 `(vars.collected or '') + str(vars.one)` 就是这样,
于是 vars.one 被报成"产出了但没有任何地方引用",而它明明在用。
"""
report = analyze(_graph([
{"id": "a", "type": "llm", "config": {"prompt": "x", "assign_to": "one"}},
{"id": "b", "type": "transform", "config": {
"mode": "expression",
"expression": "(vars.collected or '') + str(vars.one)",
"assign_to": "collected",
}},
], [("a", "b")]))
one = next(v for v in report.variables if v.path == "vars.one")
assert one.refs, "表达式里用了就不该说没人用"
assert not [i for i in report.issues if "vars.one" in i.message]
def test_branch_case_conditions_count_as_usage() -> None:
report = analyze(_graph([
{"id": "a", "type": "llm", "config": {"prompt": "x", "assign_to": "score"}},
{"id": "b", "type": "branch", "config": {
"mode": "expression",
"cases": [{"key": "hi", "condition": "vars.score > 3"}],
}},
], [("a", "b")]))
assert next(v for v in report.variables if v.path == "vars.score").refs
def test_auto_provided_variables_are_not_nagged_about() -> None:
"""系统自动提供的变量不报"没人用"——作者没声明过它,提醒他等于噪音。
内置变量(last_message 等)本来就不在这条路上;循环的 item_index 和
入口字段的 vars.<name> 别名以前会被报,而真正的 input.<name> 用着呢。
"""
report = analyze(_loop_graph())
unused = [i.message for i in report.issues if "没有任何地方引用" in i.message]
assert not any("item_index" in m for m in unused), unused
assert not any("vars.items" in m for m in unused), unused