当前位置：首页 > news >正文

拆解 FastGPT：知识库 + 工作流 + 对话的三合一架构

news 2026/6/1 12:17:52

🦞 一只用 AI Agent 搭副业产线的程序员

上篇拆 Dify 的时候，有个读者留言：虾哥，Dify 是做通用 AI 应用平台的，那 FastGPT 跟它区别在哪？

好问题。如果 Dify 是"低代码 AI 应用工厂"，FastGPT 就是"开箱即用的知识库问答系统"。它不是"什么都能做"，而是把"知识库 + 工作流 + 对话"这三个场景做到深。

这篇文章拆开 FastGPT 的代码，看它怎么把三条主线拧成一股绳。

项目简介

FastGPT（GitHub 20k+ Stars）是国内最流行的开源知识库问答平台。核心能力：上传文档 → 自动切片 → 向量化 → 对话时检索 → LLM 生成答案。2024 年后加入了工作流（Flow）能力，让用户可以在知识库 RAG 的基础上编排自定义流程。全栈 TypeScript（Next.js + Fastify + PostgreSQL + MongoDB）。

架构全景

┌──────────────────────────────────────────────────────────────┐ │ 前端（Next.js + React Flow） │ │ 知识库管理 · 工作流编辑器 · 对话调试 · 数据看板 │ ├──────────────────────────────────────────────────────────────┤ │ API 服务（Fastify + TypeScript） │ │ /api/core/dataset · /api/core/app · /api/core/chat · ... │ ├──────────────────────────────────────────────────────────────┤ │ 核心三大引擎 │ │ ┌─────────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Dataset 引擎 │ │ Workflow 引擎 │ │ Chat 引擎 │ │ │ │ 文档分割→向量化 │ │ 节点编排→执行 │ │ 会话管理→生成 │ │ │ │ →混合检索→重排 │ │ →变量流转 │ │ →上下文拼接 │ │ │ └─────────────────┘ └──────────────┘ └──────────────┘ │ ├──────────────────────────────────────────────────────────────┤ │ 数据层 │ │ PostgreSQL（元数据）· MongoDB（文档/chunk/对话）· Milvus（向量）│ └──────────────────────────────────────────────────────────────┘

注意：FastGPT 的三个引擎不是独立的三个微服务——它们在同一个 Node 进程里协同工作。下面分别拆开。

关键设计一：Dataset 的分块策略——不是简单的 Split

大多数 RAG 教程的做法是：文档 → split by separator → 固定窗口切块 → 存向量。FastGPT 的切块引擎比这复杂。看核心逻辑：

// packages/service/core/dataset/data/ —— 切块逻辑（概念性重建）interfaceChunkSplitConfig{chunkSize:number;// 每块最大长度（token 数）chunkOverlap:number;// 重叠 token 数customSplitSymbols?:string[];// 用户自定义分隔符}interfaceChunkResult{chunks:Array<{q:string;// 索引内容（用于向量匹配的文本）a:string;// 期望答案（QA 模式）chunkIndex:number;// 当前块在文档中的序号pageContent:string;// 块完整原文metadata:{source:string;sourceName:string;fileId:string;parentId?:string;// 上一层级的块 ID（层级关系）childrenId?:string[];// 下一层级的块 ID};}>;}classDataSplitter{// 核心：不只是切，还维护块的层级关系asyncsplitText(text:string,config:ChunkSplitConfig):Promise<ChunkResult>{// Step 1: 按段落自然分割（不是固定窗口傻切）constparagraphs=this.splitByNaturalBoundary(text);// Step 2: 在段落基础上按 token 数合并// 宁可少切、不可断句letchunks=this.mergeByTokenLimit(paragraphs,config.chunkSize);// Step 3: 加入重叠——前后块各留一段尾巴chunks=this.addOverlap(chunks,config.chunkOverlap);// Step 4: 构建块的层级树// 子块记录 parentId，父块记录 childrenId// 检索命中了块可以顺藤摸瓜找回上下文chunks=this.buildHierarchy(chunks);returnchunks;}}

