Plan-and-execute 原理、架构与代码实现

1. 介绍

为挖掘大型语言模型的推理能力，以解决多步复杂推理任务。研究人员（Wei 等人，2022）提出了少样本思维链（CoT）提示法，通过少量人工构建的分步推理示例，能让大型语言模型明确生成推理步骤，提高解决推理任务的准确率。同时，为了省去提示中人工构建示例的操作，一些研究人员（ Kojima 等人，2022）提出零样本思维链（Zero-shot-CoT）：通过将目标问题陈述后附加 “让我们一步步思考” 作为输入提示给大型语言模型，出人意料地让大语言模型取得了与少样本思维链提示法相近的性能。但尽管零样本思维链取得了成功，但它仍存在以下三个缺陷：

计算错误：中间计算过程出错，导致最终答案偏差。
步骤缺失错误：复杂任务中遗漏关键推理环节，逻辑链断裂。
语义误解错误：出现对问题的语义理解错误和推理步骤不连贯等其它错误。

为解决零样本思维链的上述缺陷，尤其是 “步骤缺失” 与 “推理不连贯” 问题，由 Lei Wang 等人于 2023 年提出Plan-and-Solve（简称 PS，计划-求解）框架，其论文《Plan-and-Solve Prompting: Improving Zero-Shot Chain-of-Thought Reasoning by Large Language Models》（论文地址：https://arxiv.org/pd f/230 5.04 091）于 2023 年 5 月发表。Plan-and-Solve架构作为一种改进零样本思维链（ Chain-of-Thought ，CoT）推理的方法。其核心思想是通过引导模型先制定解决问题的计划，再按照计划逐步执行，这一模式显著提高了大模型在复杂任务上的成功率和效率。

2 论文实验

Plan-and-Solve方法本质上是先计划再执行，即先把用户的问题分解成一个个的子任务，然后再执行这些子任务，最后合并输出得到结果。在论文中的实验作法是：简单地将零样本思维链中的 “让我们一步步思考” 替换为 “让我们首先理解问题并制定解决计划，然后按照计划一步步解决问题”。如下图所示：

图：GPT-3 在（a）零样本思维链提示、（b）Plan-and-Solve提示、（c）答案提取提示的示例。

上图中，分别演示了三种提示：（a）零样本思维链提示、（b）Plan-and-Solve提示、（c）答案提取提示的示例。零样本思维链（Zero-shot-CoT）通过 “让我们一步步思考” 促使大语言模型生成多步骤推理，但在问题复杂时仍可能产生错误的推理步骤，最终结果错误。与零样本思维链不同，Plan-and-Solve提示首先要求大语言模型通过生成分步计划并执行该计划来设计解题方案，得出最终正确的结果。答案提取提示在Plan-and-Solve提示的基础上，进一步提炼最终答案的格式：因此，答案（阿拉伯数字形式）是60%。

以下是图片文字的中文翻译：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39


（a）零样本思维链提示
问题（Q）：在一个有 20 名学生的舞蹈班中，20% 的学生报名参加当代舞，剩下学生中的 25% 报名参加爵士舞，其余的报名参加嘻哈舞。整个班级中报名参加嘻哈舞的学生占比是多少？
回答（A）：逐步思考。

首先，班级中 20% 的学生报名参加当代舞，这意味着剩下 80% 的学生。
接下来，剩下 80% 学生中的 25% 报名参加爵士舞，所以这意味着原始班级中 25% 的学生报名参加爵士舞。这样就剩下原始班级的 55%。
最后，班级剩下的（55%）学生报名参加嘻哈舞。所以，整个班级中 55% 的学生报名参加嘻哈舞。（错误答案 ）

（b）Plan-and-Solve提示
问题（Q）：在一个有 20 名学生的舞蹈班中，20% 的学生报名参加当代舞，剩下学生中的 25% 报名参加爵士舞，其余的报名参加嘻哈舞。整个班级中报名参加嘻哈舞的学生占比是多少？
回答（A）：首先，理解问题并制定解决问题的计划。
然后，逐步执行计划并解决问题。

计划（Plan）：
步骤 1：计算报名参加当代舞和爵士舞的学生总数。
步骤 2：计算报名参加嘻哈舞的学生总数。
步骤 3：计算报名参加嘻哈舞的学生占比。

解答（Solution）：
步骤 1：20 名学生的 20% 是 4 名学生，剩下 16 名学生的 25% 是 4 名学生。所以，报名参加当代舞和爵士舞的学生总数是 8 名。
步骤 2：剩下的 12 名学生报名参加嘻哈舞。
步骤 3：报名参加嘻哈舞的学生占比是 12/20 = 60%。（正确答案 ）

（c）Plan-and-Solve答案提取提示
问题（Q）：在一个有 20 名学生的舞蹈班中，20% 的学生报名参加当代舞，剩下学生中的 25% 报名参加爵士舞，其余的报名参加嘻哈舞。整个班级中报名参加嘻哈舞的学生占比是多少？
回答（A）：首先，理解问题并制定解决问题的计划。
然后，逐步执行计划并解决问题。

计划（Plan）：
步骤 1：计算报名参加当代舞和爵士舞的学生总数。
步骤 2：计算报名参加嘻哈舞的学生总数。
步骤 3：计算报名参加嘻哈舞的学生占比。

解答（Solution）：
步骤 1：20 名学生的 20% 是 4 名学生，剩下 16 名学生的 25% 是 4 名学生。所以，报名参加当代舞和爵士舞的学生总数是 8 名。
步骤 2：剩下的 12 名学生报名参加嘻哈舞。
步骤 3：报名参加嘻哈舞的学生占比是 12/20 = 60%。

因此，答案（阿拉伯数字形式）是 60% 。

为解决零样本思维链的计算错误并提升生成的推理步骤质量，我们为 PS 提示法添加了更详细的指令。具体作法是：提示词中增加了 “提取相关变量及其对应数值” 和 “计算中间结果（注意计算和常识）” 的指令。这种提示变体被称为 PS + 提示策略（如下图（b）所示）。

图:（a）Plan-and-Solve提示（PS 提示）和（b）带有更详细指令的Plan-and-Solve提示（PS + 提示）

上图展示了（a）Plan-and-Solve提示（PS 提示）和（b）带有更详细指令的Plan-and-Solve提示（PS + 提示）的示例， PS + 策略仅在提示中增加了 “提取相关变量及其对应数值” 和 “计算中间结果（注意计算和常识）” 的指令，尽管简单，却显著提升了生成的推理过程的质量，并获得正确答案。

