198 lines
7.1 KiB
Rust
198 lines
7.1 KiB
Rust
//! Plan 执行器（Phase 2 · DAG 分层调度）
|
|
//!
|
|
//! 按 Plan::to_layers 输出的层级顺序执行子任务：
|
|
//! - 层间串行：上层全部完成才进下一层（deps 保证）
|
|
//! - 层内并行：同层 SubTask 用 tokio::JoinSet 并发执行
|
|
//!
|
|
//! **当前状态**：骨架实现，PLAN_EXECUTION_ENABLED 门控（默认关）。
|
|
//! 未接入 agentic loop 主路径，翻 true 启用后仍需与 run_agentic_loop 对接。
|
|
//!
|
|
//! 设计依据：`docs/02-架构设计/单对话并行多轮-设计-2026-06-20.md`
|
|
|
|
use std::sync::Arc;
|
|
|
|
use crate::planner::{Plan, SubTask};
|
|
|
|
/// Plan 执行总开关。false = 不启用（走单链 ReAct 旧行为）。
|
|
///
|
|
/// 运行时原子门控,支持热切换(通过 IPC 或前端设置开关)。
|
|
/// 默认关,翻 true 后 process_tool_calls 内以 JoinSet 并行执行同层工具。
|
|
/// 与 PLANNING_ENABLED 解耦:plan_hint 独立使能,编排可见性始终可开。
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
pub static PLAN_EXECUTION_ENABLED: AtomicBool = AtomicBool::new(false);
|
|
|
|
/// 设置 Plan 执行开关。
|
|
pub fn set_plan_execution(enabled: bool) {
|
|
PLAN_EXECUTION_ENABLED.store(enabled, Ordering::SeqCst);
|
|
tracing::info!(enabled, "[PLAN-EXEC] 执行开关已更新");
|
|
}
|
|
|
|
/// 读取 Plan 执行开关。
|
|
pub fn plan_execution_enabled() -> bool {
|
|
PLAN_EXECUTION_ENABLED.load(Ordering::SeqCst)
|
|
}
|
|
|
|
/// 单个 SubTask 的执行结果。
|
|
#[derive(Debug, Clone)]
|
|
pub struct SubTaskResult {
|
|
/// 对应 SubTask id
|
|
pub id: String,
|
|
/// 执行是否成功
|
|
pub success: bool,
|
|
/// 执行产出（工具结果摘要 / LLM 回复）
|
|
pub output: String,
|
|
}
|
|
|
|
/// Plan 执行器：按 DAG 层级调度子任务。
|
|
///
|
|
/// 泛型参数 `F` 为子任务执行函数：
|
|
/// - 输入：`&SubTask`（子任务定义）
|
|
/// - 输入：`&str`（父任务结果摘要，供子任务上下文参考）
|
|
/// - 输出：`Future<Output = anyhow::Result<String>>`（子任务产出）
|
|
///
|
|
/// 调用方（Phase 3 对接 agentic loop 时）传入真实的执行闭包，
|
|
/// 每个 SubTask 独立跑一轮 agentic loop，共享父上下文快照。
|
|
pub struct PlanExecutor;
|
|
|
|
impl PlanExecutor {
|
|
/// 按 Plan DAG 分层执行。
|
|
///
|
|
/// - 层间串行：上层全部完成才进下一层
|
|
/// - 层内并行：同层 SubTask 用 JoinSet 并发
|
|
/// - 失败传播：任一 SubTask 失败时，记录错误但继续执行同层其他任务（容错）；
|
|
/// 若需中断策略，调用方据返回结果自行判断。
|
|
///
|
|
/// 返回每个 SubTask 的执行结果（按 Plan 原始顺序排列）。
|
|
pub async fn execute<F, Fut>(
|
|
plan: &Plan,
|
|
mut run_subtask: F,
|
|
) -> anyhow::Result<Vec<SubTaskResult>>
|
|
where
|
|
F: FnMut(Arc<SubTask>, String) -> Fut,
|
|
Fut: std::future::Future<Output = anyhow::Result<String>>,
|
|
{
|
|
let layers = plan
|
|
.to_layers()
|
|
.map_err(|e| anyhow::anyhow!("Plan 环依赖: {:?}", e.cycle))?;
|
|
|
|
let mut results: Vec<SubTaskResult> = Vec::new();
|
|
let mut parent_summary = String::new();
|
|
|
|
for (layer_idx, layer) in layers.iter().enumerate() {
|
|
tracing::info!(
|
|
layer_idx,
|
|
task_count = layer.len(),
|
|
"[plan-exec] 执行第 {} 层, {} 个子任务",
|
|
layer_idx,
|
|
layer.len()
|
|
);
|
|
|
|
// 层内并行：JoinSet 收集
|
|
let _join_set: tokio::task::JoinSet<SubTaskResult> = tokio::task::JoinSet::new();
|
|
let summary_clone = parent_summary.clone();
|
|
|
|
for task in layer {
|
|
let task = Arc::new(task.clone());
|
|
let summary = summary_clone.clone();
|
|
// 注意:run_subtask 是 FnMut,不能直接 move 进 JoinSet(多任务并发调用)。
|
|
// Phase 3 对接时,run_subtask 改为 Arc<Fn> 或 channel 模式。
|
|
// 当前骨架用串行执行模拟(层内不并行),验证分层逻辑正确性。
|
|
let result = run_subtask(task.clone(), summary).await;
|
|
let sr = match result {
|
|
Ok(output) => SubTaskResult {
|
|
id: task.id.clone(),
|
|
success: true,
|
|
output,
|
|
},
|
|
Err(e) => SubTaskResult {
|
|
id: task.id.clone(),
|
|
success: false,
|
|
output: e.to_string(),
|
|
},
|
|
};
|
|
results.push(sr);
|
|
}
|
|
|
|
// 汇总本层结果作为下层 parent_summary
|
|
let layer_summary: Vec<String> = results
|
|
.iter()
|
|
.filter(|r| layer.iter().any(|t| t.id == r.id))
|
|
.map(|r| format!("[{}] {}", r.id, r.output))
|
|
.collect();
|
|
parent_summary = layer_summary.join("\n");
|
|
}
|
|
|
|
Ok(results)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::planner::{Plan, SubTask};
|
|
|
|
#[tokio::test]
|
|
async fn execute_simple_linear_plan() {
|
|
// a → b → c (三层,各一个任务)
|
|
let plan = Plan::from_tasks(vec![
|
|
SubTask::new("a", "task-a"),
|
|
SubTask::new("b", "task-b").with_deps(vec!["a".to_string()]),
|
|
SubTask::new("c", "task-c").with_deps(vec!["b".to_string()]),
|
|
]);
|
|
|
|
let results = PlanExecutor::execute(&plan, |task, _parent| async move {
|
|
Ok(format!("done:{}", task.id))
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(results.len(), 3);
|
|
assert!(results.iter().all(|r| r.success));
|
|
assert_eq!(results[0].id, "a");
|
|
assert_eq!(results[1].id, "b");
|
|
assert_eq!(results[2].id, "c");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn execute_parallel_layer() {
|
|
// a → {b, c} → d (三层,第二层两个并行)
|
|
let plan = Plan::from_tasks(vec![
|
|
SubTask::new("a", "task-a"),
|
|
SubTask::new("b", "task-b").with_deps(vec!["a".to_string()]),
|
|
SubTask::new("c", "task-c").with_deps(vec!["a".to_string()]),
|
|
SubTask::new("d", "task-d").with_deps(vec!["b".to_string(), "c".to_string()]),
|
|
]);
|
|
|
|
let results = PlanExecutor::execute(&plan, |task, _parent| async move {
|
|
Ok(format!("done:{}", task.id))
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(results.len(), 4);
|
|
assert!(results.iter().all(|r| r.success));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn execute_with_failure_continues() {
|
|
// a(fail) → b: a 失败,b 仍执行(容错策略)
|
|
let plan = Plan::from_tasks(vec![
|
|
SubTask::new("a", "task-a"),
|
|
SubTask::new("b", "task-b").with_deps(vec!["a".to_string()]),
|
|
]);
|
|
|
|
let results = PlanExecutor::execute(&plan, |task, _parent| async move {
|
|
if task.id == "a" {
|
|
anyhow::bail!("task-a failed")
|
|
}
|
|
Ok("ok".to_string())
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(results.len(), 2);
|
|
assert!(!results[0].success); // a 失败
|
|
assert!(results[1].success); // b 仍执行
|
|
}
|
|
}
|