FastGPT 工作流 Runtime 引擎解析基于 Tarjan SCC 与 DFS 边分类的节点调度与执行机制【免费下载链接】FastGPTFastGPT is a knowledge-based platform built on the LLMs, offers a comprehensive suite of out-of-the-box capabilities such as data processing, RAG retrieval, and visual AI workflow orchestration, letting you easily develop and deploy complex question-answering systems without the need for extensive setup or configuration.项目地址: https://gitcode.com/GitHub_Trending/fa/FastGPTFastGPT 的可视化 AI 工作流本质上是一个有向图执行引擎用户在画布上编排的节点 连线会在运行时被编译为带状态的有向图由WorkflowQueue统一调度。本文以仓库设计文档 runtime.md 为主体结合 dispatch/index.ts 与 tarjan.ts 的源码实现系统讲解 FastGPT 工作流 Runtime 如何通过 Tarjan 强连通分量算法识别循环、通过 DFS 边分类区分回边并基于节点边分组 边状态机完成分支、循环、并行、工具调用等复杂拓扑的正确调度。读完本文你将掌握 Runtime 的核心数据结构、边分组算法、节点运行状态判定规则以及它在典型工作流拓扑下的行为与测试覆盖情况。一、Runtime 整体架构FastGPT 工作流 Runtime 的设计目标是把画布上任意可连的有向图允许分支、循环、并行、交叉变成一组语义清晰、可判定的执行规则。整个执行体系围绕三块内容展开图论分析层使用 Tarjan 算法找出强连通分量SCC判断节点是否处于循环使用 DFS 对每条边做分类树边 / 回边 / 前向边 / 跨边识别循环边。边分组层为每个节点的所有输入边构建边分组NodeEdgeGroups把复杂的图结构归约为若干组或/且关系的判定条件。队列调度层WorkflowQueue维护活跃节点队列activeRunQueue与跳过节点队列skipNodeQueue用迭代循环取代递归配合并发上限持续驱动节点执行直到所有路径收敛。1.1 WorkflowQueue核心执行类WorkflowQueue定义于 dispatch/index.ts是工作流执行的心脏负责管理节点执行队列与状态。其关键属性包括属性作用runtimeNodesMap节点 ID 到运行时节点对象的映射初始化于 index.tsedgeIndex边的索引按 source 和 target 两个维度预构建bySource/byTargetnodeEdgeGroupsMap预构建的节点 → 输入边分组 Map运行时直接查询避免重复计算activeRunQueue活跃运行队列存放待检查/待运行的节点 IDskipNodeQueue跳过节点队列存放整条被跳过路径上的节点关键方法一览buildEdgeIndex()扫描全部运行时边构建bySource、byTarget两个索引 Map实现于 index.ts。buildNodeEdgeGroupsMap()对每个节点执行DFS 边分类 → Tarjan SCC 分析 → 按分支句柄分组三步流程一次性构建全图边分组index.ts。getNodeRunStatus()根据预构建的边分组判定节点应run、skip还是waitindex.ts。addActiveNode()把节点加入活跃队列若当前没有处理循环则触发startProcessing()index.ts。startProcessing()迭代式调度主循环控制并发并处理跳过队列index.ts。1.2 Tarjan 算法模块图论分析能力集中在 packages/service/core/workflow/utils/tarjan.ts对外暴露四个函数findSCCs()Tarjan 算法找出所有强连通分量返回nodeToSCC节点→SCC ID与sccSizesSCC ID→分量大小两个 Maptarjan.ts。classifyEdgesByDFS()对全图执行一次 DFS为每条边标注tree/back/forward/cross类型tarjan.ts。isNodeInCycle()通过 SCC 分量大小判断节点是否位于循环中sccSizes 1即视为循环tarjan.ts。getEdgeType()按source-target-sourceHandle三元组作为边的唯一键从边类型表中查询边的类型tarjan.ts。二、核心数据结构2.1 边的状态机运行时边的 schema 定义于 packages/global/core/workflow/type/edge.ts在存储边StoreEdgeItemTypesource/sourceHandle/target/targetHandle基础上扩展出status字段type EdgeStatus waiting | active | skipped;三种状态的语义waiting等待执行——源节点尚未完成或该边对应的执行路径尚未走到active已激活——源节点已执行完成把结果送达了这条边skipped已跳过——源节点被跳过如分支未选中该边所在路径被剪枝。初始状态在 runtime/utils.ts 的storeEdges2RuntimeEdges中统一置为waiting每次节点执行结束后由调度逻辑更新下游边状态见 index.ts命中skipHandleId的输出桩对应的边置为skipped其余置为active。2.2 边的类型图论视角type EdgeType tree | back | forward | cross;tree树边DFS 树中首次发现目标节点的边back回边从后代指向当前 DFS 路径上祖先的边——即循环边forward前向边从祖先指向后代的非树边cross跨边连接不同 DFS 子树的边。其中back边是循环判定的关键信号。2.3 节点边分组type NodeEdgeGroups RuntimeEdgeItemType[][]; type NodeEdgeGroupsMap Mapstring, NodeEdgeGroups;每个节点的所有输入边会被划分成若干组组内边是且关系全部满足才能运行组间是或关系任意一组满足即可运行。分组结构预构建一次、全流程复用是状态判定的唯一依据。三、核心算法3.1 边分组算法流程buildNodeEdgeGroupsMap的完整流程源码注释见 index.ts1. 全局 DFS 边分类 └─ 识别回边循环边 2. Tarjan SCC 算法 └─ 找出所有强连通分量 └─ 判断节点是否在循环中 3. 为每个节点构建边分组 ├─ 分类边回边 vs 非回边 ├─ 处理非回边 │ ├─ 节点在循环中 → 按 branchHandle 分组 │ └─ 节点不在循环中 → 所有非回边放在同一组 └─ 处理回边 └─ 按 branchHandle 分组源码中isBranchNode只认三种节点类型条件判断节点ifElseNode、意图分类节点classifyQuestion、用户选择节点userSelect见 index.ts。3.2 分组策略策略 1节点不在循环中所有非回边放在同一组。这类边是且的关系必须全部满足条件才能运行——典型代表是并行汇聚多条并行支路全部完成后汇合节点才执行。策略 2节点在循环中非回边按branchHandle分组回边也按branchHandle分组。不同组的边是或的关系任意一组满足即可运行——这样既能保证首次从入口进入循环也能保证循环体可以在回边激活后再次执行。3.3 branchHandle 查找findBranchHandle从边的源节点开始沿输入边向上回溯找到第一个分支节点及其输出桩句柄sourceHandle以此作为该边归属的分支路径标识若回溯不到任何分支节点则返回默认值common。回溯过程中若遇到分支节点则用其sourceHandle覆盖句柄否则保持当前句柄继续向上实现于 index.tsfindBranchHandle(edge) { // 从边的源节点开始向上回溯 queue [{ nodeId: edge.source, handle: edge.sourceHandle }] while (queue.length 0) { { nodeId, handle } queue.shift() // 如果当前节点是分支节点且有 handle返回 handle if (isBranchNode(node) handle) { return handle } // 继续向上回溯 for (inEdge of inEdges) { newHandle isBranchNode(sourceNode) ? inEdge.sourceHandle : handle queue.push({ nodeId: inEdge.source, handle: newHandle }) } } return common }groupEdgesByBranch先为每条边计算branchHandle再按句柄值聚合成组index.ts。四、节点运行状态判断节点只有三种运行状态run运行、skip跳过、wait等待。判定逻辑getNodeRunStatusindex.ts完全基于预构建的边分组getNodeRunStatus(node, nodeEdgeGroupsMap) { edgeGroups nodeEdgeGroupsMap.get(node.nodeId) // 1. 没有输入边 → 入口节点直接运行 if (!edgeGroups || edgeGroups.length 0) { return run } // 2. 检查是否可以运行任意一组边满足条件 // 每组边内至少有一个 active且没有 waiting if (edgeGroups.some(group group.some(edge edge.status active) group.every(edge edge.status ! waiting) )) { return run } // 3. 检查是否跳过所有组的边都是 skipped if (edgeGroups.every(group group.every(edge edge.status skipped) )) { return skip } // 4. 否则等待 return wait }三条判定规则总结规则条件结果运行条件任意一组边至少一个active且没有waitingrun跳过条件所有组的边全部为skippedskip等待条件不满足以上两者wait在调度主循环中index.tsrun状态的节点会先把所有指向它的边重置为waiting再执行节点逻辑nodeRunWithActiveskip状态的节点同样重置入边状态后执行跳过处理nodeRunWithSkip并消耗maxRunTimes预算跳过节点扣减 0.1正常运行扣减实际运行次数见 index.ts 与 index.tsmaxRunTimes 0时整个工作流终止——这是防止死循环的兜底机制。五、Tarjan 强连通分量算法Tarjan 算法用于在有向图中找出所有强连通分量Strongly Connected Components, SCC。强连通分量定义在有向图中若从节点 A 能到达节点 B且从节点 B 也能到达节点 A则 A、B 同属一个强连通分量。SCC 大小 1 即表示存在循环自环节点自身构成大小为 1 的分量但大小为 1 的分量不一定在循环中因此循环判定只看size 1。核心实现tarjan.ts使用lowLink与discoveryTime两个辅助表以栈 迭代遍历的方式为每个节点分配 SCC IDfunction findSCCs(runtimeNodes, edgeIndex) { nodeToSCC new Map() sccSizes new Map() sccId 0 stack [] inStack new Set() lowLink new Map() discoveryTime new Map() time 0 function tarjan(nodeId) { // 初始化 discoveryTime.set(nodeId, time) lowLink.set(nodeId, time) time stack.push(nodeId) inStack.add(nodeId) // 遍历所有出边 for (edge of outEdges) { targetId edge.target if (!discoveryTime.has(targetId)) { // 未访问过递归访问 tarjan(targetId) lowLink.set(nodeId, min(lowLink.get(nodeId), lowLink.get(targetId))) } else if (inStack.has(targetId)) { // 在栈中更新 lowLink lowLink.set(nodeId, min(lowLink.get(nodeId), discoveryTime.get(targetId))) } } // 如果是 SCC 的根节点 if (lowLink.get(nodeId) discoveryTime.get(nodeId)) { sccNodes [] do { w stack.pop() inStack.delete(w) nodeToSCC.set(w, sccId) sccNodes.push(w) } while (w ! nodeId) sccSizes.set(sccId, sccNodes.length) sccId } } // 从所有未访问节点开始 for (node of runtimeNodes) { if (!discoveryTime.has(node.nodeId)) { tarjan(node.nodeId) } } return { nodeToSCC, sccSizes } }六、DFS 边分类算法classifyEdgesByDFS使用深度优先搜索对每条边进行分类tarjan.ts分类规则function classifyEdgesByDFS(runtimeNodes, edgeIndex) { edgeTypes new Map() visited new Set() inStack new Set() discoveryTime new Map() finishTime new Map() time 0 function dfs(nodeId) { visited.add(nodeId) inStack.add(nodeId) discoveryTime.set(nodeId, time) for (edge of outEdges) { targetId edge.target if (!visited.has(targetId)) { // 未访问 → 树边 edgeTypes.set(edgeKey, tree) dfs(targetId) } else if (inStack.has(targetId)) { // 在当前路径上 → 回边循环边 edgeTypes.set(edgeKey, back) } else if (discoveryTime.get(source) discoveryTime.get(targetId)) { // 从祖先指向后代 → 前向边 edgeTypes.set(edgeKey, forward) } else { // 跨边 edgeTypes.set(edgeKey, cross) } } inStack.delete(nodeId) finishTime.set(nodeId, time) } // 从所有入口节点开始 DFS for (node of entryNodes) { if (!visited.has(node.nodeId)) { dfs(node.nodeId) } } return edgeTypes }注意两点源码细节边的唯一键是source-target-sourceHandle三元组sourceHandle缺失时用default兜底见 tarjan.ts这意味着同一对节点之间允许存在多条不同输出桩的连线入口节点定义为没有输入边的节点tarjan.tsDFS 结束后还会补遍历孤立节点tarjan.ts确保图中所有节点都被分类。七、典型场景分析设计文档给出了 5 个代表性拓扑且每个场景在测试目录 packages/service/test/core/workflow/dispatch/checkNodeRunStatus 下都有对应的用例文件base.test.ts、toolcall.test.ts、safe.test.ts、boundary.test.ts、case.test.ts逐一验证。7.1 简单分支汇聚┌─ if ──→ B ──┐ start ──→ A ├──→ D └─ else ─→ C ──┘边分组D 节点只有一组[B→D, C→D]。运行逻辑A 走 if 分支B→D active, C→D skipped→ D 运行A 走 else 分支B→D skipped, C→D active→ D 运行B 还在执行B→D waiting, C→D skipped→ D 等待。对应测试见 base.test.ts场景 1简单分支汇聚。7.2 简单循环start ──→ A ──→ B ──→ C ──┐ ↑ | └────────────────┘边分组A 节点分为两组组1[start→A]、组2[C→A]C→A是回边。运行逻辑第一次执行start→A active, C→A waiting→ A 运行组 1 满足循环执行start→A skipped, C→A active→ A 运行组 2 满足两条边都 waitingstart→A waiting, C→A waiting→ A 等待。对应测试见 base.test.ts场景 2简单循环。7.3 分支 循环┌─ if ──→ B ──┐ start ──→ A ├──→ D ──┐ └─ else ─→ C ──┘ | ↑ | └──────────────────────┘边分组D 分为组1[B→D]、组2[C→D]循环内按 branchHandle 分组A 分为组1[start→A]、组2[D→A]。运行逻辑第一次走 if 分支B→D active, C→D skipped→ D 运行第一次走 else 分支B→D skipped, C→D active→ D 运行循环回来start→A skipped, D→A active→ A 运行。对应测试见 base.test.ts场景 3分支 循环。7.4 并行汇聚无分支节点start ──→ A ──→ C └──→ B ──→ C边分组C 只有一组[A→C, B→C]不在循环中非回边合为同组且关系。运行逻辑A 和 B 都完成A→C active, B→C active→ C 运行只有 A 完成A→C active, B→C waiting→ C 等待只有 B 完成A→C waiting, B→C active→ C 等待。对应测试见 base.test.ts场景 4并行汇聚。7.5 工具调用场景┌──selectedTools──→ Tool1 ──┐ start → Agent ─┤ ├──→ End └──────────────────────────→ ┘边分组Tool1 一组[Agent→Tool1 (selectedTools)]End 分为组1[Agent→End]、组2[Tool1→End]。运行逻辑Agent 调用 Tool1Agent→Tool1 active→ Tool1 运行Agent 不调用工具Agent→Tool1 skipped, Agent→End active→ End 运行Tool1 执行完成Tool1→End active, Agent→End active→ End 运行注意此时 Agent→End 也是 active两组同时满足体现分支与并行的混合语义。工具相关全部用例集中在 toolcall.test.ts覆盖单工具、多工具并行、嵌套工具调用、工具与分支结合四类场景。八、调度主循环从队列到执行的完整链路startProcessingindex.ts是驱动整个工作流运转的迭代式主循环源码注释index.ts概括了其设计使用activeRunQueue记录待检查的节点可能可以运行并控制并发数量每次添加新节点以及节点运行结束后均会执行一次processActiveNode检查若未触发跳出条件必定会继续从队列取节点处理节点运行后将 target 节点加入activeRunQueue等待下一轮处理。循环内的关键控制点结束条件activeRunQueue与运行中的 Promise 集合都为空时进入收尾——调试模式下若已无下一步运行节点且skipNodeQueue非空则先处理跳过节点index.ts。并发限制runningNodePromises.size this.maxConcurrency时用Promise.race等待最早完成的节点从而在每次节点结束的瞬间立刻推进下一轮index.ts。去重addActiveNode通过 Set 天然去重同一节点不会在队列中出现两次index.ts。边状态写入节点执行完成后依据skipHandleId批量更新所有下游边的active/skipped状态并把下一批活跃/跳过节点收集起来index.ts。九、性能优化9.1 预构建边分组优化前每次判断节点状态时都要重新计算边分组时间复杂度 O(n × m)n 为节点数m 为边数优化后在WorkflowQueue初始化时一次性构建所有节点的边分组运行时直接查Map状态判定退化为 O(1)。9.2 边索引优化前每次查找节点的输入/输出边都要遍历所有边时间复杂度 O(m)优化后构建bySource与byTarget两个 Map查询降为 O(1)EdgeIndex类型定义于 tarjan.ts。9.3 迭代替代递归优化前使用递归处理节点队列深层图可能导致调用栈溢出优化后startProcessing使用while (true)迭代循环替代递归的processActiveNode从根本上避免栈溢出问题。十、测试覆盖设计文档标注的测试文件路径为test/cases/global/core/workflow/dispatch/checkNodeRunStatus.test.ts在当前仓库中对应实际路径为 packages/service/test/core/workflow/dispatch/checkNodeRunStatus按文件拆分base、toolcall、safe、boundary、case。已覆盖场景对应设计文档清单简单分支汇聚简单循环分支 循环并行汇聚无分支节点所有边都 skipped多层分支嵌套嵌套循环多个独立循环汇聚复杂有向有环图多入口多循环自循环节点用户工作流 - 多层循环回退复杂分支与循环混合多层嵌套循环退出极度复杂多分支多循环交叉部分场景工具调用 - 单工具场景工具调用 - 多工具并行场景工具调用 - 嵌套工具调用场景工具调用 - 工具与分支结合场景测试规模设计文档记录总测试数 72、通过 72、失败 0针对核心checkNodeRunStatus判定逻辑。按当前仓库拆分后的文件统计仅base.test.ts即含 72 个用例加上toolcall22、safe28、boundary11、case10覆盖了从基础拓扑到边界与安全性如死循环防护的完整矩阵。10.1 场景 14 的问题分析问题场景 14.7 测试失败——期望节点 F 在只有一条边 active 时等待但实际返回run。原因场景 14 包含 D→E 的交叉路径导致 F 的两条输入边D→F 和 E→F被分成了不同的组。当 D→F active 时第一组已满足条件F 因此可以运行。解决方案删除场景 14.7 测试理由有三场景 14 是极端复杂的测试场景不应出现在实际工作流中在当前分组逻辑下D→F 与 E→F 来自不同分支它们是或的关系当 D→F active 时 F 可以运行这本身符合分支逻辑的语义。这一取舍体现了设计意图运行时语义以分支路径为准而非以汇聚完整性为准——来自不同分支的边天然是或关系不做强制合流等待。十一、设计原则11.1 分支语义或关系来自不同分支的边是或的关系任意一个分支满足条件即可运行——如 if-else 分支且关系来自同一分支的边是且的关系所有边都必须满足条件才能运行——如并行汇聚。11.2 循环处理循环识别使用 Tarjan SCC 算法识别循环SCC 大小 1 即表示存在循环循环边分组回边循环边按branchHandle分组不同循环路径的边分入不同组保证循环体的重入能力。11.3 应避免的复杂场景跨分支的交叉路径如 D→E多个循环出口如 G→A 和 G→C过度嵌套的分支和循环。原因难以理解和维护、容易出现逻辑错误、性能开销大、用户体验差。这与场景 14.7 被删除的决策一脉相承——Runtime 在正确性与图可理解性之间做了明确取舍。十二、未来优化方向设计文档从三个方向展望了演进空间性能优化并行执行优化更智能的并发控制、内存优化减少中间状态存储、缓存优化缓存常用计算结果功能增强更丰富的分支类型支持、更灵活的循环控制、更强大的错误处理可观测性更详细的执行日志、更直观的执行可视化、更完善的性能监控。十三、相关文件索引核心代码packages/service/core/workflow/dispatch/index.ts —WorkflowQueue类边索引、边分组、状态判定、队列调度packages/service/core/workflow/utils/tarjan.ts — Tarjan SCC 算法、DFS 边分类packages/global/core/workflow/runtime/type.ts — 运行时节点类型RuntimeNodeItemType与节点响应 schemapackages/global/core/workflow/runtime/utils.ts — 运行时工具函数含storeEdges2RuntimeEdges初始化边状态packages/global/core/workflow/type/edge.ts — 存储边与运行时边的 schema 定义测试文件packages/service/test/core/workflow/dispatch/checkNodeRunStatus/base.test.ts — 基础拓扑分支、循环、并行状态判定测试packages/service/test/core/workflow/dispatch/checkNodeRunStatus/toolcall.test.ts — 工具调用场景测试packages/service/test/core/workflow/dispatch/checkNodeRunStatus/safe.test.ts — 边界与安全场景测试packages/service/test/core/workflow/dispatch/checkNodeRunStatus/boundary.test.ts — 边界条件测试packages/service/test/core/workflow/dispatch/checkNodeRunStatus/case.test.ts — 复杂组合场景测试总结FastGPT 工作流 Runtime 采用基于图论的设计通过 Tarjan SCC 算法识别循环、DFS 边分类区分回边再以节点边分组为桥梁把任意复杂的有向图归约为组间或、组内且的判定模型最终由WorkflowQueue的迭代式调度主循环统一驱动节点执行。分支、循环、并行、工具调用等场景在预构建边分组与三层状态判定run / skip / wait下均能获得确定且正确的执行语义。通过预构建边分组、边索引、迭代替代递归等优化Runtime 在保证正确性的同时具备良好的性能表现与栈安全特性并配有覆盖 18 类场景的测试矩阵当前仓库拆分后合计 143 个用例作为正确性保障。未来可在并行执行、错误处理与可观测性方面继续演进进一步释放复杂工作流编排的能力边界。【免费下载链接】FastGPTFastGPT is a knowledge-based platform built on the LLMs, offers a comprehensive suite of out-of-the-box capabilities such as data processing, RAG retrieval, and visual AI workflow orchestration, letting you easily develop and deploy complex question-answering systems without the need for extensive setup or configuration.项目地址: https://gitcode.com/GitHub_Trending/fa/FastGPT创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考