论文在三个推理问题类别的十个基准数据集上评估所提出的Plan-and-Solve提示法，包含：在六个数学推理数据集（包括 AQuA（代数应用题数据集）、GSM8K（小学数学应用题数据集）、MultiArith（需要多步推理和运算的数学应用题数据集）、AddSub（加减法算术应用题数据集）、SingleEq（单方程小学代数应用题的数据集）和 SVAMP（ 4 年级及以下学生的单未知数算术应用题基准））；两个常识推理数据集（CommonsenseQA（多项选择题基准数据集）和 StrategyQA（需要多步推理但未给出推理步骤的基准数据集））以及两个符号推理数据集（Last Letter（单词的最后一个字母拼接起来的问题数据集）和 Coin Flip（关于硬币经过翻转或未翻转后是否仍正面朝上的问题数据集））。

总体而言，实验评估结果表明：

零样本 PS 提示法能生成比零样本思维链提示法质量更高的推理过程，因为 PS 提示提供了更详细的指令，引导大型语言模型进行正确推理。
零样本 PS + 提示法在所有推理问题和数据集上均大幅优于零样本思维链。此外，尽管 PS + 提示法不需要人工示例，但在算术推理上的性能与少样本思维链提示法相近，在部分数据集上优于少样本人工思维链提示法。

2. 核心思想

2.1 原论文的`Plan-and-Solve(规划-求解)`提示策略

原论文中提出的Plan-and-Solve(规划-求解)提示策略（简：PS 和 PS + 提示法），是一种新的零样本思维链提示方法，核心是让大语言模型在最终答案前，引导大型语言模型制定解决问题的计划并生成中间推理过程（将整个任务分解为更小的子任务），然后按照计划逐步执行求解，以提高复杂推理任务的准确性和稳定性。其思路通俗来讲，就是让大模型在解决问题时模拟人的过程 “先想后做”，如我们在解决一道复杂的数学题时，不会一下子就写出答案，而是会先在脑海里规划出解题的步骤，比如先分析题目条件、确定使用的公式、分步骤计算等，然后再按照这个规划一步步去计算，最终得到答案。

原论文中提出的Plan-and-Solve(规划-求解)提示策略的实现比较简单，主要是通过构建提示词指令模板，该模板需满足以下两个标准：

模板应能促使大型语言模型生成确定计划（子任务）并完成这些子任务。
模板应引导大型语言模型更关注计算和中间结果，尽可能确保其正确性。

为满足第一个标准，借鉴零样本思维链。在零样本思维链中，输入指令包括触发指令 “让我们一步步思考”。而零样本 PS 提示法包含 “制定计划” 和 “执行计划” 的指令，因此，提示为 “ 让我们首先理解问题并制定解决计划，然后按照计划一步步解决问题。”

为满足第二个标准，在基于计划的触发语句中扩展更详细的指令。具体来说，添加 “注意计算” 到触发句中，要求大型语言模型尽可能准确地进行计算。为减少因缺失必要推理步骤导致的错误，加入 “提取相关变量及其对应数值”，明确指示大型语言模型不要忽略输入问题中的相关信息。此外，在提示中添加 “计算中间结果”，以增强大型语言模型生成相关重要推理步骤的能力。如提示：“先理解问题，提取相关变量与对应数值，制定计划；再执行计划，计算中间结果（关注计算和常识），逐步解题并给出答案。”

总体来说，这种Plan-and-Solve(规划-求解)提示策略，尽管简单，却显著提升了生成的推理过程的质量。

2.2 改进的计划与求解（Plan-and-Solve)框架

随着构建基于大语言模型（LLM）的代理（Agent）在复杂任务中的应用不断深化，涌现了许多适配复杂场景的优质设计模式，其中比较典型的是ReAct设计模式，它通过 “思考（Reasoning）→行动（Action）→观察（Observation）” 的循环机制实现动态决策：先对任务需求进行分析（Reason）并决策出下一步行动，然后执行行动(Action)（如工具调用或信息获取等），再基于行动结果调整策略（Observation），最终逐步逼近目标。

虽然ReAct 对一些任务有效，然而，ReAct 模式存在显著局限 —— 其决策逻辑高度依赖 “局部判断”，每一步仅聚焦单个行动选择，缺乏对任务全局的统筹规划与前瞻性布局。这使得它在复杂任务中易出现两类问题：一是 “路径冗余”，例如处理 “年度经营分析报告生成” 任务时，可能反复调用同一数据工具补充信息；二是 “目标偏移”，比如客服代理在处理用户复合诉求时，因过度关注某一细节问题，偏离了整体需求的解决方向。其在复杂场景下暴露出两个核心缺点：

LLM 调用效率低：每次执行工具调用前，都需触发一次 LLM 调用以确定行动，导致任务流程中 LLM 调用频次过高，不仅增加时间成本，还提升了资源消耗；
全局规划缺失：LLM 仅针对当前子问题进行规划，未对整个任务的完整流程进行系统性推理，易形成次优执行轨迹，难以保障任务最终效果。

通过明确的规划步骤可以克服这两个缺点，借鉴了论文《Plan-and-Solve Prompting: Improving Zero-Shot Chain-of-Thought Reasoning by Large Language Models》中的思想，在 ReAct 的基础上添加 “全局规划” 能力，形成新的plan-and-execute框架，并涌现了类似的框架项目如Baby-AGI，其核心逻辑可概括为 “先全局规划、后分步执行、再动态调整” 的闭环逻辑：借助 LLM 对任务进行整体规划，生成完整分步计划；再按照计划有序执行，并根据执行反馈（如工具调用结果）对计划进行动态优化，形成 “规划 - 执行 - 反馈 - 调整” 的完整链路。既保留了 ReAct 的灵活性，又弥补了其全局统筹能力的不足。

Plan-and-Execute 的架构由规划器（Planner）、执行器（Executor）、重规划器（Replanner） 三部分组成：

规划器（Planner）：接收用户需求后，调用 LLM 进行深度拆解与全局规划，生成完整分步计划。
执行器（Executor）：接收规划器输出的分步计划，按序执行每个子任务。在实现层面，执行器可为具备工具调用能力的独立 Agent— 根据子任务需求自主选择 API 接口、数据检索工具或自动化组件，完成具体操作并返回执行结果（如调用海关数据库获取目标市场关税数据）。
重规划器（Replanner）：负责根据执行器的实际执行情况和信息反馈来调整计划。

