# 工作流节点变量作用域与输入变量关联设计 ## 1. 背景与目标 当前工作流引擎使用**扁平变量表** `WorkflowContext.variables` 传递节点输出：每个节点执行完成后，把输出 Map 中的 key-value 直接写入上下文，下游节点通过 `{{变量名}}` 模板或 `ContextPromptHelper` 自动注入来消费变量。 该机制存在两个问题： 1. **同名变量覆盖**：若多个前驱节点都输出同名变量（例如都叫 `result`），后执行的节点会覆盖前者，导致下游节点只能看到最近一个同名输出。 2. **无法精确关联**：节点输入变量无法显式指定来源；隐式关联仅依赖同名匹配，不支持类型转换、JSON 嵌套路径、模糊匹配等场景。 本设计目标： - 为每个节点构建**隔离的节点级工作区**，仅包含当前节点可达前驱节点的输出变量。 - 所有前驱节点输出按 `节点ID.变量名` 命名空间化，避免同名覆盖。 - 支持输入变量**显式关联**前驱输出，并自动按类型转换规则转换。 - 支持四级**隐式关联 fallback**：精确同名 → 忽略分隔符/大小写同名 → JSON 嵌套路径同名 → JSON 嵌套路径模糊同名。 --- ## 2. 核心概念 ### 2.1 节点工作区（NodeWorkspace） 节点 `N` 开始执行前，引擎为其构建一个工作区： ``` NodeWorkspace(N) = { variables: Map, // 可直接通过 {{var}} 引用的扁平变量 scopedOutputs: Map>, // nodeId -> {varName -> value} files: List, // 工作空间文件列表 runId: String } ``` 其中 `variables` 的构造规则： - 保留初始运行输入（`request.inputs`）。 - 对 `N` 的每个可达前驱节点 `P`，把 `scopedOutputs[P]` 中的变量以 `P_varName` 形式注入（双下划线或点号分隔，见下文）。 - 是否同时保留扁平变量（无节点前缀）作为向后兼容，通过 feature flag 控制。 ### 2.2 节点级命名空间 在 `WorkflowContext` 中新增独立的节点输出存储： ```java private final Map> nodeScopedOutputs = new ConcurrentHashMap<>(); ``` 每个节点执行成功后： ```java context.setNodeOutput(nodeId, output); // 现有：扁平合并（保留兼容） context.setNodeScopedOutput(nodeId, output); // 新增：命名空间隔离 ``` ### 2.3 变量引用语法 为兼容现有 `{{varName}}` 模板，同时支持精确引用某个节点的输出，引入两种引用方式： | 语法 | 含义 | 示例 | |------|------|------| | `{{varName}}` | 引用当前工作区中的 `varName`（可能是扁平变量，也可能是最近同名变量） | `{{question}}` | | `{{nodeId.varName}}` | 引用指定节点 `nodeId` 的输出变量 `varName` | `{{node_2.result}}` | | `{{nodeId.var.path}}` | 引用指定节点输出变量的嵌套 JSON 路径 | `{{node_2.output.total.result}}` | > 注：`nodeId` 本身可能包含下划线，为消除歧义，内部实现使用 `__NODE__` 作为分隔符或采用 Map 结构，前端展示使用 `.` 语法。 --- ## 3. 数据模型 ### 3.1 后端 WorkflowContext 扩展 ```java public class WorkflowContext { private final Map variables = new ConcurrentHashMap<>(); private final Map> nodeScopedOutputs = new ConcurrentHashMap<>(); private final List nodeOutputs = ...; private final Path workingDir; // ... public void setNodeScopedOutput(String nodeId, Map output); public Map getNodeScopedOutput(String nodeId); public Map> getAllNodeScopedOutputs(); } ``` ### 3.2 前端节点 data 扩展 节点 `data.inputs[]` 中每个字段增加 `mapping`： ```json { "name": "result", "type": "string", "description": "", "required": true, "mapping": { "sourceNodeId": "node_2", "sourceField": "result", "sourcePath": null, "autoMapped": false } } ``` 当 `sourcePath` 非空时，表示从 `sourceField` 的 JSON 嵌套路径取值，例如 `output.total.result`。 `autoMapped` 标记该关联是否由隐式关联算法自动产生，用户可覆盖。 ### 3.3 边 mapping 数据 保留现有边 `data.mapping` 用于精确源→目标字段映射，同时新增 `autoResolved` 标记： ```json { "mapping": [ {"sourceField": "result", "targetField": "result"} ], "autoResolved": true } ``` --- ## 4. 算法设计 ### 4.1 可达前驱计算 基于现有 `DagResolver.ResolvedDag` 增加反向邻接表： ```java Map> incomingEdges ``` 静态可达前驱（规划阶段与前端编辑阶段使用）： ```java Set getReachablePredecessors(String nodeId) { Set visited = new HashSet<>(); Queue queue = new LinkedList<>(); queue.add(nodeId); while (!queue.isEmpty()) { String cur = queue.poll(); for (EdgeInfo edge : incomingEdges.getOrDefault(cur, emptyList())) { if (visited.add(edge.getSource())) { queue.add(edge.getSource()); } } } visited.remove(nodeId); return visited; } ``` > 前端 `ioInference.js` 中已实现等价的 `getReachablePredecessors(targetId, allNodes, allEdges)`（BFS 反向遍历边图），用于： > - 节点"前置数据关联"下拉选项（`WorkflowEditor.vue` 的 `inputMappingOptions`）：列出所有可达前序节点的输出字段供用户显式关联，**不仅限直接上游**。 > - 边映射状态判断（`collectReachableOutputs`）：合并所有可达前驱的输出做同名匹配。 > > 这保证编辑期可见的关联选项与运行期 `NodeWorkspace` 的可达变量集合对齐。 动态可达前驱（运行时，条件分支生效后）： - 条件节点仅把实际激活的分支下游加入可达集。 - 实现方式：`WorkflowLevelExecutor` 在节点执行后根据 `selectedBranch` 维护 `activeEdges`，动态可达集基于活跃边计算。 ### 4.2 节点工作区构建 ```java NodeWorkspace buildWorkspace(String currentNodeId, ResolvedDag dag, WorkflowContext context) { NodeWorkspace ws = new NodeWorkspace(); ws.getVariables().putAll(context.getInitialInputs()); Set reachable = dag.getReachablePredecessors(currentNodeId); for (String nodeId : topologicalOrder(reachable)) { Map scoped = context.getNodeScopedOutput(nodeId); if (scoped == null) continue; ws.getScopedOutputs().put(nodeId, scoped); for (Map.Entry e : scoped.entrySet()) { String scopedKey = nodeId + "__" + e.getKey(); ws.getVariables().put(scopedKey, e.getValue()); // 向后兼容：同时扁平注入（可选，受 feature flag 控制） ws.getVariables().putIfAbsent(e.getKey(), e.getValue()); } } ws.setFiles(scanWorkingDir(context.getWorkingDir())); return ws; } ``` ### 4.3 输入变量解析算法 对节点 `N` 的每个输入字段 `input`： ``` 1. 若 input.mapping 存在且 sourceNodeId 属于 reachable(N)： value = resolvePath(scopedOutputs[sourceNodeId], sourceField, sourcePath) return convert(value, input.type) 2. 否则进入隐式关联： candidates = 按拓扑逆序排列的 reachable(N) 中的前驱节点 2.1 精确同名：在 candidates 中查找第一个输出字段名 == input.name 2.2 模糊同名：忽略 [-_] 和大小写后匹配 2.3 JSON 嵌套精确同名：遍历 candidates 的所有输出变量，按 JSON Path 查找字段名 == input.name 2.4 JSON 嵌套模糊同名：忽略 [-_] 和大小写后匹配 3. 若找到匹配： 记录 mapping = {sourceNodeId, sourceField, sourcePath, autoMapped: true} return convert(value, input.type) 4. 若未找到： 若 input.required 为 true，前置条件校验失败 否则返回 null ``` ### 4.4 模糊匹配归一化 ```java static String normalizeForMatch(String s) { return s.replaceAll("[-_]", "").toLowerCase(Locale.ROOT); } ``` ### 4.5 JSON Path 遍历 支持点号路径，数组使用数字下标： - `output.total.result` → `value["output"]["total"]["result"]` - `items.0.name` → `value["items"][0]["name"]` 实现基于 Jackson `JsonNode` 或递归 Map，缺失路径返回 `null`。 --- ## 5. 类型转换规则 基于 `docs/workflow-node-fields.md` 实现 `VariableConverter`。 ### 5.1 通用转换 | 源类型 | 目标 string | 目标 number | 目标 boolean | 目标 object | 目标 array | |--------|------------|-------------|--------------|-------------|-----------| | String | 原值 | parseDouble / NaN | true/false/1/0/yes/no | JSON parse / wrap | JSON parse / wrap | | Number | toString | 原值 | != 0 | wrap | wrap | | Boolean | toString | 1/0 | 原值 | wrap | wrap | | Map | JSON string | NaN | !empty | 原值 | entry list | | Collection | JSON string | size | !empty | firstOrWrap | 原值 | ### 5.2 节点特定转换 #### KnowledgeRetrieval.evidences（List） | 目标类型 | 行为 | |----------|------| | string | 取 Top-1 的 content（可配置 top1/concat/json） | | number | evidenceCount | | boolean | 非空 | | object | Top-1 证据 Map | | array | Top-K 证据 Map 列表 | #### LLM/Agent/SmartAction.result（String） | 目标类型 | 行为 | |----------|------| | string | 原值 | | number | 提取首个数字 / NaN | | boolean | true/false/1/0/yes/no / 非空 | | object | JSON parse / wrap | | array | JSON parse / lines split | --- ## 6. 接口设计 ### 6.1 新增 Java 类 | 类名 | 职责 | |------|------| | `NodeWorkspace` | 节点工作区数据对象 | | `NodeWorkspaceBuilder` | 根据 DAG 和上下文构建工作区 | | `NodeInputResolver` | 解析节点输入变量，支持显式/隐式关联 | | `VariableConverter` | 类型转换器 | | `JsonPathExtractor` | JSON 嵌套路径取值 | | `WorkflowScopeResolver` | 可达前驱计算（静态+动态） | ### 6.2 修改的 Java 类 | 类名 | 修改点 | |------|--------| | `WorkflowContext` | 增加 `nodeScopedOutputs` 及相关方法 | | `DagResolver.ResolvedDag` | 增加 `incomingEdges` | | `WorkflowLevelExecutor` | 节点执行前构建工作区、解析输入；条件分支维护活跃边 | | `TemplateRenderer` | 支持 `{{nodeId.varName}}` 和 `{{nodeId.var.path}}` 语法 | | 各 Executor | 从 NodeWorkspace 读取输入，不再直接读扁平 variables | ### 6.3 前端改动 | 文件 | 修改点 | |------|--------| | `ioInference.js` | 新增 `getReachablePredecessors(targetId, allNodes, allEdges)`：通过边反向 BFS 计算可达前驱；`getNodeInputs` 返回 mapping 字段；`inferEdgeMapping` 增加隐式关联逻辑 | | `WorkflowEditor.vue` | `inputMappingOptions` 基于 `getReachablePredecessors` 列出**所有可达前驱节点**的输出（不仅限直接上游）；边变化时刷新相关节点 input mapping | | 节点配置面板 | 每个 input 增加"关联方式"选择器，选项来自全部可达前驱 | | `workflowNode.js` | 默认 data 中 inputs 增加 mapping 字段 | > **关联范围说明**：节点 `N` 的"前置数据关联"下拉选项 = `getReachablePredecessors(N.id)` 返回的所有节点 × 各节点的输出字段。即：从 `N` 反向 BFS 边图，凡是画布中可达的前序节点（直接上游、上游的上游、分叉合并的另一支等）输出都可显式关联。这与运行时 `NodeWorkspace` 的可达前驱定义一致（见 4.1 节），保证编辑期选项与运行期可见变量对齐。 --- ## 7. 兼容策略 ### 7.1 向后兼容 - 保留 `WorkflowContext.variables` 扁平表，旧工作流不开启命名空间也能运行。 - 新增 feature flag `workflow.node-scoped-variables.enabled`，默认 `false`。 - 当 flag 关闭时： - 仍然写入 `nodeScopedOutputs`（无影响）。 - `NodeWorkspaceBuilder` 仍然构造工作区，但 `variables` 同时保留扁平注入。 - 各执行器优先从工作区读取，未找到时 fallback 到扁平 variables。 ### 7.2 数据迁移 - 旧 graphData 中的 `inputs` 没有 `mapping` 字段，解析时按 `null` 处理，走隐式关联。 - 前端保存时自动补全 `mapping` 字段。 ### 7.3 条件分支 - 静态规划阶段使用保守可达集（所有分支都视为可达）。 - 运行时根据 `selectedBranch` 动态过滤，确保未激活分支的变量不进入下游工作区。 --- ## 8. 测试策略 ### 8.1 单元测试 - `WorkflowScopeResolverTest`：线性 DAG、分支 DAG、循环检测。 - `NodeInputResolverTest`：四级隐式关联 fallback、显式 mapping、关联失效。 - `VariableConverterTest`：通用转换表、RagEvidence 转换、LLM result 转换。 - `JsonPathExtractorTest`：对象嵌套、数组下标、缺失路径。 ### 8.2 集成测试 - 复杂 DAG 执行后，验证下游节点工作区只包含可达前驱变量。 - 同名变量场景：多个前驱输出同名变量，下游通过 `nodeId.varName` 精确引用。 ### 8.3 回归测试 - 旧工作流（无 mapping）执行结果与改造前一致。 --- ## 9. 实现顺序建议 为控制风险，建议按以下 MVP → 增强 → 完善的顺序实现： 1. **MVP**： - `WorkflowContext` 节点级输出存储 - `DagResolver` 反向索引 - `NodeWorkspace` 构建 - `VariableConverter` 通用转换 - `NodeInputResolver` 显式 mapping + 精确同名隐式关联 - `TemplateRenderer` 支持 `{{nodeId.varName}}` - 修改 `WorkflowLevelExecutor` 和主要 Executor 2. **增强**： - 2.2 / 2.3 / 2.4 级隐式关联 - 动态条件分支可达集 3. **完善**： - 前端 mapping UI - 自动保存 mapping - feature flag 默认开启并移除兼容代码 --- ## 10. 相关文档 - `docs/workflow-node-fields.md`：节点自动输入输出字段与转换规则 - `docs/workflow-scheduling-engine-plan.md`：工作流调度引擎设计 # TODO ☐ 将设计写入 docs/workflow-variable-scope-design.md ☐ 阶段1：后端基础设施（WorkflowContext、DagResolver、NodeWorkspace、VariableConverter） ☐ 阶段2：输入变量解析（NodeInputResolver、WorkflowLevelExecutor、各执行器） ☐ 创建 NodeInputResolver（显式 mapping + 四级隐式 fallback） ☐ 更新 TemplateRenderer 支持 {{nodeId.varName}} 语法 ☐ 更新 ContextPromptHelper 支持 NodeWorkspace / Map 变量 ☐ 更新 NodeExecutor 接口与各执行器使用 NodeWorkspace ☐ 更新 NodeWorkspaceBuilder 为 Spring Bean 并支持 activeEdges ☐ 更新 WorkflowLevelExecutor 构建工作区、解析输入、维护活跃边 ☐ 编译并修复错误 ☐ 阶段3：前端配置界面（data结构、ioInference、mapping UI） ☐ 阶段4：兼容性与测试 ☐ 添加 NodeInputResolver / TemplateRenderer / VariableConverter / WorkflowScopeResolver 测试