Plan-and-Execute 是大模型 Agent 最经典分层架构：先全局定完整步骤计划，再按序落地执行，出错自动重规划，完美平衡全局视野与动态容错，解决 ReAct 走一步看一步、容易跑偏循环的问题。

目录

一、核心定义

二、标准工作闭环（4 步）

三、三大核心模块

四、优缺点

✅ 优点

❌ 缺点

五、与 ReAct 模式对比（面试高频）

六、代码实战

📝 案例拆解：AI 是如何安排出差的？

阶段一：需求接收与“工具箱”打包

阶段二：“大脑”思考，生成严格的 JSON 计划表

阶段三：“手脚”按表操课（执行器流水线）

阶段四：汇总汇报，任务结束

【完整代码】

一、核心定义

把复杂长链路任务拆分层解耦：

  • Planner（规划器）：一次性生成完整、有序子任务清单，把控全局目标
  • Executor（执行器）：按计划调用工具、串行 / 并行完成每一步
  • Replanner（重规划器）：执行异常 / 结果不符时，动态修改方案

二、标准工作闭环（4 步）

  1. 目标接收：用户输入复杂多步骤任务
  2. 全局规划 Planning：LLM 拆解任务→输出结构化步骤、依赖关系、执行顺序
  3. 分步执行 Execution：按计划调用 API、搜索、计算、检索，记录每步结果
  4. 校验 + 重规划 Replan：步骤失败 / 信息变更→更新计划→重新执行；全部完成后汇总答案

通俗例子（比价买手机）

  • 规划：列平台→查各平台实时价→对比排序→输出最低价
  • 执行：依次调用电商接口爬价格、计算对比
  • 重规划：某平台无货→新增备选平台、调整步骤

三、三大核心模块

  1. 规划层：长上下文全局推理，不碰具体工具，只定路线
  2. 执行层：专注工具调用、数据处理、状态流转，不做宏观决策
  3. 反馈层：监控执行结果，判断是否需要修改计划、回溯重试

四、优缺点

✅ 优点

  • 全局视野强，不易目标漂移、不无限循环
  • 步骤清晰、可解释性高，方便调试
  • 无依赖子任务支持并行执行，省 Token、速度更快
  • 局部失败不崩盘，单独重规划即可，鲁棒性强

❌ 缺点

  • 前期计划过长，计划赶不上变化，易无效执行
  • 分层调用更多 LLM，推理开销更高
  • 步骤依赖复杂时，重规划逻辑设计难度大

五、与 ReAct 模式对比（面试高频）

维度 Plan-and-Execute ReAct
逻辑 先全盘规划，再批量执行 边思考、边行动、走一步看一步
全局能力 极强，长链路稳定 弱，易陷入局部循环
灵活性 中等，靠重规划适配变化 极高，实时响应环境
适用场景 长流程、多工具、固定复杂任务 短交互、实时多变、探索类任务

六、代码实战

【案例】

📝 案例拆解：AI 是如何安排出差的？

前情提要： 用户输入的目标是："帮我安排一次从北京到上海的出差：先查上海天气，然后订明天的机票，最后发邮件通知经理"

阶段一：需求接收与“工具箱”打包

当这句指令传给主函数 run_plan_and_execute 时，系统其实并没有马上动手。它首先把用户的这句“大白话”，连同公司现有的三个工具的说明书（查天气、订机票、发邮件的名字、用途和参数要求），一起打包交给了“部门经理”（规划器 Planner）。

阶段二：“大脑”思考，生成严格的 JSON 计划表

这一步是整段代码的“灵魂”。大模型（DeepSeek）在这一步扮演了一个**“只写计划，不干活”**的规划师。 它阅读了工具说明书，开始拆解任务，并输出了一份严格的 JSON 格式的执行计划。

{
  "steps": [
    {
      "id": 1,
      "description": "查询上海当前的天气情况",
      "tool_name": "search_weather",
      "arguments": {
        "city": "上海"
      }
    },
    {
      "id": 2,
      "description": "预订明天从北京到上海的机票",
      "tool_name": "book_flight",
      "arguments": {
        "origin": "北京",
        "destination": "上海",
        "date": "明天"
      }
    },
    {
      "id": 3,
      "description": "发送邮件通知经理出差行程和天气",
      "tool_name": "send_email",
      "arguments": {
        "to": "manager@company.com",
        "subject": "北京到上海出差安排",
        "body": "{{2}}" 
      }
    }
  ]
}

步骤 3 的 body 参数里的 "{{2}}"。大模型非常聪明，它知道自己现在还没查机票呢，所以它填了一个占位符，意思是“发邮件的时候，把步骤 2 的结果填进来”。

阶段三：“手脚”按表操课（执行器流水线）

计划表生成后，交给了执行器（Executor）。执行器是个没有感情的打工人，它不管逻辑对不对，只管按顺序从步骤 1 跑到步骤 3。

  • 执行步骤 1（查天气）：

    • 提取工具：search_weather

    • 检查参数：有没有 {} 占位符？没有。

    • 直接执行：调用 search_weather("上海")。

    • 获得结果："上海当前晴朗，气温25°C"。

    • 存入档案库：results[1] = "上海当前晴朗..."

  • 执行步骤 2（订机票）：

    • 提取工具：book_flight

    • 检查参数：没有占位符。

    • 直接执行：调用 book_flight("北京", "上海", "明天")。

    • 获得结果："已预订 明天 从 北京 到 上海 的航班"。

    • 存入档案库：results[2] = "已预订 明天..."

  • 执行步骤 3（发邮件）：

    • 提取工具：send_email

    • 关键动作（参数解析）：执行器发现 body 参数的值是 "{{2}}"！它立刻去档案库里翻出 results[2] 的内容，把参数替换成真实的机票信息。

    • 直接执行：调用 send_email("manager@company.com", "...", "已预订 明天...")

    • 存入档案库：results[3] = "邮件已发送至..."