Plan-and-Execute框架是在Plan-and-Solve Prompting框架上的扩展。通过以上这三个部分，形成 “规划→执行→反馈→调整” 的完整循环链路，既保证了任务处理的全局统筹性，又赋予系统应对复杂场景的动态适应能力。

3. 工作流程

Plan-and-Execute（规划 - 执行）框架的工作流程图如下：

输入任务：接收来自用户用户或系统提交待处理的任务。
Planner（规划器）：接收任务后，调用模型（LLM）对任务进行深入分析，生成清晰的分步执行计划（即 Task list ）。存储生成的计划列表，作为后续执行的 “行动指南”，让复杂任务变 “可落地步骤”。
Executor（执行器）：按任务列表逐个子任务执行，每次取出任务列表中的第一个任务进行处理，并将返回执行的结果传给Replanner。
Replanner（重规划器）：接收执行结果后，结合任务目标进行评估执行偏差，判断是否已经完成任务，还是需要调整计划。若以完成任务，则生成最终响应；若未完成，则生成调整计划列表。
更新计划清单：更新Replanner生成的调整计划列表。进入2-3循环。
输出结果：任务完成后，输出最终成果（如完整报告、问题解决方案），结束流程。

4. 与其它模式的对比

Plan-and-Execute 架构与类似框架ReAct、ReWOO的核心对比如下：

ReAct：通过 “思考→行动→观察” 的循环来实现动态决策。智能体先对任务需求进行思考分析，做出下一步行动决策并执行，再观察行动的结果进行下一轮的思考分析，逐步逼近目标。但本质是 “局部决策驱动”，每步仅聚焦单个行动选择，缺乏对任务全局的统筹规划，复杂场景下易出现路径冗余或目标偏移，且每次工具调用前均需触发 LLM 调用，Token 消耗与效率成本较高。
ReWOO ：ReWOO 是对 ReAct 框架的优化，主要是引入规划器，将核心组件拆分为三个独立模块：规划器、工作器和求解器。先使用规划器分解任务并制定相互依赖的计划（子任务）列表；再交给工作器依次执行；执行完成后由求解器整合结果。解决了 ReAct 中 “每次执行工具调用前，都需触发一次 LLM 调用以确定行动，导致 LLM 调用频次过高"的问题，从而显著降低 Token 消耗。但生成计划后，计划执行较为刚性，缺乏动态调整机制。若某子任务执行结果与预期偏差（如检索到的信息失效、工具调用失败），无法对原有计划进行调整，易导致 “一步错、步步错”的问题。
Plan-and-Execute 框架：采用 “规划 - 执行 - 反馈 - 调整” 的闭环逻辑。与 ReWOO的关键区别，是新增了 Replan（重规划）机制，弥补了 ReWOO “计划生成后缺乏动态调整机制” 的局限。流程上，规划器先借助 LLM 完成任务全局拆解，生成完整分步计划；执行器按计划调用工具执行任务；重规划器根据执行反馈判断是否调整计划，若需调整则重新规划，确保任务朝着目标推进。由于可根据实时反馈优化计划，更适配复杂多变的任务场景。

6. 代码实现

6.1 架构设计

Plan-and-Solve 的架构由以下几个核心组件构成：

规划器 (Planner)：负责将复杂任务拆解为可落地的多步骤执行计划，为后续行动明确路径。通常由具备强推理能力的大语言模型（LLM）承担。
执行器 (Executor)：负责执行规划器生成的具体计划任务。执行器可以是另一个大语言模型，也可以是一个专门设计的 Agent，如 ReAct 模式的Agent，通过 “思考 - 行动 - 观察” 的循环，自主思考、决策下一步的行动、调用工具（如调用API，查询数据库、代码执行器）来完成计划任务。
重新规划器（replanner）：接收执行器反馈的子任务结果，通过 LLM 评估 “实际结果是否符合计划预期”，若出现偏差（如数据缺失、工具调用失败、新信息出现），则触发重新规划，并更新计划列表。
状态管理(State)：扮演 “记忆中枢”，负责跟踪任务的当前状态，包括当前计划、已执行的步骤及其结果、当前的响应等。状态管理器确保 Agent 在执行过程中能够保持上下文感知能力。

6.2 编写代码

6.2.1 前置准备

环境要求：
- Python 3.11+
- 安装依赖库：pip install openai requests python-dotenv langchain_community langgraph langchain_openai langchain-tavily

基础配置：

设置模型调用API Key

创建.env文件，配置模型 API访问密钥，这里使用千问模型，需配置QWEN_API_KEY及QWEN_BASE_URL。

1
2
3
4
5


# 千问模型接口访问key
# 如何获取API Key：https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key
QWEN_API_KEY="sk-*******"
# 千问模型接口访问地址
QWEN_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"

搜索引擎 API 密钥（这里使用tavily）

.env文件中配置搜索引擎tavily API 密钥，Tavily国内需科学上网才能访问：

1
2
3


# tavily 搜索API KEY
# 访问 https://tavily.com 注册并获取API Key,用户每月有1000次免费调用额度.
TAVILY_API_KEY="tvly-ZuSNSW4CehsNizV****"

6.2.2 引入依赖包

引入相关依赖，读取.env文件中配置的参数转化为环境变量。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


import os
import operator
from typing import Annotated, List, Tuple, Union

# 导入LangChain相关组件：输出解析器、图结构、类型定义等
from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph.graph import CompiledGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict

# 环境变量加载、语言模型接口、提示模板等
from dotenv import load_dotenv
from langchain_core.language_models import BaseChatModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.constants import END
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field

# 导入搜索工具

from tools.web_search.langchain_tavily_search import TavilySearch


load_dotenv()

6.2.3 配置与常量（Config 类）

该类集中管理所有静态配置，便于维护。

1
2
3
4
5
6


class Config:
    DEFAULT_MODEL = "qwen-plus-2025-04-28"  # 推荐使用千问Plus模型（效果优于Turbo）
    QWEN_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"  # 千问API兼容地址
    DEFAULT_TEMPERATURE = 0  # 温度=0,工具调用/规划场景需确定性输出，故设为0;创意场景可设 0.7-1.0
    TAVILY_MAX_RESULTS = 2  # 搜索工具返回的最大结果数（减少冗余信息）
    RECURSION_LIMIT = 10  # 递归限制（防止工作流无限循环，如计划永远执行不完）

6.2.4 定义状态

LangGraph 工作流的核心是状态（State）在节点间的流转，而 state 则定义了节点间传递的数据结构，主要包含三部分：