FastGPT 切块的两个核心差异：

语义边界优先：先按段落、标题等自然边界分割，再在 token 限制内合并。这样切出来的块是"一段完整的话"，而不是"在句子中间被切断的半句话"。
块的层级关系：parentId/childrenId维护了一棵树。检索命中一个叶子节点后，能沿着 parentId 链找回父级块的完整上下文。这是 LlamaIndex Node 的relationships思路，但在 FastGPT 里做得更实用——就是树，不用复杂的图。

设计洞察：好的切块策略不是"怎么切"，而是"怎么在检索时找回上下文"。切得再好，丢失了上下文关系也是白搭。parentId 链就是最低成本的关系维护方案。

关键设计二：Workflow 的节点系统——可暂停、可分支、有状态

FastGPT 的工作流引擎和 Dify 的有本质区别：FastGPT 的节点是运行时状态机，不是无状态的函数。

// packages/service/core/workflow/ —— 工作流节点定义（概念性重建）enumWorkflowNodeTypeEnum{systemInput='systemInput',// 用户输入chatNode='chatNode',// LLM 对话（可挂载知识库）datasetSearch='datasetSearch',// 知识库搜索codeNode='codeNode',// 代码执行（沙箱）ifElseNode='ifElseNode',// 条件判断httpNode='httpNode',// HTTP 请求toolNode='toolNode',// 插件/工具调用answerNode='answerNode',// 指定回复（不调 LLM）userSelect='userSelect',// 用户交互——暂停等用户选择loopNode='loopNode',// 循环节点variableUpdate='variableUpdate',// 更新变量}// 每个节点定义了自己的输入输出 Schema 和运行时行为interfaceWorkflowNodeDefinition{type:WorkflowNodeTypeEnum;inputs:VariableSchema[];// 输入变量的 JSON Schemaoutputs:VariableSchema[];// 输出变量的 JSON Schema// 核心：节点的运行函数——接收输入、更新运行时状态、返回输出run:(ctx:NodeRunContext)=>Promise<NodeRunResult>;}

关键的设计差异：userSelect节点。它不是"执行完就往下走"，而是暂停工作流、把控制权交给用户。用户选了选项之后，工作流从暂停点继续执行。

// 简化的运行时状态interfaceWorkflowRuntimeState{workflowId:string;currentNodeId:string;nodeStates:Map<string,{status:'waiting'|'running'|'completed'|'paused'|'error';input:Record<string,any>;output:Record<string,any>;}>;variables:Record<string,any>;// 全局变量池history:ChatHistoryItem[];// 对话历史// 暂停信息pauseReason?:'user_select'|'error_retry'|'approval';pauseData?:any;}

这个"可暂停"的设计让 FastGPT 的 Workflow 不只是个批处理引擎——它支持人机协作。比如"客服机器人检测到用户要投诉 → 暂停 → 弹通知给人工客服 → 人工客服介入 → 工作流继续"。

设计洞察：不是所有工作流都应该全自动。真实场景里，"AI 处理大部分 + 关键时刻转人工"是更务实的方案。FastGPT 把"暂停"作为一等公民设计到引擎里，而不是事后打补丁加的。

关键设计三：Chat 的上下文拼接——不是简单的"全量塞进去"

对话引擎的核心挑战：用户可能开了很长一段对话，但 LLM 有上下文窗口限制。怎么裁剪对话历史？

FastGPT 的做法不是截断最近 N 条——而是按节点动态拼接：

