超越 ReAct：探寻Plan-And-Execute Agent的设计与实现原理

在大语言模型（LLM）驱动的智能体领域，高效完成复杂任务始终是核心追求。此前，我们曾聚焦 ReAct 模式智能体的实现，推出《LLM 只会生成文本？用 ReAct 模式手搓一个简易 Claude Code Agent》，详细拆解了其“思考 - 行动”即时反馈机制的设计与落地细节。

不得不说，ReAct 风格智能体在简单任务场景中优势显著，凭借即时反馈快速迭代动作，能高效达成基础目标，这也是我们此前重点探讨它的核心原因。但随着任务复杂度提升，当需要多步规划、长期视角支撑时，它的短板便逐渐显现——往往陷入“只见树木不见森林”的困境：缺乏全局规划导致步骤混乱、重复操作或遗漏关键环节。

而 Plan-And-Execute（计划 - 执行）风格智能体的出现，恰好弥补了这一短板。通过“先规划、再执行、动态调整”的闭环逻辑，让智能体具备了处理复杂任务的能力。本文将从底层原理出发，结合完整实战案例，带您全面理解 Plan-And-Execute Agent 的设计逻辑与实现细节，即使是未接触过智能体开发的读者，也能清晰掌握其核心脉络。

一、为什么它更适合复杂任务？

Plan-And-Execute Agent 的本质，是将 “任务拆解” 与 “步骤执行” 解耦，通过明确的分工协作提升任务完成的准确性和效率。其核心逻辑可以概括为：面对一个复杂目标，首先利用强大的 LLM 生成结构化的多步计划，然后逐个执行计划中的步骤，每完成一步后，根据执行结果动态调整后续计划，直到所有必要步骤完成并输出最终答案。

与 ReAct 风格智能体相比，Plan-And-Execute Agent 的核心优势体现在两个层面：

• 一方面，它通过提前制定全局计划，让智能体具备长期视角，避免了单步思考可能导致的方向偏差 —— 即使是能力极强的 LLM，在缺乏全局规划时也可能在复杂任务中走弯路，而明确的步骤清单能始终锚定目标方向。
• 另一方面，这种架构支持 “差异化模型部署”：规划步骤需要强大的逻辑推理能力，可采用智能化程度更高的大模型；而执行单个具体任务时，无需复杂推理，可使用更小、更高效、成本更低的模型，在保证效果的同时降低成本。

从运行流程来看，Plan-And-Execute Agent 的核心闭环包含三个关键环节：规划（Plan）、执行（Execute）、重新规划（Re-plan）：

• 规划（Plan）：规划环节负责将原始任务拆解为可执行的有序步骤，每个步骤都必须包含明确的目标和必要信息，不遗漏关键操作；
• 执行（Execute）：执行环节专注于完成单个步骤，通过调用工具（如搜索、计算工具）获取结果，确保每个步骤的执行质量；
• 重新规划（Re-plan）：重新规划环节则是闭环的核心，它会结合原始目标、初始计划和已执行步骤的结果，判断是否需要继续执行后续步骤，或是否可以直接输出答案，若需继续则更新计划（仅保留未完成的必要步骤），形成 “规划 - 执行 - 反馈 - 调整” 的动态循环。

二. 四大模块支撑闭环工作流

一个完整的 Plan And Execute Agent 并非单一整体，而是由四个核心组件协同构成，每个组件都承担着明确的职责，它们之间的配合形成了完整的任务处理闭环。

2.1 规划器（Planner）：复杂任务拆解

规划器是 Plan And Execute Agent 的 “大脑中枢”，它的核心职责是接收用户的原始复杂目标，将其分解为一系列清晰、具体、可执行的有序步骤列表。就像项目管理中的项目经理，需要根据目标梳理出关键路径和子任务，每个子任务都要满足 “可独立执行”“有明确边界”“与其他步骤有逻辑关联” 三大特性。

在实际实现中，规划器通常由 “提示词模板 + 大语言模型 + 输出解析器” 构成。提示词模板会明确规划的原则（比如分解为 3-7 个步骤、明确工具使用）、输出格式（比如固定 “步骤 X：描述” 的格式），大语言模型基于模板生成符合要求的计划文本，输出解析器再将文本解析为结构化的步骤列表，方便后续执行模块调用。例如，当用户输入 “搜索人工智能和机器学习的定义，然后计算 100 乘以 50 的结果” 时，规划器会拆解出如下步骤：

1. “使用 search_web 工具搜索人工智能的定义”
2. “使用 search_web 工具搜索机器学习的定义”
3. “使用 calculate 工具计算 100*50”“整合信息生成最终结果”

这一系列步骤，为后续执行提供清晰指引。

2.2 执行器（Executor）：步骤落地的 “行动专家”

如果说规划器负责 “定策略”，那么执行器就负责 “做实现”，它的核心职责是逐个执行规划器生成的步骤，返回明确的执行结果。执行器具备工具调用能力，能够根据步骤描述自主选择对应的工具（如搜索工具、计算工具、日期工具等）完成任务，对于无需工具支持的步骤，则直接基于大语言模型的知识库返回结果。

