From c769dd20f7767be08efce324a19990d5dda6654c Mon Sep 17 00:00:00 2001 From: hkdong Date: Wed, 27 May 2026 18:07:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=80=BB=E6=8E=A7?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E5=85=B1=E8=AF=86=E5=90=8E=E4=BB=8D?= =?UTF-8?q?=E8=A2=AB=E9=94=99=E5=88=A4=E4=B8=BA=E7=BB=A7=E7=BB=AD=E6=B4=BE?= =?UTF-8?q?=E6=B4=BB=E7=9A=84=E6=AD=BB=E5=BE=AA=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _find_dispatch_message 原实现会一路向前翻历史,直到撞到任意一条带 tool_calls 的 AIMessage,导致本轮 LLM 即使共识(tool_calls=[]),也会 被上一轮的派活 AI 翻出来再用一次,前端因此把同样的 task 又发给子智能体, 形成"嘴上共识、手上派活"的死循环。 改成只看最新一条 AIMessage(跳过末尾 skill_stop / clarification ack ToolMessage 即可),让 tool_calls 是否为空作为本轮的唯一权威信号。 Co-Authored-By: Claude Opus 4.7 --- .claude/settings.json | 10 ++++++ .../app/gateway/routers/multi_agent.py | 33 +++++++++++-------- offline-backend-20260512/backend/config.yaml | 2 +- 3 files changed, 30 insertions(+), 15 deletions(-) create mode 100644 .claude/settings.json diff --git a/.claude/settings.json b/.claude/settings.json new file mode 100644 index 00000000..e02ee52c --- /dev/null +++ b/.claude/settings.json @@ -0,0 +1,10 @@ +{ + "permissions": { + "allow": [ + "PowerShell(uv --version)", + "PowerShell(netstat -ano)", + "PowerShell(Get-NetTCPConnection -LocalPort 8001 -ErrorAction SilentlyContinue)", + "PowerShell(Get-Process -Id 20560 -ErrorAction SilentlyContinue | Select-Object Id,ProcessName,Path,StartTime; try { \\(Invoke-WebRequest -Uri \"http://127.0.0.1:8001/health\" -UseBasicParsing -TimeoutSec 5\\).Content } catch { \"health check failed: $_\" })" + ] + } +} diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 97854298..bcf6bb01 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -764,25 +764,30 @@ def _find_trailing_ai_text(messages: list[dict[str, Any]], *, exclude_text: str def _find_dispatch_message(messages: list[dict[str, Any]]) -> dict[str, Any]: - """Locate the latest AIMessage that carries dispatch tool calls. + """Locate the latest AIMessage from THIS turn. - The very last message can be a skill-stop ToolMessage injected by - ``SkillStopMiddleware`` when it intercepts the dispatch, or a clarification - ToolMessage. Either way, the actionable tool_calls live on the AIMessage - *before* it, so we walk backwards until we hit one with ``tool_calls``. - Returns the latest AIMessage when none of them carry tool_calls so the - caller can still extract its text content. + Trailing ToolMessages (skill_stop / clarification ack injected by + ``SkillStopMiddleware`` or ``ClarificationMiddleware``) are skipped because + the actionable tool_calls live on the AIMessage that emitted them. We stop + at the **first AIMessage encountered** — its ``tool_calls`` (or lack + thereof) is the authoritative signal for this turn. + + 历史踩坑:之前实现是"一直往前翻直到撞到带 tool_calls 的 AI",看似优雅, + 实际上让 leader 永远收敛不了:本轮 LLM 不论怎么共识,只要历史上派过活, + 上一轮的派活 AI 就会被翻出来再用一次,前端会把同样的 task 又发给子智能体, + 陷入死循环。现在停在最新一条 AI 上,把判定交给调用方:`tool_calls=[]` → + 共识分支,有 tool_calls → 派活分支。 """ - last_ai: dict[str, Any] = {} for msg in reversed(messages): if not isinstance(msg, dict): continue - if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"): - if not last_ai: - last_ai = msg - if msg.get("tool_calls"): - return msg - return last_ai + msg_type = (msg.get("type") or "").lower() + if msg_type in ("tool", "toolmessage"): + continue # 跳过末尾 skill_stop / clarification ack + if msg_type in ("ai", "aimessage", "aimessagechunk"): + return msg + break # 撞到 Human/其他类型,本轮没 AI(异常路径,留空让上游报错) + return {} async def _leader_run( diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index ac83d28f..fbf628ce 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -14,7 +14,7 @@ models: display_name: DeepSeek Chat use: langchain_openai:ChatOpenAI model: deepseek-chat - api_key: sk-e1c22b7931dc446f8b67e57878edd105 + api_key: sk-1520028668cb4cc388ed8315ff1e856e base_url: https://api.deepseek.com request_timeout: 600.0 max_retries: 2 From b7da062a4c2bab622bad6a1c643cd19c7ed71914 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 27 May 2026 19:11:36 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E5=A4=9A=E6=99=BA=E8=83=BD=E4=BD=93?= =?UTF-8?q?=E4=BC=9A=E5=95=86=E6=B2=99=E7=AE=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../roundtable-planning/api/multi-agent.ts | 171 ++++++++++++++++-- .../hooks/useStep2Orchestration.ts | 35 +++- 2 files changed, 191 insertions(+), 15 deletions(-) diff --git a/frontend-web/src/roundtable-planning/api/multi-agent.ts b/frontend-web/src/roundtable-planning/api/multi-agent.ts index 7a46211e..ef186438 100644 --- a/frontend-web/src/roundtable-planning/api/multi-agent.ts +++ b/frontend-web/src/roundtable-planning/api/multi-agent.ts @@ -260,6 +260,69 @@ export interface StreamMultiAgentResult { clarification: ClarificationPayload | null; } +/** + * Best-effort partial JSON parser for accumulated `tool_call_chunks.args`. + * + * LangChain emits tool args as a series of string fragments (one delta per + * chunk). To get progressive `args.path` / `args.content` during streaming — + * the way `useStream` does in the main chat — we concatenate the deltas and + * try to parse. When the JSON is still mid-token (e.g. `{"path":"/mnt`) we + * close any open string + brackets and retry, so callers can read partially + * available keys even before the full payload arrives. + * + * Returns null when nothing useful can be extracted yet. + */ +function parsePartialJson(s: string): Record | null { + if (!s) return null; + try { + const v = JSON.parse(s); + return v && typeof v === "object" ? (v as Record) : null; + } catch { + // fall through to best-effort close + retry + } + let trimmed = s; + // A dangling `\` would turn our injected closing `"` into an escape; drop it. + if (trimmed.endsWith("\\")) trimmed = trimmed.slice(0, -1); + let inString = false; + let escaped = false; + const stack: string[] = []; + for (const ch of trimmed) { + if (escaped) { + escaped = false; + continue; + } + if (ch === "\\") { + escaped = true; + continue; + } + if (inString) { + if (ch === '"') inString = false; + continue; + } + if (ch === '"') { + inString = true; + continue; + } + if (ch === "{") stack.push("}"); + else if (ch === "[") stack.push("]"); + else if (ch === "}" || ch === "]") stack.pop(); + } + let closed = trimmed; + if (inString) { + closed += '"'; + } else { + // Mid-key / dangling separator (`{"path":`) can't be balanced cleanly. + closed = closed.replace(/[,:]\s*$/, ""); + } + while (stack.length) closed += stack.pop(); + try { + const v = JSON.parse(closed); + return v && typeof v === "object" ? (v as Record) : null; + } catch { + return null; + } +} + /** * Identify the final status frame our backend appends after the LangGraph * stream. Required keys are `status` (array for leader, string for special) @@ -394,6 +457,61 @@ export async function streamMultiAgent( req.onPhaseChange?.(next, name, args); }; + /** + * 累积 tool_call_chunks(delta args 字符串) —— 与 LangChain `AIMessageChunk` + * 合并的内部行为对齐,补主聊天靠 `useStream` 暗中做的那部分工作。Key 是 chunk.index + * (并发 tool 调用通常 0/1/2 区分);chunk.id 变化时视作"同一 index 上的新调用"并重置。 + * 每帧累积后用 `parsePartialJson` 尽力 parse,成功就把合成的 `tool_calls[]` 注入 + * 到 messages-tuple 帧的 chunkObj 上,让上层 `onUpstreamFrame` 和 `onPhaseChange` + * 都能拿到完整的 `args.path` / `args.content` 来做"流式期间打开沙箱 + 实时预览"。 + */ + const toolCallAcc = new Map< + number, + { id?: string; name: string; argsStr: string } + >(); + + type SyntheticToolCall = { + id?: string; + name: string; + args: Record; + }; + + const ingestToolCallChunks = (chunks: unknown): void => { + if (!Array.isArray(chunks)) return; + for (const tcc of chunks) { + if (!tcc || typeof tcc !== "object") continue; + const tccObj = tcc as Record; + const idx = typeof tccObj.index === "number" ? tccObj.index : 0; + let entry = toolCallAcc.get(idx); + if (!entry) { + entry = { name: "", argsStr: "" }; + toolCallAcc.set(idx, entry); + } + if (typeof tccObj.id === "string" && tccObj.id) { + if (entry.id && entry.id !== tccObj.id) { + // 同一 index 上换了新 tool_call,重置累积(不太常见,defensive)。 + entry.name = ""; + entry.argsStr = ""; + } + entry.id = tccObj.id; + } + if (typeof tccObj.name === "string" && tccObj.name) entry.name = tccObj.name; + if (typeof tccObj.args === "string") entry.argsStr += tccObj.args; + } + }; + + const buildSyntheticToolCalls = (): SyntheticToolCall[] => { + const out: SyntheticToolCall[] = []; + for (const entry of toolCallAcc.values()) { + if (!entry.name) continue; + const parsed = parsePartialJson(entry.argsStr); + if (parsed) { + out.push({ id: entry.id, name: entry.name, args: parsed }); + } + } + return out; + }; + while (true) { const { done, value } = await reader.read(); if (done) break; @@ -441,26 +559,51 @@ export async function streamMultiAgent( const toolCalls = chunkObj.tool_calls; const toolCallChunks = chunkObj.tool_call_chunks; const content = chunkObj.content; + + // 1) 累积本帧的 tool_call_chunks delta(若有),再合成可能可读的 tool_calls。 + // 主聊天 useStream 在 SDK 层做了这件事,我们在自定义 SSE 里手动补上。 + ingestToolCallChunks(toolCallChunks); + const hasCompleteToolCalls = + Array.isArray(toolCalls) && toolCalls.length > 0; + let synthetic: SyntheticToolCall[] = []; + if (!hasCompleteToolCalls && toolCallAcc.size > 0) { + synthetic = buildSyntheticToolCalls(); + if (synthetic.length > 0) { + // 把累积出的 tool_calls 注入回帧的 chunkObj —— 这样下面 onUpstreamFrame + // 那边的 handleUpstreamFrame 只读 tool_calls 字段就能拿到完整 args, + // 不必再单独做一份累积。 + chunkObj.tool_calls = synthetic; + } + } + + const effectiveToolCalls: unknown[] | null = hasCompleteToolCalls + ? (toolCalls as unknown[]) + : synthetic.length > 0 + ? synthetic + : null; + const hasToolCall = - (Array.isArray(toolCalls) && toolCalls.length > 0) || + (effectiveToolCalls !== null && effectiveToolCalls.length > 0) || (Array.isArray(toolCallChunks) && toolCallChunks.length > 0); + if (chunkType === "tool") { setPhase("tool_result"); } else if (hasToolCall) { - // Pick the most informative name we can find in either field. - // 优先用 tool_calls(完整对象,含解析好的 args),tool_call_chunks 的 args - // 通常是局部 JSON 字符串(LangGraph 流式期间逐字符发),没法直接用。 - const firstCall = Array.isArray(toolCalls) && toolCalls.length > 0 - ? (toolCalls[0] as Record) - : Array.isArray(toolCallChunks) && toolCallChunks.length > 0 - ? (toolCallChunks[0] as Record) - : null; - const toolName = firstCall && typeof firstCall.name === "string" ? firstCall.name : ""; - // 仅当来自 tool_calls(完整对象)时透传 args;tool_call_chunks 的 args 是 - // string 拼接中,不传上去给 UI,等下一个完整 chunk 再带。 - const fromCompleteToolCalls = Array.isArray(toolCalls) && toolCalls.length > 0; + // Pick the most informative name + args we have so far. + // 优先用累积出的 / 完整的 tool_calls(含可读 args 对象);若都还不到火候, + // 回退到 tool_call_chunks 拿名字(args 还是 string,不透传)。 + const firstCall = + effectiveToolCalls && effectiveToolCalls.length > 0 + ? (effectiveToolCalls[0] as Record) + : Array.isArray(toolCallChunks) && toolCallChunks.length > 0 + ? (toolCallChunks[0] as Record) + : null; + const toolName = + firstCall && typeof firstCall.name === "string" ? firstCall.name : ""; const args = - fromCompleteToolCalls && firstCall && typeof firstCall.args === "object" && firstCall.args !== null + firstCall && + typeof firstCall.args === "object" && + firstCall.args !== null ? (firstCall.args as Record) : undefined; setPhase("tool_calling", toolName, args); diff --git a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts index 7758b912..2c61e515 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts @@ -223,11 +223,21 @@ export function useStep2Orchestration( select: selectArtifact, setOpen: setArtifactsOpen, deselect: deselectArtifact, + autoOpen, + autoSelect, } = useArtifacts(); const artifactPathsRef = useRef([]); useEffect(() => { artifactPathsRef.current = artifacts; }, [artifacts]); + // 用户手动关掉沙箱后,autoOpen 会被 ArtifactsProvider 置 false —— 把它们打到 ref + // 上,让 openSeatVirtualArtifact 不用进 useCallback deps,避免每次 toggle 重建函数。 + const autoOpenFlagsRef = useRef({ autoOpen, autoSelect }); + useEffect(() => { + autoOpenFlagsRef.current = { autoOpen, autoSelect }; + }, [autoOpen, autoSelect]); + /** 已自动打开过的 virtualUrl 集合 —— 流式期间一秒会回调几十次,防止重复 schedule 开沙箱。 */ + const hasAutoOpenedRef = useRef>(new Set()); // ─── State ───────────────────────────────────────────────────────── const [activeArtifactThreadId, setActiveArtifactThreadId] = useState( @@ -381,9 +391,20 @@ export function useStep2Orchestration( /** * 用「虚拟 URL」(write-file: scheme)打开沙箱 —— 内容来自 stub messages 的 args.content, * 不污染 `artifacts[]` 真实路径列表(列表后续可由 present_files / 显式 setArtifacts 填充)。 + * + * auto=true 流式期间: + * - 已自动打开过该 virtualUrl 就直接跳过(防止 onPhaseChange / onUpstreamFrame 每帧重复 schedule); + * - 用户手动关闭过沙箱(`autoOpen` 已被 ArtifactsProvider 置 false)也尊重决定,不再自动开。 + * auto=false(用户点击步骤 / 文件卡)始终强制打开。 */ const openSeatVirtualArtifact = useCallback( (seatThreadId: string, virtualUrl: string, auto: boolean) => { + if (auto) { + if (hasAutoOpenedRef.current.has(virtualUrl)) return; + const flags = autoOpenFlagsRef.current; + if (!flags.autoOpen || !flags.autoSelect) return; + hasAutoOpenedRef.current.add(virtualUrl); + } setActiveArtifactThreadId(seatThreadId); const open = () => { selectArtifact(virtualUrl, auto); @@ -625,7 +646,9 @@ export function useStep2Orchestration( const steps: Step2Dialogue["steps"] = d.steps ?? []; const last = steps[steps.length - 1]; if (last && last.key === key) { - if (args && !last.args) { + // 同 phase:tool 重复回调 —— 流式累积期 args 会持续长大(present_files + // 多文件分批、write_file 内容增长),拿最新 args 覆盖,不要早早定死。 + if (args && args !== last.args) { const next = [...steps]; next[next.length - 1] = { ...last, args }; return { ...d, steps: next }; @@ -804,11 +827,15 @@ export function useStep2Orchestration( safetyCounter += 1; highlightSpeaker(2); + // 总控也可能直接 write_file —— 之前只接了 special 流,导致总控写文件时沙箱不自动打开。 + // 这里把 leader thread id 绑到气泡 + artifact handler 上,与 special 同样的行为。 + const leaderThreadId = resolveSeatThreadId(ids, coord); const leaderBubbleId = appendStreamingDialogue({ sender: "总控协调", avatarType: "blue", initialContent: "", confidence: "调度中", + agentThreadId: leaderThreadId, }); const leaderAborter = new AbortController(); @@ -832,7 +859,10 @@ export function useStep2Orchestration( `${phase}:${toolName ?? ""}`, args, ); + handleArtifactToolPhase(phase, toolName, args, leaderThreadId); }, + onUpstreamFrame: (frame) => + handleUpstreamFrame(frame, leaderThreadId), }); } catch (err) { activeAbortersRef.current.delete(leaderAborter); @@ -1281,6 +1311,8 @@ export function useStep2Orchestration( const restoredStubs = snapshot.seatStubMessages ?? {}; seatStubMessagesRef.current = restoredStubs; setSeatStubMessages(restoredStubs); + // 草稿里历史 write_file 的虚拟 URL 与当前会话无关 —— 清空守卫,允许新 run 触发自动打开。 + hasAutoOpenedRef.current = new Set(); // 调用方根据 furthestStep 决定是否置 init 闸门为已 init;这里默认置上, // 调用方需要的话再 resetInitGate()。 step2InitRef.current = true; @@ -1307,6 +1339,7 @@ export function useStep2Orchestration( setArtifacts([]); seatStubMessagesRef.current = {}; setSeatStubMessages({}); + hasAutoOpenedRef.current = new Set(); pendingInjectPromptRef.current = null; deselectArtifact(); step2InitRef.current = false;