拆解 Warp AI Agent（二）：风险分级执行——Agent 如何做到安全并行、危险排队

系列第二篇。上篇讲了 Action 的类型安全设计，本篇看这些 Action怎么被调度执行——Warp 的BlocklistAIActionModel实现了一个精巧的风险分级执行引擎：只读操作并行跑，危险操作串行排队等用户确认。

一、问题：AI 一次返回多个 Action，怎么调度？

LLM 的一个回复可能包含多个工具调用：

用户：帮我在 src/ 下找到所有 TODO 注释并修复 AI 回复： 1. Grep("TODO", path="src/") ← 只读搜索 2. ReadFiles("src/main.rs") ← 只读读取 3. RequestFileEdits([替换 TODO]) ← 写入编辑 4. RequestCommandOutput("cargo test") ← 执行命令

4 个 Action，风险等级完全不同。如何调度？

• 粗暴方案：全部串行执行 → 慢，只读操作被写操作阻塞
• 粗暴方案：全部并行执行 → 危险！用户还没确认编辑，测试就已经跑起来了
• Warp 方案：风险分级 + 阶段调度

二、BlocklistAIActionModel：四队列架构

// app/src/ai/blocklist/action_model.rspubstructBlocklistAIActionModel{executor:ModelHandle<BlocklistAIActionExecutor>,/// 等待预处理的 Action（解析、校验）pending_preprocessed_actions:HashMap<AIConversationId,PendingPreprocessedActions>,/// 等待执行的 Action 队列（FIFO）pending_actions:HashMap<AIConversationId,VecDeque<AIAgentAction>>,/// 正在执行的 Action（按阶段分组）running_actions:HashMap<AIConversationId,RunningActions>,/// 已完成的 Action 结果finished_action_results:HashMap<AIConversationId,Vec<Arc<AIAgentActionResult>>>,/// 保持 Action 原始顺序（并行执行后结果需重新排序）action_order:HashMap<AIConversationId,HashMap<AIAgentActionId,usize>>,/// 历史结果（跨 Exchange 累积）past_action_results:HashMap<AIAgentActionId,Arc<AIAgentActionResult>>,}

数据流：

LLM 返回 Actions │ ▼ pending_preprocessed_actions ← 预处理（解析参数、校验） │ ▼ pending_actions ← 排队等待 │ ├─ 并行阶段 ──▶ running_actions (Parallel) │ │ │ ▼ (所有并行 Action 完成后) │ sort_finished_results ← 按原始顺序重排结果 │ └─ 串行阶段 ──▶ running_actions (Serial) │ ▼ (一个一个执行，等用户确认) finished_action_results

三、RunningActionPhase：并行 vs 串行的判定

// app/src/ai/blocklist/action_model/execute.rspub(super)enumRunningActionPhase{/// 屏障操作，必须独占执行Serial,/// 同一兼容组内的操作可以并行Parallel(ParallelExecutionPolicy),}pub(super)enumParallelExecutionPolicy{/// 只读、仅查本地上下文的操作，可以安全并行ReadOnlyLocalContext,}

3.1 哪些操作可以并行？