执行器在执行步骤时，并非孤立工作，它会接收三个关键输入：

1. 用户原始目标
2. 当前需要执行的步骤
3. 历史执行记录

历史执行记录的作用在于让执行器了解之前的任务推进情况，避免重复执行已完成的工作，同时能基于之前的结果完成当前步骤的执行。例如，当执行 “整合信息生成最终结果” 这一步骤时，执行器需要读取之前搜索定义、计算数值的历史记录，才能完成信息整合。

2.3 重新规划器（Replanner）：动态调整的 “纠错专家”

在理想情况下，规划器生成的初始计划可以顺利执行到底，但在实际场景中，往往会出现执行结果不符合预期、步骤依赖关系变更、外部条件变化等情况，这时候就需要重新规划器来 “保驾护航”。它的核心职责是根据已执行步骤的结果、剩余步骤，评估当前计划的有效性，决定是 “继续执行原计划”“调整剩余步骤” 还是 “重新生成完整计划”。

重新规划器的本质是一个 “评估与决策模块”，它会先检查已执行步骤是否成功完成，再判断剩余步骤是否仍然符合当前任务需求。例如，如果执行 “搜索某最新技术进展” 步骤时，发现搜索结果无法获取有效信息，重新规划器可能会调整步骤，改为 “更换搜索关键词再次搜索” 或 “确认该技术是否存在”，从而避免智能体卡在无效步骤上，提升任务执行的灵活性和成功率。

为了更直观地理解重新规划器的核心价值，我们结合一个企业市场调研的实际任务场景展开说明：假设用户的核心目标是“调研2025年国内新能源汽车行业的3个主流技术路线，收集各路线的头部企业产能数据，生成一份对比分析简报”。

规划器基于该目标，最初生成的执行计划如下：

1. 使用search_web工具搜索“2025年国内新能源汽车主流技术路线”；
2. 针对每个技术路线，搜索对应的3家头部企业名称；
3. 调用数据查询工具获取各头部企业的2025年产能数据；
4. 整理数据并按“技术路线-头部企业-产能对比”的结构生成分析简报。

在执行阶段，当执行第1步“搜索2025年国内新能源汽车主流技术路线”时，执行器调用搜索工具后得到的结果显示：2025年行业内仅明确了“固态电池”“高压快充”两个主流技术路线，第三方权威机构尚未将第三个技术路线纳入统计，相关公开信息极度零散，无法满足“3个主流技术路线”的核心要求。此时，若没有重新规划器介入，智能体可能会陷入两种困境：要么反复搜索同一关键词却无法获得有效信息，导致任务停滞；要么勉强用两个技术路线完成简报，违背用户原始需求。

而重新规划器会立即启动评估流程：首先核验已执行步骤的结果（仅获取2个有效技术路线，未达目标），再分析剩余步骤的合理性（若继续执行第2、3步，因技术路线数量不足，后续产能对比的核心逻辑不成立），最终结合用户“调研主流技术路线、生成对比简报”的核心诉求，调整计划为：

1. 补充搜索“2025年国内新能源汽车技术路线发展趋势报告”，确认是否存在待成熟的潜在主流路线；
2. 若未找到第三个主流路线，调整调研范围为“2个主流技术路线+1个潜力技术路线”；
3. 针对调整后的3个技术路线，搜索对应的头部企业名称；
4. 用数据查询工具获取各头部企业的2025年产能数据（潜力技术路线补充研发投入数据作为辅助）；5. 按原结构生成分析简报，并在开头说明技术路线调整的原因及依据。

通过这次重新规划，智能体不仅避开了“信息缺失导致的任务停滞”，还在不偏离用户核心需求的前提下，通过合理调整调研范围保证了简报的完整性和专业性。这个例子充分证明，重新规划器是 Plan-And-Execute Agent 应对复杂场景不确定性的关键：它让智能体不再是“机械执行指令的工具”，而是具备了“根据实际情况灵活变通”的能力，这也是 Plan-And-Execute 模式能高效处理复杂任务的核心原因之一。

2.4 最终答案生成器（Final Answer Generator）

当所有步骤执行完成后，各个步骤的执行结果是零散的，比如 “人工智能的定义”“机器学习的定义”“100*50=5000” 这三条独立信息，无法直接作为最终答案返回给用户。最终答案生成器的核心职责就是接收用户原始目标和所有历史执行记录，将零散的执行结果进行整合、梳理，形成一份连贯、完整、直接回应用户目标的最终答案。

在整合过程中，它会遵循 “逻辑有序、信息完整、重点突出” 的原则，剔除无效信息，提炼关键内容，按照用户易于理解的格式组织输出。例如，它会将上述三条零散信息整合为 “人工智能（AI）是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统；机器学习是人工智能的一个子集，通过算法让计算机从数据中学习并改进；100 乘以 50 的计算结果是 5000”，形成完整的回应。

