FastGPT — 架构与原理
30 秒导读： FastGPT 是一个用「拖拽画布」搭 AI 应用的平台。你在画布上连出一张流程图（LLM 对话、知识库检索、判断、工具调用……），后端把这张图编译成一张有向图，再用一个叫
WorkflowQueue的调度内核一个节点一个节点地跑，边跑边通过 SSE 把结果流式吐给前端。理解 FastGPT，核心就是理解这台「图执行引擎」怎么决定谁先跑、谁被跳过、卡在用户输入时怎么存档续跑。
1. 这是什么（零基础也能懂）
一句话定义： FastGPT 是一个 AI Agent 构建平台——用可视化的 Flow（工作流）编排，把 LLM、知识库、工具串成一个能对话的应用（依据：README.md:16）。
解决什么问题 / 给谁用：
假设你要做一个「公司客服机器人」：它得先判断用户问的是售前还是售后，售后就去知识库查文档、必要时调一个查订单的 API，最后用大模型把答案组织成话说出来。
- 不用 FastGPT，你得自己写调度、写检索、写工具调用循环、写流式返回。
- 用 FastGPT，你在画布上把这些能力拖成节点、连成线，平台替你把它跑起来。
面向的是搭应用的人（低代码搭 AI 应用）和读源码的工程师（想知道这台引擎怎么实现 ）。
它能做什么（功能）：
| 能力 | 对应节点（举例） |
|---|---|
| 大模型对话 | AI 对话节点（chatNode） |
| 让模型自己调工具 | 工具调用节点（toolCall）、规划 Agent（agent） |
| 知识库检索（RAG） | 知识库搜索节点（datasetSearchNode） |
| 流程控制 | 判断器（ifElseNode）、问题分类（classifyQuestion）、循环（loopRun）、并行（parallelRun） |
| 人机交互 | 用户选择（userSelect）、表单输入（formInput）——会暂停工作流等用户 |
| 外部集成 | HTTP 请求、代码沙箱、读文件、子应用 / 插件 |
（依据：节点类型到执行函数的总注册表 packages/service/core/workflow/dispatch/constants.ts:38）
用起来什么样： 对外是一个 OpenAI 风格的对话接口。前端 / API 调用者往 /api/v2/chat/completions 发消息，服务端跑完工作流，用 SSE 把答案流式返回：
# 示意，非源码：一次流式对话请求
curl -N https://<host>/api/v2/chat/completions \
-H 'Authorization: Bearer <apikey>' \
-d '{ "chatId": "abc", "stream": true,
"messages": [{ "role": "user", "content": "退货怎么退？" }] }'
# 返回：一串 SSE 事件（node 状态、answer 文本片段、node 响应详情……）
一句话直觉 / 类比： 把 FastGPT 想成一台流程图解释器。你画的图是「程序」，WorkflowQueue 是「CPU」，它按边的连接关系决定下一条指令跑哪个节点；遇到「等用户点选」就像断点——把现场存进数据库，下次请求再从断点续跑。
2. 顶层全景（它大概怎么转）
先看主线： 一次对话从 HTTP 请求进来，到 SSE 流式吐出，中间是「加载图 → 转成运行态 → 队列调度逐节点跑 → 边跑边流式回传」。
怎么读下面这张图： 从上到下是一次请求的时间顺序；虚线框 WorkflowQueue 是核心，它内部是一个「取节点 → 判断跑/跳/等 → 更新边 → 激活下游」的循环。
HTTP POST /api/v2/chat/completions
│
▼
┌───────────────────────────────────────────────┐
│ API 入口层 (completions.ts) │
│ · 鉴权 / 取历史 / 取 App 最新版本的图 │
│ · store 图 ──► runtime 图（节点+边+入口标记） │
│ · 建 SSE 流式响应上下文 │
└───────────────────────────────┬────────────────┘
│ dispatchWorkFlow(...)
▼
┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
调度内核 WorkflowQueue
│ ┌─────────────────────────────────────────────┐ │
│ activeRunQueue ← 入口节点 │
│ │ │ 取一个节点 │ │
│ ▼ │
│ │ 按「入边状态」判定 → run / skip / wait │ │
│ │run │skip │
│ │ ▼ ▼ │ │
│ 执行节点(callbackMap) 标记出边 skipped │
│ │ │ │ │ │
│ ▼ 出边 active/skipped ▼ │
│ │ 激活下游 → 回到取节点 跳过传播给下游 │ │
└─────────────────────────────────────────────┘
└ ─ ─ ─ ─ ─ ─ ─ ─ ─┬─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
│ 每个节点结果
▼
SSE 事件流（flowNodeStatus / answer / flowNodeResponse …）
│
▼
前端逐字渲染 + 落库(chat item)
部件一句话职责：
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
| API 入口 | 鉴权、拼历史、加载图、初始化 SSE、调 dispatchWorkFlow | projects/app/src/pages/api/v2/chat/completions.ts |
dispatchWorkFlow | 一次运行的外壳：查余额、建变量态、装停止/超时检查、收尾落库 | packages/service/core/workflow/dispatch/index.ts:125 |
WorkflowQueue | 真正的调度内核：队列 + 并发 + 跳过传播 + 交互暂停 | .../dispatch/index.ts:343 |
callbackMap | 「节点类型 → 执行函数」的总路由表 | .../dispatch/constants.ts:38 |
各 dispatchXxx 节点 | 具体能力：对话 / 检索 / 工具 / 判断… | .../dispatch/{ai,dataset,tools,...}/ |
| 运行态数据模型 | RuntimeNodeItemType / RuntimeEdgeItemType / 引用解析 | packages/global/core/workflow/runtime/ |
主线走一遍（高层，不进代码）：
- 请求进
completions.ts：鉴权、取会话历史、取该 App「最新版本」的图（nodes+edges+chatConfig）。 - store 图 → runtime 图：把存储态节点/边转成运行态，并标出入口节点（
workflowStart等；若上次是交互暂停，则入口是存档里的续跑点）。 - 调
dispatchWorkFlow：建好变量状态、SSE 写入器、停止信号轮询，然后把入口节点丢进WorkflowQueue。 - 队列反复「取节点 → 判 run/skip/wait → 跑 → 按出边激活下游」，跑到没有可跑节点为止。
- 期间每个节点通过
workflowStreamResponse把状态 / 答案 / 详情作为 SSE 事件推给前端；结束后把 AI 回复和运行详情落库。
3. 阅读地图（按这个顺序读）
FastGPT 的仓库很大，但「把一张图跑成一次对话」这条主线只涉及五块。建议顺序：
- 工作流的数据模型：节点 / 边 / 引用 / 变量 —— 打地基。先搞清楚一个节点长什么样、边怎么带状态、节点之间怎么用
[nodeId, outputId]互相引用取值。不懂数据模型，后面调度看不懂。 - 一次对话的端到端路径：completions API → 调度 → SSE 流式返回 —— 走通主干。从 HTTP 请求追到 SSE 返回，看清入口层做了哪些准备、SSE 有哪些事件、结果怎么落库。
- 调度内核：WorkflowQueue 怎么把一张图跑起来 —— 全库最核心。队列怎么运转 、节点凭什么判「跑 / 跳过 / 等待」、跳过怎么沿边传播、遇到环怎么办、交互怎么暂停与续跑。
- 让平台成为 Agent 的 LLM 节点：对话、工具调用循环、规划 Agent —— 上层能力。AI 对话节点怎么拼 prompt / 流式输出，工具调用节点怎么把「模型选工具 → 跑工具 → 回喂结果」跑成一个循环，规划 Agent 怎么做多步计划。
- 知识库（RAG）：从文档摄入到混合检索 —— 检索侧。向量 + 全文的混合召回、RRF 融合、重排、相似度过滤。
只想快速判断「这项目和我的任务相关吗」：读完 §1、§2 即可；要改引擎行为，直奔 03;要接 LLM/工具能力，看 04;要调检索效果，看 05。
4. 巧妙之处（值得带走的设计）
这一节挑几个「不显然、但很聪明」的设计，每条先说妙在哪，再给源码锚点。
4.1 用「边的状态」而不是「拓扑排序」来驱动执行
一般图执行会先做拓扑排序再按序跑。FastGPT 不排序，而是给每条边一 个状态：waiting（待定）/ active（激活，该走）/ skipped（跳过）。一个节点跑不跑，只看它入边的状态组合（依据：WorkflowQueue.getNodeRunStatus dispatch/index.ts:626）:
- 有一组入边「至少一条 active 且没有 waiting」→ run；
- 所有入边都 skipped → skip；
- 否则 → wait（等上游先出结果）。
妙在：这天然支持分支（判断器只把命中分支的出边设 active、其余设 skipped）和汇聚（多路汇合时按「边组」判断），不需要全局排序，节点可以在数据就绪的那一刻就被激活。
4.2 「跳过」也要沿着图传播，否则分支后的节点会被误跑
如果只是「不跑」被跳过的节点，那它下游的节点会一直 waiting、永远不结束。FastGPT 的做法是把跳过也当成一种「执行」：跳过一个节点时，把它的出边全标 skipped，并把下游节点加进专门的跳过队列 skipNodeQueue，让跳过状态继续往下传，直到整张图收敛（依据：跳过处理 dispatch/index.ts:775 processSkipNodes；出边标记与下游收集 dispatch/index.ts:1222 的 nodeOutput）。
这里还有个坑点处理：分支节点可能被「跳过递归」反向传播影响，所以跑过 / 跳过的节点会记进 skippedNodeIdList，保证递归 skip 至多传到当前分支节点、不越界（依据：dispatch/index.ts:1364-1374 附近注释与逻辑）。
4.3 交互节点 = 可持久化的「断点」，请求结束也能续跑
「用户选择 / 表单输入 / 余额不足暂停」这类节点会中断工作流。FastGPT 把当前现场——入口节点 id、边的状态、跳过队列、各节点已产出的 output——打包成一个 interactive 对象存进会话历史（依据：handleInteractiveResult dispatch/index.ts:1426，其中 memoryEdges 把入口前的边重新激活以保证下次能跑）。
下次请求进来，入口层发现上次是交互暂停，就用存档的 entryNodeIds / memoryEdges 重建运行态，从断点续跑（依据：storeEdges2RuntimeEdges runtime/utils.ts:207、getWorkflowEntryNodeIds runtime/utils.ts:221 里对 lastInteractive 的分支）。等于给一个无状态 HTTP 接口做了可恢复的执行。
4.4 用回调队列代替深递归，天然带并发限流
早期图执行容易写成深递归，节点一多就爆栈。WorkflowQueue 改成迭代式回调循环：activeRunQueue 存待检查节点，startProcessing 用一个 while 循环取节点、异步跑、跑完再触发下一轮（依据：dispatch/index.ts:694 startProcessing）。同一台队列里同时在跑的节点数由 maxConcurrency（默认 10）限流（依据：dispatch/index.ts:388 构造参数、:729 的并发判断），并且同一个节点同时只会跑一个实例。
4.5 有环也能跑：Tarjan SCC + DFS 回边识别
工作流允许「循环」（loopRun）等结构，图里就会出现环。若不识别环，「跳过传播」和「入边判定」会死循环。FastGPT 在构图阶段一次性做两件事：DFS 边分类找出回边、Tarjan 算法找出强连通分量（SCC，互相可达的一组节点），据此决定某个节点的入边要不要按分支句柄分组（依据：buildNodeEdgeGroupsMap dispatch/index.ts:454，调用 classifyEdgesByDFS / findSCCs / isNodeInCycle，来自 ../utils/tarjan）。这是把「本科图论」真正用在了产品调度里。
4.6 工具调用是「LLM 循环」，工具本身又是一段子工作流
工具调用节点不是调一次模型就完。它跑一个 agent loop（最多 50 轮）：模型选工具 → 执行工具 → 把结果回喂 → 模型再决定继续调还是收尾（依据：toolCall.ts:129 传入 maxRunAgentTimes: 50，循环体 runAgentLoop 在 ai/llm/agentLoop/loop/base.ts:186）。而「执行工具」本身往往是再跑一遍子工作流——于是 FastGPT 用 workflowDispatchDeep 限制递归深度（>20 直接返回），防止工具套工具无限展开（依据：runWorkflow dispatch/index.ts:1498，深度上限 :1504）。
5. 代码地图（导航索引）
按「主题 → 文件 → 关键符号」组织，符号名比行号抗漂移，可直接 grep。
引擎主线（最核心）
| 主题 | 文件 | 关键符号 |
|---|---|---|
| 一次运行的外壳（余额/变量态/停止信号/收尾） | packages/service/core/workflow/dispatch/index.ts | dispatchWorkFlow |
| 调度内核类 | packages/service/core/workflow/dispatch/index.ts | WorkflowQueue |
| 迭代式队列循环（取节点/并发/结束判定） | .../dispatch/index.ts | startProcessing |
| 单节点检查：run / skip / wait | .../dispatch/index.ts | checkNodeCanRun、getNodeRunStatus |
| 执行节点并格式化响应 | .../dispatch/index.ts | nodeRunWithActive |
| 跳过传播 | .../dispatch/index.ts | processSkipNodes、addSkipNode、nodeRunWithSkip |
| 出边状态更新 + 下游收集 | .../dispatch/index.ts | nodeOutput（在 checkNodeCanRun 内） |
| 交互暂停/存档现场 | .../dispatch/index.ts | handleInteractiveResult |
| 递归运行 + 深度限制(>20) | .../dispatch/index.ts | runWorkflow |
| 环检测（回边 + SCC） | .../dispatch/index.ts、.../workflow/utils/tarjan | buildNodeEdgeGroupsMap、classifyEdgesByDFS、findSCCs |
| 节点类型 → 执行函数 路由表 | .../dispatch/constants.ts | callbackMap |
对话管道与数据模型
| 主题 | 文件 | 关键符号 |
|---|---|---|
| 对话 API 入口（v2） | projects/app/src/pages/api/v2/chat/completions.ts | handler |
| store 图 → runtime 图 | packages/global/core/workflow/runtime/utils.ts | storeNodes2RuntimeNodes、storeEdges2RuntimeEdges |
| 入口节点判定（含交互续跑） | .../runtime/utils.ts | getWorkflowEntryNodeIds |
节点间引用取值 [nodeId, outputId] | .../runtime/utils.ts | getReferenceVariableValue |
| 运行态节点/调度上下文类型 | .../runtime/type.ts | RuntimeNodeItemType、ChatDispatchProps、ModuleDispatchProps |
| SSE 事件枚举 / 响应键枚举 | .../runtime/constants.ts | SseResponseEventEnum、DispatchNodeResponseKeyEnum |
AI 节点与知识库
| 主题 | 文件 | 关键符号 |
|---|---|---|
| AI 对话节点 | .../dispatch/ai/chat.ts | dispatchChatCompletion |
| 工具调用节点（外壳） | .../dispatch/ai/toolcall/index.ts | dispatchRunTools |
| 工具调用循环（编排 LLM loop + 跑工具） | .../dispatch/ai/toolcall/toolCall.ts | runToolCall |
| 通用 agent 循环（最多 50 轮） | .../ai/llm/agentLoop/loop/base.ts | runAgentLoop |
| 规划 Agent 节点 | .../dispatch/ai/agent/index.ts | dispatchRunAgent、dispatchPiAgent |
| 知识库搜索节点 | .../dispatch/dataset/search.ts | dispatchDatasetSearch |
| 检索核心（向量+全文+重排） | .../dataset/search/defaultRecall/index.ts | searchDatasetData、multiQueryRecall |
| RRF 融合（k=60，加权倒数排名） | packages/global/core/dataset/search/utils.ts | datasetSearchResultConcat |
| 检索模式枚举 | packages/global/core/dataset/constants.ts | DatasetSearchModeEnum |
说明：上表符号均来自 commit
03ff1b5d的真实源码。若上游更新导致行号漂移，用符号名grep即可重新定位；语义未变则本文结论仍成立。