变量静态分析：把拼错一个变量名的静默失败说破

这套编排靠 {{ }} 串数据流，而模板取不到值时**渲染成空字符串**
（resolve_path 返回 None → _stringify → ""）。那是为了让半成品的图也能跑，
代价是：拼错一个变量名不报错、不警告、不留任何痕迹——下游拿到一段空文本，
然后模型基于空内容一本正经地编一段回答出来。这是这套东西最大的调试黑洞，
而界面上此前没有任何地方提过它。

app/engine/variables.py 做纯静态分析（不用跑图、不用有运行记录——用户在
编排到一半时最需要它，而那时候还没有任何一次运行可看）：

- 产出侧：input 字段、各节点的 assign_to、每个节点的 nodes.<id>
- 引用侧：递归扫所有配置字符串（output.fields[].value、code 的 files 都是
  嵌套的，只扫顶层等于没扫），认得过滤器和下标
- 三类问题：
  · 引用了没人产出的 —— 十有八九是拼错，用 difflib 给出"是不是想写…"
  · 引用早于产出 —— 变量确实存在、名字也对，但那一步在这之后才跑，这里
    取到的是空。这类最难查，因为表里查得到
  · 产出了没人用 —— 只报 info，不拦着跑

前两类接进 validate_graph，画布上直接标红/标黄。另外开了
POST /api/workflows/variables 返回完整的变量表（谁产出、谁在哪个字段引用），
给接下来的变量表界面和输入补全用。

tests/test_variables.py 8 条，包括循环图不死循环（loop 的 body 连回自身很
常见）、节点读自己上一轮写的变量不该误报。

backend/app/api/workflows.py

@@ -292,3 +292,19 @@ async def validate(payload: ValidateIn) -> dict[str, Any]:
         return {"ok": False, "issues": [{"level": "error", "message": f"图结构非法：{e}"}]}
     result = validate_graph(spec)
     return result.model_dump()
+
+
+@router.post("/variables")
+async def variables(payload: ValidateIn) -> dict[str, Any]:
+    """这张图里有哪些变量、谁产出、谁引用。
+
+    纯静态分析，不用跑图也不用有运行记录——用户在编排到一半时最需要它，
+    而那时候还没有任何一次运行可看。运行期的实际取值由前端从事件里补上。
+    """
+    from app.engine.variables import analyze
+
+    try:
+        spec = GraphSpec.model_validate(payload.graph)
+    except Exception as e:  # noqa: BLE001
+        return {"variables": [], "issues": [{"level": "error", "message": f"图结构非法：{e}"}]}
+    return analyze(spec).model_dump()

backend/app/engine/schema.py

@@ -228,4 +228,15 @@ def validate_graph(spec: GraphSpec) -> ValidationResult:
         if not spec.incoming(node.id) and not spec.outgoing(node.id):
             result.add("节点没有任何连线，不会被执行", level="warning", node_id=node.id)
 
+    # 变量问题也进校验结果：模板取不到值只会渲染成空字符串，不报错，
+    # 所以拼错一个变量名在运行时是完全静默的——只能在这里说破。
+    # 放在最后导入，避免 variables 反过来 import schema 形成环。
+    from app.engine.variables import analyze as _analyze_vars
+
+    for issue in _analyze_vars(spec).issues:
+        # info 级是"产出没人用"这类提示，不进校验（它不影响能不能跑）
+        if issue.level == "info":
+            continue
+        result.add(issue.message, level=issue.level, node_id=issue.node_id)
+
     return result

backend/app/engine/variables.py

