From 6c0a32330c787312e2aa04c098ac18f7ffb6ebfc Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 31 Aug 2026 19:58:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(stream):=20=E6=B5=81=E5=BC=8F=E6=96=AD?= =?UTF-8?q?=E6=B5=81=E3=80=8C=E6=B8=85=E9=99=A4=E9=87=8D=E6=9D=A5=E3=80=8D?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E9=87=8D=E8=AF=95=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端↔API 流式连接在输出中途断开时，旧行为直接终止任务。现改为： - 网络类断流（status_code 为空）有接收也重试，与零接收共用 5 次 额度、间隔 10s；4xx 业务错误保持原逻辑 - 重试前发 stream_reset 事件，前端清除本轮已渲染的半截内容 （思考块含已闭合的/文本/preparing 工具条目），显示「连接中断， 正在重试（第 n/5 次）…」占位，socket 与 REST 轮询双通道生效 - 修复断流作废轮 usage 重复计数 - 子智能体同构改造：网络断流重试耗尽转 idle 汇报 Team Leader， 重试前将已推送的「正在调用」工具条目标记 cancelled --- modules/sub_agent/task.py | 70 +++++++++++++++---- server/chat_flow_stream_loop.py | 28 ++++++-- static/src/app.ts | 3 +- .../src/app/methods/taskPolling/aiStream.ts | 22 ++++++ .../src/app/methods/taskPolling/lifecycle.ts | 6 ++ static/src/composables/useLegacySocket.ts | 35 ++++++++++ static/src/locales/en-US/appTasks.ts | 2 + static/src/locales/zh-CN/appTasks.ts | 2 + static/src/stores/chat.ts | 45 ++++++++++++ 9 files changed, 191 insertions(+), 22 deletions(-) diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index 67c2549c..d7a82874 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -37,14 +37,19 @@ class SubAgentModelCallError(RuntimeError): """子智能体模型请求失败。 received_any=False：请求阶段失败（未收到任何文本/思考/工具调用），可重试； - received_any=True：已开始收到流内容后断开（输出期间断开），直接失败不重试。 - 语义与主智能体 run_streaming_attempts 的 can_retry 判定一致 - （not full_response and not tool_calls and not current_thinking）。 + received_any=True：已开始收到流内容后断开（输出期间断开）—— + 网络类断流（is_network_error=True）同样可重试（清除重来，与主智能体 + run_streaming_attempts 的放宽判定一致）；非网络错误（4xx 等业务错误） + 有接收时不重试。 + calling_tools：断流时流中已推送「正在调用」进度事件的工具列表， + 重试前由 _run_loop 标记取消，避免前端进度弹窗永久卡在「调用中」。 """ - def __init__(self, message: str, *, received_any: bool = False): + def __init__(self, message: str, *, received_any: bool = False, is_network_error: bool = True, calling_tools: Optional[List[Dict[str, Any]]] = None): super().__init__(message) self.received_any = received_any + self.is_network_error = is_network_error + self.calling_tools = calling_tools or [] # 模型请求失败重试策略（与主智能体 max_api_retries=4 / retry_delay_seconds=10 对齐： @@ -331,9 +336,10 @@ class SubAgentTask: # 安全点：写入延迟的上下文通知（不触发新一轮工作，仅让下一轮模型调用看到） self._flush_pending_notifications() - # 模型请求（带失败重试，与主智能体同构：最多 5 次尝试、重试间隔 10s、 - # 仅当零接收——未收到任何文本/思考/工具调用——时重试； - # 已开始收到内容后断开属于输出期间故障，直接失败不重试） + # 模型请求（带失败重试，与主智能体同构：最多 5 次尝试、重试间隔 10s）。 + # 重试范围：零接收一律可重试；有接收（输出中途断开）仅网络类断流可重试， + # 采取「清除重来」——半截内容只在 _call_model 局部变量与已推送的 calling + # 进度事件里，转发给主智能体的输出只发生在流完整结束后，无副作用。 assistant_message = "" reasoning = "" tool_calls = [] @@ -348,7 +354,7 @@ class SubAgentTask: break except SubAgentModelCallError as exc: call_error = exc - can_retry = api_attempt < _SUB_AGENT_MAX_API_RETRIES and not exc.received_any + can_retry = api_attempt < _SUB_AGENT_MAX_API_RETRIES and (exc.is_network_error or not exc.received_any) ma_debug( "sub_agent_api_attempt_failed", task_id=self.task_id, @@ -363,6 +369,17 @@ class SubAgentTask: ) if not can_retry: break + # 有接收断流重试（清除重来）：本轮已推送的「正在调用」工具进度 + # 条目标记取消，避免前端进度弹窗永久卡在「调用中」； + # 重试成功后模型若再次调用会推送新的条目（新一轮工具 id 不同）。 + for calling_tool in exc.calling_tools: + self.emit("progress", { + "id": calling_tool.get("id"), + "tool": calling_tool.get("name", ""), + "status": "cancelled", + "args": {}, + "ts": int(time.time() * 1000), + }) # 重试本身也是一次额外 API 请求 self.stats["api_calls"] += 1 # 重试等待期间必须响应软停止/硬取消 @@ -376,8 +393,8 @@ class SubAgentTask: continue if call_error is not None: - if self.multi_agent_mode and not call_error.received_any: - # 5 次尝试全部失败（均为零接收）：子智能体不终止，转为 idle + if self.multi_agent_mode and (call_error.is_network_error or not call_error.received_any): + # 5 次尝试全部失败（零接收或网络断流）：子智能体不终止，转为 idle # 并向 Team Leader 汇报错误，等待检查网络后重新下达指令 error_report = tr( "sub_agent_task.model_call_failed_idle", @@ -396,8 +413,8 @@ class SubAgentTask: self._idle = True self._persist_conversation(partial_summary=error_report[:200]) continue - # 输出期间断开（已开始收到内容）或传统后台模式：直接失败， - # 异常上抛由 run() 捕获后 _write_failure 落盘 failed 状态 + # 非网络业务错误（有接收，理论罕见）或传统后台模式（重试已耗尽）： + # 直接失败，异常上抛由 run() 捕获后 _write_failure 落盘 failed 状态 if self.multi_agent_mode: # 同步把失败状态告知 Team Leader，避免主智能体无感知空等 self._forward_output_to_master( @@ -779,15 +796,30 @@ class SubAgentTask: raise asyncio.CancelledError() if chunk.get("error"): error_info = chunk.get("error") + is_network = True if isinstance(error_info, dict): error_text = ( error_info.get("error_message") or error_info.get("error_text") or str(error_info) ) + # status_code 为空 = 网络类断流（连接错误/超时/远端断开）； + # 有值（4xx 等 HTTP 业务错误）不参与有接收重试放宽 + is_network = error_info.get("status_code") is None else: error_text = str(error_info) - raise SubAgentModelCallError(tr("sub_agent_task2.api_call_failed", error=error_text), received_any=received_any) + # 收集本轮已推送「正在调用」进度的工具，重试前需标记取消 + calling_tools = [ + {"id": tc.get("id"), "name": tc.get("function", {}).get("name", "")} + for tc in tool_calls + if tc.get("id") and tc.get("function", {}).get("name") + ] + raise SubAgentModelCallError( + tr("sub_agent_task2.api_call_failed", error=error_text), + received_any=received_any, + is_network_error=is_network, + calling_tools=calling_tools, + ) choice = (chunk.get("choices") or [{}])[0] delta = choice.get("delta") or {} if delta.get("content"): @@ -844,7 +876,17 @@ class SubAgentTask: raise except Exception as exc: # chat 流本身抛出的漏网异常（如底层连接错误未被转为 error chunk 时） - raise SubAgentModelCallError(tr("sub_agent_task2.api_call_exception", error=exc), received_any=received_any) from exc + calling_tools = [ + {"id": tc.get("id"), "name": tc.get("function", {}).get("name", "")} + for tc in tool_calls + if tc.get("id") and tc.get("function", {}).get("name") + ] + raise SubAgentModelCallError( + tr("sub_agent_task2.api_call_exception", error=exc), + received_any=received_any, + is_network_error=True, + calling_tools=calling_tools, + ) from exc return assistant_message, reasoning, tool_calls, usage diff --git a/server/chat_flow_stream_loop.py b/server/chat_flow_stream_loop.py index 09cb663f..d0807d45 100644 --- a/server/chat_flow_stream_loop.py +++ b/server/chat_flow_stream_loop.py @@ -319,7 +319,9 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien } # === API响应完成后只计算输出token === - if last_usage_payload: + # 断流作废轮不计 usage：本论响应不完整，重试成功后会按新一轮 usage 计数， + # 若在 error 分支判断之前 apply 会导致同一回答重复计数。 + if last_usage_payload and not api_error: try: web_terminal.context_manager.apply_usage_statistics(last_usage_payload) debug_log( @@ -367,12 +369,13 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien error_message = f"API 请求失败（HTTP {error_status}）" else: error_message = "API 请求失败" - can_retry = ( - api_attempt < max_api_retries - and not full_response - and not tool_calls - and not current_thinking - ) + # 重试判定：网络类断流（status_code 为空：连接错误/超时/远端断开， + # 区别于 HTTP 业务错误）允许「有接收」重试——半截内容只存在于局部变量 + # 与前端，后端历史与磁盘均未写入，清除重来无副作用；HTTP 业务错误 + # （4xx 等，实际上不可能有接收）保持原逻辑仅零接收重试。 + has_partial_output = bool(full_response or tool_calls or current_thinking) + is_network_error = error_status is None + can_retry = api_attempt < max_api_retries and (is_network_error or not has_partial_output) sender('error', { 'message': error_message, 'status_code': error_status, @@ -388,6 +391,17 @@ async def run_streaming_attempts(*, web_terminal, messages, tools, sender, clien 'max_attempts': max_api_retries + 1 }) if can_retry: + if has_partial_output: + # 「清除重来」：通知前端清掉本轮 attempt 已渲染的半截内容 + # （思考块含已闭合的 / 文本 / preparing 工具条目），重试的新内容 + # 将由新一轮 thinking_start/text_start/tool_preparing 重新推送。 + # 后端内存历史与磁盘均无半截状态，无需清理。 + sender('stream_reset', { + 'reason': 'stream_disconnected', + 'attempt': api_attempt + 2, + 'max_attempts': max_api_retries + 1, + 'retry_in': retry_delay_seconds, + }) try: profile = get_model_profile(getattr(web_terminal, "model_key", None)) web_terminal.apply_model_profile(profile) diff --git a/static/src/app.ts b/static/src/app.ts index f8adf3bb..893b45a5 100644 --- a/static/src/app.ts +++ b/static/src/app.ts @@ -93,7 +93,8 @@ const appOptions = { chatCompleteTextAction: 'completeText', chatAddSystemMessage: 'addSystemMessage', chatEnsureAssistantMessage: 'ensureAssistantMessage', - chatClearStreamingResidualState: 'clearStreamingResidualState' + chatClearStreamingResidualState: 'clearStreamingResidualState', + chatResetStreamingAttemptActions: 'resetStreamingAttemptActions' }), ...mapActions(useInputStore, { inputSetFocused: 'setInputFocused', diff --git a/static/src/app/methods/taskPolling/aiStream.ts b/static/src/app/methods/taskPolling/aiStream.ts index 8e3d14ea..fbebd173 100644 --- a/static/src/app/methods/taskPolling/aiStream.ts +++ b/static/src/app/methods/taskPolling/aiStream.ts @@ -260,5 +260,27 @@ export const aiStreamMethods = { this.chatCompleteTextAction(full); this.$forceUpdate(); this.monitorEndModelOutput(); + }, + handleStreamReset(data: any) { + // 断流重试「清除重来」：移除本轮 attempt 已渲染的半截内容 + // （思考块含已闭合的 / 文本 / preparing 工具条目），重试内容将重新推送。 + debugLog('[TaskPolling] 断流重试，清理本轮半截内容'); + const retryAttempt = Number(data?.attempt) || 0; + const retryMax = Number(data?.max_attempts) || 0; + const retryLabel = retryAttempt && retryMax + ? t('appTasks.streamRetrying', { attempt: retryAttempt, max: retryMax }) + : t('appTasks.streamRetryingGeneric'); + const removedActions = this.chatResetStreamingAttemptActions?.(retryLabel) || []; + for (const action of removedActions) { + if (action?.id) { + this.preparingTools?.delete(action.id); + } + if (action?.type === 'tool') { + this.toolUnregisterAction?.(action); + } + } + this.chatClearThinkingLocks?.(); + this.monitorEndModelOutput(); + this.$forceUpdate(); } }; diff --git a/static/src/app/methods/taskPolling/lifecycle.ts b/static/src/app/methods/taskPolling/lifecycle.ts index 17500b11..7b3ac38a 100644 --- a/static/src/app/methods/taskPolling/lifecycle.ts +++ b/static/src/app/methods/taskPolling/lifecycle.ts @@ -230,6 +230,12 @@ export const lifecycleMethods = { this.handleTextEnd(eventData, eventIdx); break; + case 'stream_reset': + // 断流重试「清除重来」：清理本轮 attempt 半截内容； + // 历史重放时同样安全（后续事件会把新一轮内容重新渲染出来） + this.handleStreamReset(eventData, eventIdx); + break; + case 'tool_preparing': this.handleToolPreparing(eventData, eventIdx); break; diff --git a/static/src/composables/useLegacySocket.ts b/static/src/composables/useLegacySocket.ts index 0e2e46a0..f58e7fe9 100644 --- a/static/src/composables/useLegacySocket.ts +++ b/static/src/composables/useLegacySocket.ts @@ -1142,6 +1142,41 @@ export async function initializeLegacySocket(ctx: any) { ctx.apiRequestPending = true; }); + // 断流重试「清除重来」：后端与 API 的流式连接在输出中途断开、即将自动重试， + // 清除本轮 attempt 已渲染的半截内容（思考块含已闭合的 / 文本 / preparing 工具条目）， + // 重试的新内容会由新一轮 thinking_start/text_start/tool_preparing 重新推送。 + ctx.socket.on('stream_reset', (data) => { + if (data?.conversation_id && data.conversation_id !== ctx.currentConversationId) { + return; + } + // 轮询模式下由任务事件流重放处理（lifecycle.ts case 'stream_reset'） + if (ctx.usePollingMode && !ctx.waitingForSubAgent) { + return; + } + socketLog('断流重试，清理本轮半截内容', data); + // 清逐字流缓冲，防止半截缓冲在重置后继续渲染 + resetStreamingBuffer(); + const retryAttempt = Number(data?.attempt) || 0; + const retryMax = Number(data?.max_attempts) || 0; + const retryLabel = retryAttempt && retryMax + ? t('appTasks.streamRetrying', { attempt: retryAttempt, max: retryMax }) + : t('appTasks.streamRetryingGeneric'); + const removedActions = ctx.chatResetStreamingAttemptActions?.(retryLabel) || []; + for (const action of removedActions) { + if (action?.id) { + ctx.preparingTools?.delete(action.id); + } + if (action?.type === 'tool') { + ctx.toolUnregisterAction?.(action); + } + } + if (typeof ctx.chatClearThinkingLocks === 'function') { + ctx.chatClearThinkingLocks(); + } + ctx.monitorEndModelOutput?.(); + ctx.$forceUpdate(); + }); + // 思考流开始 ctx.socket.on('thinking_start', (data) => { // 对话定向事件：只响应当前对话，避免运行中对话的流式输出渲染到 /new 等空白页。 diff --git a/static/src/locales/en-US/appTasks.ts b/static/src/locales/en-US/appTasks.ts index b3f8138d..5d5bb5a0 100644 --- a/static/src/locales/en-US/appTasks.ts +++ b/static/src/locales/en-US/appTasks.ts @@ -64,6 +64,8 @@ export default { // ── Task errors & retry (lifecycle.ts) ── retrySoonTitle: 'Retrying soon', retryInSeconds: 'Retrying in {n}s (attempt {attempt}/{max})\nError: {error}', + streamRetrying: 'Connection interrupted, retrying (attempt {attempt}/{max})…', + streamRetryingGeneric: 'Connection interrupted, retrying…', toolCallFailed: 'Tool call failed', taskFailedTitle: 'Task failed', apiErrorTitle: 'API call failed', diff --git a/static/src/locales/zh-CN/appTasks.ts b/static/src/locales/zh-CN/appTasks.ts index 515e5560..96a86b74 100644 --- a/static/src/locales/zh-CN/appTasks.ts +++ b/static/src/locales/zh-CN/appTasks.ts @@ -65,6 +65,8 @@ export default { // ── 任务错误与重试（lifecycle.ts） ── retrySoonTitle: '即将重试', retryInSeconds: '将在 {n} 秒后重试（第 {attempt}/{max} 次）\n错误：{error}', + streamRetrying: '连接中断，正在重试（第 {attempt}/{max} 次）…', + streamRetryingGeneric: '连接中断，正在重试…', toolCallFailed: '工具调用失败', taskFailedTitle: '任务执行失败', apiErrorTitle: 'API 调用失败', diff --git a/static/src/stores/chat.ts b/static/src/stores/chat.ts index 5046f530..0d89d5c2 100644 --- a/static/src/stores/chat.ts +++ b/static/src/stores/chat.ts @@ -419,6 +419,51 @@ export const useChatStore = defineStore('chat', { } } }, + // 断流重试「清除重来」：移除当前 assistant 消息中本轮 API 尝试已渲染的 + // 半截内容——从 actions 末尾往前删除思考（含已闭合）/文本/preparing 工具条目， + // 遇到上一轮迭代的已完成工具结果等非本轮内容即停。复位流式标志；若消息 + // 因此被清空则恢复「生成中」占位（label 通常为重试提示文案）。 + // 返回被移除的 action 对象列表，供调用方清理 preparingTools / 工具注册表。 + resetStreamingAttemptActions(label: string = ''): any[] { + const removed: any[] = []; + let msg: any = null; + if ( + this.currentMessageIndex >= 0 && + this.messages[this.currentMessageIndex]?.role === 'assistant' + ) { + msg = this.messages[this.currentMessageIndex]; + } else { + for (let i = this.messages.length - 1; i >= 0; i--) { + if (this.messages[i]?.role === 'assistant') { + msg = this.messages[i]; + break; + } + } + } + if (!msg || !Array.isArray(msg.actions)) return removed; + while (msg.actions.length) { + const last = msg.actions[msg.actions.length - 1]; + const isThinking = last?.type === 'thinking'; + const isText = last?.type === 'text'; + const isPreparingTool = + last?.type === 'tool' && String(last?.tool?.status || '').toLowerCase() === 'preparing'; + if (!isThinking && !isText && !isPreparingTool) break; + removed.push(last); + msg.actions.pop(); + } + msg.currentStreamingType = null; + msg.activeThinkingId = null; + msg.streamingThinking = ''; + msg.streamingText = ''; + if (msg.actions.length === 0) { + msg.awaitingFirstContent = true; + msg.generatingLabel = label; + } else { + msg.awaitingFirstContent = false; + msg.generatingLabel = ''; + } + return removed; + }, addSystemMessage(content: string, meta: any = null) { // 与历史重建保持一致：子智能体/后台完成通知作为独立 assistant 消息渲染， // 避免运行时与刷新后在垂直间距上不一致。