FastGPT Workflow 節(jié)點(diǎn)響應(yīng)持久化改造Append-Only 存儲(chǔ)與交互恢復(fù) NodeResponse ID 設(shè)計(jì)解析【免費(fèi)下載鏈接】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.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/fa/FastGPTFastGPT 的 workflow 運(yùn)行詳情通過chat_item_responses集合平鋪持久化每條 row 的data即一個(gè)節(jié)點(diǎn)響應(yīng)nodeResponse。本文以倉(cāng)庫(kù)內(nèi)權(quán)威設(shè)計(jì)文檔node-response-append-only-interactive-id.md為主線結(jié)合 nodeResponseStorage.ts、nodeResponseSink.ts、mergeNode.ts 等實(shí)現(xiàn)源碼系統(tǒng)講解 append-only 數(shù)據(jù)模型、讀取時(shí)的增量合并算法、交互恢復(fù)場(chǎng)景下 nodeResponse ID 的復(fù)用規(guī)則以及運(yùn)行前preChatRound的職責(zé)邊界。讀完你將掌握 FastGPT 節(jié)點(diǎn)詳情從“運(yùn)行期可更新”遷移到“只追加 讀取時(shí)折疊”的完整設(shè)計(jì)思路以及交互恢復(fù)如何避免展示節(jié)點(diǎn)重復(fù)。背景從“可更新存儲(chǔ)”到“只追加存儲(chǔ)”的演進(jìn)workflow 運(yùn)行詳情通過chat_item_responses平鋪保存。每條 row 的data是一個(gè) nodeResponse包含三個(gè)關(guān)鍵身份字段data.id展示節(jié)點(diǎn) ID標(biāo)識(shí)一個(gè)節(jié)點(diǎn)響應(yīng)實(shí)例data.parentId父展示節(jié)點(diǎn) ID讀取時(shí)用于還原childrenResponses樹形結(jié)構(gòu)chatItemDataId所屬 AI chat item 的dataId即本輪響應(yīng)消息 ID。早期設(shè)計(jì)依賴{ appId, chatId, chatItemDataId, data.id }唯一索引并在運(yùn)行期先刪除同data.id的舊 row 再寫入新 row或用 replace 模式清空舊詳情。該方案有兩個(gè)核心痛點(diǎn)大表唯一索引成本高在承載海量節(jié)點(diǎn)詳情的大表上維護(hù)復(fù)合唯一索引寫入吞吐和鎖競(jìng)爭(zhēng)壓力大違背運(yùn)行期只追加的性能目標(biāo)運(yùn)行中頻繁 delete/update 增加寫放大且并行 retry 時(shí)“刪除舊 rows 再寫入”的時(shí)序很難保證一致性。因此當(dāng)前權(quán)威方案將 nodeResponse 表調(diào)整為append-onlyworkflow 運(yùn)行過程中只createrows不更新、不刪除。重復(fù)展示節(jié)點(diǎn)不再依賴數(shù)據(jù)庫(kù)去重而是通過讀取時(shí)按(data.id, data.parentId)fold折疊合并來還原最終形態(tài)。該設(shè)計(jì)文檔合并并替代了歷史文檔node-response-stream-persistence.md其中的data.idunique 索引、運(yùn)行期 delete 后 create、replace/append 模式、parallel retry 刪除舊 rows 等描述已過時(shí)以及舊版 append-only 討論稿。核心結(jié)論速覽設(shè)計(jì)文檔沉淀的結(jié)論如下chat_item_responses運(yùn)行期只追加 rows對(duì)話刪除、應(yīng)用刪除、過期清理等外部清理流程可以批量刪除。data.id不再是數(shù)據(jù)庫(kù)唯一鍵只表示前端展示節(jié)點(diǎn)身份。同一個(gè)data.id且parentId相同的多條 rows 表示同一個(gè)展示節(jié)點(diǎn)的多次增量讀取時(shí)合并成一個(gè)節(jié)點(diǎn)兩條 row 都沒有parentId時(shí)也視為同一個(gè) parent。mergeSignId已廢棄不再寫入、不再讀取、不再兼容舊合并語(yǔ)義。舊數(shù)據(jù)若依賴mergeSignId展示異?？山邮苓w移或回放另行處理。dispatchWorkFlow.responseChatItemId是必填運(yùn)行參數(shù)dispatch 不生成兜底 ID也不查詢MongoChatItem或MongoChatItemResponse判斷是否重復(fù)。保存對(duì)話記錄的新運(yùn)行必須先走preChatRound由業(yè)務(wù)入口完成最終chatId/responseChatItemId解析、生成鎖、AI dataId 沖突檢查和 Human/AI placeholder 預(yù)創(chuàng)建。普通新運(yùn)行中 Human 和 AI 使用同一個(gè)roundDataId responseChatItemId。Human/AI 同 dataId 是預(yù)期行為同一個(gè)obj下重復(fù) dataId 才是不合法語(yǔ)義。本輪運(yùn)行前只阻塞 AI dataId 沖突。數(shù)據(jù)模型與索引設(shè)計(jì)row 結(jié)構(gòu)chat_item_responses的核心字段定義如下對(duì)應(yīng) chatItemResponseSchema.tstype ChatItemResponseSchema { teamId: ObjectId; appId: ObjectId; chatId: string; chatItemDataId: string; data: ChatHistoryItemResType; time: Date; };在真實(shí) Schema 中appId字段注釋說明了其歷史物理字段名語(yǔ)義為sourceIdApp 場(chǎng)景才是真實(shí) appIdsourceType來自ChatSourceTypeEnumtime默認(rèn)為當(dāng)前時(shí)間。保留的索引當(dāng)前chat_item_responses保留兩個(gè)索引ChatItemResponseSchema.index({ appId: 1, chatId: 1, chatItemDataId: 1, _id: 1 }); ChatItemResponseSchema.index({ teamId: 1, time: -1 });索引用途{ appId, chatId, chatItemDataId, _id }按 AI chat item 拉取完整 nodeResponse rows并按_id: 1保持寫入順序源碼中復(fù)合索引包含_id避免詳情讀取時(shí)額外排序{ teamId, time: -1 }過期清理或團(tuán)隊(duì)維度清理。源碼中還額外定義了一個(gè){ sourceType, appId, chatId, chatItemDataId, _id }索引帶 TODO 注釋暫未全面檢查操作故未加 sourceType 索引的完整方案說明數(shù)據(jù)訪問正在向 sourceType 維度演進(jìn)。明確不再創(chuàng)建的索引ChatItemResponseSchema.index( { appId: 1, chatId: 1, chatItemDataId: 1, data.id: 1 }, { unique: true } );chat_items當(dāng)前保留普通索引ChatItemSchema.index({ appId: 1, chatId: 1, dataId: 1 }); ChatItemSchema.index({ appId: 1, chatId: 1, deleteTime: 1 }); ChatItemSchema.index({ appId: 1, chatId: 1, _id: -1 }); ChatItemSchema.index({ appId: 1, chatId: 1, obj: 1, _id: -1 });{ appId, chatId, dataId }不能改成 unique因?yàn)槠胀ㄐ逻\(yùn)行中 Human 和 AI 會(huì)共享同一個(gè)dataId同一輪消息的 Human/AI 記錄同 ID。如果后續(xù) AI dataId 沖突檢查需要優(yōu)化可以補(bǔ)普通索引{ appId, chatId, dataId, obj }但不加 unique。寫入路徑WorkflowNodeResponseWriter寫入封裝在WorkflowNodeResponseWriternodeResponseStorage.ts其工作流程為一個(gè) workflow 請(qǐng)求復(fù)用一個(gè) writer子 workflow、loop、parallel、toolcall 等共享該 writer。record()接收本次要保存的 nodeResponses補(bǔ)齊id/parentId、裁剪 dataset quote、計(jì)算childResponseCount轉(zhuǎn)成 flat rows對(duì)應(yīng)createChatItemResponseRows。recordWithParent()只給沒有parentId的 root child 補(bǔ)外層 parent已有parentId的響應(yīng)保持內(nèi)部層級(jí)避免破壞更細(xì)的層級(jí)結(jié)構(gòu)。writer 通過 promise queuewriteQueue串行化并發(fā)record保證 Mongo_id順序接近運(yùn)行期寫入順序——因?yàn)樽?workflow、parallel 分支可能并發(fā)調(diào)用同一個(gè) writer串行化后才能保證詳情展示順序穩(wěn)定。默認(rèn)batchSize 5達(dá)到閾值或 close 時(shí) flush。flush 只執(zhí)行create(rowsWithTime, { ordered: true, session, ...writePrimary })不做任何 delete/update/replace。普通寫入失敗重試 3 次NODE_RESPONSE_WRITE_RETRY_TIMES 3仍失敗則寫 slim rows只保留節(jié)點(diǎn)身份、名稱、類型、父子關(guān)系、運(yùn)行時(shí)間和消耗統(tǒng)計(jì)等關(guān)鍵字段的瘦身版本slim 仍失敗時(shí)丟棄本批詳情 rows 并記錄日志不阻斷主 workflow。saveChat需要的引用citeCollectionIds、錯(cuò)誤數(shù)和根節(jié)點(diǎn)積分由 writer 在運(yùn)行期維護(hù) summarysummaryContributionsMap按id parentId覆蓋避免 retry/完成態(tài)重復(fù)累計(jì)詳情 rows 寫庫(kù)失敗不影響這些摘要。值得注意的是寫入前不做 JSON/BSON 體積預(yù)估BSON 大小、不可序列化字段等問題統(tǒng)一交給 Mongo 寫入校驗(yàn)失敗后進(jìn)入 retry/slim fallback避免正常路徑額外 CPU 與臨時(shí)內(nèi)存開銷。flush 后會(huì)立即釋放 buffer降低運(yùn)行期內(nèi)存占用。另外數(shù)據(jù)集搜索節(jié)點(diǎn)datasetSearchNode的quoteList在入庫(kù)前會(huì)被瘦身slimQuoteListForStorage只保留id/chunkIndex/datasetId/collectionId/sourceId/sourceName/score等引用關(guān)聯(lián)、來源和分?jǐn)?shù)元信息移除 q/a 完整文本——因?yàn)橥暾?quote 體積很大且詳情展示只需要來源元信息瘦身可降低單條 row 過大導(dǎo)致 Mongo 寫失敗的概率。實(shí)時(shí)發(fā)布路徑WorkflowNodeResponseSinkNodeResponse 的持久化和實(shí)時(shí)發(fā)布統(tǒng)一由請(qǐng)求級(jí)WorkflowNodeResponseSink協(xié)調(diào)nodeResponseSink.ts一個(gè) workflow 請(qǐng)求只創(chuàng)建一個(gè) sink內(nèi)部復(fù)用同一個(gè)WorkflowNodeResponseWriter。root workflow、child workflow、Agent、ToolCall、LoopRun、ParallelRun共享該 sink。節(jié)點(diǎn)和 Agent adapter 只上交標(biāo)準(zhǔn) nodeResponse不直接操作 writer也不直接發(fā)送flowNodeResponseSSE。sink 為缺少 parentId 的響應(yīng)補(bǔ)調(diào)用方顯式傳入的 parentIdWorkflowNodeResponseInput.parentId調(diào)用 writer 規(guī)范化并寫入再按請(qǐng)求可見性配置發(fā)布本次響應(yīng)。writer 仍按batchSize批量物理寫 Mongo“接收一個(gè)、返回一個(gè)”指每個(gè)邏輯 nodeResponse 都產(chǎn)生獨(dú)立 SSE 事件不要求每條 response 單獨(dú)執(zhí)行 Mongo create。sink不負(fù)責(zé)RuntimeNodeResponseSummary、usage、計(jì)費(fèi)、child count 或控制流判斷這些仍由 WorkflowQueue/Agent collector 在各自運(yùn)行作用域內(nèi)計(jì)算避免跨作用域重復(fù)累計(jì)。同(id, parentId)的多條響應(yīng)仍是 append-only 增量sink 不去重、不覆蓋、不改變數(shù)值字段的增量語(yǔ)義。輸出協(xié)議矩陣V2streamtrue, detailtrue可見 nodeResponse 逐條發(fā)送flowNodeResponse客戶端按(id, parentId)拼樹結(jié)束時(shí)不再發(fā)送完整 nodeResponse 數(shù)組。V1streamtrue, detailtrue運(yùn)行期不發(fā)送單個(gè) nodeResponse結(jié)束時(shí)一次性發(fā)送flowResponses。V1/V2streamfalse, detailtrue結(jié)束時(shí)在 JSONresponseData中一次性返回。V2 Share 流式完整 nodeResponse 逐條寫庫(kù)對(duì)外先按 public node/field 規(guī)則過濾再逐條發(fā)送為保持pushResult2Remote原有回調(diào)契約運(yùn)行期間仍保留最終詳情數(shù)組。Share 可見性分層處理Share 可見性必須分層處理不能只依賴一個(gè)字段過濾函數(shù)responseAllDatafalsesink 只發(fā)布 public node 類型和字段并保留客戶端拼樹需要的id/parentId。Share workflow 內(nèi)部始終保留回答中的引用 IDwriter 始終接收完整 nodeResponse普通 API 保持retainDatasetCite原有語(yǔ)義。datasetquoteList入庫(kù)時(shí)繼續(xù)移除 q/a只保留引用關(guān)聯(lián)、來源和分?jǐn)?shù)等元信息。showCite控制公開 nodeResponse 是否包含quoteList關(guān)閉時(shí) SSE 與非流式 JSON 都不返回quoteList但不改寫 SSE 回答文本也不改變持久化數(shù)據(jù)?？蛻舳藳]有 quoteList 時(shí)不展示引用之后重新開啟配置并刷新 Share可以根據(jù)已保存的引用 ID 和 quote 元信息恢復(fù)展示。showRunningStatus控制flowNodeStatus/toolCall/toolParams/toolResponse等過程事件不直接禁止引用展示依賴的 publicflowNodeResponse。showSkillReferences繼續(xù)由 Agent 輸出鏈路控制并受showRunningStatus約束。showWholeResponse/showFullText/canDownloadSource繼續(xù)由前端能力和詳情/引用/文件接口鑒權(quán)sink 不替代這些權(quán)限檢查。明確隱藏內(nèi)部 workflow 的系統(tǒng)插件繼續(xù)既不寫入 child rows也不發(fā)布 child 事件只保留外層工具節(jié)點(diǎn)響應(yīng)。pushResult2Remote不屬于本次 SSE 改造范圍繼續(xù)使用運(yùn)行期finalResponseData調(diào)用/shareAuth/finish不增加延遲讀庫(kù)或回調(diào)協(xié)議變化。運(yùn)行期明確刪除的行為不按data.iddelete 舊 rows不做updateOne upsert不做 replace 模式不在持久化 buffer 中按data.id去重不依賴data.idunique 索引不為 retry 預(yù)生成 row_id做冪等極低概率重復(fù) create 產(chǎn)生的冗余 rows 由讀取 fold 吸收。persistToDb false 的場(chǎng)景persistToDb false的 writer 不寫 Mongo只保留 summary 和可選內(nèi)存詳情retainInMemory適用于 debug、eval、臨時(shí)運(yùn)行等不保存歷史的入口。這類入口仍必須給 dispatch 傳隨機(jī)responseChatItemId只是該 ID 不參與數(shù)據(jù)庫(kù)查重。讀取與合并按 (data.id, parentId) 折疊增量讀取時(shí)先按 chat item 拉 rowsMongoChatItemResponse.find( { appId, chatId, chatItemDataId }, { data: 1 } ).sort({ _id: 1 });然后composeNodeResponseDetail()調(diào)用mergeNodeResponseDataByIdAndParent()做 fold實(shí)現(xiàn)見 mergeNode.ts規(guī)則如下只處理存在data.id的 rows無 id 的 row 會(huì)被丟棄無法參與合并。合并 identity 是(data.id, data.parentId)parentId不存在時(shí)歸一為同一個(gè)空值getNodeResponseIdentityKey用\u0000分隔 id 與 parentId。同 identity 的多條 rows 合并為一個(gè)展示節(jié)點(diǎn)。數(shù)值字段按增量累加包括runningTime保留兩位小數(shù)、totalPoints、childResponseCount、tokens含 input/output/toolCall/embedding/reRank/extension等。llmRequestIds去重合并。compressTextAgent、deepSearchResult這類結(jié)構(gòu)化用量字段按現(xiàn)有規(guī)則累加。普通標(biāo)量字段以后到的 incoming 為準(zhǔn)。childrenResponses遞歸按同一規(guī)則合并。child row 早于 parent row 到達(dá)時(shí)先作為臨時(shí) rootparent 到達(dá)后回收掛到childrenResponses對(duì)應(yīng)appendNodeResponseByParent的 orphan 回收邏輯。批量讀取時(shí)使用mergeNodeResponseListByParent一次性掛樹算法先按id parentId合并同層增量再按 parentId 掛到childrenResponses避免每條 row 遞歸掃描已構(gòu)建的整棵樹在 loop/parallel 產(chǎn)生大量 rows 時(shí)把復(fù)雜度從接近 O(n2) 降到以線性掃描為主。歷史兼容邊界新數(shù)據(jù)統(tǒng)一使用childrenResponsespluginDetail/toolDetail/loopDetail/parallelDetail/loopRunDetail只作為歷史 detail 字段讀取和遞歸統(tǒng)計(jì)來源getChildrenResponses會(huì)把這些舊字段與childrenResponses一并收集不再作為新鏈路的通用寫入結(jié)構(gòu)chat_items.responseData已廢棄。讀取時(shí)如果獨(dú)立表沒有 rows才回退舊內(nèi)聯(lián)詳情getChatItemResponseData的 fallback 邏輯避免歷史數(shù)據(jù)被空結(jié)果覆蓋childTotalPoints不再對(duì)外保留mergeNodeResponseDataByIdAndParent最后會(huì)stripChildTotalPoints子節(jié)點(diǎn)積分展示由客戶端基于childrenResponses現(xiàn)場(chǎng)計(jì)算。NodeResponse ID 語(yǔ)義與交互恢復(fù)普通節(jié)點(diǎn)隨機(jī) ID普通節(jié)點(diǎn)首次運(yùn)行時(shí)生成隨機(jī)data.idgetNanoid()。這類 ID 不需要可預(yù)測(cè)也不需要數(shù)據(jù)庫(kù)唯一約束。交互恢復(fù)復(fù)用暫停前 ID交互恢復(fù)時(shí)需要復(fù)用暫停前記錄的 nodeResponse ID避免同一個(gè)展示節(jié)點(diǎn)在恢復(fù)后拆成兩個(gè)節(jié)點(diǎn)const nodeResponseId lastInteractive?.nodeResponseId lastInteractive.entryNodeIds?.includes(node.nodeId) ? lastInteractive.nodeResponseId : getNanoid();WorkflowInteractiveResponseType增加通用字段定義于 interactive/type.tsnodeResponseId?: string;該字段與entryNodeIds平級(jí)表示觸發(fā)本次暫停的當(dāng)前 workflow 節(jié)點(diǎn)對(duì)應(yīng)的 nodeResponsedata.id。同一時(shí)間只允許一個(gè)暫停模式因此一個(gè)字符串即可表示當(dāng)前恢復(fù)入口。嵌套交互嵌套交互繼續(xù)沿用childrenResponse每一層 interactive 都可以攜帶自己的nodeResponseId。例如 ToolCall 包裝的子 workflow 暫停時(shí){ type: toolChildrenInteractive, entryNodeIds: [toolCallNodeId], nodeResponseId: tool-call-node-response-id, params: { childrenResponse: { type: userInput, entryNodeIds: [formNodeId], nodeResponseId: form-node-response-id }, toolParams: { toolCallId: call_xxx } } }恢復(fù)時(shí)ToolCall 節(jié)點(diǎn)復(fù)用toolChildrenInteractive.nodeResponseId子 workflow 復(fù)用childrenResponse.nodeResponseId新增 rows 繼續(xù)寫到同一條 AI chat item 的chatItemDataId下讀取時(shí)父 ToolCall 和子節(jié)點(diǎn)都按(data.id, parentId)合并頁(yè)面只展示一個(gè) ToolCall 節(jié)點(diǎn)用量和運(yùn)行時(shí)間按增量累加。LoopRun 恢復(fù)iteration wrapper 的 ID 派生LoopRun 的 iteration wrapper 是虛擬展示節(jié)點(diǎn)ID 由 loopRun 父 nodeResponse ID 派生id ${loopRunNodeResponseId}:iter:${iteration};這樣同一個(gè) loop 節(jié)點(diǎn)在不同父作用域下運(yùn)行不會(huì)因?yàn)閚ode.nodeId iteration沖突。交互恢復(fù)時(shí)只要 loopRun 父節(jié)點(diǎn)復(fù)用interactive.nodeResponseId同一輪 iteration wrapper 也會(huì)自然復(fù)用同一個(gè)data.id。LoopRun 暫停時(shí)會(huì)寫一次當(dāng)前 iteration wrapper作為暫停前 child nodeResponses 的 parent并把pendingIterationSummary存到 interactive params?；謴?fù)后同一個(gè) wrapper ID 再寫本次 resume 的增量統(tǒng)計(jì)。由于讀取會(huì)累加數(shù)值字段恢復(fù)后的 wrapper 必須只寫本次 resume 片段的增量值不能寫暫停前后合并后的累計(jì)值——這是防止數(shù)值雙算的關(guān)鍵約束文檔在“后續(xù)關(guān)注”中明確要求 LoopRun、ToolCall 等恢復(fù)場(chǎng)景必須持續(xù)保證寫入的是本次運(yùn)行片段增量。運(yùn)行前 preChatRound業(yè)務(wù)入口的職責(zé)邊界保存歷史的新運(yùn)行進(jìn)入 workflow 前只調(diào)用preChatRound實(shí)現(xiàn)見 prepare.ts。它負(fù)責(zé)解析最終chatId?？誧hatId自動(dòng)生成隨機(jī) chatIdgetNanoid(24)NO_RECORD_HISTORIES即NO_RECORD_CHAT_ID NO_RECORD_HISTORIES表示不保存歷史。解析最終responseChatItemId。請(qǐng)求未傳時(shí)生成隨機(jī) ID。判斷是否持久化 chat items 和 nodeResponse rows。持久化運(yùn)行占用MongoChat.chatGenerateStatus generating。普通新運(yùn)行檢查 AIdataId沖突。普通新運(yùn)行嚴(yán)格創(chuàng)建本輪 Human AI placeholder。交互繼續(xù)復(fù)用上一條 AI 的dataId不創(chuàng)建新的 Human/AI placeholder。失敗時(shí)如果已經(jīng)占用生成狀態(tài)立刻置為error。返回值type PreChatRoundResult { chatId: string; responseChatItemId: string; shouldPersistChatRound: boolean; shouldFinalizePreparedRound: boolean; };持久化判斷統(tǒng)一為const finalChatId chatId NO_RECORD_CHAT_ID ? chatId : chatId || getNanoid(24); const shouldPersistChatRound finalChatId ! NO_RECORD_CHAT_ID;入口后續(xù)必須使用preparedRound.chatId和preparedRound.responseChatItemId不能繼續(xù)使用請(qǐng)求里的原始值。nodeResponseWriteConfig.persistToDb應(yīng)等于preparedRound.shouldPersistChatRound。普通新運(yùn)行順序解析最終chatId/responseChatItemIdNO_RECORD_HISTORIES直接返回不持久化結(jié)果不占用生成鎖調(diào)用tryStartGenerateChat占用生成鎖已有 generating 時(shí)拋ChatErrEnum.chatIsGenerating校驗(yàn)已有 AI chat item 中不存在同responseChatItemId嚴(yán)格 create 本輪 Human AI placeholder二者使用同一個(gè)dataId responseChatItemIdprepareChatRound使用嚴(yán)格 create 而非 upsert檢查或創(chuàng)建失敗時(shí)寫生成狀態(tài)error并拋錯(cuò)創(chuàng)建成功后才進(jìn)入 workflow。AI dataId 沖突檢查口徑MongoChatItem.findOne( { appId, chatId, obj: ChatRoleEnum.AI, dataId: responseChatItemId }, dataId );只檢查 AI 的原因Human/AI 同dataId是新運(yùn)行的正常結(jié)構(gòu)本輪 nodeResponse rows 歸屬于 AIchatItemDataIdHuman 歷史重復(fù)不影響 nodeResponse append-only 的安全性可以離線審計(jì)不作為運(yùn)行前阻塞條件。preChatRound保持在業(yè)務(wù)入口不下沉到dispatchWorkFlow。dispatch 被 debug、skill debug、MCP、outLink、定時(shí)觸發(fā)等入口復(fù)用不應(yīng)該感知source/sourceName/shareId/outLinkUid/userContent等 chat 保存字段。此外源碼中stripUserContentFileUrls會(huì)清理用戶消息里的文件臨時(shí) URL只保留 file key 參與持久化避免歷史記錄保存過期訪問地址。刪除與清理運(yùn)行期 writer 不刪除 nodeResponse rows。外部刪除規(guī)則刪除整條對(duì)話或批量日志時(shí)可以按chatId刪除MongoChatItemResponse局部消息刪除繼續(xù)保持MongoChatItem軟刪除語(yǔ)義新數(shù)據(jù) Human/AI 同dataId刪除一輪消息時(shí)前端可以繼續(xù)收集 Human 和 AI 的 dataId但發(fā)請(qǐng)求前應(yīng)去重刪除接口支持 bodycontentIdsbody 優(yōu)先body 為空時(shí)兼容 querycontentId。OpenAPI 默認(rèn)聲明 body。客戶端約束普通新運(yùn)行一輪只生成一個(gè)roundDataIdHuman/AI 共用該值交互繼續(xù)復(fù)用上一條 AIdataId不是新一輪 Human/AIReact list key不能只用dataId因?yàn)?Human/AI 可能相同應(yīng)包含obj或_id/id前端按dataId更新 AI 記錄時(shí)需要帶 AI 語(yǔ)義避免命中同 ID HumannodeResponse SSE 合并和詳情彈窗都應(yīng)使用(id, parentId)合并語(yǔ)義不再依賴mergeSignId。測(cè)試要求與回歸保障倉(cāng)庫(kù)為本次改造配套了完整的測(cè)試覆蓋核心測(cè)試文件包括 nodeResponseStorage.test.ts、nodeResponseSink.test.ts、index.persistence.test.ts 與 mergeNode.test.ts。preChatRound相關(guān)普通新運(yùn)行成功創(chuàng)建MongoChat、Human、AI placeholderHuman/AI 同dataId responseChatItemId初始responseChatItemId命中已有 AI直接拋錯(cuò)不進(jìn)入 workflow不隨機(jī)兜底初始responseChatItemId只命中 Human不按重復(fù) ID 報(bào)錯(cuò)生成鎖沖突拋ChatErrEnum.chatIsGenerating不創(chuàng)建 placeholderplaceholder 創(chuàng)建失敗或重復(fù)校驗(yàn)失敗生成狀態(tài)置為error空chatId自動(dòng)生成隨機(jī) chatId 并保存記錄NO_RECORD_HISTORIES不寫 chat、不寫 chat item、不占用生成鎖仍返回 dispatch 可用的隨機(jī)responseChatItemId非 query 交互繼續(xù)復(fù)用上一條 AIdataId不創(chuàng)建新 placeholder找不到上一條 AI 時(shí)拋錯(cuò)interactive query按新一輪創(chuàng)建 Human/AI placeholderfinalizeChatRound能在 Human/AI 同 dataId 時(shí)按obj更新兩條記錄。nodeResponse append-only 相關(guān)writer 寫入只調(diào)用 create不執(zhí)行運(yùn)行期 delete/update/replacebuffer 中同data.id多條 rows 全部寫入不預(yù)去重retry 保留 3 次普通重試和 slim fallback不依賴預(yù)生成_id讀取按_id順序 fold同(data.id, parentId)合并為一個(gè)展示節(jié)點(diǎn)相同data.id但不同parentId不合并parentId都不存在時(shí)視為同 parent 并合并數(shù)值字段按增量累加標(biāo)量以后到為準(zhǔn)llmRequestIds去重child 先于 parent 到達(dá)時(shí)最終能掛回 parentmergeSignId不參與合并。交互恢復(fù)相關(guān)暫停時(shí)interactive.nodeResponseId寫入當(dāng)前節(jié)點(diǎn)data.id交互繼續(xù)時(shí)恢復(fù)入口復(fù)用interactive.nodeResponseId頁(yè)面只展示一個(gè)節(jié)點(diǎn)ToolCall 子 workflow 暫停后繼續(xù)父 ToolCall 和子 workflow 分別復(fù)用對(duì)應(yīng)層級(jí)nodeResponseIdLoopRun 暫停后繼續(xù)父 loopRun 復(fù)用interactive.nodeResponseIditeration wrapper 使用${loopRunNodeResponseId}:iter:${iteration}恢復(fù)后只寫本次片段增量避免數(shù)值雙算。索引回歸ChatItemResponseSchema不聲明{ appId, chatId, chatItemDataId, data.id }unique 索引保留{ appId, chatId, chatItemDataId, _id }讀取索引ChatItemSchema.index({ appId, chatId, dataId })保持普通索引不改 unique如新增{ appId, chatId, dataId, obj }也只能是普通索引。后續(xù)關(guān)注與實(shí)施邊界設(shè)計(jì)文檔同時(shí)記錄了需要持續(xù)關(guān)注的運(yùn)維與演進(jìn)事項(xiàng)append-only 會(huì)增加 rows 數(shù)量需要依賴對(duì)話刪除、應(yīng)用刪除和過期清理控制表規(guī)模歷史mergeSignId數(shù)據(jù)不遷移異常展示風(fēng)險(xiǎn)已接受如果線上 AI dataId 沖突檢查成為熱點(diǎn)再評(píng)估普通索引{ appId, chatId, dataId, obj }LoopRun、ToolCall 等恢復(fù)場(chǎng)景必須持續(xù)保證寫入的是本次運(yùn)行片段增量而不是累計(jì)值。本次實(shí)施 TODO 清單均已勾選完成表明改造范圍是新增請(qǐng)求級(jí)WorkflowNodeResponseSink統(tǒng)一 writer 與 V2 SSE 發(fā)布WorkflowQueue 的 root/child runtime 通過 sink 逐條發(fā)布 nodeResponseAgent collector 移除 writer 依賴改為上交 sinkLoopRun/ParallelRun 虛擬任務(wù)節(jié)點(diǎn)改走 sink系統(tǒng)插件內(nèi)部 workflow 使用禁用 sink 的作用域保持隱藏語(yǔ)義V1、非流式 JSON、Share public 過濾和引用/文件權(quán)限保持兼容最終全量測(cè)試中全倉(cāng)并發(fā)出現(xiàn) 4 個(gè) 20 秒超時(shí)相關(guān)文件單獨(dú)復(fù)跑全部通過?？偨Y(jié)FastGPT 的 nodeResponse 持久化從“唯一索引 運(yùn)行期更新”演進(jìn)為“append-only 寫入 讀取時(shí)按 (data.id, parentId) 折疊”本質(zhì)上是把“寫時(shí)去重”的復(fù)雜度轉(zhuǎn)移到了“讀時(shí)合并”從而換取運(yùn)行期穩(wěn)定、低成本的只追加寫入。配合請(qǐng)求級(jí)WorkflowNodeResponseSink統(tǒng)一持久化與 SSE 發(fā)布、preChatRound在業(yè)務(wù)入口完成 chat 語(yǔ)義校驗(yàn)與 placeholder 預(yù)創(chuàng)建、交互恢復(fù)時(shí)復(fù)用nodeResponseId保證展示節(jié)點(diǎn)唯一這套設(shè)計(jì)同時(shí)解決了大表寫入性能、并行運(yùn)行寫入順序、Share 可見性分層以及暫停/恢復(fù)場(chǎng)景的展示一致性問題。對(duì)于需要深入理解 FastGPT workflow 運(yùn)行鏈路或設(shè)計(jì)類似對(duì)話式 AI 工作流引擎持久化方案的開發(fā)者建議進(jìn)一步閱讀 nodeResponseStorage.ts、nodeResponseSink.ts、mergeNode.ts 以及 prepare.ts 中對(duì)應(yīng)的測(cè)試用例?！久赓M(fèi)下載鏈接】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.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/fa/FastGPT創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考