@@ -0,0 +1,255 @@
+"""工作流里的变量：谁产出、谁引用、现在是什么。
+
+这套编排靠 `{{ }}` 模板串起数据流，而模板取不到值时 **渲染成空字符串**
+（resolve_path 返回 None → _stringify → ""）。那是为了让半成品的图也能跑，
+代价是一个拼错的变量名不会报错、不会警告、不会在任何地方留下痕迹——它只是
+让下游拿到一段空文本，然后模型一本正经地基于空内容编一段回答出来。
+
+所以需要有人在跑之前把这件事说破：
+
+- **引用了没人产出的变量** —— 十有八九是拼错了
+- **引用早于产出** —— 变量确实存在，但那个节点在它之后才跑，这时候取到的是空
+- **产出了没人用** —— 不一定是错，但常常是改图改了一半的残留
+
+顺带把"这张图里到底有哪些变量"列出来，界面上的变量表和输入补全都用它。
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from app.engine.schema import GraphNode, GraphSpec
+
+_TEMPLATE_RE = re.compile(r"\{\{(.*?)\}\}", re.DOTALL)
+
+# 这些是状态里天然存在的命名空间，不需要谁去产出
+_BUILTIN_ROOTS = {
+    "input": "工作流的入口输入",
+    "nodes": "各节点的输出",
+    "vars": "节点用 assign_to 写入的变量",
+    "output": "已收集的成果",
+    "loops": "循环的当前项和轮次",
+    "usage": "token 与成本累计",
+    "messages": "对话历史",
+    "last_message": "最近一条消息的文本",
+}
+
+
+class VarRef(BaseModel):
+    """一处引用。"""
+
+    node_id: str
+    node_label: str
+    field: str
+    expr: str
+
+
+class Variable(BaseModel):
+    """一个可以被 {{ }} 引用的东西。"""
+
+    path: str
+    kind: str          # input | var | node | builtin | loop
+    label: str
+    #: 谁产出它。builtin 没有产出者
+    produced_by: str | None = None
+    produced_by_label: str | None = None
+    #: 产出者在拓扑序里的层级，用来判断引用是不是太早
+    order: int = -1
+    refs: list[VarRef] = Field(default_factory=list)
+    #: 这个变量声明时带的说明（input 字段的 description）
+    description: str = ""
+
+
+class VarIssue(BaseModel):
+    level: str         # error | warning | info
+    message: str
+    node_id: str | None = None
+    path: str = ""
+
+
+class VariableReport(BaseModel):
+    variables: list[Variable] = Field(default_factory=list)
+    issues: list[VarIssue] = Field(default_factory=list)
+
+
+# --------------------------------------------------------------------------
+
+
+def _depths(spec: GraphSpec) -> dict[str, int]:
+    """每个节点的拓扑层级（最长路径）。有环就在环上停住，不死循环。"""
+    incoming: dict[str, list[str]] = {n.id: [] for n in spec.nodes}
+    for e in spec.edges:
+        if e.target in incoming and e.source in incoming:
+            incoming[e.target].append(e.source)
+
+    depth: dict[str, int] = {}
+    visiting: set[str] = set()
+
+    def walk(node_id: str) -> int:
+        if node_id in depth:
+            return depth[node_id]
+        if node_id in visiting:      # 环：当成同层，别递归下去
+            return 0
+        visiting.add(node_id)
+        parents = incoming.get(node_id) or []
+        depth[node_id] = 0 if not parents else max(walk(p) for p in parents) + 1
+        visiting.discard(node_id)
+        return depth[node_id]
+
+    for n in spec.nodes:
+        walk(n.id)
+    return depth
+
+
+def _iter_strings(value: Any, prefix: str = "") -> list[tuple[str, str]]:
+    """把配置里所有字符串摊平成 (字段路径, 文本)。"""
+    out: list[tuple[str, str]] = []
+    if isinstance(value, str):
+        if "{{" in value:
+            out.append((prefix or "config", value))
+    elif isinstance(value, dict):
+        for k, v in value.items():
+            out.extend(_iter_strings(v, f"{prefix}.{k}" if prefix else str(k)))
+    elif isinstance(value, list):
+        for i, v in enumerate(value):
+            out.extend(_iter_strings(v, f"{prefix}[{i}]"))
+    return out
+
+
+def _root_of(expr: str) -> tuple[str, str]:
+    """把 `vars.sales | json` 拆成 ('vars.sales', 'vars')。"""
+    path = expr.split("|")[0].strip()
+    # 去掉下标，vars.rows[0].name → vars.rows.name
+    clean = re.sub(r"\[[^\]]*\]", "", path)
+    root = clean.split(".")[0].strip()
+    return clean, root
+
+
+def _label_of(node: GraphNode) -> str:
+    return node.data.label or node.id
+
+
+def analyze(spec: GraphSpec) -> VariableReport:
+    """静态分析一张图的变量。不需要跑，也不需要有运行记录。"""
+    depth = _depths(spec)
+    report = VariableReport()
+    index: dict[str, Variable] = {}
+
+    def ensure(path: str, **kw: Any) -> Variable:
+        if path not in index:
+            index[path] = Variable(path=path, **kw)
+        return index[path]
+
+    # ---- 产出侧 ----
+    for node in spec.nodes:
+        cfg = node.data.config or {}
+        label = _label_of(node)
+        order = depth.get(node.id, 0)
+
+        if node.type == "input":
+            for field in cfg.get("fields") or []:
+                name = str((field or {}).get("name") or "").strip()
+                if not name:
+                    continue
+                ensure(
+                    f"input.{name}", kind="input", label=f"入口输入「{name}」",
+                    produced_by=node.id, produced_by_label=label, order=order,
+                    description=str((field or {}).get("description") or ""),
+                )
+
+        var_name = str(cfg.get("assign_to") or "").strip()
+        if var_name:
+            ensure(
+                f"vars.{var_name}", kind="var", label=f"「{label}」写入的变量",
+                produced_by=node.id, produced_by_label=label, order=order,
+            )
+
+        # 每个节点的输出都能用 nodes.<id> 引用
+        ensure(
+            f"nodes.{node.id}", kind="node", label=f"「{label}」这一步的输出",
+            produced_by=node.id, produced_by_label=label, order=order,
+        )
+
+    for root, desc in _BUILTIN_ROOTS.items():
+        if root in ("input", "nodes", "vars"):
+            continue   # 这三个的具体成员上面已经列过了
+        ensure(root, kind="builtin", label=desc)
+
+    # ---- 引用侧 ----
+    for node in spec.nodes:
+        cfg = node.data.config or {}
+        label = _label_of(node)
+        order = depth.get(node.id, 0)
+        for field, text in _iter_strings(cfg):
+            for m in _TEMPLATE_RE.finditer(text):
+                expr = m.group(1).strip()
+                if not expr:
+                    continue
+                path, root = _root_of(expr)
+                if not path:
+                    continue
+                ref = VarRef(node_id=node.id, node_label=label, field=field, expr=expr)
+
+                # 精确命中（input.x / vars.x / nodes.x）或者根命中（last_message）
+                target = index.get(path)
+                if target is None and root in _BUILTIN_ROOTS:
+                    # nodes.foo.text 这种，退到 nodes.foo 这一层
+                    parent = ".".join(path.split(".")[:2])
+                    target = index.get(parent) or index.get(root)
+                if target is not None:
+                    target.refs.append(ref)
+                    # 变量存在，但产出它的节点排在引用者之后——跑到这儿时它还是空的
+                    if (
+                        target.produced_by
+                        and target.produced_by != node.id
+                        and target.order > order
+                    ):
+                        report.issues.append(VarIssue(
+                            level="warning", node_id=node.id, path=path,
+                            message=f"{{{{ {path} }}}} 由「{target.produced_by_label}」产出，"
+                                    f"但那一步在这之后才跑，这里取到的会是空值",
+                        ))
+                    continue
+
+                # 谁都不认识它。模板取不到值会渲染成空字符串，不报错——
+                # 所以这条必须在跑之前说出来
+                hint = _did_you_mean(path, index)
+                report.issues.append(VarIssue(
+                    level="error", node_id=node.id, path=path,
+                    message=(
+                        f"{{{{ {path} }}}} 没有任何节点产出，"
+                        f"是不是想写 {{{{ {hint} }}}}？取不到值会渲染成空字符串，不报错"
+                        if hint else
+                        f"{{{{ {path} }}}} 没有任何节点产出。"
+                        f"取不到值会渲染成空字符串，不报错"
+                    ),
+                ))
+
+    # ---- 产出了没人用 ----
+    for var in index.values():
+        if var.kind in ("builtin", "node"):
+            continue      # nodes.<id> 是自动存在的，没人引用很正常
+        if not var.refs and var.produced_by:
+            report.issues.append(VarIssue(
+                level="info", node_id=var.produced_by, path=var.path,
+                message=f"{{{{ {var.path} }}}} 产出了但没有任何地方引用",
+            ))
+
+    report.variables = sorted(
+        index.values(), key=lambda v: (_KIND_ORDER.get(v.kind, 9), v.order, v.path)
+    )
+    return report
+
+
+_KIND_ORDER = {"input": 0, "var": 1, "node": 2, "loop": 3, "builtin": 4}
+
+
+def _did_you_mean(path: str, index: dict[str, Variable]) -> str:
+    """拼错是这里最常见的错误，能猜就直说，别只讲"不存在"。"""
+    import difflib
+
+    matches = difflib.get_close_matches(path, list(index), n=1, cutoff=0.7)
+    return matches[0] if matches else ""

backend/tests/test_variables.py

@@ -0,0 +1,123 @@
+"""变量静态分析。
+
+这套模板取不到值时渲染成空字符串（为了让半成品的图也能跑）。代价是拼错一个
+变量名不会报错、不会警告、不留痕迹——下游拿到空文本，模型基于空内容一本正经
+地编一段回答出来。这些测试守的就是"在跑之前把这件事说破"。
+"""
+
+from __future__ import annotations
+
+from app.engine.schema import GraphSpec
+from app.engine.variables import analyze
+
+
+def _graph(nodes, edges=()):
+    return GraphSpec.model_validate({
+        "nodes": [
+            {"id": n["id"], "type": n["type"], "position": {"x": 0, "y": 0},
+             "data": {"label": n.get("label", ""), "config": n.get("config", {})}}
+            for n in nodes
+        ],
+        "edges": [{"source": s, "target": t} for s, t in edges],
+    })
+
+
+def test_lists_what_each_node_produces() -> None:
+    g = _graph([
+        {"id": "start", "type": "input", "label": "输入",
+         "config": {"fields": [{"name": "question", "description": "用户的问题"}]}},
+        {"id": "q", "type": "llm", "label": "取数", "config": {"assign_to": "sales"}},
+    ], [("start", "q")])
+    paths = {v.path: v for v in analyze(g).variables}
+
+    assert "input.question" in paths
+    assert paths["input.question"].description == "用户的问题"
+    assert "vars.sales" in paths
+    assert paths["vars.sales"].produced_by == "q"
+    # 每个节点的输出都能被引用，不用显式声明
+    assert "nodes.start" in paths and "nodes.q" in paths
+
+
+def test_typo_is_reported_with_a_suggestion() -> None:
+    """拼错是这里最常见的错误。只说"不存在"没用，能猜就直说。"""
+    g = _graph([
+        {"id": "start", "type": "input", "config": {"fields": [{"name": "question"}]}},
+        {"id": "a", "type": "llm", "label": "作答",
+         "config": {"prompt": "回答：{{ input.quesiton }}"}},
+    ], [("start", "a")])
+    errs = [i for i in analyze(g).issues if i.level == "error"]
+
+    assert len(errs) == 1
+    assert "input.quesiton" in errs[0].message
+    assert "input.question" in errs[0].message, "应该给出拼写建议"
+    assert errs[0].node_id == "a"
+    # 必须说清后果，否则用户不知道为什么"没报错但结果不对"
+    assert "空字符串" in errs[0].message
+
+
+def test_reference_before_production_is_a_warning() -> None:
+    """变量确实存在，但那一步在这之后才跑——这时候取到的是空。
+
+    这类最难查：变量名拼对了，表里也有，就是值不对。
+    """
+    g = _graph([
+        {"id": "start", "type": "input", "config": {"fields": [{"name": "q"}]}},
+        {"id": "use", "type": "llm", "label": "用它", "config": {"prompt": "{{ vars.later }}"}},
+        {"id": "make", "type": "llm", "label": "产出它", "config": {"assign_to": "later"}},
+    ], [("start", "use"), ("use", "make")])
+    warns = [i for i in analyze(g).issues if i.level == "warning"]
+
+    assert len(warns) == 1
+    assert "vars.later" in warns[0].message
+    assert "产出它" in warns[0].message, "要说清是谁产出的"
+    assert warns[0].node_id == "use"
+
+
+def test_same_node_reading_its_own_var_is_fine() -> None:
+    """循环里节点读自己上一轮写的变量是正常写法，不该报。"""
+    g = _graph([
+        {"id": "n", "type": "llm", "label": "累加",
+         "config": {"assign_to": "acc", "prompt": "上一轮：{{ vars.acc }}"}},
+    ])
+    assert [i for i in analyze(g).issues if i.level in ("error", "warning")] == []
+
+
+def test_unused_variable_is_only_info() -> None:
+    """产出没人用不一定是错，常常是改图改了一半——提一句，但别拦着跑。"""
+    g = _graph([
+        {"id": "a", "type": "llm", "label": "算一下", "config": {"assign_to": "unused"}},
+    ])
+    issues = analyze(g).issues
+    assert [i.level for i in issues] == ["info"]
+    assert "vars.unused" in issues[0].message
+
+
+def test_builtins_and_filters_and_indexes_resolve() -> None:
+    g = _graph([
+        {"id": "a", "type": "llm", "config": {
+            "prompt": "{{ last_message }} {{ vars.rows[0].name }} {{ nodes.a.text | json }}",
+            "assign_to": "rows",
+        }},
+    ])
+    assert [i for i in analyze(g).issues if i.level != "info"] == []
+
+
+def test_scans_nested_config_not_just_top_level() -> None:
+    """output.fields[].value、code 节点的 files 都是嵌套的，漏扫等于没扫。"""
+    g = _graph([
+        {"id": "out", "type": "output", "config": {
+            "fields": [{"name": "结果", "value": "{{ vars.nope }}"}],
+        }},
+    ])
+    errs = [i for i in analyze(g).issues if i.level == "error"]
+    assert len(errs) == 1 and "vars.nope" in errs[0].message
+    assert "fields[0].value" in errs[0].path or errs[0].path == "vars.nope"
+
+
+def test_cycle_does_not_hang() -> None:
+    """循环图很常见（loop 的 body 连回 loop 本身），拓扑分析不能死循环。"""
+    g = _graph(
+        [{"id": "a", "type": "llm"}, {"id": "b", "type": "llm"}],
+        [("a", "b"), ("b", "a")],
+    )
+    assert analyze(g).variables      # 跑得完就行