当前计划列表：用字符串列表记录待执行的步骤；
已执行步骤记录：用元组列表存储（包含步骤内容 + 执行结果）；
基础信息：包括最终响应和原始输入数据。

1
2
3
4
5
6


class PlanExecute(TypedDict):
    input: str  # 用户原始查询（如“2024澳网男单冠军家乡”）
    plan: List[str]  # 待执行的步骤列表（如["查冠军是谁", "查冠军家乡"]）
    # 已执行步骤记录：用Annotated+operator.add实现“累加”（而非覆盖）
    past_steps: Annotated[List[Tuple], operator.add]
    response: str  # 最终返回给用户的答案

6.2.5 LLM 与工具初始化（核心能力封装）

该部分负责定义智能体的 “大脑”（LLM）和 “手脚”（工具），是智能体具备 “思考” 和 “行动” 能力的基础。

（1）initialize_llm：初始化语言模型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


def initialize_llm(model: str = Config.DEFAULT_MODEL, temperature: float = Config.DEFAULT_TEMPERATURE) -> BaseChatModel:
    api_key = os.getenv("QWEN_API_KEY")  # 从.env文件获取千问API密钥
    if not api_key:
        raise ValueError("缺少环境变量QWEN_API_KEY")  # 异常处理：防止密钥缺失
    
    base_url = os.getenv("QWEN_BASE_URL") or Config.QWEN_BASE_URL  # 优先用环境变量的API地址
    
    # 用ChatOpenAI调用千问（千问兼容OpenAI格式）
    return ChatOpenAI(
        model=model,
        api_key=api_key,
        openai_api_base=base_url,
        temperature=temperature
    )

环境变量管理：通过os.getenv从.env文件读取密钥，避免硬编码（安全最佳实践）。
这里使用 LangChain 的ChatOpenAI类来访问千问模型，可以替换为兼容 OpenAI 接口格式的其它模型。替换为其它模型，需要修改这里的api_key（模型API访问密钥）、base_url（模型API地址）、model（模型名称）参数。

（2）initialize_tools：初始化工具

1
2
3
4


def initialize_tools() -> List:
    # 初始化Tavily搜索工具（设置最大结果数）
    search = TavilySearch(max_results=Config.TAVILY_MAX_RESULTS)
    return [search]

这里默认只添加了网络搜索工具Tavily，可根据业务场景添加其它的工具。

6.2.6 定义规划器节点

规划器节点根据用户输入，生成完成任务的计划步骤列表。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


class Plan(BaseModel):
    """未来要遵循的计划步骤"""
    steps: List[str] = Field(description="要遵循的不同步骤，应按顺序排列")


    @staticmethod
    async def plan_step(state: PlanExecute):
        """规划步骤节点：生成初始计划
        根据用户输入生成完成任务所需的步骤计划
        参数:
            state: 当前工作流状态，包含用户输入等信息
        返回:
            包含生成的计划步骤的字典
        """

        # 创建解析器生成格式说明
        parser = PydanticOutputParser(pydantic_object=Plan)

        # 生成初始计划的提示模板，指导模型为目标生成完整的分步计划，确保计划能够直接导向正确答案。
        planner_prompt = ChatPromptTemplate.from_messages([
            (
                "system",
                """该计划应包含单个任务，若执行正确，将得出正确答案。不要添加任何多余步骤。\
                最后一步的结果应为最终答案。确保每个步骤都包含所需的所有信息——不要跳过步骤。\n
                必须严格按照以下格式要求返回数据，不要添加任何额外内容：\n
                {format_instructions}\n\n
                请确保输出完全符合上述格式，字段类型和约束要严格遵守。
                """,  # 添加了JSON格式要求
            ),
            ("placeholder", "{messages}"),
        ])
        # 创建规划器：提示模板 + LLM（带结构化输出）
        planner = planner_prompt | initialize_llm().with_structured_output(Plan)
        # 调用规划器生成计划
        result = await planner.ainvoke({
            "messages": [("user", state["input"])],
            "format_instructions": parser.get_format_instructions()  # 解析器生成格式说明
        })
        # 返回包含计划步骤，将被合并到工作流状态中
        return {"plan": result.steps}

PydanticOutputParser.get_format_instructions()：生成 LLM 可理解的格式说明，指导LLM输出相应的响应格式（如 “输出应为 JSON，包含 steps 字段，类型为列表”），确保 LLM 输出合规。
with_structured_output(Plan)：LangChain 的便捷方法，替代手动解析 LLM 输出（无需自己写 JSON 解析逻辑）。

6.2.7 定义执行器节点

执行器节点从当前计划列表中，取出第一个任务进行执行，并记录执行结果。这里使用ReAct模式的智能体作为执行器。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43


# 创建执行器，用来执行计划。
def _create_agent_executor() -> CompiledGraph:
    """创建执行器,使用React模式创建一个能够使用工具执行具体步骤的智能体。
    """
    llm = initialize_llm()
    tools = initialize_tools()
    prompt = "你是一个专业的助手。"
    # 创建并返回React模式的智能体
    return create_react_agent(llm, tools, prompt=prompt)


@staticmethod
async def execute_step(state: PlanExecute):
	 """执行步骤节点：执行计划中的第一步
        调用智能体执行器处理计划中的第一个步骤，并记录执行结果
        参数:
            state: 当前工作流状态，包含计划步骤等信息
        返回:
            包含已执行步骤记录的字典
    """
    # 1. 获取React智能体执行器
    executor = _create_agent_executor()
    
    # 2. 处理计划：格式化步骤列表，提取第一步任务
    plan = state["plan"]
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))  
    task = plan[0]  # 执行计划的第一个步骤（避免多步骤并行，简化逻辑）
    
    # 3. 格式化任务描述
    task_formatted = f"""对于以下计划：
                        {plan_str}
                        你的任务是执行第1步：{task}。"""
    
    # 4. 调用执行器执行任务（debug=True：输出调试信息，正式环境需关闭）
    agent_response = executor.invoke(
        {"messages": [("user", task_formatted)]},
        debug=True
    )
    
    # 5. 记录已执行步骤（(任务内容, 执行结果)），返回给LangGraph合并状态
    return {
        "past_steps": [(task, agent_response["messages"][-1].content)],
    }

agent_response["messages"][-1].content：React 智能体的输出存储在messages列表中，最后一条消息是最终执行结果。
单步执行：每次只执行计划的第一步，后续通过 “重新规划” 决定下一步，避免步骤混乱。

6.2.8 定义重新规划节点

重新规划节点根据 “已执行步骤结果”，通过 LLM 评估下一步的操作：