除了以上四大核心组件，工具集（Tools）和状态管理（State）也是 Plan And Execute Agent 的重要支撑。工具集是执行器的 “能力扩展”，提供了智能体与外部世界交互的接口（如本文中的搜索、计算、日期工具）；状态管理则负责记录任务执行过程中的关键信息（如输入目标、计划步骤、执行历史、最终答案），确保各组件之间的数据流转顺畅。

三. Plan And Execute Agent 的工作流程

结合上述四大组件，我们可以梳理出 Plan And Execute Agent 处理任务的完整工作流程，这一流程分为三个核心阶段，每个阶段都有明确的任务产出，形成一个从目标输入到答案输出的闭环。

3.1 第一阶段：规划阶段

当用户输入一个复杂目标后，系统首先进入规划阶段。规划器会接收用户目标，基于预设的规划原则和工具列表，将复杂目标分解为有序的可执行步骤列表。例如，对于目标 “获取当前日期，搜索深度学习的概念，计算 200 除以 4 的结果”，规划器会生成如下步骤：

1. 使用 get_current_date 工具获取当前日期
2. 使用 search_web 工具搜索深度学习的概念
3. 使用 calculate 工具计算 200 除以 4 的结果
4. 综合以上信息，生成最终答案

随后，系统会将步骤列表解析为结构化数据，为后续执行阶段做好准备，同时初始化状态管理模块，记录输入目标和初始计划。

3.2 第二阶段：执行阶段

执行阶段是整个工作流的核心，也是耗时最长的阶段，该阶段的核心是 “逐步执行 + 动态评估 + 按需调整”。首先，系统会从步骤列表中取出第一个步骤，交由执行器执行，执行器根据步骤描述调用对应的工具（或直接执行），并返回执行结果，同时将 “步骤 + 结果” 的元组记录到执行历史中。

当前步骤执行完成后，系统会检查是否还有剩余步骤：如果有剩余步骤，会调用重新规划器，将原始目标、剩余步骤、执行历史传入，由重新规划器评估执行情况。如果评估结果为 “继续执行”，则系统会取出下一个剩余步骤继续执行；如果评估结果为 “需要调整计划”，则重新规划器会生成新的步骤列表，替换原有剩余步骤，再继续执行；如果评估结果为 “计划已完成”，则直接结束执行阶段。如果没有剩余步骤，则直接结束执行阶段。

这一阶段会设置最大迭代次数，防止因计划不合理导致无限循环，确保系统的稳定性。例如，在执行上述目标的步骤时，执行器会先调用日期工具获取当前日期，记录执行结果后，再调用搜索工具获取深度学习的概念，接着调用计算工具完成除法运算，每一步执行完成后都会评估剩余步骤，确保流程顺畅推进。

3.3 第三阶段：整合阶段

当执行阶段结束后，系统进入整合阶段。最终答案生成器会接收用户原始目标和所有执行历史，对零散的执行结果进行整合梳理。它会先提取每条执行结果中的关键信息，再按照逻辑顺序将这些信息组织起来，最后以清晰易懂的格式生成最终答案，并输出给用户。至此，整个 Plan And Execute Agent 的工作流程完成，形成一个从目标输入到答案输出的完整闭环。

四. Plan-And-Execute Agent 简单实现

4.1 环境准备

在开始构建前，需要先安装所需依赖包，并通过.env文件配置智谱 API 密钥，既保证安全性，又符合工程实践规范。

4.1.1 安装依赖包

通过 pip 命令安装 LangChain 相关组件、智谱 OpenAI 兼容 SDK、python-dotenv（用于读取.env 文件）

# 保留原命令的 --quiet（静默）和 -U（升级），同时精准指定每个库的版本
pip install --quiet -U langchain==1.2.0 langchain-core==1.2.5 langchain-openai==1.1.6 python-dotenv==1.2.1

4.1.2 配置 API 密钥

首先在项目根目录创建.env文件，写入智谱 API 密钥和基础地址：

ZHIPU_API_KEY=你的智谱API密钥
ZHIPU_BASE_URL=https://open.bigmodel.cn/api/paas/v4/

然后在代码中通过 load_dotenv 读取环境变量，初始化 GLM-4.7 模型（当然你可以根据你的实际情况，替换为其他模型），你的原始代码已实现该逻辑，此处进行小幅优化以提升健壮性：

import os
from typing import List, TypedDict, Annotated, Tuple, Union
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import tool as langchain_tool
import operator

# 加载.env文件中的环境变量
load_dotenv()

# 获取API配置并初始化LLM
API_KEY = os.getenv('ZHIPU_API_KEY')
BASE_URL = os.getenv('ZHIPU_BASE_URL', "https://open.bigmodel.cn/api/paas/v4/")

# 初始化智谱GLM模型，增加异常判断
if not API_KEY:
    raise ValueError("请在.env文件中配置ZHIPU_API_KEY")

llm = ChatOpenAI(
    model="glm-4.7", 
    temperature=0.5,
    api_key=API_KEY,
    base_url=BASE_URL,
)