implBlocklistAIActionExecutor{pubfnaction_phase(&self,action:&AIAgentAction,ctx:&AppContext)->RunningActionPhase{match&action.action{// ✅ 并行：只读 + 本地上下文AIAgentActionType::ReadFiles(..)|AIAgentActionType::SearchCodebase(..)|AIAgentActionType::ReadSkill(_)=>RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext),// ✅ 条件并行：Grep/FileGlob 需要运行时判断AIAgentActionType::Grep{..}ifself.grep_executor.as_ref(ctx).can_execute_in_parallel(ctx)=>RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext),AIAgentActionType::FileGlob{..}|AIAgentActionType::FileGlobV2{..}ifself.file_glob_executor.as_ref(ctx).can_execute_in_parallel(ctx)=>RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext),// ❌ 串行：所有写操作、命令执行、MCP 调用_=>RunningActionPhase::Serial,}}}

规则很清晰：只有纯只读 + 纯本地的操作才能并行。任何涉及写入、命令执行、远程调用的操作都是串行。

3.2 阶段切换规则

fncan_start_action_with_current_phase(current_phase:RunningActionPhase,next_phase:RunningActionPhase,can_autoexecute:bool,)->bool{matchcurrent_phase{// 串行阶段：屏障，不允许新 Action 加入RunningActionPhase::Serial=>false,// 并行阶段：只有同组 + 可自动执行才允许加入RunningActionPhase::Parallel(group)=>{next_phase==RunningActionPhase::Parallel(group)&&can_autoexecute}}}

执行循环的核心逻辑：

fntry_to_execute_available_actions(&mutself,conversation_id:AIConversationId,ctx:...){loop{letfront_action=self.pending_actions.get(&conversation_id).and_then(|q|q.front());// 如果当前有正在执行的阶段，检查新 Action 能否加入ifletSome(current_phase)=self.action_execution_phase(conversation_id){if!self.can_start_action_in_current_phase(&front_action,conversation_id,current_phase,ctx){return;// 等当前阶段完成}}// 执行下一个 Actionletresult=self.start_pending_action_by_id(&front_action.id,conversation_id,false,ctx);// 如果启动了串行 Action，必须等它完成ifmatches!(result,StartedAction::Async{phase:RunningActionPhase::Serial}){return;}// 并行 Action 可以继续循环，尝试启动更多}}

四、20+ 独立执行器：每个工具一个

BlocklistAIActionExecutor不是一个大 switch，而是20+ 个独立执行器的组合：

好处：每个执行器只关心自己的 Action 类型，状态隔离，不会互相干扰。新增执行器不需要修改现有代码。

五、LLM 自评风险：is_read_only + is_risky

回到第一篇提到的RequestCommandOutput：

RequestCommandOutput{command:String,is_read_only:Option<bool>,// LLM 标注：只读？is_risky:Option<bool>,// LLM 标注：有风险？rationale:Option<String>,// LLM 给出的执行理由...}

设计意图：让 LLM 在生成 Action 时就评估风险，而不是由宿主事后判断。这有几个好处：

1. UI 可以直接展示风险等级— “AI 认为这个命令是只读的” / “AI 认为这个命令有风险”
2. 自动执行策略基于 LLM 自评— 只读命令自动执行，有风险命令等用户确认
3. 审计日志— 每个操作都有 AI 的"理由"，事后可追溯

与 Claude Code 的对比：Claude Code 的自动执行策略基于硬编码的工具名（read_file自动，write_file需确认），而 Warp 让 LLM 自己判断——同一个RequestCommandOutput，ls可以自动执行，rm -rf需要确认。

六、并行结果的顺序保证

并行执行会导致结果乱序，但 Agent 期望按原始顺序接收结果。Warp 的解法：

pubstructBlocklistAIActionModel{/// 记录每个 Action 在原始列表中的位置action_order:HashMap<AIConversationId,HashMap<AIAgentActionId,usize>>,}fnsort_finished_results(&mutself,conversation_id:AIConversationId){ifletSome(action_order)=self.action_order.get(&conversation_id){ifletSome(finished_results)=self.finished_action_results.get_mut(&conversation_id){finished_results.sort_by_key(|result|{action_order.get(&result.action_id).copied().unwrap_or(usize::MAX)});}}}

执行过程：

原始顺序: [ReadFiles(0), Grep(1), ReadFiles(2)] 并行执行: ReadFiles(0) + Grep(1) 同时启动 完成顺序: Grep(1) 先完成, ReadFiles(0) 后完成 排序还原: [ReadFiles(0), Grep(1), ReadFiles(2)] ← 保持原始顺序

七、阶段排空机制

并行阶段的所有 Action 完成后，才会推进到下一个阶段：

// Action 完成回调fnhandle_action_finished(&mutself,...){// 把结果加入完成列表self.finished_action_results.entry(conversation_id).or_default().push(result);// 检查当前阶段是否还有运行中的 Actionifself.running_actions.get(&conversation_id).is_some_and(|r|!r.is_empty()){// 阶段未排空，等待其他并行 Action 完成return;}// 阶段排空 → 排序结果 → 尝试启动下一阶段self.sort_finished_results(conversation_id);self.try_to_execute_available_actions(conversation_id,ctx);}

时间线: t0: [ReadFiles, Grep] ← 并行启动 t1: Grep 完成 ← 阶段未排空，等待 t2: ReadFiles 完成 ← 阶段排空！排序结果 t3: [RequestFileEdits] ← 串行启动（需要用户确认） t4: 用户确认 t5: [cargo test] ← 串行启动

八、与业界方案对比

Warp 是唯一支持只读操作并行执行的终端 Agent。

九、可复用模式：Risk-Graded Execution

┌─────────────────────────────────────────┐ │ Risk-Graded Execution Engine │ ├─────────────────────────────────────────┤ │ 1. Action 分类 │ │ - ReadOnlyLocalContext → Parallel │ │ - 其他 → Serial │ │ │ │ 2. 阶段调度 │ │ - 同组并行 Action 可同时执行 │ │ - 串行 Action 是屏障，必须独占 │ │ - 阶段排空后才推进到下一阶段 │ │ │ │ 3. LLM 自评风险 │ │ - is_read_only + is_risky │ │ - rationale 审计字段 │ │ │ │ 4. 结果重排序 │ │ - 记录原始顺序 │ │ - 并行完成后按原始顺序排回 │ │ │ │ 5. 独立执行器 │ │ - 每种 Action 类型一个执行器 │ │ - 新增执行器不影响现有代码 │ └─────────────────────────────────────────┘

十、总结

Warp 的执行引擎回答了一个核心问题：AI Agent 既要效率（并行读文件），又要安全（危险操作排队确认），如何兼得？

答案是风险分级：

1. 编译期：Action 类型确保参数安全（上篇）
2. 调度期：风险分级决定并行/串行（本篇）
3. 执行期：独立执行器隔离状态
4. 完成期：结果重排序保证顺序一致性

一句话总结：只读并行提速、危险串行保安全、LLM 自评风险、阶段排空才推进——Warp 的执行引擎用最小的复杂度实现了"又快又安全"。

系列导航：

• （一）类型即协议
• （二）风险分级执行 ← 你在这里
• （三）对话状态机
• （四）Merkle Tree 增量索引
• （五）跨生态联邦