阶段四：汇总汇报，任务结束

执行器跑完了这三步，没有发生任何报错（如果中途出错，它内置的 while 循环会尝试重试两次）。最后，它把整个 results 字典打包，作为最终结果返回给用户。

【完整代码】


from typing import List, Dict, Any
from langchain_deepseek import ChatDeepSeek
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from dotenv import load_dotenv
load_dotenv()

# ===================== 工具定义 =====================
@tool
def search_weather(city: str) -> str:
    """获取指定城市的天气信息。
    Args:
        city: 城市名称
    """
    return f"{city}当前晴朗，气温25°C"

@tool
def book_flight(origin: str, destination: str, date: str) -> str:
    """预订航班。
    Args:
        origin: 出发地城市
        destination: 目的地城市
        date: 出发日期，例如'明天'或'2026-04-21'
    """
    if origin == destination:
        raise ValueError("出发地与目的地不能相同")
    return f"已预订 {date} 从 {origin} 到 {destination} 的航班"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件。
    Args:
        to: 收件人邮箱地址
        subject: 邮件主题
        body: 邮件正文
    """
    return f"邮件已发送至 {to}，主题：{subject}"


tools = [search_weather, book_flight, send_email]


# ===================== 结构化输出模型 =====================
class PlanStep(BaseModel):
    id: int = Field(description="步骤ID")
    description: str = Field(description="步骤描述")
    tool_name: str = Field(description="要调用的工具名称")
    arguments: Dict[str, Any] = Field(description="工具参数字典")


class PlanOutput(BaseModel):
    steps: List[PlanStep] = Field(description="任务步骤列表")


# ===================== 规划器 =====================
class Planner:
    def __init__(self):
        self.llm = ChatDeepSeek(model="deepseek-reasoner", temperature=0)
        self.parser = JsonOutputParser(pydantic_object=PlanOutput)

        # 【修改这里】：在系统提示词中加入 {tools} 和 {format_instructions} 占位符
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """你是专业任务规划专家。
                可用工具如下：
                {tools}
                {format_instructions}
                要求：
                1. 严格使用上述提供的可用工具。
                2. 只输出严格的JSON格式，不要任何解释、不要标题、不要文字说明、只返回JSON格式的数据！"""),
            ("human", "我的目标是：{goal}")
        ]).partial(
            format_instructions=self.parser.get_format_instructions()
        ) | self.llm | self.parser

    def generate_plan(self, goal: str) -> PlanOutput:
        tools_desc = "\n".join([f"-{t.name}: {t.description}-" for t in tools])

        result = self.prompt.invoke({
            "goal": goal,
            "tools": tools_desc
        })
        return PlanOutput(**result)


# ===================== 执行器 =====================
class Executor:
    def __init__(self, tools: List):
        self.tools = {t.name: t for t in tools}

    def execute_plan(self, plan: PlanOutput, max_retries=2) -> Dict[str, Any]:
        results = {}
        for step in plan.steps:
            tool = self.tools.get(step.tool_name)
            if not tool:
                raise ValueError(f"未知工具: {step.tool_name}")

            retry_count = 0
            while retry_count <= max_retries:
                try:
                    args = self._resolve_args(step.arguments, results)
                    res = tool.invoke(args)
                    results[step.id] = res
                    print(f"✅ 步骤 {step.id}: {res}")
                    break
                except Exception as e:
                    retry_count += 1
                    print(f"❌ 步骤 {step.id} 失败 ({retry_count}/{max_retries+1}): {e}")
                    if retry_count > max_retries:
                        raise RuntimeError(f"步骤 {step.id} 执行失败: {e}")
        return results

    def _resolve_args(self, args: Dict, results: Dict) -> Dict:
        resolved = {}
        for k, v in args.items():
            if isinstance(v, str) and v.startswith("{{") and v.endswith("}}"):
                ref_id = int(v.strip("{}"))
                resolved[k] = results.get(ref_id, "")
            else:
                resolved[k] = v
        return resolved


# ===================== 主流程 =====================
def run_plan_and_execute(goal: str):
    planner = Planner()
    executor = Executor(tools)

    try:
        plan = planner.generate_plan(goal)
        print("\n📋 生成的执行计划：")
        for step in plan.steps:
            print(f"  {step.id}. {step.description} → {step.tool_name}({step.arguments})")

        results = executor.execute_plan(plan)
        return {"status": "success", "results": results}
    except Exception as e:
        return {"status": "error", "message": str(e)}

# ===================== 运行 =====================
if __name__ == "__main__":
    goal = "帮我安排一次从北京到上海的出差：先查上海天气，然后订明天的机票，最后发邮件通知经理"
    output = run_plan_and_execute(goal)
    print("\n🎯 最终结果：", output)

Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

Flink 系列第13篇：Flink 生产环境中的并行度与资源配置

本文系统介绍了 Flink 计算资源管理机制，重点解析了 Task Slot 的内存隔离特性、Slot 共享优化和 Operator Chain 机制。详细阐述了并行度的四层设置方式及其优先级，推荐生产环境采用算子级并行度配置。通过源码分析揭示了资源申请与分配的核心流程，包括 Scheduler、SlotPool 和 JobMaster 的协作机制。最后给出生产环境配置建议：合理设置 Slot 内

avatar 2048 AI社区
cover

Gemini 3.1 Pro技术架构深度拆解：200万token上下文背后的工程秘密

avatar 2048 AI社区

2026扩大健康领域的人工智能应用报告：AI医疗落地缓慢的真相是“治理缺位”｜附60+报告、数据下载

原文出处：拓端抖音号@拓端tecdat。

avatar 2048 AI社区