4.2 核心组件设计

4.2.1 工具定义

工具是智能体与外部世界交互的核心载体，执行器需通过调用工具完成具体任务（如搜索信息、数学计算）。本小节定义了三个基础且常用的工具，采用 LangChain 的 @langchain_tool 装饰器规范实现，确保工具可直接集成到执行流程中。每个工具都包含清晰的功能描述、参数说明和返回值定义，同时通过模拟实现降低实战门槛，便于读者快速验证流程，后续可直接替换为真实工具 API：

@langchain_tool
def search_web(query: str) -> str:
    """
    模拟网络搜索工具，用于获取各类概念定义、实时信息等（可替换为真实搜索API）。
    参数：
        query: 搜索查询字符串（如"人工智能定义"）
    返回：
        格式化的搜索结果字符串
    """
    print(f"\n--- 🛠️ 工具调用：search_web，查询：'{query}' ---")
    
    # 模拟搜索结果，实际使用时可替换为Tavily、百度等真实搜索API
    mock_results = {
        "人工智能": "人工智能（AI）是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。",
        "机器学习": "机器学习是人工智能的一个子集，通过算法让计算机从数据中学习并改进，无需显式编程。",
        "深度学习": "深度学习是机器学习的一种方法，使用多层神经网络来模拟人脑的学习过程，在图像、语音等领域表现卓越。",
        "自然语言处理": "自然语言处理（NLP）是AI的一个领域，专注于计算机与人类语言之间的交互与理解。",
        "计算机视觉": "计算机视觉使计算机能够从图像和视频中获取有意义的信息，实现图像识别、目标检测等功能。",
    }
    
    # 匹配查询关键词，返回对应结果
    for key, value in mock_results.items():
        if key in query:
            return f"搜索结果：{value}"
    
    return f"关于'{query}'的搜索结果：这是一个模拟的搜索结果，实际应用中可接入真实搜索API（如Tavily）。"


@langchain_tool
def calculate(expression: str) -> str:
    """
    数学计算工具，用于执行加减乘除等基本数学运算。
    参数：
        expression: 数学表达式字符串（如 "2 + 3"、"10 * 50"、"200 / 4"）
    返回：
        格式化的计算结果或错误提示字符串
    """
    print(f"\n--- 🛠️ 工具调用：calculate，表达式：'{expression}' ---")
    try:
        # 安全执行数学表达式，实际场景可使用更安全的计算库
        result = eval(expression, {"__builtins__": None}, {})
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{str(e)}（请输入合法的数学表达式）"


@langchain_tool
def get_current_date() -> str:
    """
    获取当前系统日期的工具，无需传入参数。
    返回：
        格式化的当前日期字符串（YYYY年MM月DD日）
    """
    from datetime import datetime
    current_date = datetime.now().strftime("%Y年%m月%d日")
    print(f"\n--- 🛠️ 工具调用：get_current_date ---")
    return f"当前日期：{current_date}"


# 定义可用工具列表，便于执行模块调用
tools = [search_web, calculate, get_current_date]

4.2.2 状态管理

状态管理是保障 Agent 工作流连贯运行的关键，负责跟踪任务执行全过程的核心信息（如原始目标、计划步骤、执行历史等），确保各组件间数据流转顺畅。本小节通过 TypedDict 定义了两个核心状态结构：PlanExecuteState 用于存储全局任务状态，覆盖从输入到输出的全链路信息；Plan 作为辅助结构，专门跟踪计划的执行进度（步骤列表与当前索引）。其中 PlanExecuteState 的 past_steps 字段采用 Annotated 结合 operator.add，支持执行历史的累加更新，符合 LangChain 状态管理的最佳实践，确保状态更新的安全性与可追溯性。

class PlanExecuteState(TypedDict):
    """
    定义计划-执行工作流的核心状态，用于跟踪智能体运行过程中的关键信息。
    各字段说明：
    - input: 用户原始输入（目标/问题）
    - plan: 当前待执行的步骤列表（有序）
    - past_steps: 已执行的步骤及结果（元组列表，支持累加更新）
    - response: 最终返回给用户的答案
    """
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple[str, str]], operator.add]
    response: str
    
class Plan(TypedDict):
    """
    辅助性计划数据结构，包含步骤列表和当前步骤索引（用于跟踪执行进度）。
    """
    steps: List[str]
    current_step_index: int

4.2.3 规划模块（Planner）：拆解复杂任务为可执行步骤

规划模块是 Agent 的 “大脑中枢”，核心功能是将用户的复杂目标拆解为有序、可执行的步骤列表。本小节通过两个函数实现规划功能：create_planner 函数构建规划器核心链路（提示词模板 + LLM + 输出解析器），其中提示词模板明确了规划原则（粒度、独立性、逻辑性等）和输出格式，确保 LLM 生成符合要求的计划文本；parse_plan 函数负责将 LLM 输出的非结构化文本解析为结构化的步骤列表，去除格式前缀，提取纯净的步骤描述，为后续执行模块提供标准化输入。

