From cdb17a0d73dd57bf636b3da539eb5a831c6adb13 Mon Sep 17 00:00:00 2001 From: xiaohei Date: Mon, 22 Jun 2026 05:44:00 +0000 Subject: [PATCH] =?UTF-8?q?fix(agent+orchestrator):=20=E5=85=B3=20qwen=20?= =?UTF-8?q?=E6=80=9D=E8=80=83=E6=A8=A1=E5=BC=8F=E5=87=BA=E5=B9=B2=E5=87=80?= =?UTF-8?q?=20JSON=20+=20=E4=BF=AE=20bounce=20=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BUG-A: qwen3.7-max 默认思考模式,重 impl 子任务把预算花在 reasoning_content 上不出最终 content JSON → json.loads("") 失败。task_executor._complete 两处 chat.completions.create 注入 extra_body={"enable_thinking": False}(env AGENT_ENABLE_THINKING=1 可opt-in)。对齐 Qwen 官方结构化输出用法。 BUG-B: bounce 仅重开含 impl 文件的任务;impl 失败(0文件)+test成功时 reopens=0 → run 永远 running 空转。加兜底:reopens==0 时重开 FAILED 任务 (受 MAX_REVIEW_CYCLES 约束);仍无可重开则 accept_no_rework 走正常收敛终结。 契约测试全绿(runtime-contract/merge-smoke/workflow-e2e/contract-freeze)。 Co-Authored-By: Claude Opus 4.8 --- agent/task_executor.py | 17 ++++++++++++++++- orchestrator/main.py | 30 +++++++++++++++++++++++------- 2 files changed, 39 insertions(+), 8 deletions(-) diff --git a/agent/task_executor.py b/agent/task_executor.py index 06119e8..979f86f 100644 --- a/agent/task_executor.py +++ b/agent/task_executor.py @@ -610,9 +610,19 @@ Return ONLY the JSON, no other text.""" rc = extra.get("reasoning_content") if isinstance(extra, dict) else None return (rc or "").strip() + @staticmethod + def _thinking_extra_body() -> Optional[dict]: + """qwen3.7-max defaults to THINKING mode: on heavy impl subtasks it spends the token + budget on `reasoning_content` and never emits the final `content` JSON, so json.loads("") + fails ("Expecting value: line 1 column 1"). Disable thinking for deterministic structured + execution. Set AGENT_ENABLE_THINKING=1 to opt back in (e.g. for non-structured analysis).""" + enabled = os.getenv("AGENT_ENABLE_THINKING", "false").strip().lower() in ("1", "true", "yes", "on") + return None if enabled else {"enable_thinking": False} + async def _complete(self, prompt: str, max_tokens: int) -> str: """Call the LLM with optional Jina MCP tools; handles the tool-call loop.""" extra_headers = self._model_attribution_headers() + extra_body = self._thinking_extra_body() tools = await self._load_jina_tools() messages = [{"role": "user", "content": prompt}] @@ -623,6 +633,8 @@ Return ONLY the JSON, no other text.""" max_tokens=max_tokens, extra_headers=extra_headers or None, ) + if extra_body: + kwargs["extra_body"] = extra_body if tools: kwargs["tools"] = tools kwargs["tool_choice"] = "auto" @@ -656,10 +668,13 @@ Return ONLY the JSON, no other text.""" # Fallback: ask for a final answer without tools messages.append({"role": "user", "content": "Please provide your final answer now."}) - response = await self.client.chat.completions.create( + fallback_kwargs: dict = dict( model=self.model, messages=messages, max_tokens=max_tokens, extra_headers=extra_headers or None, ) + if extra_body: + fallback_kwargs["extra_body"] = extra_body + response = await self.client.chat.completions.create(**fallback_kwargs) self._record_openai_usage(response) return self._message_text(response.choices[0].message) diff --git a/orchestrator/main.py b/orchestrator/main.py index a23bd2c..33fec87 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -776,13 +776,29 @@ async def refresh_swarm_run_status(run): if collect_generated_files([t]).get("impl"): if await task_queue.reopen_task(t.task_id): reopened += 1 - run.metadata["review_cycles"] = agg_cycles + 1 - run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + reopened - await swarm_runtime.save_run(run) - logger.info("aggregation: bounced run %s — reopened %d for rework (cycle %d, pass_rate=%s)", - run.swarm_id, reopened, agg_cycles + 1, - (agg.get("validation") or {}).get("pass_rate")) - return + # Deadlock guard: when the impl subtask(s) FAILED (0 impl files) and only test tasks + # produced output, the loop above finds nothing to reopen → reopens=0 → the run would + # spin 'running' forever. Reopen the FAILED tasks so the next cycle retries the impl. + if reopened == 0: + for t in tasks: + if t.status == TaskStatus.FAILED and await task_queue.reopen_task(t.task_id): + reopened += 1 + if reopened > 0: + run.metadata["review_cycles"] = agg_cycles + 1 + run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + reopened + await swarm_runtime.save_run(run) + logger.info("aggregation: bounced run %s — reopened %d for rework (cycle %d, pass_rate=%s)", + run.swarm_id, reopened, agg_cycles + 1, + (agg.get("validation") or {}).get("pass_rate")) + return + # Nothing reopenable (no impl files and no failed tasks to retry): bouncing here would + # deadlock the run in 'running'. Stop bouncing and accept the existing artifacts so the + # run terminates via the normal completion/convergence path below. + agg["decision"] = "accept_no_rework" + run.metadata["aggregation"] = agg + logger.warning("aggregation: bounce on run %s but nothing reopenable (reopens=0) — " + "accepting existing artifacts to avoid deadlock (cycle %d)", + run.swarm_id, agg_cycles + 1) await swarm_runtime.save_run(run) if run.status == next_status: