# 多 Agent 并行执行与仲裁合并设计 — 2026-07-01 > 性质:架构设计 / 数据模型 / 事件协议 / UI 交互 > 关联: [AI-Native方向与路线图-2026-06-29.md](../构想审查/AI-Native方向与路线图-2026-06-29.md)(P0 智能体人设 + 多 Agent 协作) > 关联: [单对话并行多轮-设计-2026-06-20.md](../单对话并行多轮-设计-2026-06-20.md)(Plan DAG 结构与分层调度) > 关联: [Agent架构说明-2026-06-14.md](Agent架构说明-2026-06-14.md)(当前 Agent 能力边界) > 关联: [三层模型-流程模板与人设体系-2026-06-28.md](三层模型-流程模板与人设体系-2026-06-28.md)(模板/工作流/人设三层抽象) > 用途:Batch 37-38 的实施依据 --- ## 一、设计目标 DevFlow 的定位是 AI Native——AI 是系统的主要操作者,人的角色是「决策者 + 政策制定者」。 多 Agent 并行执行要解决的核心问题: | 问题 | 现状(单 Agent) | 多 Agent 后 | |------|-----------------|------------| | 复杂任务拆解 | LLM 自行在单轮内拆,上下文窗口受限 | Coordinator 拆为 SubTask,各自独立上下文 | | 并行产出 | 串行,无冲突 | 多 Agent 同时改文件 → 数据竞争 | | 用户感知 | 单条流式输出 | 多 Agent 产出交织,用户分不清谁是谁 | | 错误恢复 | 一个工具失败 → LLM 下一轮决策 | 一个 SubTask 失败 → Coordinator 决定是否继续 | --- ## 二、数据模型(方案 C:混合) ### 2.1 设计原则 1. **messages 不膨胀** — 只加 `subtask_id` 外键列,不把 Plan/Conflict 字段塞进去 2. **重实体独立** — Plan / SubTask / Conflict 各自独立表,各自演进 3. **历史兼容** — `subtask_id = NULL` 表示单 Agent 时期消息,零回归 4. **FK 三层链路** — messages → SubTask → Plan 可追溯完整生命周期 ### 2.2 ER 图 ```mermaid erDiagram ai_conversations ||--o{ ai_plans : "1:N 触发" ai_plans ||--o{ ai_subtasks : "1:N 拆解" ai_plans ||--o{ ai_conflicts : "1:N 冲突" ai_subtasks ||--o{ ai_messages : "1:N 产出(subtask_id FK)" ai_subtasks ||--o{ ai_tool_executions : "1:N 工具调用(subtask_id FK)" ai_conflicts }o--|| ai_subtasks : "subtask_a" ai_conflicts }o--|| ai_subtasks : "subtask_b" ai_plans { TEXT id PK TEXT conversation_id FK TEXT user_message_id "触发 Plan 的用户消息" TEXT status "planning/executing/merging/done/error" INTEGER subtask_count TEXT created_at TEXT completed_at "可空" } ai_subtasks { TEXT id PK TEXT plan_id FK TEXT persona_id "coder/reviewer/architect/..." TEXT intent "子任务意图描述" TEXT status "pending/running/done/error" INTEGER layer "DAG 层级(0起)" TEXT deps "JSON 数组: 依赖的 SubTask id 列表" TEXT created_at TEXT completed_at "可空" } ai_conflicts { TEXT id PK TEXT plan_id FK TEXT file_path "冲突文件路径" TEXT subtask_a FK TEXT subtask_b FK TEXT diff_a "Agent A 对该文件的改动摘要" TEXT diff_b "Agent B 对该文件的改动摘要" TEXT resolution "pending/a/b/merged/manual" TEXT resolved_by "reviewer/user" TEXT created_at TEXT resolved_at "可空" } ``` ### 2.3 现有表改动 ```sql -- V36 迁移 ALTER TABLE ai_messages ADD COLUMN subtask_id TEXT; ALTER TABLE ai_tool_executions ADD COLUMN subtask_id TEXT; -- 新建 3 张表(见 ER 图字段) CREATE TABLE ai_plans (...); CREATE TABLE ai_subtasks ( ... -- worktree 分支名(git worktree 隔离机制,见 §4.2) branch TEXT -- 如 "subtask/{plan_id}/A" ); CREATE TABLE ai_conflicts ( ... -- 冲突类型(见 §4.2 四层防线) conflict_type TEXT DEFAULT 'file' -- 'file'(同文件) / 'semantic'(编译失败) ); ``` ### 2.4 三层关联查询示例 ```sql -- 查某 Plan 的全部产出(消息 + 工具) SELECT m.* FROM ai_messages m JOIN ai_subtasks s ON m.subtask_id = s.id WHERE s.plan_id = ?; -- 查某 Plan 的所有冲突 SELECT c.* FROM ai_conflicts c WHERE c.plan_id = ? AND c.resolution = 'pending'; -- 查某对话的全部 Plan 历史 SELECT p.* FROM ai_plans p WHERE p.conversation_id = ? ORDER BY created_at DESC; ``` --- ## 三、事件协议 ### 3.1 新增事件 | 事件 | 时机 | 载荷 | 前端响应 | |------|------|------|---------| | `AiPlanCreated` | Coordinator.decompose 完成后 | `{ plan_id, layers: [[{id, persona, intent, status}],...] }` | PlanProgress 展示 DAG | | `AiSubTaskStatusChanged` | SubTask 状态变更(pending→running→done/error) | `{ subtask_id, plan_id, status, persona_id }` | PlanProgress 更新节点状态 + 工具卡分组徽章 | | `AiMergeCompleted` | Coordinator.merge 完成后 | `{ plan_id, merged_output, conflicts: [{id, file, subtask_a, subtask_b}] }` | 展示合并结果 + 冲突徽章 | | `AiConflictResolved` | 用户/reviewer 解决冲突后 | `{ conflict_id, resolution, resolved_by }` | 冲突徽章消失 + 最终输出更新 | ### 3.2 事件流时序 ``` 用户发送消息 ↓ AiPlanCreated { plan_id, layers } ↓ (PlanProgress 立即展示) Layer 0 启动 ↓ AiSubTaskStatusChanged { subtask_0, running } AiSubTaskStatusChanged { subtask_1, running } ← 层内并行 ↓ (工具卡片按 subtask_id 分组,带 persona 徽章) AiSubTaskStatusChanged { subtask_0, done } AiSubTaskStatusChanged { subtask_1, done } ↓ Layer 1 启动 ... ↓ AiMergeCompleted { plan_id, merged_output, conflicts } ↓ (有冲突 → 冲突徽章;无冲突 → 直接展示合并输出) 用户点击冲突 → 展示 diff_a/diff_b → 选择 ↓ AiConflictResolved { conflict_id, resolution } ↓ 最终输出展示 ``` --- ## 四、并行执行策略 ### 4.1 调度规则 ``` Plan.to_layers() → [[subtask_0, subtask_1], [subtask_2], [subtask_3, subtask_4]] Layer 0(并行) Layer 1(串行) Layer 2(并行) ``` | 规则 | 说明 | |------|------| | 层间串行 | 上层全部 done + **Git merge 到 plan 分支后** 才创建下一层 worktree(fork 天然包含上层改动) | | 层内并行 | 同层 SubTask 各自创建 Git worktree,JoinSet 并发执行 | | 写工具天然隔离 | 每个 SubTask 在独立 worktree 内工作,write_file/patch_file 写到 worktree 目录(不重定向) | | 只读工具天然可见 | read_file 读 worktree 内文件(含本 worktree 改动 + fork 基点的全部历史) | | 层内隔离 | 同层并行 SubTask 互相看不到对方的 worktree 改动(独立分支,完全隔离) | ### 4.2 Git worktree 隔离机制 **核心思路**:用 Git worktree 替代自建暂存区,复用 Git 原生的分支隔离/diff/merge/conflict 能力。 #### 优势(vs 自建暂存区) | 维度 | 自建暂存区 | Git worktree | |------|-----------|-------------| | 隔离 | 手动 cp 文件 | `git worktree add` 天然隔离 | | diff | 自己实现比对 | `git diff` 原生(含上下文/行级/二进制) | | 合并 | 手动 cp 或自写 merge | `git merge` 三方合并 | | 冲突解决 | 自建 diff 展示 | `git merge-tree` 预检 + conflict markers | | 回滚 | 手动备份/恢复 | `git reset` / `git checkout` | | 审计 | 自建日志 | commit history 天然审计链 | | AI 工具兼容 | write_file 需重定向路径 | worktree 就是普通目录,所有工具原样工作 | | PR 集成 | 需额外实现 | plan 分支直接创建 PR | #### 数据流 ``` Plan 启动: git worktree add .devflow/wt/{plan_id} -b plan/{plan_id} (基于工程当前分支创建 plan 工作分支) Layer 0 并行: SubTask A → git worktree add .devflow/wt/{plan_id}/A -b subtask/{plan_id}/A (基于 plan 分支创建,A 的所有工具操作在此 worktree 内,天然隔离) SubTask B → git worktree add .devflow/wt/{plan_id}/B -b subtask/{plan_id}/B (同上,B 完全隔离,A 看不到 B 的改动,反之亦然) SubTask A/B done 后各自 git add + git commit(改动落入各自分支) Layer 0 merge(本层全部 done 后): ① cd .devflow/wt/{plan_id} (plan 分支) ② git diff plan..subtask/{plan_id}/A → A 的改动集 ③ git diff plan..subtask/{plan_id}/B → B 的改动集 ④ 同文件交集 → git merge-tree (三方合并预检) 无冲突 → git merge subtask/{plan_id}/A && git merge subtask/{plan_id}/B 有冲突 → Conflict 表记录 + Reviewer 仲裁 / 用户选择 ⑤ 合并完成 → plan 分支已含 Layer 0 全部改动 Layer 1 fork: 基于 plan 分支最新 commit 创建新 worktree(天然包含 Layer 0 改动) Plan 完成: ① 编译检查:在 plan worktree 跑 cargo check / tsc --noEmit 失败 → semantic 冲突标记 ② plan 分支创建 PR(可选,或直接 merge 回主分支) ③ git worktree remove 清理所有子 worktree + plan worktree ``` #### 读工具兼容规则 | 场景 | 行为 | |------|------| | 同层并行 SubTask 互读 | 不可见(独立 worktree + 独立分支) | | 下一层读上一层产出 | 可见(上一层已 merge 到 plan 分支,新 worktree 基于此创建) | | SubTask 读自己 worktree 内的写入 | 可见(worktree 是真实目录,写完即读) | #### 非 Git 工程的降级 未绑定 Git 的工程(无 .git 目录)降级为串行执行(单 SubTask 逐个跑,无 worktree 隔离)。 此时 ai_subtasks.branch 为 NULL,Coordinator.dispatch 走原串行路径。 ### 4.3 冲突防护四层防线 | 防线 | 机制 | 覆盖冲突类型 | |------|------|------------| | **第一层:层间串行 + merge 后才 fork** | 上一层 merge 到 plan 分支后,下一层 worktree 基于此创建 | 跨层同文件覆盖 | | **第二层:同层 worktree 隔离 + merge-tree 预检** | 独立分支 + git 三方合并检测 | 同文件并发写 | | **第三层:编译检查** | merge 完成后在 plan worktree 跑 cargo check / tsc --noEmit,失败 → 标记语义冲突 | 跨文件语义冲突(删函数/改签名) | | **第四层:命令互斥锁** | run_command 对同目录的 npm/cargo 加 mutex | 资源竞争(npm install 并发) | ### 4.4 Token 预算管控 ```rust struct TokenBudgetPool { total: AtomicU64, // 全局预算(来自设置项) consumed: AtomicU64, // 已消耗 } impl TokenBudgetPool { fn try_reserve(&self, estimate: u64) -> bool { // CAS 循环:consumed + estimate <= total } } ``` - 每个 SubTask 启动前向预算池申请估算额度 - 超限时 Coordinator 拒绝启动新 SubTask(降级为串行顺序执行剩余任务) - 预算来源:设置项 `df-ai-plan-token-budget`(默认 100k tokens) ### 4.5 错误传播 | 场景 | 策略 | |------|------| | SubTask 执行失败 | 记录 error 状态,**不中断其他同层 SubTask**(容错) | | 全部 SubTask 失败 | Coordinator 标记 Plan 状态为 error,前端展示错误 | | 部分 SubTask 失败 | merge 时跳过失败 SubTask 的分支(不 merge),只合并成功的 | | 子 Agent 超时 | 单 SubTask 超时(默认 120s)→ 标记 error,不影响其他 | | Git merge 冲突 | conflict markers 保留在 plan worktree,Reviewer Agent 仲裁或用户手动解决 | | 编译检查失败 | 标记 semantic 冲突,Reviewer Agent 尝试修复或标记给用户 | | worktree 创建失败 | 降级为串行(无 worktree 隔离,逐 SubTask 在主目录执行) | --- ## 五、UI 交互设计 ### 5.1 PlanProgress 组件(发送即展示) ``` ┌─────────────────────────────────────────┐ │ 📋 执行计划 2/4 │ ├─────────────────────────────────────────┤ │ Layer 1 │ │ ▶ 🔵 [coder] 重构代码 running │ │ 🟢 [architect] 分析结构 done │ │ ↓ │ │ Layer 2 │ │ ⏸ [tester] 补测试 pending │ │ ↓ │ │ Layer 3 │ │ ⏸ [reviewer] 审查 pending │ └─────────────────────────────────────────┘ ``` - **发送即展示**:用户发消息后 Coordinator 分解完成立即展示 - **实时更新**:AiSubTaskStatusChanged 驱动节点状态变化 - **折叠工具卡**:每个 SubTask 下的工具卡片折叠归组(点击展开) - **层间箭头**:DAG 层级关系可视化 ### 5.2 工具卡片分组(subtask_id 归组) ``` ┌─ 🔵 [coder·重构代码] ──────────────┐ │ ▸ read_file main.rs ✓ done │ │ ▸ patch_file utils.rs ✓ done │ └────────────────────────────────────┘ ┌─ 🟢 [architect·分析结构] ──────────┐ │ ▸ list_directory ✓ done │ │ ▸ 分析结论: 模块耦合度偏高... │ └────────────────────────────────────┘ ``` - 每个 SubTask 一个折叠容器,带 persona 颜色徽章 - 工具卡片归入对应 SubTask 容器 - 默认折叠,有审批/错误时自动展开 ### 5.3 冲突展示(徽章非阻塞) ``` ┌─────────────────────────────────────────┐ │ ✅ 执行完成 ⚠ 2 处冲突待处理 │ ├─────────────────────────────────────────┤ │ 合并输出: │ │ ... │ ├─────────────────────────────────────────┤ │ ⚠ main.rs — Agent A vs Agent B │ │ [查看 diff] [接受 A] [接受 B] [手动] │ │ ⚠ utils.rs — Agent A vs Agent B │ │ [查看 diff] [接受 A] [接受 B] [手动] │ └─────────────────────────────────────────┘ ``` - 冲突用**徽章**提示,不打断阅读流 - 点击展开 diff 对比 + resolution 按钮 - reviewer Agent 可自动给出推荐(`resolved_by=reviewer`),用户确认即可 --- ## 六、取消与中断 | 操作 | 行为 | |------|------| | 用户点停止 | 停止所有并行 SubTask(整体取消) | | 单 SubTask 超时 | 只标记该 SubTask error,不影响其他 | | 会话切换 | 后台 SubTask 继续执行(F-09 并发语义一致) | | 会话删除 | 所有关联 SubTask 停止 + Plan 标记 error | --- ## 七、实施分批 ### 数据层 + Git worktree 隔离 + 并行执行 | # | 任务 | 文件 | |---|------|------| | 1 | V36 迁移(3 新表 + 2 ALTER + subtasks.branch + conflicts.conflict_type) | migrations.rs | | 2 | PlanRepo / SubTaskRepo / ConflictRepo CRUD | 新 repo 文件 | | 3 | Git worktree 生命周期管理(create/commit/merge/remove) | 新 git_worktree.rs | | 4 | Coordinator.dispatch JoinSet 层内并行(每 SubTask 绑 worktree) | coordinator.rs | | 5 | Token 预算池(AtomicU64 CAS,超限降级串行) | coordinator.rs | | 6 | 子 Agent 独立 ContextManager + fork 快照 + worktree_path | coordinator.rs | | 7 | 层间 merge 到 plan 分支 + 下一层基于 plan 创建 worktree | coordinator.rs | | 8 | 4 个新事件 + 事件双写(emit + publish_event) | AiChatEvent | | 9 | 前端类型定义(PlanRecord/SubTaskRecord/ConflictRecord) | api/types.ts | | 10 | PlanProgress 接入真实状态 + 发送即展示 | PlanProgress.vue | | 11 | 工具卡按 subtask_id 折叠分组 + persona 徽章 | MessageList.vue | | 12 | 编译警告清理(coordinator_plan unused / audit 子模块 unused imports) | 各文件 | ### 仲裁合并 + 冲突 UI + 编译检查 | # | 任务 | 文件 | |---|------|------| | 1 | Coordinator.merge: git merge-tree 三方合并预检 | coordinator.rs + git_worktree.rs | | 2 | 冲突检测:同文件路径(file 类型) + 编译失败(semantic 类型) | coordinator.rs | | 3 | Reviewer Agent 仲裁(persona 扩展:读 diff → 推荐 resolution + 理由) | persona.rs | | 4 | 合并产出落回主对话(方案 C:可展开单条) | coordinator.rs + MessageList.vue | | 5 | ConflictResolver.vue(git diff 展示 + resolution 按钮) | 新组件 | | 6 | resolve_conflict IPC + AiConflictResolved 事件闭环 | AiChatEvent | | 7 | 命令互斥锁(run_command 对同目录 npm/cargo 加 mutex) | audit/approval.rs | | 8 | agentic/mod.rs 拆分(2222 行 → 提取 compress/title/knowledge_inject 独立模块) | agentic/*.rs | | 9 | AiChat 进度条补全(后端 AiAgentRound 补 max_rounds + 前端 completed 计数) | AiChat.vue | | 10 | context.rs 拆分(1956 行 → 提取 sanitize/budget/compress 独立模块) | context/*.rs | --- ## 八、子 Agent 上下文隔离 ### 8.1 双重隔离:上下文 + 文件系统 多 Agent 并行执行时需要两层隔离: | 层 | 隔离对象 | 机制 | |---|---------|------| | 上下文隔离 | messages / currentText / tool_results / pending_approvals | 独立 ContextManager 实例 | | 文件隔离 | 工程文件(worktree 内的源码) | Git worktree(见 §4.2) | ### 8.2 上下文隔离 ```rust struct SubTaskContext { /// 从主对话 fork 的快照(只读父上下文 + 用户原始消息) parent_snapshot: Vec, /// 独立的 ContextManager(不写回主对话) messages: ContextManager, /// 分配的人设 persona: AgentPersona, /// 子任务 id(产出消息标记 subtask_id) subtask_id: String, /// Git worktree 路径(工具操作的工作目录,见 §4.2) worktree_path: Option, // None = 非 Git 工程降级串行 } ``` **隔离规则**: | 维度 | 主对话 | 子 Agent | |------|--------|----------| | messages | 用户消息 + 合并产出 | fork 快照 + 独立执行轨迹 | | ContextManager | 主对话的 | 每个 SubTask 独立实例 | | pending_approvals | 主对话的 | 各 SubTask 独立(无并发写竞争) | | tool_results | 写入子 Agent 的 messages | 合并时提取摘要写入主对话 | | **文件系统** | 主工程目录 | **Git worktree 隔离**(§4.2) | | **工作目录** | 主工程路径 | **worktree_path**(工具的 cwd 重定向到此) | ### 8.3 合并产出落回主对话(方案 C:可展开单条) ``` SubTask 完成 → ExecutionResult { output, success } ↓ Coordinator.merge(results) → MergeResult { merged_output, conflicts } ↓ 主对话 push 单条可展开 assistant 消息 ``` **展示策略**: - **默认折叠**:主对话只展示一条 assistant 消息(merged_output 摘要) - **点击展开**:展开后按 SubTask 分组,每组显示完整执行轨迹(从 ai_messages WHERE subtask_id=? 拉取) - **冲突内联**:合并消息底部内联冲突列表(点击展开 diff + resolution 按钮) - **子 Agent 执行轨迹不堆入主对话流**:只在 ai_messages 表保留(带 subtask_id),展开时按需加载 --- ## 九、前端类型定义 ```typescript // api/types.ts 新增 /** Plan 执行状态 */ export type PlanStatus = 'planning' | 'executing' | 'merging' | 'done' | 'error' /** SubTask 执行状态 */ export type SubTaskStatus = 'pending' | 'running' | 'done' | 'error' /** 冲突解决状态 */ export type ConflictResolution = 'pending' | 'a' | 'b' | 'merged' | 'manual' /** Plan 记录 */ export interface PlanRecord { id: string conversation_id: ConvId user_message_id: MessageId status: PlanStatus subtask_count: number created_at: string completed_at: string | null } /** SubTask 记录 */ export interface SubTaskRecord { id: string plan_id: string persona_id: string intent: string status: SubTaskStatus layer: number deps: string[] // JSON 解析后的数组 /** Git worktree 分支名(NULL=非 Git 工程降级串行) */ branch: string | null created_at: string completed_at: string | null } /** 冲突类型 */ export type ConflictType = 'file' | 'semantic' /** 冲突记录 */ export interface ConflictRecord { id: string plan_id: string /** 冲突文件路径(semantic 类型时为触发编译失败的入口文件) */ file_path: string /** 冲突类型:file=同文件路径 / semantic=编译失败 */ conflict_type: ConflictType subtask_a: string subtask_b: string /** git diff 输出(A 的改动) */ diff_a: string /** git diff 输出(B 的改动) */ diff_b: string resolution: ConflictResolution resolved_by: 'reviewer' | 'user' | null created_at: string resolved_at: string | null } /** PlanProgress 展示用 DAG 层结构 */ export interface PlanLayer { items: Array<{ id: string persona: string intent: string status: SubTaskStatus }> } ``` --- ## 十、状态机 ### 10.1 Plan 状态流转 ```mermaid stateDiagram-v2 [*] --> planning: Coordinator.decompose() planning --> executing: Plan 写入 DB + emit AiPlanCreated executing --> merging: 所有 SubTask done/error merging --> done: merge 完成 + 无冲突 merging --> done: 所有冲突已解决 merging --> error: merge 失败 executing --> error: 全部 SubTask 失败 done --> [*] error --> [*] ``` ### 10.2 SubTask 状态流转 ```mermaid stateDiagram-v2 [*] --> pending: Plan.decompose 创建 pending --> running: dispatch 分配 + JoinSet 启动 running --> done: agentic loop 收敛(无 tool_calls) running --> error: 超时 / 全部工具失败 done --> [*] error --> [*] ``` ### 10.3 Conflict 状态流转 ```mermaid stateDiagram-v2 [*] --> pending: merge 检测到冲突 pending --> resolved: reviewer 推荐 + 用户确认 pending --> resolved: 用户手动选择 a/b/merged resolved --> [*] ``` --- ## 十一、事件总线集成 4 个新事件均走现有双写机制(`app_handle.emit` + `ai_event_bus.publish_event`), 与 AiToolCallStarted/AiCompleted 等关键状态变更事件一致: ```rust let ev = AiChatEvent::AiPlanCreated { ... }; let _ = app_handle.emit("ai-chat-event", ev.clone()); let _ = app_handle.state::().ai_event_bus.publish_event(ev); ``` - `emit`:桌面端前端监听 - `publish_event`:tunnel subscriber 透传小程序 --- ## 十二、不做的方向 | 方向 | 放弃理由 | 可能的时机 | |------|---------|-----------| | 工具级并行(同轮多 tool_call JoinSet) | 并发写 pending_approvals + LLM 对乱序 tool_result 行为不可预测 | 永不(LLM 已能单轮多 tool_call) | | 单 SubTask 取消 | MVP 简化,整体取消即可 | P1 以后按需 | | 双栏 diff 对比 UI | 违背 AI Native 定位(让用户做 AI 该做的合并) | 永不(reviewer Agent 仲裁替代) | | SubTask 级审批门控 | 审批在工具级(RiskLevel)已足够 | P2 审批政策配置 |