直接返回答案：如果已经做完的步骤，已足够信息能得出用户要的结果，则直接生成响应答案给用户。
继续执行新计划：如果还未完成，则重新生成计划，并更新state中的计划列表。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54


class Response(BaseModel):
    """用户响应数据模型"""
    response: str


class Act(BaseModel):
    """要执行的动作,用于在重新规划阶段决定下一步动作：是直接响应还是继续执行计划"""
    action: Union[Response, Plan] = Field(
        description="要执行的动作。若要响应用户，使用Response。若需要进一步使用工具，使用Plan。"
    )


@staticmethod
async def replan_step(state: PlanExecute):
    # 1. 定义重新规划的提示模板
    replanner_prompt = ChatPromptTemplate.from_template("""针对给定目标，制定一个简单的分步计划。
        该计划应包含单个任务，若执行正确，将得出正确答案。不要添加任何多余步骤。
        最后一步的结果应为最终答案。确保每个步骤都包含所需的所有信息——不要跳过步骤。

        你的目标是：
        {input}

        你最初的计划是：
        {plan}

        你目前已完成以下步骤：
        {past_steps}

        相应地更新你的计划。如果无需更多步骤且可以回复用户，则进行回复。否则，填写计划。仅在计划中添加仍需完成的步骤。不要将已完成的步骤作为计划的一部分返回。
        
        必须严格按照以下格式要求返回数据，不要添加任何额外内容：\n
        {format_instructions}\n\n
        请确保输出完全符合上述格式，字段类型和约束要严格遵守。
        """)  # 添加了JSON格式要求
    
    # 2. 创建Act模型的解析器（判断动作类型：Response/Plan）
    parser = PydanticOutputParser(pydantic_object=Act)
    
    # 3. 创建重新规划器：提示模板 + LLM（带结构化输出）
    replanner = replanner_prompt | initialize_llm().with_structured_output(Act)
    
    # 4. 调用重新规划器生成新的动作指令
    output = await replanner.ainvoke({
        "input": state["input"],
        "plan": state["plan"],
        "past_steps": state["past_steps"],
        "format_instructions": parser.get_format_instructions()
    })
    
    # 5. 根据输出的动作类型返回不同的结果
    if isinstance(output.action, Response):
        return {"response": output.action.response}  # 有答案了，返回最终响应
    else:
        return {"plan": output.action.steps}  # 还需继续，返回新计划

6.2.9 定义路由节点

判断工作流是否结束，是 “继续执行” 还是 “终止”。

1
2
3
4


@staticmethod
def should_end(state: PlanExecute) -> str:
    # 若状态中有response（最终答案），则返回END（结束）；否则返回"execute"（继续执行）
    return END if "response" in state and state["response"] else "execute"

6.2.10 定义图（LangGraph 核心）

图定义工作流的 “节点连接关系”，即 “从哪里来，到哪里去”。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


def build_plan_and_execute_workflow() -> CompiledStateGraph:
    # 1. 初始化状态图（指定状态类型为PlanExecute）
    workflow = StateGraph(PlanExecute)
    
    # 2. 添加节点（将Nodes类的静态方法与节点名绑定）
    workflow.add_node("planner", Nodes.plan_step)    # 规划节点
    workflow.add_node("execute", Nodes.execute_step)  # 执行节点
    workflow.add_node("replan", Nodes.replan_step)    # 重新规划节点
    
    # 3. 定义“无条件边”（固定流向）
    workflow.add_edge(START, "planner")       # 起始点 → 规划节点（第一步先规划）
    workflow.add_edge("planner", "execute")   # 规划后 → 执行节点（执行第一步）
    workflow.add_edge("execute", "replan")    # 执行后 → 重新规划节点（判断下一步）
    
    # 4. 定义“条件边”（根据重新规划结果决定流向）
    workflow.add_conditional_edges(
        "replan",                # 源节点：从“重新规划”节点出发
        Nodes.should_end,        # 路由函数：判断流向
        ["execute", END]         # 可能的目标节点：继续执行 或 结束
    )
    
    # 5. 编译工作流（生成可执行的状态图）
    return workflow.compile()

6.2.11 主程序（运行入口）

main函数是代码的执行入口，负责初始化环境、构建工作流。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


async def main():
    # 1. 加载环境变量（.env文件）
    load_environment()
    
    # 2. 构建工作流
    app = build_plan_and_execute_workflow()
    
    # 3. （可选）保存工作流图片（需科学上网）
    # save_grath_image(app)
    
    # 4. 测试查询（用户输入）
    inputs = {"input": "2024年澳大利亚网球公开赛男子单打冠军的家乡是哪里？"}
    
    # 5. 设置递归限制（防止无限循环）
    config = {"recursion_limit": Config.RECURSION_LIMIT}
    
    # 6. 流式运行工作流（实时输出每一步结果）
    async for event in app.astream(inputs, config=config):
        for k, v in event.items():
            if k != "__end__":  # 过滤掉“结束”标记，只输出有效结果
                print(v)

# 运行主程序（异步函数需用asyncio.run启动）
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

6.2.12 完整代码

完整代码位于项目根目录下：cognitive_pattern/plan_and_solve/plan_and_execute.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364


import os
import operator
from typing import Annotated, List, Tuple, Union

# 导入LangChain相关组件：输出解析器、图结构、类型定义等
from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph.graph import CompiledGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict

# 环境变量加载、语言模型接口、提示模板等
from dotenv import load_dotenv
from langchain_core.language_models import BaseChatModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.constants import END
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field

# 导入搜索工具

from tools.web_search.langchain_tavily_search import TavilySearch


load_dotenv()


# ----------------------------
# 配置与常量
# ----------------------------
class Config:
    """应用配置常量"""

    DEFAULT_MODEL = "qwen-plus-2025-04-28"  # 推荐"qwen-plus" 使用qwen-turbo系列效果差
    DEFAULT_TEMPERATURE = 0  # 默认温度参数（0表示确定性输出）
    TAVILY_MAX_RESULTS = 2  # 搜索工具返回的最大结果数
    QWEN_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"  # 千问API基础地址
    RECURSION_LIMIT = 10  # 递归限制，防止无限循环


# ----------------------------
# 状态与数据模型
# ----------------------------
class PlanExecute(TypedDict):
    """智能体运行状态数据结构,用于在工作流的不同节点之间传递状态信息"""
    input: str  # 用户输入的原始查询
    plan: List[str]  # 待执行的计划步骤列表
    # 已执行步骤记录，使用Annotated和operator.add实现状态累加
    past_steps: Annotated[List[Tuple], operator.add]
    response: str  # 最终要返回给用户的响应