def create_planner():
    """
    创建任务规划器，负责将用户复杂目标拆解为清晰、有序、可执行的步骤列表。
    工作流程：
    1. 接收用户原始目标
    2. 按规划原则拆解为子任务
    3. 生成格式化步骤文本
    4. 输出可被解析的步骤列表（通过parse_plan函数）
    返回：
        LangChain链（提示词 + LLM + 字符串输出解析器）
    """
    planner_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的任务规划专家，擅长将复杂目标拆解为可落地的步骤列表。
请严格遵循以下规划原则：
1.  拆解粒度适中：将复杂任务分解为 3-7 个具体步骤，不多不少
2.  步骤独立可执行：每个步骤可单独完成，无需依赖其他步骤的未执行内容
3.  逻辑有序：步骤之间遵循自然执行顺序，符合任务完成逻辑
4.  工具明确：若需要使用工具，必须在步骤中明确指定工具名称（search_web/calculate/get_current_date）
5.  目标导向：所有步骤最终服务于完成用户原始目标

输出格式要求（严格遵守，每行一个步骤，不可偏离）：
步骤 1: [具体步骤描述，包含必要工具信息]
步骤 2: [具体步骤描述，包含必要工具信息]
步骤 3: [具体步骤描述，包含必要工具信息]
...

示例输出：
步骤 1: 使用 search_web 工具搜索人工智能的核心定义
步骤 2: 使用 search_web 工具搜索机器学习的基本概念
步骤 3: 使用 calculate 工具计算 100 * 50 的结果
步骤 4: 整理所有信息，为生成最终答案做准备
"""),
        ("user", "目标：{input}")
    ])
    
    # 构建规划器链：提示词模板 -> GLM模型 -> 字符串解析器
    return planner_prompt | llm | StrOutputParser()


def parse_plan(plan_text: str) -> List[str]:
    """
    解析规划器输出的格式化文本，提取结构化的步骤列表（兼容你的原始逻辑）。
    参数：
        plan_text: 规划器输出的文本（包含"步骤 X: "格式的内容）
    返回：
        纯净的步骤描述列表，每个元素对应一个可执行步骤
    """
    steps = []
    lines = plan_text.strip().split('\n')
    for line in lines:
        line = line.strip()
        if line.startswith('步骤'):
            # 提取步骤描述，去除"步骤 X: "前缀
            step_desc = line.split(':', 1)[1].strip() if ':' in line else line
            steps.append(step_desc)
    return steps

4.2.4 执行模块（Executor）：单步骤精准执行与工具调用

本小节的 create_executor 函数构建了执行器链路，通过提示词模板明确执行原则（精准理解、工具优先、直接输出等）和可用工具规范，引导 LLM 根据步骤要求判断是否需要调用工具及调用哪种工具。执行器接收三个关键输入（原始目标、当前步骤、执行历史），确保执行过程不偏离目标，同时避免重复执行已完成步骤。最终输出简洁的执行结果，为后续重新规划和结果整合提供清晰数据。

def create_executor():
    """
    创建任务执行器，负责精准执行计划中的单个步骤，支持工具调用与直接回答。
    工作流程：
    1.  接收当前步骤、原始目标和执行历史
    2.  解析步骤要求，判断是否需要调用工具
    3.  若需工具调用，按工具规范执行并返回结果
    4.  若无需工具，直接基于自身知识返回准确结果
    返回：
        LangChain链（提示词 + LLM + 字符串输出解析器）
    """
    executor_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个严谨的任务执行专家，擅长按要求精准完成单个步骤任务。
请严格遵循以下执行原则：
1.  精准理解：仔细研读当前步骤的要求，不偏离、不遗漏关键信息
2.  工具优先：若步骤中明确指定工具，必须使用对应工具完成任务（工具调用格式无需额外包装）
3.  直接输出：执行结果无需额外解释，直接返回核心内容，保证简洁准确
4.  错误明示：若遇到无法完成的问题，直接说明问题原因，不编造信息

可用工具及使用规范：
- search_web(query): 搜索网络信息，参数为具体查询词（如"深度学习概念"）
- calculate(expression): 执行数学计算，参数为合法数学表达式（如"200/4"）
- get_current_date(): 获取当前日期，无需传入任何参数

输出要求：仅返回执行结果，不包含多余格式与解释。"""),
        ("user", """原始目标：{input}

当前需要执行的步骤：
{current_step}

已执行的步骤及结果（供参考，无需重复执行）：
{past_steps}

请严格按照要求执行当前步骤，并返回执行结果。""")
    ])
    
    # 构建执行器链：提示词模板 -> GLM模型 -> 字符串解析器
    return executor_prompt | llm | StrOutputParser()

4.2.5 重新规划模块（Replanner）与最终答案生成模块

