"凌晨3点47分,我盯着那个ReAct循环第17次原地打转。它像只无头苍蝇,每次都说'让我想想下一步',然后...又绕回了原点。那一刻我明白:有些任务,必须先看清全局,才能迈出第一步。"
什么是 Plan-and-Execute 模式?
Plan-and-Execute(规划-执行模式)是 AI Agent 的一种核心设计范式。与 ReAct 那种"走一步看一步"的思维方式不同,Plan-and-Execute 强调先制定完整计划,再按步骤执行。
简单来说,它就像:
- 🗺️ ReAct = 探险家摸着石头过河,随时调整方向
- 📋 Plan-and-Execute = 项目经理先写项目计划书,再按里程碑推进
核心原理:双层架构
┌─────────────────────────────────────────────────────────────┐
│ Plan-and-Execute 架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Planner │───▶│ Plan │───▶│ Executor │ │
│ │ (规划器) │ │ (计划) │ │ (执行器) │ │
│ │ │ │ │ │ │ │
│ │ • 理解目标 │ │ • 任务分解 │ │ • 执行步骤 │ │
│ │ • 分析资源 │ │ • 依赖关系 │ │ • 调用工具 │ │
│ │ • 制定策略 │ │ • 执行顺序 │ │ • 收集结果 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ◀──────────────────────────────────────────────┘ │
│ 反馈循环(可选) │
│ │
└─────────────────────────────────────────────────────────────┘
工作流程详解
- 输入理解:Agent 接收用户指令
- 全局规划:Planner 模块将大任务分解为子任务序列
- 计划存储:生成的计划被保存(在内存或外部存储)
- 顺序执行:Executor 按步骤执行每个子任务
- 结果聚合:合并各步骤结果,生成最终输出
- 动态调整(可选):如遇问题,可重新规划
什么时候用 Plan-and-Execute?
| 场景特征 | 推荐模式 | 原因 |
|---|---|---|
| 任务步骤已知、可预见 | ✅ Plan-and-Execute | 提前规划效率更高 |
| 需要多步骤协作 | ✅ Plan-and-Execute | 明确步骤依赖关系 |
| 任务探索性强、不可预测 | 🔄 ReAct | 灵活应变更重要 |
| 单步工具调用 | 🔄 ReAct | 规划开销不划算 |
| 需要长期状态管理 | ✅ Plan-and-Execute | 计划作为状态载体 |
💡 最佳实践
很多生产级 Agent 采用混合策略:先用 Plan-and-Execute 制定大方向,然后在每个子任务内使用 ReAct 处理细节。这样既保证全局可控,又保留局部灵活性。
很多生产级 Agent 采用混合策略:先用 Plan-and-Execute 制定大方向,然后在每个子任务内使用 ReAct 处理细节。这样既保证全局可控,又保留局部灵活性。
OpenClaw 实战应用
在 OpenClaw 中,我们可以用 sessions_spawn 和 sessions_send 来实现 Plan-and-Execute 模式。下面是一个完整示例:
场景:研究并撰写一份行业报告
任务分解:
- 搜索行业最新动态
- 分析竞争对手
- 整理关键数据
- 撰写报告摘要
# OpenClaw Plan-and-Execute 模式示例
# 场景:多步骤研究任务
import asyncio
async def planner(task):
"""
规划器:将大任务分解为可执行的子任务
在 OpenClaw 中,这可以是一个技能或子 Agent
"""
plan = {
"task": task,
"steps": [
{
"id": 1,
"name": "搜索最新动态",
"tool": "web_search",
"query": "AI Agent 2026 最新趋势"
},
{
"id": 2,
"name": "获取详细内容",
"tool": "web_fetch",
"depends_on": 1
},
{
"id": 3,
"name": "生成报告",
"tool": "write",
"depends_on": 2
}
]
}
return plan
async def executor(step, context):
"""
执行器:执行单个步骤
可以使用 OpenClaw 的 tools
"""
if step["tool"] == "web_search":
# 调用 web_search 工具
results = await web_search(query=step["query"])
return {"search_results": results}
elif step["tool"] == "web_fetch":
# 获取上一步的URL
urls = [r["url"] for r in context["search_results"][:3]]
contents = []
for url in urls:
content = await web_fetch(url=url)
contents.append(content)
return {"contents": contents}
elif step["tool"] == "write":
# 生成报告
report = await generate_report(context["contents"])
return {"report": report}
# 主执行流程
async def plan_and_execute(task):
# 1. 规划阶段
plan = await planner(task)
print(f"📋 生成计划: {len(plan['steps'])} 个步骤")
# 2. 执行阶段
context = {}
for step in plan["steps"]:
print(f"▶️ 执行步骤 {step['id']}: {step['name']}")
result = await executor(step, context)
context.update(result)
print(f"✅ 完成: {step['name']}")
return context.get("report")
# 运行
result = await plan_and_execute("生成AI行业趋势报告")
使用 OpenClaw Subagent 实现
更高级的做法是用 sessions_spawn 将每个子任务派发给专门的子 Agent:
# 使用 OpenClaw subagent 实现 Plan-and-Execute
# 1. 规划 Agent - 负责任务分解
planner = await sessions_spawn(
task="将'研究OpenClaw竞品'分解为3-5个可执行步骤",
runtime="subagent",
mode="run"
)
# 2. 执行 Agent 池 - 并行执行独立任务
executors = []
for step in plan.steps:
executor = await sessions_spawn(
task=step.description,
runtime="subagent",
mode="run"
)
executors.append(executor)
# 3. 收集结果并整合
results = [await ex.result for ex in executors]
final_report = await sessions_spawn(
task=f"整合以下研究结果生成报告: {results}",
runtime="subagent",
mode="run"
)
Plan-and-Execute vs ReAct
| 对比维度 | Plan-and-Execute | ReAct |
|---|---|---|
| 决策时机 | 先规划全部步骤 | 每步后重新思考 |
| token 消耗 | 规划阶段一次性消耗较多 | 逐步累积,可能更多 |
| 适应性 | 中等(可重规划) | 高(实时调整) |
| 透明度 | 高(计划清晰可见) | 中等(思考过程分散) |
| 适用场景 | 结构化、可预测任务 | 探索性、开放式任务 |
| 失败恢复 | 可从失败步骤重试 | 自然回退到上一步 |
⚠️ 常见踩坑
1. 过度规划:简单任务也用复杂计划,token 浪费严重
2. 计划僵化:不处理执行中的意外情况
3. 依赖地狱:步骤间依赖关系复杂,难以并行
4. 状态丢失:没有持久化计划,Agent 重启后忘记进度
1. 过度规划:简单任务也用复杂计划,token 浪费严重
2. 计划僵化:不处理执行中的意外情况
3. 依赖地狱:步骤间依赖关系复杂,难以并行
4. 状态丢失:没有持久化计划,Agent 重启后忘记进度
生产级优化技巧
1. 动态重规划
当执行失败时,不是简单重试,而是重新规划剩余步骤:
async def execute_with_replan(plan, max_retries=3):
for step in plan.steps:
for attempt in range(max_retries):
result = await execute(step)
if result.success:
break
elif attempt == max_retries - 1:
# 重规划剩余步骤
plan = await replan(plan, failed_step=step)
2. 计划持久化
使用 OpenClaw 的 memory 或外部存储保存计划状态:
# 保存计划到记忆
await memory.save({
"type": "execution_plan",
"plan_id": plan_id,
"steps": plan.steps,
"current_step": current_idx,
"context": accumulated_context
})
3. 并行执行
分析步骤依赖关系,并行执行无依赖的步骤:
# 依赖图分析后并行执行
ready_steps = [s for s in steps if not s.dependencies]
await asyncio.gather(*[
execute(step) for step in ready_steps
])
总结
Plan-and-Execute 模式是 Agent 系统从"玩具"走向"生产"的关键进化。它教会了 AI 一个朴素的道理:磨刀不误砍柴工。在这个充满不确定性的世界里,有时候我们需要先停下来,看看地图,再出发。
"凌晨3点47分的那个循环,后来我用 Plan-and-Execute 重写了。它不再打转,而是一步步走向了终点。原来,Agent 和人类一样,也需要一个计划。"