// packages/service/core/chat/ —— 上下文拼接逻辑（概念性重建）classChatContextBuilder{asyncbuildContext(chatHistory:ChatItem[],workflow:Workflow,runtimeState:WorkflowRuntimeState):Promise<ChatCompletionMessage[]>{letmessages:ChatCompletionMessage[]=[];// Step 1: 拼接系统提示词// 由两部分组成：工作流的 systemPrompt + 知识库的 contextconstsystemPrompt=awaitthis.buildSystemPrompt(workflow,runtimeState);// Step 2: 拼接知识库检索结果// 每个 datasetSearch 节点的结果都被注入到对话上下文constkbContext=runtimeState.nodeStates.filter(n=>n.status==='completed'&&n.type==='datasetSearch').map(n=>n.output.searchResults).flat();messages.push({role:'system',content:`${systemPrompt}\n\n参考资料：\n${kbContext.join('\n')}`});// Step 3: 动态裁剪历史对话// 不是硬截断，而是智能压缩consttokenBudget=this.getMaxTokens()-this.estimateTokens(messages);consttrimmedHistory=this.smartTrim(chatHistory,tokenBudget);messages.push(...trimmedHistory);returnmessages;}smartTrim(history:ChatItem[],budget:number):ChatMessage[]{// 策略：// 1. 始终保留最近的 N 条（"工作记忆"）// 2. 对更早的对话做摘要压缩（"长期记忆"）// 3. 如果还超，逐条丢弃最旧的// 这样保证关键信息不丢失constrecent=history.slice(-KEEP_RECENT_N);constolder=history.slice(0,-KEEP_RECENT_N);if(this.estimateTokens(older)>budget*0.3){// 老对话太多——压缩成摘要constsummary=awaitthis.summarizeHistory(older);return[{role:'system',content:`对话摘要：${summary}`},...recent];}return[...older,...recent];}}

三种上下文拼接策略：

系统提示词：来自工作流配置 + 知识库检索结果。这是"静态知识"，每次对话不变。
最近 N 条对话：始终保留，是 LLM 的"工作记忆"。保证它知道刚才在聊什么。
历史对话压缩：对更早的对话生成一句话摘要。这是"长期记忆"——用最小 token 代价保留关键信息。

设计洞察：上下文窗口不是无限大，但对话可能无限长。"保留最近 + 压缩历史"是人类记忆的工作原理——FastGPT 把它用到了 LLM 的上下文管理里。

核心代码拆解：知识库的混合检索

FastGPT 的检索不是纯向量搜索——它做了向量 + 关键词的混合检索，然后 Rerank 重排：

// packages/service/core/dataset/search/ —— 混合检索（概念性重建）classDatasetSearchEngine{asyncsearch(params:SearchParams):Promise<SearchResult[]>{// Step 1: 向量检索（ANN —— 近似最近邻）constvectorResults=awaitthis.vectorSearch(params.datasetId,params.queryEmbedding,params.topK*2// 多召回一倍，给后面的重排留余地);// Step 2: 关键词检索（倒排索引 —— PostgreSQL 的 ts_vector）constkeywordResults=awaitthis.keywordSearch(params.datasetId,params.query,params.topK*2);// Step 3: RRF（Reciprocal Rank Fusion）融合// 不是简单合并，而是用倒数排名加权constfusedResults=this.rrfFusion(vectorResults,keywordResults,RRF_K=60// 控制排名衰减速度);// Step 4: Rerank 重排（Cross-Encoder 模型）constrerankedResults=awaitthis.rerank(fusedResults.slice(0,params.topK*2),// 取融合后的前 N 条params.query);// 返回最终 topKreturnrerankedResults.slice(0,params.topK);}rrfFusion(vectorResults:ScoredDoc[],keywordResults:ScoredDoc[],k:number):ScoredDoc[]{// RRF 公式: score = sum(1 / (k + rank))// 在两边都排得高的文档会得到最高分constscoreMap=newMap<string,number>();for(leti=0;i<vectorResults.length;i++){scoreMap.set(vectorResults[i].id,1/(k+i+1));}for(leti=0;i<keywordResults.length;i++){constprev=scoreMap.get(keywordResults[i].id)||0;scoreMap.set(keywordResults[i].id,prev+1/(k+i+1));}// 按融合分降序return[...scoreMap.entries()].sort((a,b)=>b[1]-a[1]).map(([id,score])=>({id,score}));}}