Files
deerflow-code/frontend-web/docs/roundtable-dag-orchestration.md
T
董宏凯andClaude Opus 4.8 2e689e4482 docs(roundtable): 圆桌 DAG 编排 + 并行执行显示分步需求文档
锁定三项设计决策:分层 Stage(非自由连线)、并行跳过 leader 前端直派、主区固定 2 列(窄屏降 1 列)。含数据模型/编辑器交互/并行调度/禁写文件/高性能流式渲染/分步实现计划(Phase 1-8)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-07 21:20:58 +08:00

25 KiB
Raw Blame History

圆桌 DAG 编排 + 并行执行显示 —— 分步实现需求文档

目标读者:实现本功能的前端 / 后端工程师。 本文是可逐步落地的需求 + 设计文档,每个 Phase 都能独立交付、独立验收。


0. 背景与目标

0.1 现状

Step 2 圆桌目前有两种编排模式(orchestrationMode,见 hooks/useStep2Orchestration.ts):

  • free(自由模式):总控(leader)自由决定派活给谁,runOrchestration 循环驱动。
  • chain(线性链条):用户把席位排成一条线性链 A → B → C,runChainOrchestration 按顺序逐个派活,最后总控收口。链路用 DispatchChainModal.tsx 的横向流程图展示(总控 → 席位1 → 席位2 → … → 共识 → 报告)。

链条数据现在只是一个有序数组 selectedAgents(线性),节点状态类型见 lib/job-types.ts(ChainNodeState = pending | active | done)。

0.2 目标(本次新增)

  1. DAG 编排:用户可视化指定智能体之间的直接依赖关系——有的串行(A → B → C),有的并行(A B C 同时跑)。编排结果是一张流程图(DAG)。
  2. 并行执行:DAG 里同一层(互不依赖)的智能体并行跑,墙钟时间从「串行求和」降到「并行取最慢」。
  3. 并行禁止写文件:并行的智能体禁用文件写入 / 沙箱命令,避免它们同时打开沙箱互相打架。产物以消息正文交付;落文件留给串行 / 收口阶段。
  4. 并行高性能显示:并行时收起左侧、顶部头像条切换、主区同时显示 2 个智能体的消息流,其余只接流不渲染,避免页面卡死。

0.3 设计总原则

  • 复用 现有 special run / leader run / rollback 抢占 / 轻量交付状态接口(GET /threads/{id}/last-ai-message),不另起炉灶。
  • 线性链是 DAG 的特例:chain 模式可平滑迁移成"每层 1 个节点"的 DAG,老数据不破坏。
  • 分层(Stage)优先于自由连线:编辑交互首选「分层」模型(每个 stage 放若干并行席位,stage 间串行),实现成本低、覆盖 95% 的"串 / 并"诉求;自由连线画布作为后续可选增强(见 §2.3)。

0.4 已确认的设计决策(✅ 本期按这些实现)

以下三项已与需求方确认锁定,实现时直接按此,不再二选一。

  1. ✅ 编排模型 = 分层 Stage:本期只做分层(每个 stage 放若干并行席位、stage 间串行)。自由连线画布本期不做,仅在数据模型层预留(§1.2)以免将来返工。
  2. ✅ 并行跳过 leader、前端直派:DAG 由用户显式指定,执行时前端直接对 stage 内席位发起 special run,不每个 stage 再跑一轮总控派活;只在最后 finalSynthesis 跑一轮 leader 综合(即 §3.2 的方案 B)。
  3. ✅ 主区固定 2 列,窄屏降 1 列:并行时主区最多并排显示 2 个被选中智能体的消息流;窄屏(断点见 §5.3)降为 1 列 + 头像条横向滚动。不做 3+ 列。

1. 数据模型

1.1 DAG 的「分层」表示(推荐,主模型)

把 DAG 表示成有序的 stage 列表,每个 stage 内的席位并行、stage 之间串行:

// lib/dag-types.ts (新建)

/** 一个并行批次:stage 内所有席位并行执行。 */
export interface OrchestrationStage {
  /** stage 稳定 id(拖拽重排 / diff 用)。 */
  id: string;
  /** 本 stage 并行执行的席位(agent_id 列表)。1 个 = 退化为串行节点。 */
  agentIds: string[];
}

/** 用户编排出的执行计划。stages 顺序即串行顺序。 */
export interface OrchestrationPlan {
  /** 编排模式标识,持久化进 step2。 */
  mode: "dag";
  /** 串行执行的批次;每批内并行。 */
  stages: OrchestrationStage[];
  /** 收口:是否在所有 stage 完成后由总控综合(默认 true)。 */
  finalSynthesis: boolean;
}
  • 串行 A → B → C = stages: [{agentIds:[A]}, {agentIds:[B]}, {agentIds:[C]}]。
  • 并行 A B C = stages: [{agentIds:[A,B,C]}]。
  • 混合 A →(B C)→ D = stages: [{[A]}, {[B,C]}, {[D]}]。

为什么用分层而不是「邻接表 / 边」:执行调度本就是「拓扑分批」,分层就是分批的结果,前端不必再做拓扑排序,也天然避免环;编辑器交互更直观(拖卡片进 stage)。真正的自由连线 DAG 见 §2.3(可选)。

1.2 通用 DAG 表示(§2.3 自由连线时才需要)

若将来做自由连线画布,再引入边表示,并在执行前做一次拓扑排序 → 分层,落回 §1.1 的 stages 结构给调度器用:

export interface DagEdge { from: string; to: string }   // from 完成后才能跑 to
export interface DagGraph { nodes: string[]; edges: DagEdge[] }
// 执行前:toposortToStages(graph) -> OrchestrationStage[]

1.3 持久化

  • 复用 roundtable_drafts.step2 JSON(见后端 app.gateway.routers roundtable-drafts + hooks/useStep2Orchestration.ts 里对 step2 的读写)。
  • 在 step2 增加字段:orchestrationPlan: OrchestrationPlan | null。
  • orchestrationMode 扩展为 "free" | "chain" | "dag";chain 旧数据读取时即时转换为单节点-stage 的 dag(兼容层,不改老 draft)。

1.4 校验(纯函数,可单测)

lib/dag-types.ts 内提供:

validatePlan(plan: OrchestrationPlan, selectedAgentIds: string[]): {
  ok: boolean;
  errors: string[];   // 面向用户的中文提示
}

校验项:

  • 每个 agentId 必须在本次选定席位 selectedAgents 内(防止编排了不存在的席位 → 复用后端 _partition_dispatch 同思路的"必须在 roster 内")。
  • 不允许空 stage;不允许同一席位出现在多个 stage(一个席位只跑一次)。
  • 至少 1 个 stage。
  • (可选)单 stage 并行度上限提示(如 > 6 并行给 warning:远程模型并发 + 沙箱压力)。

2. 编排编辑器(流程图构建)交互

2.1 入口与整体布局

  • 在 Step 2 顶部模式切换里新增 「依赖编排」(dag)模式(与现有 自由 / 链条 并列)。
  • 选择 dag 后弹出 / 进入编排编辑器(复用 Dialog 大弹窗,参考 DispatchChainModal 的尺寸 w-[94vw] max-w-[1240px])。

2.2 推荐交互:分层(Stage)编辑器(Phase 2 实现)

┌ 依赖编排 ───────────────────────────────────────── [校验✓] [保存] ┐
│  可选席位(点/拖加入某个 stage)                                   │
│  [情报收集] [环境评估] [方案设计] [风险审查] [执行规划] [前端实现]  │
│ ─────────────────────────────────────────────────────────────── │
│  Stage 1(并行)              Stage 2(并行)       Stage 3        │
│  ┌──────────┐  ─────▶  ┌──────────┐ ┌────────┐ ─▶ ┌──────────┐    │
│  │ 情报收集 │          │ 方案设计 │ │风险审查│    │ 执行规划 │    │
│  └──────────┘          └──────────┘ └────────┘    └──────────┘    │
│   [+ 拖席位到此]        [+]                       [+]            │
│  [+ 新建 Stage(串在最后)]                                       │
│ ─────────────────────────────────────────────────────────────── │
│  说明:同一 Stage 内的席位会**并行**执行;Stage 之间**串行**       │
│  (后一个 Stage 能看到前面所有 Stage 的交付摘要)。                │
└───────────────────────────────────────────────────────────────────┘

交互要点:

  • 加入 stage:从顶部可选席位拖拽到某个 stage,或点席位 →「加入 Stage N」。
  • 移动 / 重排:席位可在 stage 间拖动;stage 可整体左右拖动重排(改串行顺序)。
  • 删除:席位 / 空 stage 可删。
  • 并行度提示:stage 内 ≥2 席位即标「并行」徽标;过多给 warning。
  • 实时校验:validatePlan 实时跑,错误在顶部红条提示,「保存」在校验通过前禁用。
  • 预览流程图:编辑区下方实时渲染只读流程图缩略(复用 §7 的 DAG 流程图组件)。

拖拽可用现成轻量方案(HTML5 DnD 或 @dnd-kit,看项目是否已有依赖;没有就先用「点选 + 加入/移出按钮」的无拖拽版,Phase 2 先上无拖拽、Phase 2.5 再加拖拽)。

2.3 可选增强:自由连线画布(后续,非本期必须)

  • 节点自由摆放 + 拉线建边 → DagGraph(§1.2)。
  • 实现成本高(画布 / 连线 / 命中检测 / 自动布局),且分层模型已覆盖绝大多数诉求。本期不做,仅在数据模型上预留 §1.2,避免将来返工。

2.4 验收(Phase 2)

  • 能拖 / 点把席位分配到多个 stage,保存进 step2.orchestrationPlan。
  • 刷新 / 重进 draft 能恢复编排。
  • 非法编排(空 stage / 席位重复 / 席位不在 roster)有明确中文提示且无法保存。

3. 执行调度(前端编排循环 + 后端复用)

3.1 新增 runDagOrchestration(useStep2Orchestration.ts)

与现有 runOrchestration / runChainOrchestration 并列,逻辑:

for (stageIndex, stage of plan.stages):
  if isStale() break               // 复用现有 epoch 抢占(人工干预/暂停)
  // 1) 总控按本 stage 的席位并行派活(一轮 leader,dispatched = stage.agentIds)
  //    —— 或跳过 leader,前端直接对 stage 内每个席位发起 special run(见 3.2)
  // 2) 并行跑 stage 内所有席位
  await Promise.all(stage.agentIds.map(seat =>
        streamMultiAgent({ agentType:"special", agentName:seat,
                           newMessage: buildStageTask(seat, plan, stageIndex),
                           parallelNoFile: stage.agentIds.length > 1,  // §4
                           ... })))
  // 3) 等本 stage 全部 ✅(Promise.all 天然 barrier)再进入下一 stage
// 4) finalSynthesis:最后一轮 leader(synthesisMode=true)综合所有 stage 交付

要点:

  • 批内并行:Promise.all(不是现在的 for await 串行)。
  • 批间串行:Promise.all 即 barrier,自动等齐。
  • 依赖可见性:进入 stage N 的席位时,把前面所有 stage 已交付席位的摘要拼进它的 newMessage(复用 §交付状态:网关 _collect_delivery_status / GET last-ai-message?max_chars=N,前端也可在 stage 边界拉一次摘要拼进 task)。这样 (B C) 能看到 A 的产出。
  • 抢占 / 干预 / 暂停:每个 stage 边界检查 isStale()(epoch),与现有自由 / 链条模式一致。
  • rollback:special run 已统一 multitask_strategy=rollback,并行多席位各自不同 thread,互不冲突。

3.2 是否每 stage 都走 leader? — ✅ 已定:前端直派(方案 B)

本期采用方案 B:DAG 由用户显式指定,不需要 leader 再"决定派给谁"。前端直接对 stage 内每个席位发起 special run,省掉每 stage 的 leader LLM 往返(更快、更可控);只在所有 stage 完成后、finalSynthesis 为真时跑一轮 leader(synthesisMode=true)综合所有交付。

(备选方案 A:每 stage 跑一轮 leader 派活——更贴现有架构但每 stage 多一次 leader LLM、更慢。本期不采用,留档备查。)

3.3 后端改动

  • 基本不需要新接口:special run(POST /api/multi-agent/run/stream agent_type=special)、leader run、轻量交付读(GET /threads/{id}/last-ai-message)、轻量追加(POST /threads/{id}/messages/append)都已就绪。
  • 仅需 §4 的「并行禁写文件」run policy 开关。

3.4 验收(Phase 3)

  • (B C) 两席位同时出现在执行(后端日志两条 special run 时间重叠;前端两个流并行增长)。
  • stage 间严格串行(stage 2 不早于 stage 1 全部完成)。
  • stage N 席位的输入里能看到 stage <N 的交付摘要。

4. 并行时禁止写文件(防沙箱打架)

4.1 后端 run policy

app/gateway/roundtable_run_policy.py 现有角色:leader / seat / report。新增一个角色或开关:

  • 方案:给 seat 增加一个「并行变体」seat_parallel,其 excluded_tools 在 seat 基础上再禁 write_file / bash / str_replace / present_files(与 report 角色禁写的思路一致)。
  • _special_run(multi_agent.py)按请求里的一个新 flag(如 no_file: true)选择 seat_parallel policy。

4.2 前端传参

  • streamMultiAgent 请求体新增 parallel_no_file?: boolean(或复用 no_file)。
  • runDagOrchestration 在 stage.agentIds.length > 1(真并行)时置 true;单节点 stage(实际串行)不限制,可正常写文件。

4.3 产物归属

  • 并行席位的产物 = 消息正文(被 _collect_delivery_status / 综合块读取)。
  • 需要落文件 / 画图的(如 Step3 报告、roundtable-report)放在串行的收口阶段单独跑,不在并行批次里。

4.4 验收(Phase 4)

  • 并行 stage 的席位 run 日志显示 Excluded ... ['write_file','bash','str_replace','present_files', ...]。
  • 并行期间不再出现多个席位抢同一沙箱 / 写文件冲突。

5. 并行执行的前端显示

5.1 布局总览(并行进行时)

┌ 多智能体圆桌会商中心 ─────────────────────────────────────────────┐
│ [头像条:进行中的并行智能体]            [模型选择 ▾] [暂停会商]     │
│  (情报)●  (方案)●  (风险)●  (执行)  …                              │
│   ↑选中    ↑选中    loading  hidden                                │
│ ───────────────────────────────────────────────────────────────── │
│  ┌─ 方案设计(选中)──────────┐  ┌─ 风险审查(选中)──────────┐   │
│  │ 一、整体架构…(流式渲染)  │  │ 1. 合规风险 IDFA/OAID…     │   │
│  │ …                          │  │ …                          │   │
│  └────────────────────────────┘  └────────────────────────────┘   │
└───────────────────────────────────────────────────────────────────┘
(左侧"参与角色"栏在并行时整体收起,腾出横向空间放两列消息)

5.2 关键交互(与需求逐条对齐)

  1. 收起左侧:并行执行时把左侧「参与角色 / 人工参与」栏整体折叠(一个 isParallelRunning 状态控制;非并行恢复)。
  2. 头像条:位置在消息列表上方、模型选择器左侧。展示当前并行批次的所有进行中智能体头像:
    • 进行中 → 头像加 loading 动画(复用 DispatchChainModal 的 animate-ping 光环 + Loader2Icon animate-spin 角标)。
    • 已完成 → 打勾(复用现有 CheckIcon 角标)。
    • 选中 → 高亮描边(当前正在主区显示的那 2 个)。
  3. 主区同时显示 2 个:被选中的 2 个智能体的消息列表并排两列渲染。
  4. 多于 2 个先隐藏:未选中的智能体不渲染 DOM(见 §6 性能)。
  5. 点头像切换:点击未选中的头像 → 把它设为选中之一(替换当前 2 个里的一个,或按"最近点击保留 2 个"的 LRU 策略);被换下的智能体停止渲染但继续接流。