class Plan(BaseModel):
    """未来要遵循的计划步骤"""
    steps: List[str] = Field(description="要遵循的不同步骤，应按顺序排列")


class Response(BaseModel):
    """用户响应数据模型"""
    response: str


class Act(BaseModel):
    """要执行的动作,用于在重新规划阶段决定下一步动作：是直接响应还是继续执行计划"""
    action: Union[Response, Plan] = Field(
        description="要执行的动作。若要响应用户，使用Response。若需要进一步使用工具，使用Plan。"
    )


# ----------------------------
# LLM初始化
# ----------------------------
def load_environment():
    """加载环境变量,从.env文件中加载API密钥等敏感信息，避免硬编码"""
    load_dotenv()


def initialize_llm(
        model: str = Config.DEFAULT_MODEL,
        temperature: float = Config.DEFAULT_TEMPERATURE
) -> BaseChatModel:
    """初始化语言模型(LLM)
    根据配置参数创建并返回一个LLM实例，优先使用环境变量中的配置
    参数:
        model: 模型名称
        temperature: 控制输出随机性的参数，0表示最确定，1表示最随机
    返回:
        初始化好的BaseChatModel实例
    异常:
        ValueError: 当缺少必要的API密钥时抛出
    """
    api_key = os.getenv("QWEN_API_KEY")
    if not api_key:
        raise ValueError("缺少环境变量QWEN_API_KEY")

    base_url = os.getenv("QWEN_BASE_URL") or Config.QWEN_BASE_URL

    return ChatOpenAI(
        model=model,
        api_key=api_key,
        openai_api_base=base_url,
        temperature=temperature
    )


def initialize_tools() -> List:
    """初始化工具列表
       创建并返回智能体可以使用的工具集合，目前包含Tavily搜索工具
    """
    # 初始化搜索工具，设置最大结果数
    # 请求网络时，如果是同步则使用requests库发起网络请求，异步使用aiohttp发起网络请求。
    # --同步调用的 requests 库会自动检测系统代理（如环境变量 HTTP_PROXY/HTTPS_PROXY、系统网络设置），
    # --而异步 aiohttp 默认不会。需要自行设置系统代理，实现异步网络请求。
    # 所以这里对langchain默认的TavilySearch做了修改添加自动获取系统代理并配置.
    search = TavilySearch(max_results=Config.TAVILY_MAX_RESULTS)
    return [search]


# 创建执行器，用来执行计划。
def _create_agent_executor() -> CompiledGraph:
    """创建执行器,使用React模式创建一个能够使用工具执行具体步骤的智能体执行器
    """
    llm = initialize_llm()
    tools = initialize_tools()
    prompt = "你是一个专业的助手。"
    # 创建并返回React模式的智能体
    return create_react_agent(llm, tools, prompt=prompt)


# ----------------------------
# 节点函数
# ----------------------------
class Nodes:
    """智能体图节点集合类

    封装工作流中所有节点的处理函数，使代码结构更清晰
    """

    @staticmethod
    async def plan_step(state: PlanExecute):
        """规划步骤节点：生成初始计划

        根据用户输入生成完成任务所需的步骤计划

        参数:
            state: 当前工作流状态，包含用户输入等信息

        返回:
            包含生成的计划步骤的字典
        """

        # 创建解析器生成格式说明
        parser = PydanticOutputParser(pydantic_object=Plan)

        # 生成初始计划的提示模板，指导模型为目标生成完整的分步计划，确保计划能够直接导向正确答案。
        planner_prompt = ChatPromptTemplate.from_messages([
            (
                "system",
                """该计划应包含单个任务，若执行正确，将得出正确答案。不要添加任何多余步骤。\
                最后一步的结果应为最终答案。确保每个步骤都包含所需的所有信息——不要跳过步骤。\n
                必须严格按照以下格式要求返回数据，不要添加任何额外内容：\n
                {format_instructions}\n\n
                请确保输出完全符合上述格式，字段类型和约束要严格遵守。
                """,  # 添加了JSON格式要求
            ),
            ("placeholder", "{messages}"),
        ])
        # 创建规划器：提示模板 + LLM（带结构化输出）
        planner = planner_prompt | initialize_llm().with_structured_output(Plan)
        # 调用规划器生成计划
        result = await planner.ainvoke({
            "messages": [("user", state["input"])],
            "format_instructions": parser.get_format_instructions()  # 解析器生成格式说明
        })
        # 返回包含计划步骤，将被合并到工作流状态中
        return {"plan": result.steps}

    @staticmethod
    async def execute_step(state: PlanExecute):
        """执行步骤节点：执行计划中的第一步

        调用智能体执行器处理计划中的第一个步骤，并记录执行结果

        参数:
            state: 当前工作流状态，包含计划步骤等信息

        返回:
            包含已执行步骤记录的字典
        """
        # 1.获取React智能体执行器
        executor = _create_agent_executor()

        # 2.处理计划：格式化步骤列表，提取第一步任务
        plan = state["plan"]
        plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
        task = plan[0]  # 取出计划中的第一个步骤

        # 3.格式化任务描述，明确告知执行器要完成的工作
        task_formatted = f"""对于以下计划：
                            {plan_str}
                            你的任务是执行第1步：{task}。"""

        # 4.调用执行器执行任务
        agent_response = executor.invoke(
            {"messages": [("user", task_formatted)]}
            , debug=True  # 开启调试，正式应用需关闭
        )

        # 5.记录已执行步骤（(任务内容, 执行结果)），返回给LangGraph合并状态
        return {
            "past_steps": [(task, agent_response["messages"][-1].content)],
        }

    @staticmethod
    async def replan_step(state: PlanExecute):
        """重新规划步骤节点：根据执行结果更新计划

        分析已执行步骤的结果，决定是继续执行新的步骤还是直接响应用户

        参数:
            state: 当前工作流状态，包含原始计划和已执行步骤

        返回:
            包含新计划或最终响应的字典
        """
        # 1. 定义重新规划的提示模板
        replanner_prompt = ChatPromptTemplate.from_template("""针对给定目标，制定一个简单的分步计划。
        该计划应包含单个任务，若执行正确，将得出正确答案。不要添加任何多余步骤。
        最后一步的结果应为最终答案。确保每个步骤都包含所需的所有信息——不要跳过步骤。

        你的目标是：
        {input}

        你最初的计划是：
        {plan}

        你目前已完成以下步骤：
        {past_steps}

        相应地更新你的计划。如果无需更多步骤且可以回复用户，则进行回复。否则，填写计划。仅在计划中添加仍需完成的步骤。不要将已完成的步骤作为计划的一部分返回。
        
        必须严格按照以下格式要求返回数据，不要添加任何额外内容：\n
        {format_instructions}\n\n
        请确保输出完全符合上述格式，字段类型和约束要严格遵守。
        """)  # 添加了JSON格式要求

        # 2. 创建解析器,用于生成响应格式说明
        parser = PydanticOutputParser(pydantic_object=Act)
        # 3. 创建重新规划器：提示模板 + LLM（带结构化输出）
        replanner = replanner_prompt | initialize_llm().with_structured_output(Act)
        # 4. 调用重新规划器生成新的动作指令
        output = await replanner.ainvoke({
            "input": state.get("input"),
            "plan": state.get("plan"),
            "past_steps": state.get("past_steps"),
            "format_instructions": parser.get_format_instructions()  # 解析器生成格式说明
        })
        # 5. 根据输出的动作类型返回不同的结果
        if isinstance(output.action, Response):
            # 如果是响应动作，返回最终回答
            return {"response": output.action.response}
        else:
            # 如果是计划动作，返回更新后的计划
            return {"plan": output.action.steps}

    @staticmethod
    def should_end(state: PlanExecute) -> str:
        """路由节点：决定是否结束流程

        根据当前状态判断工作流应继续执行还是结束

        参数:
            state: 当前工作流状态

        返回:
            下一步节点名称（"execute"表示继续执行，END表示结束）
        """
        # 若状态中有response（最终答案），则返回END（结束）；否则返回"execute"（继续执行）
        return END if "response" in state and state["response"] else "execute"


