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