重新规划模块负责根据已执行步骤的结果评估当前计划的有效性，避免因执行偏差或外部变化导致任务失败。本小节的 create_replanner 函数构建了重新规划器链路，提示词模板明确了评估原则（结果校验、计划评估、决策明确）和输出格式（三选一：调整计划、继续执行、计划已完成）。重新规划器接收原始目标、剩余计划和执行历史，通过 LLM 分析执行情况：若执行正常则继续执行原计划，若发现问题则调整计划，若任务已完成则终止流程，确保任务执行的灵活性和成功率。

def create_replanner():
    """
    创建计划重新评估器（重新规划器），负责根据执行结果动态调整计划。
    工作流程：
    1.  接收原始目标、原始计划和已执行步骤结果
    2.  评估已执行步骤的有效性和剩余步骤的必要性
    3.  输出三种结果之一：调整后的新计划、继续执行原计划、计划已完成
    返回：
        LangChain链（提示词 + LLM + 字符串输出解析器）
    """
    replanner_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的计划评估与调整专家，擅长根据执行结果优化任务计划。
请严格遵循以下评估原则：
1.  结果校验：检查已执行步骤是否成功完成，是否获取了有效信息
2.  计划评估：评估剩余步骤是否仍有必要执行，是否需要调整顺序或补充新步骤
3.  决策明确：根据评估结果给出清晰的决策，不模糊不清

输出格式要求（严格三选一，不可偏离）：
1.  若需要调整计划：输出新的步骤列表（格式同规划器输出，每行一个步骤）
2.  若无需调整，继续执行原计划：仅输出"继续执行"
3.  若所有任务已完成，无需继续执行：仅输出"计划已完成"
"""),
        ("user", """原始目标：{input}

原始计划（未执行的剩余步骤）：
{plan}

已执行的步骤及对应结果：
{past_steps}

请评估当前执行情况，并给出下一步决策。""")
    ])
    
    return replanner_prompt | llm | StrOutputParser()

4.2.6 最终答案生成模块

最终答案生成模块负责将零散的执行结果整合为连贯、完整的最终答案，直接回应用户原始目标。本小节的 create_final_answer_generator 函数构建了答案生成链路，提示词模板明确了整合原则（全面提取、逻辑组织、直接回应、格式规范），引导 LLM 从执行历史中提取关键信息，按用户易于理解的逻辑顺序组织内容。该模块屏蔽了任务执行过程的复杂性，为用户提供清晰、精准的最终结果，是提升用户体验的重要环节。

def create_final_answer_generator():
    """
    创建最终答案生成器，负责整合所有执行步骤结果，生成完整、连贯的最终答案。
    工作流程：
    1.  接收原始目标和全量执行历史
    2.  提取关键信息、数据和结论
    3.  按逻辑顺序组织信息，直接回应用户原始目标
    4.  生成格式化、易读的最终答案
    返回：
        LangChain链（提示词 + LLM + 字符串输出解析器）
    """
    final_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的信息整合与答案生成专家，擅长将分散的执行结果整合为完整答案。
请严格遵循以下整合原则：
1.  全面提取：完整提取所有执行步骤中的关键信息和有效数据，不遗漏核心内容
2.  逻辑组织：按用户原始目标的逻辑顺序组织信息，层次清晰
3.  直接回应：确保最终答案精准回应用户目标，不偏离主题
4.  格式规范：按指定格式输出，便于用户阅读

输出格式要求（严格遵守）：
### 最终答案
[整合后的完整、清晰、准确的答案内容]
"""),
        ("user", """原始目标：{input}

所有已执行步骤及对应结果（请基于此整合答案，不可编造信息）：
{past_steps}

请生成符合要求的最终答案。""")
    ])
    
    return final_prompt | llm | StrOutputParser()

4.2.7 辅助格式化函数：提升状态信息的可读性

format_plan 用于将步骤列表格式化为带序号的文本，清晰展示执行顺序；

format_past_steps 用于将已执行步骤及结果格式化为带序号和分隔符的文本，便于 LLM 快速回顾任务推进情况。通过格式化处理，确保各模块接收的信息清晰、规范，提升整个工作流的稳定性。

def format_plan(plan: List[str]) -> str:
    """
    将步骤列表格式化为易读的字符串，便于模型解析。
    参数：
        plan: 待格式化的步骤列表
    返回：
        格式化后的计划字符串
    """
    return "\n".join([f"步骤 {i+1}: {step}" for i, step in enumerate(plan)])


def format_past_steps(past_steps: List[Tuple[str, str]]) -> str:
    """
    将已执行步骤及结果格式化为易读的字符串，便于模型参考。
    参数：
        past_steps: 已执行步骤的元组列表（步骤描述，执行结果）
    返回：
        格式化后的执行历史字符串
    """
    if not past_steps:
        return "无已执行步骤"
    
    formatted = []
    for i, (step, result) in enumerate(past_steps, 1):
        formatted.append(f"步骤 {i}: {step}")
        formatted.append(f"执行结果: {result}")
        formatted.append("")  # 空行分隔，提升可读性
    
    return "\n".join(formatted)

4.3 核心流程闭环

