fix: tool_calls 流中断自愈——400 配对错误时移除不完整段重试一次；加消息结构诊断日志

2026-09-04 13:43:20 +08:00
parent c0c1be05f8
commit 1560a13d86
2 changed files with 61 additions and 23 deletions
@@ -132,6 +132,22 @@ class Agent:
                pass
        return text_result, image_b64

+    def _repair_tool_messages(self):
+        """修复 tool_calls 配对不完整：从尾部移除「assistant 带 tool_calls 但其后
+        tool 回应不足」的消息段（流中断可能丢失分片，400 重试前自愈）。"""
+        for i in range(len(self.messages) - 1, -1, -1):
+            m = self.messages[i]
+            if m.get("role") == "assistant" and m.get("tool_calls"):
+                # 统计其后 tool 消息数是否匹配
+                need = len(m["tool_calls"])
+                have = sum(1 for x in self.messages[i + 1:]
+                           if x.get("role") == "tool")
+                if have < need:
+                    _log.warning("修复不完整 tool_calls 段（need=%d have=%d），回退 %d 条消息",
+                                 need, have, len(self.messages) - i)
+                    self.messages = self.messages[:i]
+                return
+
    # ---------- 主循环（流式） ----------
    async def run_stream(self, prompt: str, serial: str = "",
                         history=None, on_delta=None, on_tool=None,
@@ -166,29 +182,41 @@ class Agent:
            content_parts = []
            tool_acc = {}           # index -> {id, name, args}
            has_tool = False
-            async for chunk in self._chat_stream():
-                choice = (chunk.get("choices") or [{}])[0]
-                delta = choice.get("delta") or {}
-                text = delta.get("content")
-                if text:
-                    content_parts.append(text)
-                    if on_delta:
-                        on_delta(text, "content")
-                rtext = delta.get("reasoning_content")
-                if rtext:
-                    if on_delta:
-                        on_delta(rtext, "reasoning")
-                for tc in delta.get("tool_calls") or []:
-                    has_tool = True
-                    idx = tc.get("index", 0)
-                    acc = tool_acc.setdefault(idx, {"id": "", "name": "", "args": ""})
-                    if tc.get("id"):
-                        acc["id"] = tc["id"]
-                    fn = tc.get("function") or {}
-                    if fn.get("name"):
-                        acc["name"] += fn["name"]
-                    if fn.get("arguments"):
-                        acc["args"] += fn["arguments"]
+            retried = False
+            while True:
+                try:
+                    async for chunk in self._chat_stream():
+                        choice = (chunk.get("choices") or [{}])[0]
+                        delta = choice.get("delta") or {}
+                        text = delta.get("content")
+                        if text:
+                            content_parts.append(text)
+                            if on_delta:
+                                on_delta(text, "content")
+                        rtext = delta.get("reasoning_content")
+                        if rtext:
+                            if on_delta:
+                                on_delta(rtext, "reasoning")
+                        for tc in delta.get("tool_calls") or []:
+                            has_tool = True
+                            idx = tc.get("index", 0)
+                            acc = tool_acc.setdefault(idx, {"id": "", "name": "", "args": ""})
+                            if tc.get("id"):
+                                acc["id"] = tc["id"]
+                            fn = tc.get("function") or {}
+                            if fn.get("name"):
+                                acc["name"] += fn["name"]
+                            if fn.get("arguments"):
+                                acc["args"] += fn["arguments"]
+                    break
+                except RuntimeError as e:
+                    # 流中断导致 tool_calls 分片丢失：修复后重试一次
+                    if ("tool_calls" in str(e) or "must be followed" in str(e)) and not retried:
+                        _log.warning("tool_calls 消息不完整，自愈重试")
+                        self._repair_tool_messages()
+                        retried = True
+                        continue
+                    raise

            full_content = "".join(content_parts)

@@ -367,6 +367,16 @@ def _agent_thread(run_id, prompt, serial, cfg):
                _log.warning(f"经验保存异常: {e}")
    except Exception as e:
        _log.warning(f"Agent 运行异常: {e}")
+        # 诊断：打印消息结构（tool_calls 与 tool 消息配对检查）
+        try:
+            roles = [m.get("role", "?") for m in agent.messages]
+            tcs = sum(1 for m in agent.messages
+                      if m.get("tool_calls") and isinstance(m.get("tool_calls"), list))
+            tools_msg = sum(1 for m in agent.messages if m.get("role") == "tool")
+            _log.warning(f"诊断 messages: {len(agent.messages)} 条 roles={roles[-8:]} "
+                         f"tool_calls消息={tcs} tool回应={tools_msg}")
+        except Exception:
+            pass
        with _lock:
            _run["state"] = "error"
            _run["error"] = f"{type(e).__name__}: {str(e)[:200]}"