AI Agent人工智能大模型AI 應(yīng)用工具調(diào)用本地部署MCP ClientsAgent 記憶【免費(fèi)下載鏈接】OperitThe most powerful AI agent and AI chat software on Android/Operit是一款A(yù)ndroid上能力最為強(qiáng)大、發(fā)展最久的AI Agent項(xiàng)目地址https://gitcode.com/gh_mirrors/op/Operit點(diǎn)擊查看免費(fèi)下載本文聚焦 OperitAndroid 平臺 AI Agent / AI 聊天應(yīng)用在流式 AI 輸出遭遇網(wǎng)絡(luò)錯誤并重試耗盡時的消息保真修復(fù)系統(tǒng)如何通過調(diào)整回滾事件的下發(fā)時機(jī)、讓最終失敗以可識別的異常狀態(tài)穿過服務(wù)層并在消息處理層按“中斷回合”收尾落庫從而保住用戶已經(jīng)看到的那部分 AI 回復(fù)。讀完本文你將掌握 Operit 流式聊天中SAVEPOINT/ROLLBACK事件機(jī)制、三個主流 Provider 的重試決策順序、EnhancedAIService的失敗傳播策略以及MessageProcessingDelegate中斷消息收尾的完整數(shù)據(jù)鏈路與回歸測試驗(yàn)證方法。問題背景網(wǎng)絡(luò)錯誤曾讓整輪 AI 回復(fù)變成空白在舊實(shí)現(xiàn)中流式對話一旦遭遇網(wǎng)絡(luò)錯誤數(shù)據(jù)鏈路存在一個致命的順序問題Provider 異常分支先回滾當(dāng)前請求的保存點(diǎn)——保存點(diǎn)位于本輪請求的起點(diǎn)requestSavepointId因此無論此前已經(jīng)流式輸出了多少內(nèi)容一次回滾都會把本輪文本全部截斷為空最終錯誤被EnhancedAIService吞掉不再繼續(xù)拋給下游消息層消息處理層把流的結(jié)束誤判為“正常完成”繼續(xù)走正常完成邏輯最終把空的流式快照寫回同一條 AI 消息覆蓋掉先前已經(jīng)持久化的內(nèi)容。結(jié)果就是用戶在界面上看到的文字在斷網(wǎng)瞬間全部消失數(shù)據(jù)庫里落庫的也是一條空回復(fù)。相關(guān)的問題描述與修復(fù)動機(jī)完整記錄在 01-network-failure-message-finalization.md 與該任務(wù)的 index.md 中。修正意圖把回滾放到“確認(rèn)會重試”之后本次修復(fù)的核心思路只有一句話將回滾事件的下發(fā)時機(jī)從“異常發(fā)生即回滾”推遲到“確認(rèn)會執(zhí)行下一次請求之后”。不會開始新請求的異常不可重試錯誤、重試已耗盡不再發(fā)送回滾事件已輸出的內(nèi)容原樣保留重試耗盡后的流以可識別的失敗狀態(tài)結(jié)束向上游拋出異常而不是靜默完成消息處理層通過已有的中斷消息收尾邏輯保存當(dāng)前文本與回合統(tǒng)計并禁止正常完成邏輯覆蓋該內(nèi)容。整個修復(fù)涉及四個層面的代碼下文逐一展開。Provider 層回滾事件與重試決策的解耦流事件機(jī)制SAVEPOINT 與 ROLLBACKOperit 的可回滾文本流通過TextStreamEvent攜帶兩個事件類型見 RevisableTextStream.kt事件類型語義SAVEPOINT記錄一個命名的內(nèi)容快照點(diǎn)實(shí)現(xiàn)上只記錄字符長度索引ROLLBACK攜帶保存點(diǎn) id要求把內(nèi)容截斷回該保存點(diǎn)位置每個請求發(fā)起時Provider 都會在流的最開始處登記一個保存點(diǎn)。以 OpenAIProvider.kt 為例// 用于保存當(dāng)前 attempt 已接收到的內(nèi)容一旦需要重試會整體回滾到請求起點(diǎn) val receivedContent StringBuilder() val emitter StreamEmitter(receivedContent, ::emit, eventChannel, onTokensUpdated) val requestSavepointId attempt_${UUID.randomUUID().toString().replace(-, )} emitter.emitSavepoint(requestSavepointId)TextStreamRevisionTracker見 TextStreamRevisionTracker.kt在底層維護(hù)保存點(diǎn)映射fun savepoint(id: String) { // Rollback discards a suffix, so a character index is the complete revision state. savepoints[id] contentBuffer.length } fun rollback(id: String): SmartString? { val length savepoints[id] ?: return null contentBuffer.truncate(length) savepoints.entries.removeAll { it.value length } return contentBuffer }而emitRollback見 OpenAIProvider.kt則會先把receivedContent截斷到保存點(diǎn)長度再向事件通道發(fā)出ROLLBACK事件suspend fun emitRollback(id: String): Boolean { val savepointLength savepoints[id] ?: return false if (receivedContent.length savepointLength) { receivedContent.setLength(savepointLength) } eventChannel.emit(TextStreamEvent(TextStreamEventType.ROLLBACK, id)) return true }重試決策先判定后回滾舊實(shí)現(xiàn)的錯誤在于回滾與重試決策的順序錯位。修復(fù)后回滾被收進(jìn)handleRetryableError的“確認(rèn)重試”分支見 OpenAIProvider.ktprivate suspend fun handleRetryableError( context: Context, exception: Exception, retryCount: Int, maxRetries: Int, enableRetry: Boolean, onNonFatalError: suspend (String) - Unit, onRetryAccepted: suspend () - Unit, buildRetryMessage: (String, Int) - String ): Int { if (exception is UserCancellationException || exception is CancellationException) { throw exception } checkCancellation(context, exception) val errorText resolveRetryErrorText(context, exception) if (!enableRetry) { throw IOException(errorText, exception) // 不可重試直接拋不回滾 } val newRetryCount retryCount 1 if (newRetryCount maxRetries) { // 重試耗盡拋異常結(jié)束不回滾 throw IOException( context.getString(R.string.openai_error_connection_timeout, maxRetries, errorText), exception ) } // A terminal failure must retain its streamed text; only a replacement request discards it. onRetryAccepted() // 確認(rèn)要發(fā)新請求才回滾 val retryDelayMs LlmRetryPolicy.nextDelayMs(newRetryCount) ... delay(retryDelayMs) return newRetryCount }關(guān)鍵點(diǎn)一目了然enableRetry false不可重試錯誤→直接拋出異常不回滾newRetryCount maxRetries重試耗盡→直接拋出異常不回滾只有確認(rèn)會發(fā)起下一次請求時才調(diào)用onRetryAccepted()——它綁定到emitter.emitRollback(requestSavepointId)見 OpenAIProvider.kt。同一模式在 Claude 與 Gemini 兩端同樣生效Claude 的emitRollbackClaudeProvider.kt與 Gemini 的emitRollbackGeminiProvider.kt都以onRetryAccepted回調(diào)的形式接入同一套重試流程這與 index.md 作用域中“OpenAI、Claude、Gemini Provider 的重試決策與回滾順序”的描述一致。重試參數(shù)指數(shù)退避策略Provider 層的重試參數(shù)統(tǒng)一由 LlmRetryPolicy.kt 管理參數(shù)值MAX_RETRY_ATTEMPTS5初始請求之外的額外嘗試次數(shù)RETRY_BASE_DELAY_MS1000 msRETRY_MAX_DELAY_MS16000 ms退避算法為指數(shù)增長并封頂fun nextDelayMs(retryAttempt: Int): Long { val normalizedAttempt retryAttempt.coerceAtLeast(1) val exponent (normalizedAttempt - 1).coerceAtMost(4) return (RETRY_BASE_DELAY_MS * (1L shl exponent)).coerceAtMost(RETRY_MAX_DELAY_MS) }即依次等待約 1s、2s、4s、8s、16s之后 16s 封頂。三個 Provider 均通過LlmRetryPolicy.MAX_RETRY_ATTEMPTS取得最大重試次數(shù)OpenAIProvider.kt、ClaudeProvider.kt、GeminiProvider.kt。重試耗盡后的可識別失敗狀態(tài)當(dāng)所有重試都失敗時Provider 拋出IOException并攜帶“已達(dá)最大重試次數(shù)”的明確文案見 OpenAIProvider.kt// 所有重試都失敗 lastException?.let { ex - AppLogger.e(AIService, 【發(fā)送消息】重試失敗請檢查網(wǎng)絡(luò)連接最大重試次數(shù): $maxRetries, ex) } ?: ... throw IOException( context.getString( R.string.openai_error_connection_timeout, maxRetries, lastException?.message ?: context.getString(R.string.openai_error_network_interrupted) ), lastException )流以異常結(jié)束而非正常完成——這就是“重試耗盡后的流以可識別的失敗狀態(tài)結(jié)束”的實(shí)現(xiàn)。由于此時沒有發(fā)出任何ROLLBACK事件此前所有已發(fā)出的文本塊都保留在流中等待下游按中斷收尾。EnhancedAIService 層最終失敗必須穿透服務(wù)層EnhancedAIService是 Provider 流與消息處理層之間的中樞見 EnhancedAIService.kt。舊實(shí)現(xiàn)中它捕獲錯誤后不再下拋導(dǎo)致下游誤判為正常完成修復(fù)后EnhancedAIService用hadFatalError標(biāo)志區(qū)分“致命錯誤”與“正常完成”并在確認(rèn)是致命錯誤時把異常原樣拋給下游} catch (e: Exception) { val isSocketClosed e.message?.contains(Socket closed, ignoreCase true) true if (isSocketClosed) { // 用戶取消導(dǎo)致的 Socket closed 是預(yù)期行為不應(yīng)作為錯誤處理 ... } else { hadFatalError true AppLogger.e(TAG, 發(fā)送消息時發(fā)生錯誤: ${e.message}, e) if (!providerStreamCollectionStarted) { withContext(Dispatchers.Main) { _inputProcessingState.value InputProcessingState.Error(message context.getString(R.string.enhanced_error_with_message, e.message ?: )) } } } if (!isSocketClosed) { if (!isSubTask) stopAiService() if (!providerStreamCollectionStarted) { throw e // 流尚未開始直接拋出 } } } finally { try { // 確保流處理完成后調(diào)用如果本輪已被取消則不能再繼續(xù)跑完成邏輯。 if (!hadFatalError isExecutionContextActive(execContext)) { ... processStreamCompletion(...) // 只有非致命錯誤才走正常完成 } else if (!hadFatalError) { ... } } finally { unregisterExecutionContext(execContext) ... } }要點(diǎn)錯誤被捕獲后標(biāo)記hadFatalError truefinally中的processStreamCompletion正常完成邏輯因此被跳過空快照不會再被寫入若流尚未開始收集EnhancedAIService自身發(fā)布InputProcessingState.Error狀態(tài)并直接throw e若流已開始已輸出部分內(nèi)容異常經(jīng)由共享流繼續(xù)傳播到收集端MessageProcessingDelegate的streamCollectionResult由消息處理層呈現(xiàn)錯誤狀態(tài)。這正是 01-network-failure-message-finalization.md 中所說的“最終網(wǎng)絡(luò)失敗攜帶中斷狀態(tài)到消息處理層”。MessageProcessingDelegate 層按中斷回合收尾落庫運(yùn)行態(tài)消息對象消息處理層維護(hù)ChatRuntime.activeStreamingTurn見 MessageProcessingDelegate.kt持有當(dāng)前正在流式輸出的ChatMessage。源碼注釋明確了其必要性// 取消收尾必須持有運(yùn)行態(tài)對象Room 不保存 contentStream重載后無法識別當(dāng)前流消息。 var activeStreamingTurn: ActiveStreamingTurn? null,也就是說中斷收尾必須以運(yùn)行態(tài)對象為準(zhǔn)——數(shù)據(jù)庫里只有靜態(tài)內(nèi)容無法識別“哪條消息正在流式輸出”這正是修復(fù)能生效的前提?；睾辖y(tǒng)計快照中斷時所需的回合統(tǒng)計由TurnCancellationSnapshot承載見 MessageProcessingDelegate.ktinternal data class TurnCancellationSnapshot( val inputTokens: Long, val outputTokens: Long, val cachedInputTokens: Long, val sentAt: Long, val outputDurationMs: Long, val waitDurationMs: Long, )readCurrentTurnCancellationSnapshot從EnhancedAIService.captureCurrentTurnTokenSnapshot()讀取 token 計數(shù)再結(jié)合運(yùn)行時記錄的首包時間、請求發(fā)起時間計算waitDurationMs與outputDurationMs。中斷消息完成與落庫核心完成函數(shù)completeInterruptedMessage見 MessageProcessingDelegate.ktinternal fun completeInterruptedMessage( streamingMessage: ChatMessage, finalContent: String, snapshot: TurnCancellationSnapshot?, completedAt: Long, ): ChatMessage { val messageWithMetrics snapshot?.let { stats - streamingMessage.copy( inputTokens stats.inputTokens, outputTokens stats.outputTokens, cachedInputTokens stats.cachedInputTokens, sentAt stats.sentAt.takeIf { it 0L } ?: streamingMessage.sentAt, outputDurationMs stats.outputDurationMs, waitDurationMs stats.waitDurationMs, ) } ?: streamingMessage return messageWithMetrics.copy( content finalContent, // 保存已接收內(nèi)容而不是空字符串 contentStream null, // 脫流從流式消息轉(zhuǎn)為靜態(tài)消息 completedAt completedAt, ) }隨后detachStreamingAiMessage見 MessageProcessingDelegate.kt完成完整收尾用resolveFinalContent從共享流的重放緩存重建最終文本調(diào)用completeInterruptedMessage生成帶統(tǒng)計指標(biāo)的最終消息在主線程addMessageToChat落庫同時回填對應(yīng)用戶消息的 token 與耗時統(tǒng)計若處于 waifu 分段模式則對每條分段消息同步指標(biāo)。中斷路徑如何觸發(fā)在sendUserMessage的異常處理與finally收尾中見 MessageProcessingDelegate.kt} catch (e: Exception) { if (e is kotlinx.coroutines.CancellationException) { ... cancellationToPropagate e } else { AppLogger.e(TAG, 發(fā)送消息時出錯, e) shouldFinalizeInterruptedMessage true setChatInputProcessingState( chatId, EnhancedInputProcessingState.Error(context.getString(R.string.message_send_failed, e.message)) ) withContext(Dispatchers.Main) { showErrorMessage(...) } } } finally { val interruptedTurn if (shouldFinalizeInterruptedMessage) chatRuntime.activeStreamingTurn else null val deferTurnCompleteToAsyncJob if (interruptedTurn ! null chatId ! null) { // Network errors can arrive after partial output; do not overwrite it as a completed reply. detachStreamingAiMessage( chatId chatId, activeTurn interruptedTurn, snapshot readCurrentTurnCancellationSnapshot(chatId), ) false } else if (cancellationToPropagate null) { finalizeMessageAndNotify(...) // 正常完成路徑 } else { ... // 主動取消路徑不做收尾 } ... }三段路徑清晰分離場景處理方式網(wǎng)絡(luò)失敗/普通異常shouldFinalizeInterruptedMessage true走detachStreamingAiMessage中斷收尾保留已輸出內(nèi)容與統(tǒng)計顯示 Error 狀態(tài)正常完成走finalizeMessageAndNotify正常落庫主動取消CancellationException不執(zhí)行收尾行為保持不變源碼注釋明確寫著 “Network errors can arrive after partial output; do not overwrite it as a completed reply.”——這行注釋正是整個修復(fù)的濃縮表達(dá)。同時 index.md 中“主動取消現(xiàn)有的中斷消息行為不變”這一期待結(jié)果由cancellationToPropagate分支保證?；貧w測試三個斷言鎖定修復(fù)行為Provider 級測試最終失敗不回滾、保留已輸出文本ProviderUsageCancellationTest.kt 中的assertTerminalFailureDoesNotRollback精確復(fù)現(xiàn)了修復(fù)目標(biāo)private fun assertTerminalFailureDoesNotRollback(provider: AIService) { ... val response provider.sendMessage( context context, chatHistory listOf(PromptTurn(PromptTurnKind.USER, Hi)), modelParameters emptyList(), enableThinking false, stream true, ... onUsageReported { _, _ - throw IOException(network interrupted) }, enableRetry false, ) val received StringBuilder() try { response.collect { chunk - received.append(chunk) } fail(terminal stream failure must propagate) } catch (_: IOException) { // The generated response is retained for interrupted-message finalization. } val events (response as TextStreamEventCarrier).eventChannel.replayCache assertEquals(answer, received.toString()) // 已輸出內(nèi)容保留 assertEquals(listOf(TextStreamEventType.SAVEPOINT), events.map { it.eventType }) // 只有 SAVEPOINT沒有 ROLLBACK }斷言點(diǎn)與需求一一對應(yīng)最終失敗必須傳播terminal stream failure must propagate——對應(yīng)“重試耗盡后的流以可識別的失敗狀態(tài)結(jié)束”已接收文本answer完整保留——對應(yīng)“不會開始新請求的異常不發(fā)送回滾事件”事件通道中只有SAVEPOINT沒有ROLLBACK——直接驗(yàn)證回滾未發(fā)生。同一測試類中的assertIncompleteStreamDoesNotFinalizeL240-L274還驗(yàn)證裸 EOF 等不完整流不得觸發(fā) usage 終結(jié)回調(diào)必須進(jìn)入重試/錯誤路徑避免把中斷誤判為成功。共享流級測試上游異常在重放部分內(nèi)容后繼續(xù)傳遞HotStreamTest.kt 的shareForwardsUpstreamFailureAfterReplayingPartialContent驗(yàn)證了共享流的關(guān)鍵契約——已重放的部分內(nèi)容不會丟失上游異常仍會傳給收集端val expected IOException(network interrupted) val shared streamString { emit(partial) throw expected }.share( scope scope, replay Int.MAX_VALUE, ) val received StringBuilder() try { shared.collect { chunk - received.append(chunk) } fail(shared stream must propagate the upstream failure) } catch (actual: IOException) { assertSame(expected, actual) } assertEquals(partial, received.toString())這是MessageProcessingDelegate能夠“既收到部分內(nèi)容、又收到失敗信號”的基礎(chǔ)保障也是 index.md 驗(yàn)證記錄中“已添加共享流回歸測試驗(yàn)證上游異常在重放部分內(nèi)容后繼續(xù)傳遞給收集端”所指的實(shí)現(xiàn)。配套改動回滾后的 Markdown 尾部渲染本修復(fù)的作用域還包含“回滾后的 Markdown 尾部更新”詳見配套文檔 02-tail-only-rollback-rendering.md。其舊實(shí)現(xiàn)中聊天界面收到ROLLBACK后會創(chuàng)建新的顯示流并把回滾后的完整內(nèi)容再次寫入導(dǎo)致 Markdown 渲染器以流實(shí)例為重置依據(jù)清空并重新解析整條 AI 消息——這不僅浪費(fèi)還會丟失已渲染節(jié)點(diǎn)的狀態(tài)。修復(fù)后的做法是尾部替換StreamRollbackPrefix接口RevisableTextStream.kt與withRollbackPrefix包裝L148-L153為字符流攜帶“回滾前綴”標(biāo)記UI 側(cè)rememberRevisableTextStreamRevisableTextStreamRemember.kt在收到ROLLBACK時用TextStreamRevisionTracker.rollback得到前綴快照創(chuàng)建只承載回滾后新尾部的 replacement stream并用RollbackTailStream攜帶前綴// The renderer restores this prefix itself; the replacement stream carries only new tail output. val rollbackStream RollbackTailStream( upstream replacementStream, rollbackPrefix snapshot, ) displayStream rollbackStream previousOutputStream.resetReplayCache()這樣渲染器以回滾前綴預(yù)熱流式 Markdown 會話未撤回的 Markdown 節(jié)點(diǎn)保持渲染狀態(tài)重試后的內(nèi)容繼續(xù)追加且不重播前綴——成功重試的流式顯示與工具調(diào)用撤回因此依然可用。修復(fù)后的完整行為清單綜合 01-network-failure-message-finalization.md、index.md 與上述源碼修復(fù)落地后的行為可總結(jié)為斷網(wǎng)且重試耗盡后已輸出內(nèi)容仍保留在同一條 AI 消息中——回滾只發(fā)生在確認(rèn)重試時網(wǎng)絡(luò)失敗顯示錯誤狀態(tài)不會被當(dāng)作正常完成——hadFatalError阻止正常完成邏輯異常傳播到消息層呈現(xiàn) Error主動取消的中斷消息行為不變——CancellationException路徑保持原有收尾語義成功重試的流式顯示與工具調(diào)用撤回仍可用——重試時的回滾事件正常下發(fā)UI 側(cè)尾部替換渲染保證不重播前綴回合統(tǒng)計token、等待時長、輸出時長在中斷收尾時一并落庫——TurnCancellationSnapshot全程攜帶。這條數(shù)據(jù)鏈路——Provider 保存點(diǎn)與回滾決策 →EnhancedAIService失敗穿透 →MessageProcessingDelegate中斷收尾 → UI 尾部渲染——構(gòu)成了 Operit 在弱網(wǎng)環(huán)境下保障 AI 輸出不丟失的完整閉環(huán)。贊分享AI Agent人工智能大模型AI 應(yīng)用工具調(diào)用本地部署MCP ClientsAgent 記憶【免費(fèi)下載鏈接】OperitThe most powerful AI agent and AI chat software on Android/Operit是一款A(yù)ndroid上能力最為強(qiáng)大、發(fā)展最久的AI Agent項(xiàng)目地址https://gitcode.com/gh_mirrors/op/Operit點(diǎn)擊查看免費(fèi)下載相關(guān)推薦Path of Building流放之路Build規(guī)劃的終極解決方案Path of Building流放之路Build規(guī)劃的終極解決方案 你是否曾經(jīng)在《流放之路》中花費(fèi)數(shù)小時調(diào)整裝備和天賦卻不知道哪個改動真正提升了角色實(shí)力AI Agent人工智能大模型AI 應(yīng)用工具調(diào)用本地部署MCP ClientsAgent 記憶GUI 自動化WebRTC網(wǎng)絡(luò)穿透實(shí)戰(zhàn)從連接失敗到穩(wěn)定傳輸?shù)耐暾改显谥悄苤圃旃S的監(jiān)控中心工程師小李盯著屏幕上不斷閃爍的連接中斷提示內(nèi)心充滿困惑。車間里的50臺工業(yè)相機(jī)明明在同一局域網(wǎng)內(nèi)為什么通過WebRTC傳輸?shù)揭粢曨l后端告別上傳失敗react-dropzone實(shí)現(xiàn)斷點(diǎn)續(xù)傳完整指南告別上傳失敗react dropzone實(shí)現(xiàn)斷點(diǎn)續(xù)傳完整指南 你是否經(jīng)歷過上傳GB級文件時網(wǎng)絡(luò)中斷不得不從頭開始的痛苦或者因?yàn)g覽器崩潰導(dǎo)致幾小時的上傳進(jìn)度前端UI組件上一篇history庫故障轉(zhuǎn)移機(jī)制路由降級與備份方案下一篇高杠桿操作指南4個實(shí)戰(zhàn)驗(yàn)證的達(dá)爾文.skill優(yōu)化技巧創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考