5.3 选中策略 — ✅ 固定 2 列,窄屏降 1 列

  • 维护 selectedParallelSeats: string[],最多 2(固定 2 列)。不做 3+ 列。
  • 响应式:宽屏 2 列并排;窄屏(建议断点 < 1024px / lg)降为1 列,头像条横向滚动切换;此时 selectedParallelSeats 仍可存 2 个,但主区只渲染当前 1 个(点头像切换哪个在前)。
  • 默认选中本批次最先开始的(宽屏 2 个 / 窄屏 1 个)。
  • 点头像:若已选中则取消;未选中则加入(超过上限时按 LRU 踢掉最早选中的)。被换下的席位停止渲染但继续接流(§6)。

5.4 验收(Phase 5)

  • 并行时左侧收起、出现头像条、主区两列。
  • 进行中头像有 loading;完成打勾;选中有高亮。
  • 点头像能切换主区显示的智能体,且切换是即时的(数据已在内存,见 §6)。

6. 流式高性能策略(避免页面卡死)

核心矛盾:N 个并行席位同时高频 onTextDelta,若都 setState + Markdown 渲染会卡死。

6.1 全接收、惰性渲染

  • 所有并行 run 的流都正常订阅(不丢数据):每个席位的增量文本累积进 store / ref(Map<agentId, { text, status, charCount }>)。
  • 只有被选中(主区显示)的席位才把累积文本挂到 DOM 渲染(Markdown)。
  • 未选中席位:只更新内存 + 头像条上的轻量进度(字数 / loading),不挂载消息 DOM。
  • 切换选中时:把该席位已累积的全文一次性渲染 + 继续接增量。

6.2 渲染节流

  • 选中席位的流式渲染节流/合帧:用 requestAnimationFrame 或 ~50–100ms 批量 flush 累积的 token,而不是每个 token 一次 setState。
  • Markdown 渲染对"流式中"可降级(纯文本 / 轻量渲染),完成后再做完整 Markdown(现有 MessageBubble 的 streaming 标志可复用:streaming 时简渲、done 后全渲)。

6.3 头像条进度

  • 头像条只显示廉价信息:loading 动画 + 可选字数(charCount,从内存 Map 取,节流更新,例如每 500ms 刷一次)。不触发消息体渲染。

6.4 数据流(建议结构)

streamMultiAgent(seat).onTextDelta(text) →
    parallelStore.append(seatId, text)          // 纯内存累积,O(1)
    if seatId ∈ selectedParallelSeats:          // 仅选中的进 React state(节流)
        scheduleFlush(seatId)                    // rAF / 节流 setState
    else:
        bumpCharCount(seatId)                    // 仅更新头像条进度(节流)

6.5 验收(Phase 6)

  • 6+ 席位并行时页面不卡(输入 / 滚动 / 切换流畅)。
  • 切到一个之前隐藏的席位,能立刻看到它到目前为止的全部输出(证明流没丢、只是没渲染)。

7. 流程图执行态可视化(DAG 版)

  • 把现有 DispatchChainModal.tsx(横向线性流程图)扩展 / 复刻成 DAG 流程图:
    • 按 stage 分列,stage 内席位纵向并列,stage 间用连接线。
    • 节点状态复用 ChainNodeState(pending / active / done)+ 现有动画(animate-ping / Loader2Icon / CheckIcon)。
    • 末尾接 共识 → 结果绘制 里程碑(沿用现有)。
  • 用途:编辑器里的只读预览(§2.2 底部)+ 执行时的总览(可选小窗 / 顶部缩略)。
总控 ─▶ ┌ Stage1 ┐ ─▶ ┌ Stage2 ┐ ─▶ ┌ Stage3 ┐ ─▶ 🤝共识 ─▶ 📊报告
        │ 情报●  │     │ 方案✓  │     │ 执行○  │
        └────────┘     │ 风险●  │     └────────┘
                       └────────┘

7.1 验收(Phase 7)

  • 编辑器底部 / 执行时能看到 DAG 流程图,节点状态随执行实时更新(active/done)。

8. 分步实现计划(每步可独立交付 + 验收)