# ----------------------------
# 图构建（工作流定义）
# ----------------------------
def build_plan_and_execute_workflow() -> CompiledStateGraph:
    """构建plan-execute工作流图

    定义工作流中的节点和节点之间的连接关系，构建完整的智能体工作流程

    返回:
        编译好的状态图(CompiledStateGraph)
    """
    workflow = StateGraph(PlanExecute)

    # 添加节点
    workflow.add_node("planner", Nodes.plan_step)  # 规划节点
    workflow.add_node("execute", Nodes.execute_step)  # 执行节点
    workflow.add_node("replan", Nodes.replan_step)  # 重新规划节点

    # 定义节点间的连接关系
    workflow.add_edge(START, "planner")  # 起始点 -> 规划节点
    workflow.add_edge("planner", "execute")  # 规划节点 -> 执行节点
    workflow.add_edge("execute", "replan")  # 执行节点 -> 重新规划节点
    # 添加条件边：根据重新规划的结果决定下一步
    workflow.add_conditional_edges(
        "replan",  # 源节点
        Nodes.should_end,  # 路由判断函数
        ["execute", END],  # 可能的目标节点
    )

    # 编译并返回工作流图
    return workflow.compile()


def save_grath_image(app: CompiledStateGraph):
    """保存工作流图为图片

    生成并保存工作流的可视化图表，默认使用mermaid api 生成图片，需要科学上网。

    参数:
        app: 编译好的状态图实例
    """
    from PIL import Image
    import io
    # 获取图像二进制数据（使用mermaid API生成）
    image_bytes = app.get_graph(xray=True).draw_mermaid_png()
    image_stream = io.BytesIO(image_bytes)
    # 保存为本地文件
    with Image.open(image_stream) as img:
        img.save("plan_and_execute.png")  # 保存为PNG格式


# ----------------------------
# 主程序
# ----------------------------
async def main():
    """运行计划-执行智能体示例"""
    # 1. 初始化环境变量
    load_environment()

    # 2. 构建工作流
    app = build_plan_and_execute_workflow()

    # 3. （可选）保存工作流图片（需科学上网）
    # 默认使用mermaid api 生成图片，需要科学上网,才能执行成功
    # save_grath_image(app)

    # 4. 测试查询（用户输入）
    inputs = {"input": "2024年澳大利亚网球公开赛男子单打冠军的家乡是哪里？"}
    # 5. 设置递归限制（防止无限循环）
    config = {"recursion_limit": Config.RECURSION_LIMIT}
    # 6. 流式运行工作流（实时输出每一步结果）
    async for event in app.astream(inputs, config=config):
        for k, v in event.items():
            if k != "__end__":  # 过滤掉结束标记
                print(v)


if __name__ == "__main__":
    import asyncio

    # 运行主程序
    asyncio.run(main())

6.3 运行测试

执行以下命令，由于使用tavily搜索引擎，需要配置科学上网哦：

1

python plan_and_execute.py

或者在IDE如pycharm中右键运行。

执行结果如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


D:\program\anaconda3\envs\langchain_311_env\python.exe D:\dev\guwei\tiny-agent-lab\agent-practice\cognitive_architecture\plan_and_execute\plan_and_execute.py 
{'plan': ['确定2024年澳大利亚网球公开赛男子单打冠军是谁。', '查找该冠军选手的家乡所在地。', '返回该地点信息作为最终答案。']}
[-1:checkpoint] State at the end of step -1:
{'messages': []}
[0:tasks] Starting 1 task for step 0:
- __start__ -> {'messages': [('user',
               '对于以下计划：\n'
               '                        1. 确定2024年澳大利亚网球公开赛男子单打冠军是谁。\n'
               '2. 查找该冠军选手的家乡所在地。\n'
               '3. 返回该地点信息作为最终答案。\n'
               '\n'
               '你的任务是执行第1步：确定2024年澳大利亚网球公开赛男子单打冠军是谁。。')]}
[0:writes] Finished step 0 with writes to 1 channel:
- messages -> [('user',
  '对于以下计划：\n'
  '                        1. 确定2024年澳大利亚网球公开赛男子单打冠军是谁。\n'
  '2. 查找该冠军选手的家乡所在地。\n'
  '3. 返回该地点信息作为最终答案。\n'
  '\n'
  '你的任务是执行第1步：确定2024年澳大利亚网球公开赛男子单打冠军是谁。。')]
