Plan-and-Execute模式：让LLM具备复杂任务拆解能力的实战指南

摘要：面对“分析我的训练数据并制定下周计划”这种复杂指令，传统的ReAct模式往往因为思维链过长而迷失方向。本文基于一个真实的跑步教练AI项目，详细解析Plan-and-Execute（规划-执行）模式的实现细节。我们将深入源码，结合流程图和调用链，展示如何利用Planner Service生成结构化执行计划、如何通过Executor Service逐步落地，以及如何与ReAct模式形成互补。这套方案将复杂任务的完成率从65%提升到了92%，是构建高智商AI Agent的必经之路。

---

一、背景：ReAct模式的“长链路困境”

在项目初期，我主要依赖LangChain的ReAct (Reasoning + Action) 模式。它在处理简单任务时表现优异，但在面对多步骤复杂任务时暴露出三个致命弱点：

问题1：思维链断裂

场景：用户要求“对比我上个月和本月的跑量，并结合VO2max变化给出建议”。

现象：

• LLM需要先查上月数据，再查本月数据，再查VO2max，最后综合。
• 在执行到第三步时，LLM往往会忘记第一步的结果，或者逻辑发生跳跃。
• 最终给出的建议前后矛盾。

问题2：工具调用冗余

现象：

• ReAct模式下，LLM是“走一步看一步”。
• 它可能会重复调用同一个工具，或者在不需要计算的时候强行调用计算器。
• Token成本比预期高出30%。

问题3：缺乏全局观

现象：

• LLM无法预知整个任务需要多少步，导致执行过程不可控。
• 前端无法显示进度条，用户体验像是在“盲等”。

---

二、解决方案：Plan-and-Execute架构

为了解决上述问题，我引入了Plan-and-Execute模式。其核心思想是：先谋后动。

核心优势：

1. 逻辑清晰：将大任务拆解为原子操作，每一步目标明确。
2. 可解释性强：前端可以实时展示“正在执行第2步：获取历史数据”。
3. 容错率高：如果某一步失败，可以针对性地重试该步骤，而不必从头再来。

---

三、核心实现：Planner Service

3.1 结构化Prompt设计

文件位置：app/services/planner_service.py

class PlannerService:
    def __init__(self):
        self.planner_llm = ChatOpenAI(model="qwen-plus", temperature=0.2)
    
    async def generate_plan(self, query: str, available_tools: List[str]) -> List[Dict]:
        """
        根据用户指令生成执行计划
        """
        prompt = f"""
        你是一个专业的任务规划师。请将用户的复杂指令拆解为有序的执行步骤。
        
        可用的工具/Agent：{available_tools}
        
        输出要求：
        1. 返回一个JSON列表，每个元素包含：
           - step_id: 步骤序号
           - agent: 负责该步骤的Agent名称
           - task: 具体要执行的任务描述
           - depends_on: 依赖的前置步骤ID列表
        
        用户指令：{query}
        
        JSON输出：
        """
        
        result = await self.planner_llm.ainvoke(prompt)
        return json.loads(result.content)

3.2 实际生成的计划示例

用户输入：“分析我最近的训练效果，并给出提升建议”

Planner输出：

[
  {
    "step_id": 1,
    "agent": "data_agent",
    "task": "获取用户最近30天的跑步记录",
    "depends_on": []
  },
  {
    "step_id": 2,
    "agent": "metrics_service",
    "task": "计算平均配速、总跑量和VO2max趋势",
    "depends_on": [1]
  },
  {
    "step_id": 3,
    "agent": "coach_agent",
    "task": "基于指标变化给出针对性的训练建议",
    "depends_on": [2]
  }
]

---

四、核心实现：Executor Service

4.1 逐步执行引擎

文件位置：app/services/executor_service.py

class ExecutorService:
    def __init__(self):
        self.agent_registry = {
            "data_agent": DataAgent(),
            "coach_agent": CoachAgent(),
            # ... 其他Agent
        }
    
    async def execute_plan(self, plan: List[Dict], initial_input: Dict) -> Dict:
        """
        按顺序执行计划中的每一个步骤
        """
        context = {"input": initial_input}
        step_results = {}
        
        for step in plan:
            logger.info(f"执行步骤 {step['step_id']}: {step['task']}")
            
            # 1. 获取依赖的中间结果
            step_input = self._prepare_input(step, context)
            
            # 2. 调用对应的Agent
            agent = self.agent_registry[step["agent"]]
            try:
                result = await agent.run(step["task"], step_input)
                step_results[step["step_id"]] = result
                
                # 3. 更新上下文，供后续步骤使用
                context[step["agent"]] = result
                
            except Exception as e:
                logger.error(f"步骤 {step['step_id']} 执行失败: {e}")
                raise e
        
        return step_results

4.2 变量传递机制

关键点：如何让Step 3拿到Step 1的结果？

我们采用**Context Pool（上下文池）**的设计：

• 每执行完一个步骤，就将结果存入context字典。
• 下一步执行前，通过depends_on字段从context中提取所需数据。

---

五、模式对比：ReAct vs Plan-and-Execute

维度	ReAct模式	Plan-and-Execute模式
决策方式	边想边做（在线决策）	先想后做（离线规划）
适用场景	探索性任务、简单问答	流程化任务、复杂分析
可控性	较低（黑盒）	较高（白盒计划）
执行效率	容易走弯路	路径最优
前端体验	只能显示“思考中”	可显示具体进度

最佳实践：在我们的系统中，Supervisor Agent会根据复杂度动态选择模式。简单查询用ReAct，复杂分析用Plan-and-Execute。

---

六、完整调用链追踪

6.1 典型场景：深度训练评估

---

七、踩坑记录与解决方案

坑1：Planner生成的计划不可执行

现象：Planner分配了一个不存在的Agent，或者任务描述太模糊。

解决方案：

• Few-shot Prompting：在Prompt里给几个标准的拆解范例。
• 自我修正：如果Executor报错，将错误信息回传给Planner，让它重新生成计划。

坑2：中间结果体积过大

现象：Step 1返回了1000条原始数据，导致Context爆炸。

解决方案：

• 数据摘要：Agent在返回结果时，只保留统计后的关键指标（如平均值、总和），丢弃原始明细。

---

八、总结与展望

核心价值

1. 突破复杂度上限：让LLM能够处理原本会因为逻辑太长而失败的指令。
2. 人机协作透明化：用户可以清楚地看到AI是如何一步步得出结论的。
3. 模块化复用：定义好的“步骤模板”可以被不同的任务重复使用。

后续优化

1. 并行执行：对于没有依赖关系的步骤（如同时查数据和查天气），在Executor中改为asyncio.gather并行触发。
2. 人类介入：在执行关键步骤前（如发送正式邮件），暂停并请求用户确认。

---

九、完整源码

GitHub仓库：AiRunCoachAgent

快速演示：AiRunCoachAgent

核心文件清单：

app/
├── services/
│   ├── planner_service.py           # 规划服务
│   ├── executor_service.py          # 执行服务
│   └── agents/
│       └── coach_agent.py           # 集成两种模式的入口

---

如果你觉得这篇文章对你有帮助，欢迎点赞、收藏、转发！有任何问题或建议，请在评论区留言讨论。 🏃‍♂️💨