From 00f8dd78a91da0440a94d82ce16658e1db710cf7 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 20 Sep 2026 21:09:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=97=B6=E9=97=B4=E7=BA=BF=E4=B8=8A=E8=AE=A4?= =?UTF-8?q?=E5=BE=97=E5=87=BA=E5=B9=B6=E8=A1=8C=E5=88=86=E6=94=AF=E4=BA=86?= =?UTF-8?q?=EF=BC=9A3=20=E8=B7=AF=E5=B9=B6=E8=A1=8C=20=C2=B7=20=E5=90=88?= =?UTF-8?q?=E8=AE=A1=206.6s=EF=BC=8C=E5=AE=9E=E9=99=85=203.0s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 图上一个节点连出多条边就是 fan-out，LangGraph 在同一个 superstep 里并发执行 ——这一直是真并发。但在时间线上它们只是穿插出现的几行，"这三路是同时跑的、 因此省下了 3.6 秒"一个字都没说。上一轮给协作团队做了泳道，这条同样的信息 在 fan-out 这边还是缺的。 现在按时间跨度把同时跑的节点收进一个分组，标题写明几路、meta 写明省了多少： ⑂ 3 路并行 合计 6.6s，实际 3.0s ├ 查销售库 2.2s ├ 查库存库 3.0s └ 查客户库 1.5s **判据是"同一批出发"，不是"跨度重叠"。** 第一版按包含关系排除嵌套，结果漏掉了 先跑完的那一路：三路同时开始但先后结束时，长的那条在时间上确实"包含"短的。 实测 leg0 [0.05,2.21] / leg1 [0.05,3.01] / leg2 [0.05,1.51]，被漏掉的正是 最快的 leg2——而漏掉最快那路会让"省下多少"算少。改成看开始时刻是否落在同一批 （250ms 窗口），fan-out 的几路实测相差不到 1ms，子工作流那种真嵌套则是父节点 明显更早开始，自然分得开。 组内要求两两都同批，而不是链式传递：否则 A|B 重叠、B|C 重叠但 A|C 不重叠的 三个节点也会被串成一组，而它们并不是同时在跑。瞬时节点（<30ms）不参与—— 它们谁跟谁都"重叠"，是噪音不是信息。 **误报比漏报更糟**：它会让人以为省了时间。所以断言里专门守了反向——db、human、 loop_approve 三份顺序执行的样本一个都不能被判成并行。 fixtures 里新增的 fanout 是真跑出来的，不是手写的：三路分别 sleep 1.8 / 2.6 / 1.1 秒，事件原样收进来，所以 6.6s / 3.0s 这两个数和后端量到的是同一份。 验证：check-decode 新增 6 条（含三份顺序样本的反向断言）；check-stream 新增 10 条，宽窄两栏都验了不溢出；后端 276 passed 不受影响；tsc 干净。 --- frontend/src/dev/preview.tsx | 7 + frontend/src/run/__tests__/fixtures.json | 315 +++++++++++++++++++++++ frontend/src/run/decode.ts | 90 +++++++ scripts/check-decode.mjs | 18 ++ scripts/check-stream.mjs | 24 ++ 5 files changed, 454 insertions(+) diff --git a/frontend/src/dev/preview.tsx b/frontend/src/dev/preview.tsx index b21cff93..916f3391 100644 --- a/frontend/src/dev/preview.tsx +++ b/frontend/src/dev/preview.tsx @@ -229,6 +229,13 @@ function Preview() { if (params.get('review') === '1') { return } + if (params.get('fanout') === '1') { + return + } if (params.get('team') === '1') { return } diff --git a/frontend/src/run/__tests__/fixtures.json b/frontend/src/run/__tests__/fixtures.json index 6cc267bc..aa191aed 100644 --- a/frontend/src/run/__tests__/fixtures.json +++ b/frontend/src/run/__tests__/fixtures.json @@ -2030,5 +2030,320 @@ "duration_ms": 19 } } + ], + "fanout": [ + { + "seq": 1, + "type": "run.started", + "node_id": null, + "ts": 1789909355.5191329, + "data": { + "nodes": 6, + "resumed": false + } + }, + { + "seq": 2, + "type": "node.started", + "node_id": "start", + "ts": 1789909355.528116, + "data": { + "node_type": "input", + "label": "start" + } + }, + { + "seq": 3, + "type": "node.finished", + "node_id": "start", + "ts": 1789909355.5313249, + "data": { + "duration_ms": 0, + "attempt": 1, + "preview": { + "q": "x" + }, + "vars": { + "q": "x" + }, + "artifact": "a69fbbcf7209c6f659a75067c9fa03037c2ae23f55a6f854d5994209129bbbf6" + } + }, + { + "seq": 4, + "type": "node.started", + "node_id": "leg0", + "ts": 1789909355.566488, + "data": { + "node_type": "code", + "label": "查销售库" + } + }, + { + "seq": 5, + "type": "sandbox.start", + "node_id": "leg0", + "ts": 1789909355.5680408, + "data": { + "language": "python", + "limits": { + "timeout": 30, + "memory_mb": 512, + "cpus": 1.0, + "network": false, + "max_output": 20000 + }, + "isolation": "auto", + "code": "import time;time.sleep(1.8);print('查销售库 完成')", + "code_truncated": false, + "interpolated": false + } + }, + { + "seq": 6, + "type": "node.started", + "node_id": "leg1", + "ts": 1789909355.569063, + "data": { + "node_type": "code", + "label": "查库存库" + } + }, + { + "seq": 7, + "type": "sandbox.start", + "node_id": "leg1", + "ts": 1789909355.5698738, + "data": { + "language": "python", + "limits": { + "timeout": 30, + "memory_mb": 512, + "cpus": 1.0, + "network": false, + "max_output": 20000 + }, + "isolation": "auto", + "code": "import time;time.sleep(2.6);print('查库存库 完成')", + "code_truncated": false, + "interpolated": false + } + }, + { + "seq": 8, + "type": "node.started", + "node_id": "leg2", + "ts": 1789909355.570695, + "data": { + "node_type": "code", + "label": "查客户库" + } + }, + { + "seq": 9, + "type": "sandbox.start", + "node_id": "leg2", + "ts": 1789909355.5714948, + "data": { + "language": "python", + "limits": { + "timeout": 30, + "memory_mb": 512, + "cpus": 1.0, + "network": false, + "max_output": 20000 + }, + "isolation": "auto", + "code": "import time;time.sleep(1.1);print('查客户库 完成')", + "code_truncated": false, + "interpolated": false + } + }, + { + "seq": 10, + "type": "sandbox.end", + "node_id": "leg2", + "ts": 1789909357.024351, + "data": { + "ok": true, + "exit_code": 0, + "duration_ms": 1456, + "backend": "microvm", + "stdout": "查客户库 完成\n", + "stderr": "" + } + }, + { + "seq": 11, + "type": "node.finished", + "node_id": "leg2", + "ts": 1789909357.027894, + "data": { + "duration_ms": 1456, + "attempt": 1, + "preview": { + "ok": true, + "exit_code": 0, + "stdout": "查客户库 完成\n", + "stderr": "", + "duration_ms": 1456, + "timed_out": false, + "truncated": false, + "backend": "microvm", + "files_written": "[0 项]", + "error": null, + "session_reset": false, + "data": null, + "text": "查客户库 完成\n" + }, + "vars": null, + "artifact": "696aaa83aa727152eac49f8b82dabdd860ac3d37af96c37f0854a819b0c0c350" + } + }, + { + "seq": 12, + "type": "sandbox.end", + "node_id": "leg0", + "ts": 1789909357.7229152, + "data": { + "ok": true, + "exit_code": 0, + "duration_ms": 2162, + "backend": "microvm", + "stdout": "查销售库 完成\n", + "stderr": "" + } + }, + { + "seq": 13, + "type": "node.finished", + "node_id": "leg0", + "ts": 1789909357.725617, + "data": { + "duration_ms": 2189, + "attempt": 1, + "preview": { + "ok": true, + "exit_code": 0, + "stdout": "查销售库 完成\n", + "stderr": "", + "duration_ms": 2162, + "timed_out": false, + "truncated": false, + "backend": "microvm", + "files_written": "[0 项]", + "error": null, + "session_reset": false, + "data": null, + "text": "查销售库 完成\n" + }, + "vars": null, + "artifact": "00944f0b705f77928790de8c042792bea1237d3907293ec249133e7eab0af61f" + } + }, + { + "seq": 14, + "type": "sandbox.end", + "node_id": "leg1", + "ts": 1789909358.5266712, + "data": { + "ok": true, + "exit_code": 0, + "duration_ms": 2959, + "backend": "microvm", + "stdout": "查库存库 完成\n", + "stderr": "" + } + }, + { + "seq": 15, + "type": "node.finished", + "node_id": "leg1", + "ts": 1789909358.529147, + "data": { + "duration_ms": 2959, + "attempt": 1, + "preview": { + "ok": true, + "exit_code": 0, + "stdout": "查库存库 完成\n", + "stderr": "", + "duration_ms": 2959, + "timed_out": false, + "truncated": false, + "backend": "microvm", + "files_written": "[0 项]", + "error": null, + "session_reset": false, + "data": null, + "text": "查库存库 完成\n" + }, + "vars": null, + "artifact": "5870ecb6df825e09b5b9057ff182a9c933656ed5ab8f42aa9bb0bd01fd962707" + } + }, + { + "seq": 16, + "type": "node.started", + "node_id": "merge", + "ts": 1789909358.531316, + "data": { + "node_type": "transform", + "label": "merge" + } + }, + { + "seq": 17, + "type": "node.finished", + "node_id": "merge", + "ts": 1789909358.5334458, + "data": { + "duration_ms": 0, + "attempt": 1, + "preview": "三路都回来了", + "vars": null, + "artifact": "9b4dc39b58e543ae8c5ef60ccc10c79ef2b1a720f04d40a882085731a7b28854" + } + }, + { + "seq": 18, + "type": "node.started", + "node_id": "out", + "ts": 1789909358.534973, + "data": { + "node_type": "output", + "label": "out" + } + }, + { + "seq": 19, + "type": "node.finished", + "node_id": "out", + "ts": 1789909358.537149, + "data": { + "duration_ms": 0, + "attempt": 1, + "preview": { + "r": "" + }, + "vars": null, + "artifact": "bd61075bf9ba8f23153729bbad7fada6a82f1dbeaae23f64b746441219f40327" + } + }, + { + "seq": 20, + "type": "run.finished", + "node_id": null, + "ts": 1789909358.541813, + "data": { + "output": { + "r": "" + }, + "usage": { + "duration_ms": 3021 + }, + "duration_ms": 3021 + } + } ] } \ No newline at end of file diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index ec4454e0..cdcedc8e 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -319,6 +319,84 @@ function toolStep(seq: number, tool: string, args: Record): Step { return { ...base, kind: 'tool', title: `调用 ${tool}`, detail: argText } } +/** 低于这个毫秒数的节点不参与并行分组：瞬时节点谁跟谁都"重叠"，是噪音不是信息 */ +const CONCURRENT_MIN_MS = 30 +/** 开始时刻相差在这个数以内算"同一批出发"。实测 fan-out 的几路相差不到 1ms */ +const SAME_WAVE_MS = 250 + +/** + * 认出哪几个节点是**同时**跑的，把它们收进一个分组里。 + * + * 图上一个节点连出多条边就是 fan-out，LangGraph 会在同一个 superstep 里并发 + * 执行它们——这是真并发（实测 3 个各 sleep 2 秒的节点，全部 +0.00s 开始、 + * +2.37s 结束，墙钟 2.6 秒）。但在时间线上它们只是穿插出现的几行， + * "这三路是同时跑的、因此省下了 3.6 秒"这件事一个字都没说。 + * + * 判定用时间跨度重叠，而不是看事件顺序——顺序只说明"先后发出"， + * 说明不了"同时在跑"。 + * + * 判据是"同一批出发"：fan-out 的几路由 LangGraph 在同一个 superstep 里启动， + * 开始时刻相差以毫秒计。**不能只看跨度重叠**——三路同时开始但先后结束时， + * 长的那条在时间上确实"包含"短的，按包含关系去排会把先跑完的那路漏掉 + * （实测 leg0 [0.05,2.21] / leg1 [0.05,3.01] / leg2 [0.05,1.51]， + * 漏的就是 leg2）。 + * + * 子工作流那种真嵌套则是父节点明显更早开始、更晚结束，出发时刻对不上。 + */ +function groupConcurrent( + out: Step[], spans: Map, +): void { + const idx = new Map() + out.forEach((s, i) => idx.set(s, i)) + + const candidates = out.filter((s) => { + const sp = s.nodeId ? spans.get(s.nodeId) : undefined + return !!sp && sp.ms >= CONCURRENT_MIN_MS + }) + if (candidates.length < 2) return + + const together = (a: Step, b: Step): boolean => { + const x = spans.get(a.nodeId!)! + const y = spans.get(b.nodeId!)! + if (x.start >= y.end || y.start >= x.end) return false // 压根没重叠 + return Math.abs(x.start - y.start) * 1000 <= SAME_WAVE_MS // 同一批出发 + } + + const used = new Set() + const groups: Step[][] = [] + for (const a of candidates) { + if (used.has(a)) continue + const g = [a] + for (const b of candidates) { + if (b === a || used.has(b)) continue + // 和组里每一个都重叠才算同一批，否则 A|B 重叠、B|C 重叠但 A|C 不重叠 + // 也会被串成一组，而那三个并不是同时在跑 + if (g.every((m) => together(m, b))) g.push(b) + } + if (g.length > 1) { g.forEach((m) => used.add(m)); groups.push(g) } + } + + for (const g of groups) { + g.sort((a, b) => (idx.get(a) ?? 0) - (idx.get(b) ?? 0)) + const sp = g.map((m) => spans.get(m.nodeId!)!) + const wallMs = Math.round( + (Math.max(...sp.map((x) => x.end)) - Math.min(...sp.map((x) => x.start))) * 1000) + const sumMs = sp.reduce((acc, x) => acc + x.ms, 0) + const at = Math.min(...g.map((m) => out.indexOf(m))) + const first = g[0] + for (const m of g) out.splice(out.indexOf(m), 1) + out.splice(at, 0, { + id: `par-${first.id}`, seq: first.seq, kind: 'branch', status: 'done', + title: `${g.length} 路并行`, + // 省下多少是这一组存在的理由，放 meta 里一眼看得到 + meta: sumMs > wallMs + ? `合计 ${dur(sumMs)}，实际 ${dur(wallMs)}` + : dur(wallMs), + children: g, + }) + } +} + /** * 把一次运行的事件解码成步骤树。 * @@ -337,6 +415,8 @@ export function decodeRun(events: RunEvent[]): Step[] { const pendingAgents = new Map() /** node_id → 协作团队的泳道数据。边解码边攒，最后挂到该节点的顶层 Step 上 */ const teams = new Map() + /** node_id → 起止时刻。用来事后认出"哪几个节点是同时跑的" */ + const spans = new Map() const teamOf = (nodeId: string | undefined): TeamRun | null => { if (!nodeId) return null @@ -444,6 +524,9 @@ export function decodeRun(events: RunEvent[]): Step[] { // 人工审批恢复后 LangGraph 会重放该节点，node.started 因此来第二遍。 // 那不是"又执行了一个步骤"，是同一步继续——复用原来那条， // 否则时间线上每审批一次就多一个同名节点。 + if (nodeId && !spans.has(nodeId)) { + spans.set(nodeId, { start: event.ts ?? 0, end: event.ts ?? 0, ms: 0 }) + } const existing = nodeId ? nodeSteps.get(nodeId) : undefined if (existing) { existing.status = 'running' @@ -468,6 +551,11 @@ export function decodeRun(events: RunEvent[]): Step[] { typeof p === 'object' ? p.approved : undefined, typeof p === 'object' ? String(p.note ?? '') : '') } + const span = nodeId ? spans.get(nodeId) : undefined + if (span) { + span.end = event.ts ?? span.end + span.ms = num(d.duration_ms) ?? Math.round((span.end - span.start) * 1000) + } const step = nodeId ? nodeSteps.get(nodeId) : undefined if (step) { step.status = 'done' @@ -883,6 +971,8 @@ export function decodeRun(events: RunEvent[]): Step[] { } // 还没收到 end 的工具/模型调用保持 running——它们正在进行，不是丢了 + groupConcurrent(out, spans) + // 泳道挂到 supervisor 节点的顶层 Step 上。节点内部那些"第N轮交给谁""X：指令" // 仍然留着——泳道给的是一眼看清的形状，那些步骤给的是内容，两者不重复 for (const [nodeId, team] of teams) { diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index 0a120934..b2b25ff9 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -249,6 +249,24 @@ console.log('\n=== 失败运行 ===') !steps.some((s) => s.kind === 'lifecycle' && s.status === 'running')) } +console.log('\n=== 并行分支 ===') +{ + // 图上一个节点连出多条边就是 fan-out，LangGraph 在同一个 superstep 里并发 + // 执行——这是真并发。但在时间线上它们原来只是穿插出现的几行， + // "这三路是同时跑的、因此省下了 3.6 秒"一个字都没说。 + // 这份样本是真跑出来的：三路分别 sleep 1.8 / 2.6 / 1.1 秒。 + const steps = mod.decodeRun(fixtures.fanout) + const group = steps.find((s) => /路并行/.test(s.title)) + check('认出了并行的那一批', !!group, steps.map((s) => s.title).join(' | ')) + check('三路都在组里', group?.children?.length === 3, `${group?.children?.length} 路`) + check('省下多少说出来了', /合计 6\.\ds，实际 3\.\ds/.test(group?.meta ?? ''), group?.meta) + // 顺序执行的图不能被误判成并行——误报比漏报更糟，它会让人以为省了时间 + for (const name of ['db', 'human', 'loop_approve']) { + const seq = mod.decodeRun(fixtures[name]) + check(`顺序执行的 ${name} 没有被误判`, !seq.some((s) => /路并行/.test(s.title))) + } +} + console.log('\n=== 通用规则 ===') { const all = Object.values(fixtures).flatMap((evts) => flatten(mod.decodeRun(evts))) diff --git a/scripts/check-stream.mjs b/scripts/check-stream.mjs index 9e63ec7d..2b852a8f 100644 --- a/scripts/check-stream.mjs +++ b/scripts/check-stream.mjs @@ -191,6 +191,30 @@ console.log('\n=== 复核说明 ===') await page.close() } +console.log('\n=== 并行分支 ===') +{ + // fan-out 一直是真并发（实测 3 个 sleep 2 秒的节点墙钟 2.6 秒），但时间线上 + // 原来只是穿插出现的几行，省下的时间一个字都没说 + for (const [w, dense, label] of [[1000, '0', '宽栏'], [380, '1', '窄栏']]) { + const page = await browser.newPage({ viewport: { width: w, height: 700 } }) + const errors = [] + page.on('pageerror', (e) => errors.push(e.message)) + await page.goto(`${WEB}/preview.html?fanout=1&dense=${dense}`, { waitUntil: 'networkidle' }) + await page.waitForTimeout(400) + const body = await page.locator('body').innerText() + check(`${label}：没有运行时报错`, errors.length === 0, errors.join(' | ')) + check(`${label}：说破了这是三路并行`, body.includes('3 路并行')) + check(`${label}：省下多少写出来了`, /合计 6\.\ds，实际 3\.\ds/.test(body), + body.slice(body.indexOf('3 路并行'), body.indexOf('3 路并行') + 40).replace(/\n/g, ' ')) + check(`${label}：三路都还看得见`, + ['查销售库', '查库存库', '查客户库'].every((n) => body.includes(n))) + const overflow = await page.evaluate(() => + document.documentElement.scrollWidth - document.documentElement.clientWidth) + check(`${label}：不横向溢出`, overflow <= 0, `${overflow}px`) + await page.close() + } +} + console.log('\n=== 协作团队的泳道 ===') { // 这个节点以前在界面上是一条扁平的步骤序列：看得出"谁回了什么"，