[0:checkpoint] State at the end of step 0:
{'messages': [HumanMessage(content='对于以下计划：\n                        1. 确定2024年澳大利亚网球公开赛男子单打冠军是谁。\n2. 查找该冠军选手的家乡所在地。\n3. 返回该地点信息作为最终答案。\n\n你的任务是执行第1步：确定2024年澳大利亚网球公开赛男子单打冠军是谁。。', additional_kwargs={}, response_metadata={}, id='961a5990-e723-490e-8a29-45e1ddb66451')]}
[1:tasks] Starting 1 task for step 1:
- agent -> {'is_last_step': False,
 'messages': [HumanMessage(content='对于以下计划：\n                        1. 确定2024年澳大利亚网球公开赛男子单打冠军是谁。\n2. 查找该冠军选手的家乡所在地。\n3. 返回该地点信息作为最终答案。\n\n你的任务是执行第1步：确定2024年澳大利亚网球公开赛男子单打冠军是谁。。', additional_kwargs={}, response_metadata={}, id='961a5990-e723-490e-8a29-45e1ddb66451')],
 'remaining_steps': 9}
 
 ......省略
 
 [3:checkpoint] State at the end of step 3:
{'messages': [HumanMessage(content='对于以下计划：\n                        1. 查找扬尼克·辛纳（Jannik Sinner）的家乡所在地。\n2. 返回该地点信息作为最终答案。\n\n你的任务是执行第1步：查找扬尼克·辛纳（Jannik Sinner）的家乡所在地。。', additional_kwargs={}, response_metadata={}, id='a80a2c97-c82c-4852-8767-487e18107b77'),
              AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_3aff83ab43c94e5d828682', 'function': {'arguments': '{"query": "Jannik Sinner hometown"}', 'name': 'tavily_search'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 1175, 'total_tokens': 1201, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'qwen-plus-2025-04-28', 'system_fingerprint': None, 'id': 'chatcmpl-a398dbdc-edbf-9357-9281-6f69e4fe7b25', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--85564be0-28ac-4d7f-a700-ee6050307128-0', tool_calls=[{'name': 'tavily_search', 'args': {'query': 'Jannik Sinner hometown'}, 'id': 'call_3aff83ab43c94e5d828682', 'type': 'tool_call'}], usage_metadata={'input_tokens': 1175, 'output_tokens': 26, 'total_tokens': 1201, 'input_token_details': {}, 'output_token_details': {}}),
              ToolMessage(content='{"query": "Jannik Sinner hometown", "follow_up_questions": null, "answer": null, "images": [], "results": [{"url": "https://en.wikipedia.org/wiki/Jannik_Sinner", "title": "Jannik Sinner", "content": "Jannik Sinner was born 16 August 2001 to Hanspeter and Siglinde Sinner in ... He grew up in the town of Sexten in the Dolomites, the family hometown, where ...See more", "score": 0.98563, "raw_content": null}, {"url": "https://www.britannica.com/biography/Jannik-Sinner", "title": "Jannik Sinner | Height, Age, Wimbledon, Carlos Alcaraz, & ...", "content": "Jannik Sinner has won four Grand Slam titles: the Australian Open twice (2024, 2025), Wimbledon once (2025), and the U.S. Open once (2024). **Jannik Sinner** (born August 16, 2001, San Candido, Italy) is an Italian tennis player who is a leading star in the generation of players rising in the wake of the indomitable Big Three (Novak Djokovic, Roger Federer, and Rafael Nadal). In 2024 Sinner won his first majors, at the Australian Open and the U.S. Open, and attained the number one ranking on the men’s ATP (Association of Tennis Professionals) tour. Sinner won four ATP titles in 2021 and advanced to the round of 16 in both the French Open and the U.S. Open.", "score": 0.9835, "raw_content": null}], "response_time": 1.01, "request_id": "1ded3904-8b59-43c8-b290-0f5393490872"}', name='tavily_search', id='3022a878-ae9c-4637-a84a-3fcbb984db66', tool_call_id='call_3aff83ab43c94e5d828682'),
              AIMessage(content='扬尼克·辛纳（Jannik Sinner）的家乡是意大利的桑坎迪多（San Candido）。他于2001年8月16日出生在那里。此外，他成长于多洛米蒂山脉的塞斯顿（Sexten）镇，这是他家庭的故乡。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 67, 'prompt_tokens': 1652, 'total_tokens': 1719, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'qwen-plus-2025-04-28', 'system_fingerprint': None, 'id': 'chatcmpl-f6f9b966-f2db-9624-9c6d-3306ae217dc2', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--d7f867fe-7fb2-4c76-b727-65897708a403-0', usage_metadata={'input_tokens': 1652, 'output_tokens': 67, 'total_tokens': 1719, 'input_token_details': {}, 'output_token_details': {}})]}
{'past_steps': [('查找扬尼克·辛纳（Jannik Sinner）的家乡所在地。', '扬尼克·辛纳（Jannik Sinner）的家乡是意大利的桑坎迪多（San Candido）。他于2001年8月16日出生在那里。此外，他成长于多洛米蒂山脉的塞斯顿（Sexten）镇，这是他家庭的故乡。')]}
{'response': '2024年澳大利亚网球公开赛男子单打冠军扬尼克·辛纳的家乡是意大利的桑坎迪多（San Candido）。'}

Process finished with exit code 0

最后响应结果为：{'response': '2024年澳大利亚网球公开赛男子单打冠军扬尼克·辛纳的家乡是意大利的桑坎迪多（San Candido）。'}

6.4 结论

从设计角度看，Plan-and-Execute是对ReAct和ReWOO模式的进一步优化：

相比ReAct，它的优势在于推理阶段就形成明确的长期规划，能指导执行路径，解决了ReAct缺乏对任务全局的统筹规划、复杂场景下易出现路径冗余或目标偏移的问题；
相比ReWOO，它新增eplanner（重规划器），实现了计划的动态调整机制，弥补了ReWOO一旦生成计划就无法灵活调整的缺陷 —— 避免了因某子任务执行结果偏离预期却无法修正，最终导致 “一步错、步步错” 的问题。

不过Plan-and-Execute存在一个明显局限：任务需按顺序执行，下一个任务必须等待上一个完成才能启动，总执行时间等同于所有任务耗时之和，效率较低。改进的方式是：可将任务的 “线性列表” 式结构，转化为有向无环图（DAG） 结构 —— 通过 DAG 明确任务间的依赖关系，让无依赖的任务实现并行执行，可大幅缩短总耗时。这种改进思路类似于LLM Compiler`设计模式的逻辑。

完整源码地址：

GitHub 仓库：https://github.com/tinyseeking/tidy-agent-practice/tree/main/cognitive_pattern/plan_and_solve
Gitee 仓库（国内）：https://gitee.com/tinyseeking/tidy-agent-practice/tree/main/cognitive_pattern/plan_and_solve