From fa9679b7a638f09810ae6e916aa2824350637681 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Tue, 4 Aug 2026 09:11:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(aibox):=20=E4=BF=AE=E5=A4=8D=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E7=BA=BF=E7=A8=8B=E5=AF=BC=E8=87=B4=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E6=B1=A1=E6=9F=93=E5=BC=95=E5=8F=91HTTP=2040?= =?UTF-8?q?0=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加线程锁(RLock)保护 self.messages，防止WebSocket与调度器线程并发修改 - 重写 _repair_tool_pairs 方法，支持穿透system消息查找tool响应、去重、清理孤立tool消息 - 新增 _validate_enhanced_messages API调用前最终校验，就地修复消息结构 - 调度器执行定时任务前先cancel+wait锁释放，避免与WebSocket对话冲突 - /new命令和global_exit_handler改用锁保护的reset_conversation，不再裸改messages - 新增 is_busy() 非阻塞方法供外部判断AI是否正在处理中 --- aibox/aibox.py | 262 +++++++++++++++++++++++++++++++++++++------------ 1 file changed, 199 insertions(+), 63 deletions(-) diff --git a/aibox/aibox.py b/aibox/aibox.py index b6b7f39..b8906b4 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -3029,8 +3029,18 @@ class TaskScheduler: # 修改：确保AI处于新对话状态，不加载历史上下文 if self.ai_manager.ai: - # 重置AI对话状态，但保留系统提示词 - self.ai_manager.ai.reset_conversation() + ai = self.ai_manager.ai + # 先取消任何正在进行中的对话（防止WebSocket线程与调度器线程并发修改messages） + if ai.is_busy(): + self.logger.info("AI正在处理中，先取消当前对话...") + ai.cancel() + # 短暂等待让正在进行的请求退出并释放锁（使用asyncio.sleep避免阻塞事件循环） + for _ in range(20): + await asyncio.sleep(0.1) + if not ai.is_busy(): + break + # 重置AI对话状态，但保留系统提示词（在锁保护下执行） + ai.reset_conversation() self.logger.info("AI对话状态已重置，准备执行任务") # 创建一个 Future 来等待 AI 完成 @@ -6702,6 +6712,8 @@ class DeepSeekChat: # 取消机制：threading.Event 用于跨线程信号，HTTP 中断由客户端抽象层负责 self._cancel_event = threading.Event() self._command_executor = command_executor # 保存命令执行器引用，用于取消时kill子进程 + # 消息锁：保护 self.messages 免受 WebSocket 线程与调度器线程并发修改导致对话历史污染 + self._msg_lock = threading.RLock() self._init_system_prompts(system_prompt) @@ -7066,6 +7078,11 @@ class DeepSeekChat: is_continuation: 是否为工具调用后的递归继续；为 True 时不注入冗长的工作流指令， 因为 system prompt 中已有完整行为规范，继续时只需传递工具结果 """ + with self._msg_lock: + return self._think_and_respond_impl(input_text, output_callback, is_continuation) + + def _think_and_respond_impl(self, input_text: str, output_callback: Callable = None, is_continuation: bool = False) -> Optional[str]: + """think_and_respond 的实际实现，在 _msg_lock 保护下执行。""" # 首轮开始时清除取消标志 if not is_continuation: self._cancel_event.clear() @@ -7162,6 +7179,9 @@ class DeepSeekChat: else: enhanced_messages.append(msg) + # API 调用前最终校验：确保 tool_calls ↔ tool 消息配对完整，避免 400 错误 + self._validate_enhanced_messages(enhanced_messages) + # 工具调用（v4-flash 支持思考+工具调用） tools = self.tool_registry.get_tools_schemas() or None @@ -7418,46 +7438,151 @@ class DeepSeekChat: except Exception as e: self.logger.error(f"消息回滚失败: {e}") + def _validate_enhanced_messages(self, messages): + """发送给 API 前的最终校验：检查 tool_calls 与 tool 消息的配对完整性。 + 若发现问题，就地修复（补占位 tool 消息），避免 HTTP 400。""" + if not messages: + return + pending_tool_ids = [] # 当前等待 tool 响应的 tool_call_id 列表 + seen_ids = set() + fixed = [] + i = 0 + n = len(messages) + while i < n: + msg = messages[i] + role = msg.get("role") + if role == "assistant" and msg.get("tool_calls"): + # 如果有未被响应的 tool_calls，先补占位（正常情况不应出现） + for tid in pending_tool_ids: + if tid not in seen_ids: + fixed.append({ + "role": "tool", + "tool_call_id": tid, + "content": "⚠️ 该工具调用未返回结果（校验修复）。", + }) + pending_tool_ids = [tc.get("id") for tc in msg["tool_calls"] if tc.get("id")] + seen_ids = set() + fixed.append(msg) + elif role == "tool": + tid = msg.get("tool_call_id") + if tid and tid in pending_tool_ids and tid not in seen_ids: + seen_ids.add(tid) + fixed.append(msg) + elif tid and tid not in pending_tool_ids: + # 孤立 tool 消息：不属于最近一个 assistant 的 tool_calls，跳过 + self.logger.warning(f"_validate: 跳过孤立 tool 消息 {tid}") + else: + fixed.append(msg) + else: + # 遇到非 assistant/tool 消息（user/system），如果有待响应的 tool_calls，先补占位 + for tid in pending_tool_ids: + if tid not in seen_ids: + fixed.append({ + "role": "tool", + "tool_call_id": tid, + "content": "⚠️ 该工具调用未返回结果（校验修复）。", + }) + pending_tool_ids = [] + seen_ids = set() + fixed.append(msg) + i += 1 + # 处理末尾：如果最后一条是 assistant with tool_calls 但没有 tool 响应 + for tid in pending_tool_ids: + if tid not in seen_ids: + fixed.append({ + "role": "tool", + "tool_call_id": tid, + "content": "⚠️ 该工具调用未返回结果（校验修复）。", + }) + # 原地替换 messages 内容 + messages.clear() + messages.extend(fixed) + def _repair_tool_pairs(self): """修复消息历史中的工具调用配对，保证 API 消息结构合法。 场景：assistant 消息携带 tool_calls，但对应的 tool 响应缺失（工具被中断、 - 压缩算法丢弃、端对话提前 break 等），下一次请求会因 + 压缩算法丢弃、端对话提前 break、system 消息插入等），下一次请求会因 "insufficient tool messages following tool_calls" 返回 400。 - 修复策略：为每个缺失响应的 tool_call_id 补一条占位 tool 消息，使其配对完整。 + 修复策略： + 1. 跳过 system 消息查找 tool 响应（压缩后 system 摘要可能插入在 assistant 与 tool 之间） + 2. 只收集与当前 assistant 的 tool_call_id 匹配的 tool 消息 + 3. 为缺失响应的 tool_call_id 补占位 tool 消息 + 4. 移除孤立的 tool 消息（tool_call_id 无对应前置 assistant tool_call） + 5. 同一 tool_call_id 只保留第一条 tool 响应，去重 """ if not self.messages: return repaired = [] i = 0 n = len(self.messages) + # 追踪所有已见的 assistant tool_call_id，用于识别孤立 tool 消息 + known_tool_ids = set() while i < n: msg = self.messages[i] role = msg.get("role") - tool_calls = msg.get("tool_calls") if role == "assistant" else None - if not tool_calls: + if role == "system": repaired.append(msg) i += 1 continue - # 收集紧随其后的 tool 消息 - needed = [tc.get("id") for tc in tool_calls if tc.get("id")] - j = i + 1 - tool_msgs = [] - while j < n and self.messages[j].get("role") == "tool": - tool_msgs.append(self.messages[j]) - j += 1 - have = {tm.get("tool_call_id") for tm in tool_msgs if tm.get("tool_call_id")} - # 补齐缺失的 tool 响应 - for tid in needed: - if tid not in have: - tool_msgs.append({ - "role": "tool", - "tool_call_id": tid, - "content": "⚠️ 该工具调用未返回结果（可能被中断）。", - }) - repaired.append(msg) # 保留 assistant 消息本身 - repaired.extend(tool_msgs) - i = j + if role == "assistant" and msg.get("tool_calls"): + tool_calls = msg["tool_calls"] + needed_ids = [] + for tc in tool_calls: + tid = tc.get("id") + if tid: + needed_ids.append(tid) + known_tool_ids.add(tid) + # 收集紧随其后的 tool 消息，跳过 system 消息 + collected_tools = [] # list of (index, tool_msg) to ensure ordering + collected_ids = set() + j = i + 1 + pending_systems = [] # system 消息在 assistant 和 tool 之间暂存，最后放到 tool 之后 + while j < n: + m = self.messages[j] + r = m.get("role") + if r == "system": + pending_systems.append(m) + j += 1 + continue + if r == "tool": + tid = m.get("tool_call_id") + # 只收集属于当前 assistant 且未重复的 tool 响应 + if tid and tid in needed_ids and tid not in collected_ids: + collected_tools.append(m) + collected_ids.add(tid) + elif tid and tid not in needed_ids and tid in known_tool_ids: + # 属于更早 assistant 的 tool 响应，停止收集（已越过当前 assistant 的范围） + break + # tid 为空或不属于任何已知 assistant，跳过（可能是脏数据） + j += 1 + continue + # 遇到 user 或下一个 assistant，停止收集 + break + # 为缺失的 tool_call_id 补占位消息 + for tid in needed_ids: + if tid not in collected_ids: + collected_tools.append({ + "role": "tool", + "tool_call_id": tid, + "content": "⚠️ 该工具调用未返回结果（可能被中断或压缩丢失）。", + }) + collected_ids.add(tid) + repaired.append(msg) + repaired.extend(collected_tools) + repaired.extend(pending_systems) + i = j + elif role == "tool": + tid = msg.get("tool_call_id") + if tid and tid in known_tool_ids: + repaired.append(msg) + else: + self.logger.warning(f"_repair_tool_pairs: 跳过孤立 tool 消息 id={tid}") + i += 1 + else: + # user 或普通 assistant（无 tool_calls） + repaired.append(msg) + i += 1 self.messages = repaired def reset_conversation(self, reload_prompts: bool = True): @@ -7467,37 +7592,47 @@ class DeepSeekChat: Args: reload_prompts: 是否重新加载系统提示词，默认为 True """ - self.conversation_active = True - self.history_loaded = False # 重置历史加载标记 - - if reload_prompts: - # 完全重置：清除所有消息，重新加载系统提示词 - self.messages = [] - self._init_system_prompts() - self.logger.info("对话已完全重置，系统提示词已重新加载") - else: - # 只清除非系统消息，保留系统提示词 - if self.preserve_system: - system_messages = [msg for msg in self.messages if msg["role"] == "system"] - self.messages = system_messages - else: + with self._msg_lock: + self.conversation_active = True + self.history_loaded = False # 重置历史加载标记 + + if reload_prompts: + # 完全重置：清除所有消息，重新加载系统提示词 self.messages = [] - self.logger.info("对话已重置（仅清除对话历史）") + self._init_system_prompts() + self.logger.info("对话已完全重置，系统提示词已重新加载") + else: + # 只清除非系统消息，保留系统提示词 + if self.preserve_system: + system_messages = [msg for msg in self.messages if msg["role"] == "system"] + self.messages = system_messages + else: + self.messages = [] + self.logger.info("对话已重置（仅清除对话历史）") + + def is_busy(self) -> bool: + """检查AI是否正在处理消息（非阻塞，供调度器等外部调用者判断）。""" + acquired = self._msg_lock.acquire(blocking=False) + if acquired: + self._msg_lock.release() + return False + return True def reload_prompts(self): """重新加载提示词""" - old_system_count = len([msg for msg in self.messages if msg["role"] == "system"]) - - # 移除所有旧的系统消息 - self.messages = [msg for msg in self.messages if msg["role"] != "system"] - - # 重新加载提示词 - self.prompt_manager.reload() - self._init_system_prompts() - - new_system_count = len([msg for msg in self.messages if msg["role"] == "system"]) - self.logger.info(f"提示词已重新加载: {old_system_count} -> {new_system_count} 个") - return new_system_count + with self._msg_lock: + old_system_count = len([msg for msg in self.messages if msg["role"] == "system"]) + + # 移除所有旧的系统消息 + self.messages = [msg for msg in self.messages if msg["role"] != "system"] + + # 重新加载提示词 + self.prompt_manager.reload() + self._init_system_prompts() + + new_system_count = len([msg for msg in self.messages if msg["role"] == "system"]) + self.logger.info(f"提示词已重新加载: {old_system_count} -> {new_system_count} 个") + return new_system_count # ==================== WebSocket处理器（修改版：添加重置对话功能） ==================== @@ -7812,8 +7947,17 @@ class WebSocketHandler: # 【新增】开始新对话：保存当前对话，重置AI状态，重新加载系统提示词 self.ai_manager.history_manager.archive_current_conversation() if self.ai_manager.ai: + ai = self.ai_manager.ai + # 如果AI正在处理（可能是调度器在跑定时任务），先取消 + if ai.is_busy(): + self.logger.info("/new: AI正在处理中，先取消当前对话...") + ai.cancel() + for _ in range(20): + await asyncio.sleep(0.1) + if not ai.is_busy(): + break # 重置对话状态，重新加载系统提示词 - self.ai_manager.ai.reset_conversation(reload_prompts=True) + ai.reset_conversation(reload_prompts=True) # 重置对话进行中标记 self.ai_manager.dialogue_in_progress = False # 重新初始化提示词管理器（确保最新） @@ -8135,16 +8279,8 @@ class EnhancedAIManager: self.history_manager.end_current_conversation(summary) if self.ai: - # 智能体主动结束对话后，重置对话状态但不清除系统提示词 - # 注意：这里只重置对话标记，不清除消息历史 - self.ai.conversation_active = True - # 但为了下次对话不加载旧的历史，我们清除非系统消息 - if self.ai.preserve_system: - system_messages = [msg for msg in self.ai.messages if msg["role"] == "system"] - self.ai.messages = system_messages - else: - self.ai.messages = [] - self.ai.history_loaded = False # 重置历史加载标记 + # 智能体主动结束对话后，重置对话状态但保留系统提示词（锁保护） + self.ai.reset_conversation(reload_prompts=False) self.dialogue_in_progress = False self.logger.info("对话状态已重置，可以开始新对话")