diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 7e49f5ad..499c09ce 100644 --- a/backend/app/api/workflows.py +++ b/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() diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index aad91d7e..b183bdae 100644 --- a/backend/app/engine/schema.py +++ b/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 diff --git a/backend/app/engine/variables.py b/backend/app/engine/variables.py new file mode 100644 index 00000000..baa8957f --- /dev/null +++ b/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. 引用 + 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. 是自动存在的,没人引用很正常 + 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 "" diff --git a/backend/tests/test_variables.py b/backend/tests/test_variables.py new file mode 100644 index 00000000..c1b122f2 --- /dev/null +++ b/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 # 跑得完就行