Phase 内容 主要文件 依赖 可独立验收
1 DAG 数据模型 + 校验 + 持久化字段 lib/dag-types.ts(新)、draft step2 读写 — 纯函数单测:validatePlan / chain→dag 兼容转换
2 分层编排编辑器(先无拖拽:点选加入/移出) components/DagPlanEditor.tsx(新)、Step2 模式切换 P1 能编排并保存/恢复,非法编排有提示
2.5 编辑器拖拽 同上 + @dnd-kit(若引入) P2 拖拽分配/重排
3 runDagOrchestration:批内并行 + 批间串行 + 依赖摘要注入 hooks/useStep2Orchestration.ts、api/multi-agent.ts P1 后端日志并行 special run 时间重叠;stage 严格串行
4 并行禁写文件 run policy roundtable_run_policy.py、multi_agent.py、api/multi-agent.ts P3 并行 run 的 excluded_tools 含 write_file/bash/...
5 并行显示骨架:收起左侧 + 头像条 + 两列 + 选中切换 components/Step2Panel.tsx、components/ParallelSeatStrip.tsx(新) P3 见 §5.4
6 流式高性能:全接收 / 惰性渲染 / 节流 hooks/useStep2Orchestration.ts(parallelStore)、Step2Panel P5 见 §6.5(6+ 并行不卡)
7 DAG 流程图执行态可视化 components/DagFlowChart.tsx(复刻自 DispatchChainModal) P2、P3 见 §7.1
8 边界 / 异常 各处 全部 见 §9

落地顺序建议:P1 → P3 → P4 → P5 → P6(先把"能并行跑 + 能看 + 不卡"打通),P2 编辑器可与 P3 并行开发(先用一份 hardcode/简单 UI 的 plan 驱动 P3 联调),P7 / P2.5 / 2.3 作为增强后置。


9. 边界与异常(Phase 8)

  • 某并行节点失败 / 超时:单个席位失败不拖垮整 stage;标该节点 error,stage 其余 ✅ 仍推进;收口时把"X 失败"如实告知总控(复用交付状态 ⬜ + 失败标记)。
  • 依赖未满足:调度器保证 stage 串行,理论不会发生;防御性校验 stage N 启动前 stage <N 必须全部终态(done/failed)。
  • 人工干预 / 暂停 / 恢复:在 stage 边界检查 isStale()(epoch);干预指令插入后,从"下一个未开始的 stage"继续(已完成 stage 不重跑)。
  • 并行禁写文件 vs 席位本想写文件:在席位 SOUL / 派活提示里说明"本轮为并行研讨,产出请直接写在回复正文,不要写文件",降低模型困惑。
  • 窄屏:主区两列降为一列 + 头像条横向滚动。
  • 单 stage 全并行(一次性 N 个):给并行度 warning(远程模型并发限流 / 成本)。

10. 与现有代码的衔接点速查

  • 编排循环:hooks/useStep2Orchestration.ts(runOrchestration / runChainOrchestration 旁新增 runDagOrchestration;复用 isStale() epoch、abortAllInFlight、appendStreamingDialogue / appendToDialogue)。
  • 流式 API:api/multi-agent.ts(streamMultiAgent,新增 parallel_no_file 入参;onTextDelta(text, messageId) 已是按席位回调,天然支持并行分流)。
  • 显示:components/Step2Panel.tsx(消息列表 step2RoundtableDialogues.map;新增头像条 ParallelSeatStrip + 两列容器 + 折叠左侧)。
  • 流程图:components/DispatchChainModal.tsx / lib/job-types.ts / lib/business-chain.ts(复刻成 DAG 版)。
  • 常量 / 元数据:lib/roundtable-constants.ts(BUILTIN_AGENT_META 头像/中文名、buildAgentNameToRole)。
  • 后端 run policy:app/gateway/roundtable_run_policy.py(新增 seat_parallel)、app/gateway/routers/multi_agent.py(_special_run 读 no_file flag)。
  • 交付摘要:GET /api/threads/{id}/last-ai-message?max_chars=N(依赖注入 stage N 输入用)。

11. 一句话总结

把「线性链条」升级为「分层 DAG」:编辑器用分层 stage(拖卡片)表达串/并;调度用 Promise.all 批内并行、批间 barrier 串行,依赖摘要在 stage 边界注入;并行禁写文件避免沙箱打架;显示收起左侧、头像条切换、主区两列、未选中只接流不渲染(rAF 节流)保证不卡。全部建立在现有 special/leader run + 轻量交付接口之上,后端几乎零新接口。