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