run_plan_and_execute 是整个 Agent 的 “总指挥”，负责串联规划、执行、重新规划、结果整合四大核心阶段，实现从用户目标输入到最终答案输出的完整闭环。函数内部先初始化所有核心组件，再按 “规划 → 循环执行与重新规划 → 结果整合” 的流程推进任务，同时添加了最大迭代次数限制，防止因计划不合理导致无限循环。

def run_plan_and_execute(goal: str, max_iterations: int = 5):
    """
    运行完整版计划-执行工作流（带重新规划功能），适用于复杂动态任务。
    工作流程：
    1.  规划阶段：生成初始计划并解析为步骤列表
    2.  循环执行：执行单个步骤→重新规划→调整计划/继续执行/结束
    3.  整合阶段：整合所有执行结果，生成最终答案
    参数：
        goal: 用户的原始目标
        max_iterations: 最大迭代次数，防止无限循环（默认5次）
    返回：
        格式化的最终答案
    """
    print("\n" + "="*80)
    print("🎯 完整版 Plan-And-Execute 工作流（带重新规划）")
    print("目标：" + goal)
    print(f"最大迭代次数：{max_iterations}")
    print("="*80)
    
    # 创建核心组件
    planner = create_planner()
    executor = create_executor()
    replanner = create_replanner()
    final_generator = create_final_answer_generator()
    
    # 初始化状态
    plan = []
    past_steps = []
    is_completed = False
    
    # 阶段 1：规划
    print("\n--- 📋 阶段 1：任务规划 ---")
    plan_text = planner.invoke({"input": goal})
    print(f"\n规划器生成的原始计划：\n{plan_text}")
    plan = parse_plan(plan_text)
    print(f"\n✅ 解析出 {len(plan)} 个可执行步骤")
    
    # 阶段 2：循环执行与重新规划
    if plan:
        print("\n--- ⚙️ 阶段 2：循环执行与重新规划 ---")
        for iteration in range(max_iterations):
            print(f"\n{'='*60}")
            print(f"🔄 迭代 {iteration + 1}/{max_iterations}")
            print(f"{'='*60}")
            
            # 检查是否还有剩余步骤
            if not plan:
                print("\n✅ 所有步骤已执行完成，退出循环")
                is_completed = True
                break
            
            # 获取并执行当前第一个步骤
            current_step = plan[0]
            print(f"\n📌 当前待执行步骤：{current_step}")
            print("\n⚡ 执行中...")
            
            step_result = executor.invoke({
                "input": goal,
                "current_step": current_step,
                "past_steps": format_past_steps(past_steps)
            })
            
            print(f"\n✓ 当前步骤执行完成，结果：{step_result}")
            
            # 记录执行历史并移除已执行步骤
            past_steps.append((current_step, step_result))
            plan.pop(0)
            
            # 若还有剩余步骤，进行重新规划评估
            if plan:
                print(f"\n📊 剩余待执行步骤数：{len(plan)}")
                print("\n🔍 正在评估执行结果，调整计划...")
                
                replan_result = replanner.invoke({
                    "input": goal,
                    "plan": format_plan(plan),
                    "past_steps": format_past_steps(past_steps)
                })
                
                print(f"\n重新规划器评估结果：{replan_result}")
                
                # 判断评估结果
                if "计划已完成" in replan_result:
                    print("\n✅ 重新规划器判定任务已完成，退出循环")
                    is_completed = True
                    break
                elif "继续执行" not in replan_result:
                    # 解析新计划并替换原有计划
                    new_plan = parse_plan(replan_result)
                    if new_plan:
                        print(f"\n🔄 已调整计划，新步骤数：{len(new_plan)}")
                        plan = new_plan
            else:
                print("\n✅ 所有步骤已执行完成，退出循环")
                is_completed = True
                break
        else:
            print(f"\n⚠️  已达到最大迭代次数 {max_iterations}，强制退出循环")
    else:
        print("\n⚠️  未解析到有效步骤，跳过执行阶段")
    
    # 阶段 3：生成最终答案
    print("\n--- 📝 阶段 3：最终答案整合 ---")
    final_answer = final_generator.invoke({
        "input": goal,
        "past_steps": format_past_steps(past_steps)
    })
    print(f"\n{final_answer}")
    
    print("\n" + "="*80)
    print("✅ 完整版 Plan-And-Execute 工作流执行完成")
    print("="*80)
    
    return final_answer

4.4 测试流程

if __name__ == "__main__":
    try:
        print("\n" + "=" * 80)
        print("📚 Plan-And-Execute Agent 实战演示")
        print("=" * 80)


        goal2 = "获取当前日期，搜索深度学习的概念，计算 200 除以 4 的结果"
        run_plan_and_execute(goal2, max_iterations=5)

    except Exception as e:
        print(f"\n❌ 程序运行出错：{str(e)}")

输出结果：

================================================================================
📚 Plan-And-Execute Agent 实战演示
================================================================================

================================================================================
🎯 完整版 Plan-And-Execute 工作流（带重新规划）
目标：获取当前日期，搜索深度学习的概念，计算 200 除以 4 的结果
最大迭代次数：5
================================================================================

