diff --git a/mcp_agent/agent.py b/mcp_agent/agent.py index ca8f9d3..ba80ac1 100644 --- a/mcp_agent/agent.py +++ b/mcp_agent/agent.py @@ -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) diff --git a/web/agent_api.py b/web/agent_api.py index 2dcb2a1..d8c1786 100644 --- a/web/agent_api.py +++ b/web/agent_api.py @@ -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]}"