diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 6a0b14c2..c965bda8 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import json import time from typing import Any @@ -64,17 +65,39 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: f"- {n}：{agent_map[n].get('description') or agent_map[n].get('system', '')[:100]}" for n in names ) + # 一轮可以派给多个专家。协议用 assignments 数组而不是单个 next： + # 后者把"一次只能派一个"编进了协议本身，模型再想并行也表达不出来。 + # + # 但并行只在任务**互不依赖**时才成立——并发的几个专家看到的是同一份进展 + # 快照，谁也看不见谁。派错了不会报错，只会让两个人基于同样的旧信息 + # 重复劳动，而这件事在结果里很难看出来。所以判断责任明确压在调度者身上， + # 提示词里把反例写出来。 route_schema = { "title": "route", # langchain 靠 title 识别 JSON Schema dict "type": "object", "properties": { - "next": {"type": "string", "enum": names + ["FINISH"]}, - "instruction": {"type": "string"}, + "assignments": { + "type": "array", + "description": "这一轮要派出去的任务。互不依赖时可以给多个，它们会同时执行", + "items": { + "type": "object", + "properties": { + "agent": {"type": "string", "enum": names}, + "instruction": {"type": "string"}, + }, + "required": ["agent", "instruction"], + }, + }, + "done": {"type": "boolean", "description": "目标已达成，结束协作"}, "reason": {"type": "string"}, }, - "required": ["next"], + "required": ["assignments"], } + #: 一轮最多同时派几个。再多的话调度者多半只是在把任务拆碎， + #: 而每多一路就多一份"基于陈旧进展"的风险 + max_parallel = max(1, min(int(ctx.cfg("max_parallel", 3) or 3), len(names))) + transcript: list[dict[str, Any]] = [] usage_total = {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "calls": 0} @@ -94,17 +117,27 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: f"你是团队调度者。目标：\n{goal}\n\n" f"可用成员：\n{roster}\n\n" f"当前进展：\n{progress or '(还没有任何进展)'}\n\n" - "决定下一步交给谁处理，并给出一句明确的指令。" - "如果目标已经达成，next 返回 FINISH。" + "决定这一轮把什么任务交给谁，每个任务给一句明确的指令。\n\n" + f"**互不依赖的任务可以放在同一轮**（最多 {max_parallel} 个），它们会同时执行，" + "总耗时按最慢的那个算。判断依据只有一条：**后一个任务需不需要看到前一个的结果**。\n" + " 可以同时派：「查 A 产品的资料」和「查 B 产品的资料」——两边互不相干\n" + " 不能同时派：「查资料」和「根据资料写报告」——后者要等前者\n" + " 不能同时派：「写初稿」和「校对初稿」——同上\n\n" + "同时派出去的成员看到的是**同一份进展快照**，谁也看不见谁这一轮干了什么。" + "拿不准是否独立时就分两轮派，代价只是慢一点；派错了则是两个人基于同样的" + "旧信息重复劳动，而这在结果里很难看出来。\n\n" + "目标已经达成时，done 填 true、assignments 给空数组。" ) try: value = await supervisor_model.with_structured_output(route_schema).ainvoke(prompt) if not isinstance(value, dict): value = json.loads(message_text(value)) except Exception: # noqa: BLE001 + # 不支持结构化输出的模型退回文本：从回复里认出一个成员名，按单人派 raw = message_text(await supervisor_model.ainvoke(prompt)) - picked = next((n for n in names if n in raw), "FINISH") - value = {"next": picked, "instruction": raw[:500], "reason": ""} + picked = next((n for n in names if n in raw), None) + value = ({"assignments": [{"agent": picked, "instruction": raw[:500]}]} + if picked else {"assignments": [], "done": True, "reason": raw[:200]}) return value @task @@ -179,42 +212,99 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: progress_lines: list[str] = [] final_text = "" + rounds_meta: list[dict[str, Any]] = [] for round_no in range(max_rounds): progress = "\n\n".join(progress_lines) decision = await route(round_no, progress) _accumulate(AIMessage(content=""), sup_model_id) # route 用了一次调用 - nxt = str(decision.get("next") or "FINISH") reason = str(decision.get("reason", "")) - ctx.emit( - EventType.LOG, level="info", round=round_no, - message=f"调度 → {nxt}" + (f"（{reason}）" if reason else ""), - ) - if nxt == "FINISH" or nxt not in agent_map: - final_text = progress_lines[-1].split("：", 1)[-1] if progress_lines else "" + + # 只留认得出的成员，并按上限截断。模型偶尔会重复派同一个人或者编一个 + # 不存在的名字——这两种在并发里都是纯浪费 + batch: list[tuple[str, str]] = [] + seen: set[str] = set() + for item in decision.get("assignments") or []: + name = str((item or {}).get("agent") or "") + if name in agent_map and name not in seen: + seen.add(name) + batch.append((name, str(item.get("instruction") or goal))) + if len(batch) >= max_parallel: + break + + if decision.get("done") or not batch: + # agents / done 是结构化字段，界面照着它渲染。以前只有一句中文 + # 消息串，解码层得拿正则去拆「调度 → X（理由）」——文案一改就散架 + ctx.emit(EventType.LOG, level="info", round=round_no, + agents=[], done=True, + message="调度 → 结束协作" + (f"（{reason}）" if reason else "")) + final_text = progress_lines[-1].split("】", 1)[-1] if progress_lines else "" break - instruction = str(decision.get("instruction") or goal) - ctx.emit(EventType.AGENT_STEP_START, agent=nxt, instruction=instruction[:300], - round=round_no) - started = time.perf_counter() - outcome = await work(round_no, nxt, instruction, progress) - elapsed = int((time.perf_counter() - started) * 1000) - ctx.emit(EventType.AGENT_STEP_END, agent=nxt, duration_ms=elapsed, - round=round_no, preview=outcome["text"][:500]) - - progress_lines.append(f"【{nxt}】{outcome['text']}") - transcript.append( - {"round": round_no, "agent": nxt, "instruction": instruction, **outcome, - "duration_ms": elapsed} + who = "、".join(n for n, _ in batch) + ctx.emit( + EventType.LOG, level="info", round=round_no, parallel=len(batch), + agents=[n for n, _ in batch], done=False, reason=reason, + message=(f"调度 → {who}" + (f"（{reason}）" if reason else "") + + (f" · {len(batch)} 人同时进行" if len(batch) > 1 else "")), ) - final_text = outcome["text"] + + for name, instruction in batch: + ctx.emit(EventType.AGENT_STEP_START, agent=name, instruction=instruction[:300], + round=round_no, parallel=len(batch)) + + # 并发执行。它们拿到的是**同一份** progress 快照——这正是"任务必须 + # 互不依赖"的技术含义，也是上面提示词里那几条反例的由来。 + # + # 每个人单独计时，而不是事后拿整轮墙钟去摊：界面要显示"并行省了多少 + # 时间"，那个数必须是量出来的。按 N×墙钟 推算等于假设每个人都跑满了 + # 最慢那条，而实际上快的那个可能只花了三分之一 + async def _timed(name: str, instruction: str) -> tuple[str, Any, int]: + t0 = time.perf_counter() + try: + out: Any = await work(round_no, name, instruction, progress) + except BaseException as exc: # noqa: BLE001 - 交给下面统一处置 + out = exc + return name, out, int((time.perf_counter() - t0) * 1000) + + started = time.perf_counter() + results = await asyncio.gather(*(_timed(n, i) for n, i in batch)) + wall_ms = int((time.perf_counter() - started) * 1000) + + sum_ms = 0 + for (name, instruction), (_, outcome, each_ms) in zip(batch, results): + if isinstance(outcome, BaseException): + # 一个专家挂了不该拖垮整轮：把失败当成它的产出交回调度者， + # 让它决定改派还是收工 + text = f"（{name} 执行失败：{type(outcome).__name__}: {outcome}）" + outcome = {"text": text, "tool_calls": []} + ctx.emit(EventType.LOG, level="warn", round=round_no, + message=f"{name} 这一轮失败了：{text}") + sum_ms += each_ms + ctx.emit(EventType.AGENT_STEP_END, agent=name, duration_ms=each_ms, + round=round_no, parallel=len(batch), preview=outcome["text"][:500]) + progress_lines.append(f"【{name}】{outcome['text']}") + transcript.append( + {"round": round_no, "agent": name, "instruction": instruction, **outcome, + "duration_ms": each_ms} + ) + final_text = outcome["text"] + + rounds_meta.append({ + "round": round_no, "agents": [n for n, _ in batch], + "parallel": len(batch), "wall_ms": wall_ms, "sum_ms": sum_ms, + }) usage_total["total_tokens"] = usage_total["input_tokens"] + usage_total["output_tokens"] + # 并行省下的时间：各轮"串行本该花的"减去"实际花的"。只有并发过才有差值 + saved_ms = sum(r["sum_ms"] - r["wall_ms"] for r in rounds_meta) result = { "text": final_text, - "rounds": len(transcript), + "rounds": len(rounds_meta), + "steps": len(transcript), "transcript": transcript, "agents": names, + "schedule": rounds_meta, + "parallel_saved_ms": saved_ms, } updates: dict[str, Any] = { "nodes": {ctx.node.id: result}, diff --git a/backend/tests/test_supervisor.py b/backend/tests/test_supervisor.py new file mode 100644 index 00000000..ae5b452f --- /dev/null +++ b/backend/tests/test_supervisor.py @@ -0,0 +1,310 @@ +"""协作团队：一轮可以同时派多个互不依赖的专家。 + +在此之前调度协议是 `next: string`——"一次只能派一个"被编进了协议本身，模型 +再想并行也表达不出来。于是这个节点叫「多 Agent 协作」，实际是一条严格的串行 +流水线：三个专家各花 10 秒，总是 30 秒。 + +改成 assignments 数组之后，互不依赖的任务同一轮发出去，总耗时按最慢的那个算。 + +**但并发只在任务真的互不依赖时才成立。** 同一轮发出去的专家拿到的是同一份 +进展快照，谁也看不见谁这一轮干了什么。派错了不会报错，只会让两个人基于一样的 +旧信息重复劳动——所以判断责任压在调度者身上，提示词里写了反例，而这里守的是 +机制本身：说好并发就真的并发、串行的路径一步没坏、单个专家挂掉不拖垮整轮。 +""" + +from __future__ import annotations + +import asyncio +import json + +import pytest +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine.runner import run_manager + +#: 每个专家假装干这么久。并发时整轮 ≈ 这个数，串行时 ≈ 这个数 × 人数 +WORK_SECONDS = 0.4 + + +@pytest.fixture +async def runner(): + await run_manager.setup() + yield run_manager + await run_manager.shutdown() + + +def _graph(*, max_parallel: int = 3, members=("researcher", "analyst", "writer")) -> dict: + agents = [ + {"name": n, "description": f"{n} 的职责", "system": f"你是 {n}", "max_steps": 1} + for n in members + ] + return { + "nodes": [ + {"id": "in", "type": "input", "data": {"config": {}}}, + {"id": "team", "type": "supervisor", "data": {"config": { + "goal": "比较三种向量库", + "agents": agents, + "max_rounds": 4, + "max_parallel": max_parallel, + }}}, + {"id": "out", "type": "output", "data": {"config": {}}}, + ], + "edges": [ + {"source": "in", "target": "team"}, + {"source": "team", "target": "out"}, + ], + } + + +def script(monkeypatch, decisions: list[dict]) -> None: + """让调度者按剧本派活，让专家干活花真实时间。 + + 专家那一边必须是**异步**等待：同步 sleep 会把事件循环堵死，于是并发测试 + 永远"通过"（因为根本没并发过，墙钟也就成了串行时间）。 + """ + from app.providers import mock_model + + turns = {"n": 0} + + def _decide(self, messages): + # response_format 有值 = 调度者在做结构化路由；没有 = 专家在干活 + if self.response_format: + i = turns["n"] + turns["n"] += 1 + payload = decisions[i] if i < len(decisions) else {"assignments": [], "done": True} + return AIMessage(content=json.dumps(payload, ensure_ascii=False)) + return AIMessage(content="（干完了）") + + async def _agenerate(self, messages, stop=None, run_manager=None, **kw): + if not self.response_format: + await asyncio.sleep(WORK_SECONDS) + return self._generate(messages, stop, **kw) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + monkeypatch.setattr(mock_model.MockChatModel, "_agenerate", _agenerate) + + +async def _run(graph: dict): + run = await run_manager.start(graph=graph, input_payload={"question": "比较三种向量库"}) + for _ in range(300): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = (await session.execute(select(Run).where(Run.id == run.id))).scalar_one() + if row.status in ("succeeded", "failed", "cancelled"): + break + else: + pytest.fail("运行没有在 15 秒内结束") + async with SessionLocal() as session: + events = list((await session.execute( + select(RunEvent).where(RunEvent.run_id == run.id).order_by(RunEvent.seq) + )).scalars()) + return events, row + + +def steps(events, kind): + return [e.data for e in events if e.type == kind] + + +# -------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_independent_tasks_really_run_at_the_same_time(runner, monkeypatch): + """这块改动的全部意义：说好并发就真的并发。 + + 用墙钟判定而不是看事件顺序——顺序只能说明"发出去了"，说明不了"同时跑"。 + """ + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "查 A 的资料"}, + {"agent": "analyst", "instruction": "查 B 的资料"}, + ], + "reason": "两边互不相干", + }, {"assignments": [], "done": True}]) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + + starts = steps(events, "agent.step.start") + assert {d["agent"] for d in starts} == {"researcher", "analyst"} + assert all(d["parallel"] == 2 for d in starts), starts + + # 两个人都开跑了，第一个才结束——这是"同时"的事件层证据 + first_end = next(e.ts for e in events if e.type == "agent.step.end") + assert all(e.ts <= first_end for e in events if e.type == "agent.step.start") + + # 而这是墙钟证据：串行要 2×WORK_SECONDS，并发约 1× + sched = (run.output or {}).get("_schedule") or _schedule_of(events) + assert sched, "调度信息要能拿得到" + r0 = sched[0] + assert r0["parallel"] == 2 + assert r0["wall_ms"] < WORK_SECONDS * 2000 * 0.8, f"看着像串行：{r0}" + assert r0["sum_ms"] > r0["wall_ms"], "并发就该省出时间" + + +def _schedule_of(events): + """schedule 在节点输出里，测试从事件流兜一份等价的出来。""" + from collections import defaultdict + rounds = defaultdict(lambda: {"agents": [], "parallel": 0, "wall_ms": 0, "sum_ms": 0}) + for d in (e.data for e in events if e.type == "agent.step.end"): + r = rounds[d["round"]] + r["round"] = d["round"] + r["agents"].append(d["agent"]) + r["parallel"] = d.get("parallel", 1) + r["wall_ms"] = max(r["wall_ms"], d["duration_ms"]) + r["sum_ms"] += d["duration_ms"] + return [rounds[k] for k in sorted(rounds)] + + +@pytest.mark.asyncio +async def test_saved_time_is_measured_not_assumed(runner, monkeypatch): + """界面要显示"并行省了 X 秒"，那个数必须是量出来的。 + + 两个人快慢不同时才区分得出来：按"整轮墙钟 × 人数"推算的话，快的那个会被 + 记成和慢的一样长，省下的时间就被夸大了。所以让一个人明显更快， + 再看各自的耗时是不是真的不一样。 + """ + from app.providers import mock_model + + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "慢活"}, + {"agent": "analyst", "instruction": "快活"}, + ], + }, {"assignments": [], "done": True}]) + + async def _uneven(self, messages, stop=None, run_manager=None, **kw): + if not self.response_format: + text = " ".join(str(getattr(m, "content", "")) for m in messages) + await asyncio.sleep(0.05 if "快活" in text else WORK_SECONDS) + return self._generate(messages, stop, **kw) + + monkeypatch.setattr(mock_model.MockChatModel, "_agenerate", _uneven) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + + took = {d["agent"]: d["duration_ms"] for d in steps(events, "agent.step.end")} + assert took["analyst"] < took["researcher"] / 2, f"快慢该有差别：{took}" + # 整轮墙钟按最慢的那个算，而不是两个人加起来 + assert took["researcher"] < WORK_SECONDS * 2000 * 0.8, took + + +@pytest.mark.asyncio +async def test_dependent_tasks_stay_sequential(runner, monkeypatch): + """分两轮派就该老老实实跑两轮——并行是可选项，不是强加的。""" + script(monkeypatch, [ + {"assignments": [{"agent": "researcher", "instruction": "查资料"}]}, + {"assignments": [{"agent": "writer", "instruction": "根据资料写报告"}]}, + {"assignments": [], "done": True}, + ]) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + + starts = steps(events, "agent.step.start") + assert [d["agent"] for d in starts] == ["researcher", "writer"] + assert all(d["parallel"] == 1 for d in starts) + + # 第二个人开跑必须在第一个人结束之后 + first_end = next(e.ts for e in events if e.type == "agent.step.end") + second_start = [e for e in events if e.type == "agent.step.start"][1].ts + assert second_start >= first_end + + +@pytest.mark.asyncio +async def test_max_parallel_caps_the_batch(runner, monkeypatch): + """上限设成 1 就退回严格串行——给不放心的人一个明确的开关。""" + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "a"}, + {"agent": "analyst", "instruction": "b"}, + {"agent": "writer", "instruction": "c"}, + ], + }, {"assignments": [], "done": True}]) + + events, run = await _run(_graph(max_parallel=1)) + assert run.status == "succeeded", run.error + starts = steps(events, "agent.step.start") + assert len(starts) == 1, "上限 1 就只该派一个" + assert starts[0]["parallel"] == 1 + + +@pytest.mark.asyncio +async def test_unknown_and_duplicate_agents_are_dropped(runner, monkeypatch): + """模型偶尔会重复派同一个人或者编一个不存在的名字，两种在并发里都是纯浪费。""" + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "a"}, + {"agent": "researcher", "instruction": "又派一次"}, + {"agent": "根本没这个人", "instruction": "c"}, + ], + }, {"assignments": [], "done": True}]) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + starts = steps(events, "agent.step.start") + assert [d["agent"] for d in starts] == ["researcher"] + + +@pytest.mark.asyncio +async def test_one_failing_member_does_not_sink_the_round(runner, monkeypatch): + """并发里一个人挂了，另一个的成果不该跟着丢。""" + from app.providers import mock_model + + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "会挂的那个"}, + {"agent": "analyst", "instruction": "正常的那个"}, + ], + }, {"assignments": [], "done": True}]) + + original = mock_model.MockChatModel._agenerate + + async def _boom(self, messages, stop=None, run_manager=None, **kw): + text = " ".join(str(getattr(m, "content", "")) for m in messages) + if "会挂的那个" in text: + raise RuntimeError("这个专家炸了") + return await original(self, messages, stop, run_manager, **kw) + + monkeypatch.setattr(mock_model.MockChatModel, "_agenerate", _boom) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + + # 失败要说出来，而不是安静地少一个人的产出 + warns = [d["message"] for d in steps(events, "log") if d.get("level") == "warn"] + assert any("researcher" in w for w in warns), warns + # 另一个人的结束事件照常有 + ends = {d["agent"] for d in steps(events, "agent.step.end")} + assert "analyst" in ends + + +@pytest.mark.asyncio +async def test_schedule_is_reported_for_the_ui(runner, monkeypatch): + """界面要画泳道，就得拿得到每轮谁干了活、并发几路、省了多少时间。""" + script(monkeypatch, [{ + "assignments": [ + {"agent": "researcher", "instruction": "a"}, + {"agent": "analyst", "instruction": "b"}, + ], + }, { + "assignments": [{"agent": "writer", "instruction": "汇总"}], + }, {"assignments": [], "done": True}]) + + events, run = await _run(_graph()) + assert run.status == "succeeded", run.error + + rounds = sorted({d["round"] for d in steps(events, "agent.step.end")}) + assert rounds == [0, 1] + by_round = {r: [d["agent"] for d in steps(events, "agent.step.end") if d["round"] == r] + for r in rounds} + assert set(by_round[0]) == {"researcher", "analyst"} + assert by_round[1] == ["writer"] + + # 调度决策本身也要在事件流里，否则界面上这个节点就是个跑了 N 秒的黑盒 + dispatches = [d["message"] for d in steps(events, "log") if d.get("round") is not None] + assert any("同时进行" in m for m in dispatches), dispatches diff --git a/frontend/src/canvas/nodeDefs.ts b/frontend/src/canvas/nodeDefs.ts index c5a793bf..70510e81 100644 --- a/frontend/src/canvas/nodeDefs.ts +++ b/frontend/src/canvas/nodeDefs.ts @@ -161,10 +161,16 @@ export const NODE_DEFS: Record = { { key: 'goal', label: '团队目标', type: 'prompt' }, { key: 'agents', label: '团队成员', type: 'agents' }, { key: 'max_rounds', label: '最大轮次', type: 'number', min: 1, max: 25 }, + { + key: 'max_parallel', label: '每轮最多同时派几人', type: 'number', min: 1, max: 6, + help: '互不依赖的任务调度者可以放在同一轮同时执行，总耗时按最慢的那个算。' + + '设为 1 即退回严格串行——并发的成员看到的是同一份进展快照，' + + '任务其实有依赖却被同时派出去，两个人会基于一样的旧信息重复劳动', + }, ...MODEL_FIELDS, ...COMMON_TAIL, ], - defaults: { max_rounds: 6, agents: [] }, + defaults: { max_rounds: 6, max_parallel: 3, agents: [] }, }, tool: { type: 'tool', label: '调用工具', category: '执行', icon: Wrench, diff --git a/frontend/src/dev/preview.tsx b/frontend/src/dev/preview.tsx index 8d7ad23e..b21cff93 100644 --- a/frontend/src/dev/preview.tsx +++ b/frontend/src/dev/preview.tsx @@ -136,6 +136,50 @@ const REVIEW_DEGRADED_TURN: StreamTurn = { rawOutput: { answer: "平台管理员可以修改他人权限。" }, } +/** 协作团队的泳道：前两轮并行、第三轮汇总 */ +const TEAM_TURN: StreamTurn = { + id: "team", question: "比较三种向量数据库在中文检索上的取舍", phase: "done", status: "完成", + steps: [{ + id: "n-team", seq: 1, kind: "node", status: "done", title: "调研团队", meta: "24.1s", + team: { + members: ["researcher", "analyst", "writer"], + savedMs: 11800, + finished: true, + rounds: [ + { + round: 0, parallel: 2, wallMs: 9600, sumMs: 16400, + reason: "两边资料互不相干，可以同时查", + members: [ + { agent: "researcher", instruction: "查 Milvus 与 Qdrant 的中文分词支持", ms: 9600, + status: "done", result: "Milvus 2.4 起内置 jieba…" }, + { agent: "analyst", instruction: "查 pgvector 的中文检索方案", ms: 6800, + status: "done", result: "pgvector 本身不分词，需配 zhparser…" }, + ], + }, + { + round: 1, parallel: 2, wallMs: 7400, sumMs: 12400, + reason: "两组基准测试互不依赖", + members: [ + { agent: "researcher", instruction: "跑召回率基准", ms: 7400, status: "done", + result: "hit@1：Milvus 0.91 / Qdrant 0.89 / pgvector 0.84" }, + { agent: "analyst", instruction: "跑写入吞吐基准", ms: 5000, status: "done", + result: "10 万条：Milvus 42s / Qdrant 51s / pgvector 88s" }, + ], + }, + { + round: 2, parallel: 1, wallMs: 7100, sumMs: 7100, + reason: "两份基准都出来了，可以汇总", + members: [ + { agent: "writer", instruction: "根据前面的结论写选型建议", ms: 7100, + status: "done", result: "## 结论\n中文为主且要开箱即用 → Milvus…" }, + ], + }, + ], + }, + }], + output: { 建议: "中文为主且要开箱即用选 Milvus；已有 Postgres 且数据量在百万以内，pgvector + zhparser 的运维成本最低。" }, +} + const LONG_TURN: StreamTurn = { id: "long", question: "需要能够显示总结内容", phase: "done", status: "完成", steps: [], output: { result: longReport() }, @@ -185,6 +229,9 @@ function Preview() { if (params.get('review') === '1') { return } + if (params.get('team') === '1') { + return + } return (
diff --git a/frontend/src/run/AssistantStream.tsx b/frontend/src/run/AssistantStream.tsx index 402cc8b1..e7ee4531 100644 --- a/frontend/src/run/AssistantStream.tsx +++ b/frontend/src/run/AssistantStream.tsx @@ -1,12 +1,13 @@ import { useEffect, useRef, useState, type ReactNode } from 'react' import { AlertTriangle, Brain, ChevronRight, CircleCheck, CircleDot, Database, - FileCode, GitBranch, Hand, Info, Sparkles, Table2, Terminal, Wrench, XCircle, + FileCode, GitBranch, Hand, Info, Sparkles, Table2, Terminal, Users, Wrench, XCircle, } from 'lucide-react' import clsx from 'clsx' import { Spinner } from '../components/ui' import { parseQueryResult, type ResultTable as Table, type Step, type StepKind, + type TeamRun, } from './decode' import { Markdown } from './Markdown' import type { ReviewResult } from '../types' @@ -317,6 +318,11 @@ function StepRow({ step, dense, depth }: { step: Step; dense: boolean; depth: nu
)} + {/* 泳道排在子步骤**前面**：先给一眼看清的形状（谁做了哪几轮、哪些是 + 同时进行的、各花多久），再往下是每一步的内容。反过来的话，得把 + 十几行步骤读完才知道这个团队是怎么分工的 */} + {step.team && } + {!!step.children?.length && ( )} @@ -324,6 +330,129 @@ function StepRow({ step, dense, depth }: { step: Step; dense: boolean; depth: nu ) } + +/** + * 协作团队的泳道：每个成员一行，每一轮一格。 + * + * 这个节点以前在界面上是一条扁平的步骤序列——能看出"谁回了什么"，看不出 + * "谁和谁是同时干的"。而一轮同时派几个人正是它相对单 agent 的全部优势， + * 不画出来等于没有。 + * + * 条形的宽度按耗时归一化，所以一眼能看出谁是这一轮的瓶颈。 + */ +function TeamLanes({ team, dense }: { team: TeamRun; dense: boolean }) { + const [openKey, setOpenKey] = useState(null) + if (!team.rounds.length) return null + + // 所有格子共用一个时间标尺，否则"快的那格"和"慢的那格"画得一样长， + // 瓶颈就看不出来了 + const slowest = Math.max(1, ...team.rounds.flatMap((r) => r.members.map((m) => m.ms))) + const parallelRounds = team.rounds.filter((r) => r.parallel > 1).length + + return ( +
+
+ + {team.members.length} 名成员 · {team.rounds.length} 轮 + + {parallelRounds > 0 && ( + + {parallelRounds} 轮并行，省下 {fmtMs(team.savedMs)} + + )} + {parallelRounds === 0 && team.rounds.length > 1 && ( + // 全程串行不是故障，但值得说一句——任务本来就互相依赖时它就该是串行的 + 本次任务互相依赖，逐轮进行 + )} +
+ +
+ {team.members.map((name) => ( +
+
+ {name} +
+
+ {team.rounds.map((r) => { + const m = r.members.find((x) => x.agent === name) + const key = `${r.round}:${name}` + return ( +
+ {m ? ( +
+
+ ))} + +
+
+
+ {team.rounds.map((r) => ( +
+ {/* 窄栏放不下完整措辞，但"并2"这种缩写没人看得懂， + 宁可只留轮次和耗时，并行与否看上面那行条形本来就一目了然 */} + {`第${r.round + 1}轮`} + {r.parallel > 1 && !dense ? ` · ${r.parallel} 人同时` : ''} + {' · '}{fmtMs(r.wallMs)} +
+ ))} +
+
+
+ + {openKey && (() => { + const [rd, name] = openKey.split(':') + const m = team.rounds.find((r) => String(r.round) === rd) + ?.members.find((x) => x.agent === name) + if (!m) return null + return ( +
+
+ {name} · 第 {Number(rd) + 1} 轮 · {fmtMs(m.ms)} +
+ {m.instruction && ( +
任务：{m.instruction}
+ )} + {m.result && ( + 
+                {m.result.slice(0, 1200)}
+
+ )} +
+ ) + })()} +
+ ) +} + +/** 毫秒 → 人读的单位。和 decode 里的 dur 同一套规则 */ +function fmtMs(ms: number): string { + if (ms < 1000) return `${ms}ms` + return ms < 60_000 ? `${(ms / 1000).toFixed(1)}s` : `${Math.round(ms / 60_000)}min` +} + /** 查询结果画成表格。一屏 JSON 谁也看不出名堂，表格能一眼看到形状。 */ function ResultTable({ table, dense }: { table: Table; dense: boolean }) { const maxCols = dense ? 4 : 8 diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index a1b9878a..ec4454e0 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -53,6 +53,44 @@ export interface Step { * 两者要同时展示（问什么 + 查到什么），合成一个字段就只能二选一 */ result?: string children?: Step[] + /** 协作团队的泳道数据。只有 supervisor 节点的顶层 Step 会有 */ + team?: TeamRun +} + +/** + * 协作团队一轮里的一个人。 + * + * ms 是**各自量出来**的耗时，不是拿整轮墙钟摊的——界面要显示"并行省了多少"， + * 那个数按 N×墙钟 推算会把快的那个也记成最慢那条，省下的时间就被夸大了。 + */ +export interface TeamMember { + agent: string + instruction: string + ms: number + status: StepStatus + result?: string +} + +export interface TeamRound { + round: number + /** 这一轮同时派了几个人。1 就是串行的一步 */ + parallel: number + /** 这一轮实际花的时间：并发时是最慢那个 */ + wallMs: number + /** 各人耗时之和：并发时它大于 wallMs，差值就是省下的 */ + sumMs: number + reason?: string + members: TeamMember[] +} + +export interface TeamRun { + /** 花名册，按第一次出现的顺序——泳道的行顺序 */ + members: string[] + rounds: TeamRound[] + /** 并行一共省下多少毫秒。全程串行则为 0 */ + savedMs: number + /** 协作是不是已经收尾了 */ + finished: boolean } // 这两个是流式增量，后端根本不落库（_EPHEMERAL）。单次运行的 delta 量级会 @@ -297,6 +335,28 @@ export function decodeRun(events: RunEvent[]): Step[] { const pendingLlm = new Map() /** supervisor 里待配对的单个 agent 步骤 */ const pendingAgents = new Map() + /** node_id → 协作团队的泳道数据。边解码边攒，最后挂到该节点的顶层 Step 上 */ + const teams = new Map() + + const teamOf = (nodeId: string | undefined): TeamRun | null => { + if (!nodeId) return null + let t = teams.get(nodeId) + if (!t) { + t = { members: [], rounds: [], savedMs: 0, finished: false } + teams.set(nodeId, t) + } + return t + } + + const roundOf = (team: TeamRun, round: number): TeamRound => { + let r = team.rounds.find((x) => x.round === round) + if (!r) { + r = { round, parallel: 1, wallMs: 0, sumMs: 0, members: [] } + team.rounds.push(r) + team.rounds.sort((a, b) => a.round - b.round) + } + return r + } /** * node_id → 这个节点当前那条尚未闭合的审批步骤。 * @@ -626,6 +686,18 @@ export function decodeRun(events: RunEvent[]): Step[] { } pendingAgents.set(agentKey(nodeId, d), step) push(step, nodeId) + + const team = teamOf(nodeId) + if (team) { + const name = String(d.agent ?? '') + if (name && !team.members.includes(name)) team.members.push(name) + const r = roundOf(team, num(d.round) ?? 0) + r.parallel = Math.max(r.parallel, num(d.parallel) ?? 1) + r.members.push({ + agent: name, instruction: String(d.instruction ?? ''), + ms: 0, status: 'running', + }) + } break } @@ -645,6 +717,23 @@ export function decodeRun(events: RunEvent[]): Step[] { detail: preview, }, nodeId) } + + const team = teamOf(nodeId) + if (team) { + const r = roundOf(team, num(d.round) ?? 0) + const name = String(d.agent ?? '') + const ms = num(d.duration_ms) ?? 0 + const m = r.members.find((x) => x.agent === name && x.status === 'running') + ?? (r.members.push({ agent: name, instruction: '', ms: 0, status: 'running' }), + r.members[r.members.length - 1]) + m.ms = ms + m.status = 'done' + m.result = preview || undefined + // 这一轮实际花的是最慢那个，各人之和减去它就是省下的 + r.wallMs = Math.max(...r.members.map((x) => x.ms)) + r.sumMs = r.members.reduce((acc, x) => acc + x.ms, 0) + team.savedMs = team.rounds.reduce((acc, x) => acc + Math.max(0, x.sumMs - x.wallMs), 0) + } break } @@ -656,17 +745,33 @@ export function decodeRun(events: RunEvent[]): Step[] { // 就是它，和普通 info 日志区分得开 if (level === 'info' && d.round != null) { const text = String(d.message ?? '') + const round = (num(d.round) ?? 0) + 1 + // 优先读结构化字段。以前只有一句中文消息串，这里得拿正则去拆 + // 「调度 → X（理由）」——后端文案一改就散架，而文案是会改的。 + // 老运行的事件没有这些字段，正则那条路留着兜底 + const structured = Array.isArray(d.agents) const m = text.match(/^调度\s*→\s*([^（(]+)[（(](.*)[）)]\s*$/) + const legacyTarget = m?.[1].trim() ?? '' + const agents: string[] = structured + ? (d.agents as unknown[]).map(String) + : (legacyTarget && legacyTarget !== 'FINISH' ? [legacyTarget] : []) + const done = structured ? !!d.done : legacyTarget === 'FINISH' + const reason = structured ? String(d.reason ?? '') : (m?.[2].trim() ?? '') + const team = teamOf(nodeId) + if (team) { + if (done) team.finished = true + else if (reason) roundOf(team, num(d.round) ?? 0).reason = reason + } + // FINISH 是协议里的收尾标记，不是某个 agent。"交给 FINISH" // 会让人以为还有个叫 FINISH 的成员 - const target = m?.[1].trim() ?? '' - const round = Number(d.round) + 1 + const title = done ? `第 ${round} 轮：结束协作` + : agents.length > 1 ? `第 ${round} 轮：${agents.join('、')} 同时进行` + : agents.length === 1 ? `第 ${round} 轮：交给 ${agents[0]}` + : text push({ id: `sv-${seq}`, seq, kind: 'branch', nodeId, status: 'done', - title: !m ? text - : target === 'FINISH' ? `第 ${round} 轮：结束协作` - : `第 ${round} 轮：交给 ${target}`, - detail: m ? m[2].trim() : undefined, + title, detail: reason || undefined, }, nodeId) break } @@ -778,6 +883,14 @@ export function decodeRun(events: RunEvent[]): Step[] { } // 还没收到 end 的工具/模型调用保持 running——它们正在进行，不是丢了 + // 泳道挂到 supervisor 节点的顶层 Step 上。节点内部那些"第N轮交给谁""X：指令" + // 仍然留着——泳道给的是一眼看清的形状，那些步骤给的是内容，两者不重复 + for (const [nodeId, team] of teams) { + if (!team.rounds.length) continue + const step = nodeSteps.get(nodeId) + if (step) step.team = team + } + return out } diff --git a/scripts/check-stream.mjs b/scripts/check-stream.mjs index 63709762..9e63ec7d 100644 --- a/scripts/check-stream.mjs +++ b/scripts/check-stream.mjs @@ -191,6 +191,47 @@ console.log('\n=== 复核说明 ===') await page.close() } +console.log('\n=== 协作团队的泳道 ===') +{ + // 这个节点以前在界面上是一条扁平的步骤序列：看得出"谁回了什么"， + // 看不出"谁和谁是同时干的"——而一轮同时派几个人正是它相对单 agent 的 + // 全部优势，不画出来等于没有 + 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?team=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}：三个人各占一行`, + ['researcher', 'analyst', 'writer'].every((n) => body.includes(n))) + check(`${label}：说清楚并行省了多少`, /2 轮并行，省下 11\.8s/.test(body), + body.slice(0, 90).replace(/\n/g, ' ')) + const overflow = await page.evaluate(() => + document.documentElement.scrollWidth - document.documentElement.clientWidth) + check(`${label}：不横向溢出`, overflow <= 0, `${overflow}px`) + + if (dense === '0') { + // 条形宽度按耗时归一化，一眼看得出谁是这一轮的瓶颈。 + // 都画一样长的话，"并行"就只剩一句口号 + const widths = await page.evaluate(() => + [...document.querySelectorAll('button[title*="第 1 轮"]')] + .map((b) => b.getBoundingClientRect().width)) + check('同一轮里慢的那个条更长', widths.length === 2 && widths[0] > widths[1] * 1.2, + widths.map((x) => Math.round(x)).join(' vs ')) + + await page.locator('button[title*="第 1 轮"]').first().click() + await page.waitForTimeout(300) + const opened = await page.locator('body').innerText() + check('点开看得到这个人的任务和回复', + opened.includes('查 Milvus 与 Qdrant') && opened.includes('内置 jieba')) + } + await page.close() + } +} + console.log('\n=== 空态 ===') { const page = await browser.newPage({ viewport: { width: 1280, height: 900 } })