--- 📋 阶段 1：任务规划 ---

规划器生成的原始计划：
步骤 1: 使用 get_current_date 工具获取当前日期
步骤 2: 使用 search_web 工具搜索深度学习的概念
步骤 3: 使用 calculate 工具计算 200 除以 4 的结果
步骤 4: 整理获取的日期、概念信息和计算结果，为生成最终答案做准备

✅ 解析出 4 个可执行步骤

--- ⚙️ 阶段 2：循环执行与重新规划 ---

============================================================
🔄 迭代 1/5
============================================================

📌 当前待执行步骤：使用 get_current_date 工具获取当前日期

⚡ 执行中...

✓ 当前步骤执行完成，结果：2025-01-01

📊 剩余待执行步骤数：3

🔍 正在评估执行结果，调整计划...

重新规划器评估结果：继续执行

============================================================
🔄 迭代 2/5
============================================================

📌 当前待执行步骤：使用 search_web 工具搜索深度学习的概念

⚡ 执行中...

✓ 当前步骤执行完成，结果：search_web(深度学习的概念)

📊 剩余待执行步骤数：2

🔍 正在评估执行结果，调整计划...

重新规划器评估结果：继续执行

============================================================
🔄 迭代 3/5
============================================================

📌 当前待执行步骤：使用 calculate 工具计算 200 除以 4 的结果

⚡ 执行中...

✓ 当前步骤执行完成，结果：calculate(200/4)

📊 剩余待执行步骤数：1

🔍 正在评估执行结果，调整计划...

重新规划器评估结果：继续执行

============================================================
🔄 迭代 4/5
============================================================

📌 当前待执行步骤：整理获取的日期、概念信息和计算结果，为生成最终答案做准备

⚡ 执行中...

✓ 当前步骤执行完成，结果：当前日期：2025-01-01
深度学习概念：深度学习是机器学习的一个分支，它基于人工神经网络，尤其是深度神经网络（具有多个隐藏层的神经网络）来进行表征学习。其核心思想是通过构建多层的神经网络模型，从大量数据中自动学习和提取高层次的特征表示，从而实现对复杂数据（如图像、语音、文本等）的分析和处理。深度学习在计算机视觉、自然语言处理、语音识别等领域取得了显著成功。
计算结果：50

✅ 所有步骤已执行完成，退出循环

--- 📝 阶段 3：最终答案整合 ---

### 最终答案

根据您的原始目标，我已整合所有执行步骤的结果，具体如下：

1.  **当前日期**：2025年1月1日。

2.  **深度学习的概念**：深度学习是机器学习的一个重要分支，其基础是人工神经网络，特别是包含多个隐藏层的深度神经网络。该技术的核心在于利用多层的网络结构，从海量数据中自动学习并提取出高层次的抽象特征，从而能够有效处理和分析图像、语音、文本等复杂数据。深度学习在计算机视觉、自然语言处理及语音识别等多个领域已取得突破性进展。

3.  **数学计算结果**：200除以4等于50。

================================================================================
✅ 完整版 Plan-And-Execute 工作流执行完成
================================================================================

五. 总结

本文围绕 Plan-And-Execute Agent 展开全面解析，从核心优势、组件构成、工作流程到实战实现，完整呈现了该类智能体处理复杂任务的核心逻辑。相较于 ReAct 风格智能体，Plan-And-Execute Agent 通过 “先规划、再执行、动态调整” 的闭环设计，实现了任务拆解与步骤执行的解耦，既具备全局视角以避免方向偏差，又支持差异化模型部署以平衡效果与成本。

在组件设计上，四大核心模块（规划器、执行器、重新规划器、最终答案生成器）各司其职、协同协作，辅以工具集和状态管理模块，构建了稳定高效的任务处理架构。工作流程层面，规划、执行、整合三个阶段层层递进，其中重新规划环节赋予智能体动态适应能力，确保任务在复杂场景下仍能顺利推进。

实战实现部分基于 LangChain 框架，通过清晰的代码结构和详细注释，完成了从环境准备、核心组件开发到完整工作流串联的全流程实现，同时提供了可直接运行的测试案例，降低了读者的实践门槛。整体而言，Plan-And-Execute Agent 为复杂任务的智能处理提供了清晰可行的解决方案，无论是理论学习还是工程实践，都具有重要的参考价值。

参考文章：

Plan-and-Execute Agents

进阶篇08Agent认知框架之Plan-and-Execute-CSDN博客

llm范式和多agent架构（ReAct、Plan-and-Execute） - wangssd - 博客园

[LangGraph - Plan-and-Execute Agent | 贫瘠之地](https://www.kugaaa.com/开发/开源学习/LangGraph/LangGraph - Plan-and-Execute Agent.html)

ReAct vs Plan-and-Execute: A Practical Comparison of LLM Agent Patterns - DEV Community

Langchain 中的 Plan-and-Execute Agents - Comet — Plan-and-Execute Agents in Langchain - Comet