LangChain 实验性模块 plan_and_execute 核心组件详解

langchain_experimental.plan_and_execute 是 LangChain 推出的**“规划-执行”范式 Agent 框架**，核心是把“任务拆解（规划）”和“步骤执行（执行）”拆分为两个独立模块，适合处理复杂的多步骤任务（如“查库存→算价格→安排配送”这类需要分阶段完成的场景）。

它有三个核心组件：PlanAndExecute、load_chat_planner、load_agent_executor。

---

一、整体框架：Plan-and-Execute 范式的核心思想

传统的 ReAct Agent（如 ConversationalReactAgent）是“思考-执行-思考-执行”的循环，每一步都需要重新思考下一步动作；而 Plan-and-Execute 范式将流程拆分为两个阶段：

1. 规划阶段（Planner）：用大模型将用户的复杂任务拆解为可执行的步骤列表（如“先查玫瑰库存→再算50朵价格→最后安排配送”）；
2. 执行阶段（Executor）：按规划的步骤依次执行，调用工具完成每个子任务，并反馈结果；
3. 协调阶段（PlanAndExecute）：串联规划和执行，接收用户查询→调用Planner生成计划→调用Executor执行计划→根据执行结果生成最终回答。

这种分离的设计更适合复杂、长流程的任务，避免了传统 Agent 重复思考的效率问题，步骤拆解也更清晰可追溯。

---

二、组件 1：load_chat_planner —— 创建“规划者”（Planner）

1. 核心作用

load_chat_planner 是创建规划者组件的工厂函数，它会初始化一个基于大语言模型的“规划者”，负责将用户的自然语言查询拆解为结构化的执行步骤列表。

2. 原理与实现

• 底层基于 LangChain 的 ChatModel（如 GPT-3.5/4），通过预设的提示词引导模型生成“步骤化的任务拆解”；
• 生成的计划是一个步骤列表，每个步骤包含：任务描述、是否需要调用工具、预期输出等信息；
• 规划者不会直接执行任务，仅负责“拆解问题、生成步骤”。

3. 代码示例

from langchain_openai import ChatOpenAI
from langchain_experimental.plan_and_execute import load_chat_planner

# 初始化大模型
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建规划者
planner = load_chat_planner(model)

4. 关键参数

• llm（必填）：用于生成规划的大语言模型（如 ChatOpenAI）；
• prompt（可选）：自定义规划提示词（默认使用 LangChain 预设的提示词，引导模型生成步骤化计划）；
• verbose（可选）：是否打印规划过程的日志（默认 False）。

5. 规划输出示例

当用户查询“查查玫瑰的库存然后给出50朵玫瑰的价格和当天的配送方案”时，Planner 会生成类似这样的步骤：

1. 调用 check_inventory 工具查询玫瑰的库存数量；
2. 调用 calculate_price 工具计算50朵玫瑰的总价（基础价格5元/朵，加价20%）；
3. 调用 schedule_delivery 工具安排订单10086的当天配送；
4. 汇总库存、价格、配送信息，生成最终回答。

---

三、组件 2：load_agent_executor —— 创建“执行者”（Executor）

1. 核心作用

load_agent_executor 是创建执行者组件的工厂函数，它会初始化一个“执行者”，负责按规划的步骤依次执行任务：调用工具、处理工具返回结果、记录执行日志，并将结果反馈给 Planner。

2. 原理与实现

• 底层基于 LangChain 的 AgentExecutor，结合大模型和工具列表，按步骤执行任务；
• 每个步骤执行完成后，会将工具返回的结果记录下来，作为下一步的上下文；
• 若某一步执行失败（如工具调用报错），会尝试重新执行或调整步骤（依赖大模型的判断）。

3. 代码示例

from langchain_experimental.plan_and_execute import load_agent_executor

# 工具列表（包含所有需要调用的工具）
tools = [check_inventory, calculate_price, schedule_delivery]

# 创建执行者
executor = load_agent_executor(model, tools, verbose=True)

4. 关键参数

• llm（必填）：用于判断步骤执行逻辑的大语言模型；
• tools（必填）：执行者可以调用的工具列表（如，库存查询、定价、配送工具）；
• verbose（可选）：是否打印执行过程的详细日志（默认 False，建议设为 True 方便调试）；
• agent_kwargs（可选）：自定义 Agent 的配置（如提示词、工具调用格式等）；
• return_intermediate_steps（可选）：是否返回每个步骤的执行结果（默认 False）。

5. 执行过程示例

当 Planner 生成步骤后，Executor 会依次执行：

1. 执行步骤1：调用 check_inventory("玫瑰") → 返回库存100；
2. 执行步骤2：调用 calculate_price("玫瑰", 50) → 返回总价300.0；
3. 执行步骤3：调用 schedule_delivery(10086, "玫瑰", 50, "当天") → 返回配送确认信息；
4. 执行步骤4：汇总结果，生成最终回答。

---

四、组件 3：PlanAndExecute —— 协调“规划者”与“执行者”的核心容器

1. 核心作用

PlanAndExecute 是整个 Agent 的核心协调器，负责串联“规划”和“执行”的全流程：

• 接收用户的自然语言查询；
• 调用 Planner 生成结构化的执行步骤；
• 将步骤交给 Executor 依次执行；
• 收集执行结果，生成最终的自然语言回答；
• 若执行过程中出现异常，会触发 Planner 重新调整计划（可选）。

2. 代码示例

from langchain_experimental.plan_and_execute import PlanAndExecute

# 初始化 Plan-and-Execute Agent
agent = PlanAndExecute(
    planner=planner,
    executor=executor,
    verbose=True  # 打印整个流程的日志
)

# 运行 Agent 处理用户查询
result = agent.run("查询玫瑰的库存，计算50朵玫瑰的总价（默认加价20%），并安排当天的配送，订单编号设为10086")

3. 关键参数

• planner（必填）：由 load_chat_planner 创建的规划者组件；
• executor（必填）：由 load_agent_executor 创建的执行者组件；
• verbose（可选）：是否打印整个流程的详细日志（默认 False，建议设为 True 方便调试）；
• max_iterations（可选）：最大迭代次数（默认 10，防止无限循环）；
• return_intermediate_steps（可选）：是否返回规划和执行的中间步骤（默认 False）。

4. 完整工作流程

用户查询 → PlanAndExecute → 调用Planner生成步骤 → 调用Executor执行步骤 → 收集结果 → 生成最终回答 → 返回给用户

---

五、示例及带日志的调试脚本

---

如下示例中，开启所有日志开关、返回中间步骤，并添加额外的日志解析代码，可以让你清晰看到“规划→执行→结果汇总”的每一个细节。

1、完整调试脚本

# ==============================================
# 1. 环境准备与依赖导入
# ==============================================
import os
from dotenv import load_dotenv
# 加载.env文件中的OPENAI_API_KEY（必须提前配置）
load_dotenv()
# 检查API密钥是否存在
if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("请在.env文件中配置OPENAI_API_KEY！格式：OPENAI_API_KEY='你的密钥'")

# 新版LangChain OpenAI模块
from langchain_openai import ChatOpenAI
# Plan-and-Execute核心组件
from langchain_experimental.plan_and_execute import (
    PlanAndExecute,
    load_chat_planner,
    load_agent_executor
)
# LangChain工具装饰器
from langchain.tools import tool
# 辅助打印日志（美化输出）
import pprint

# ==============================================
# 2. 定义带详细日志的工具（每个工具添加执行日志）
# ==============================================
@tool
def check_inventory(flower_type: str) -> int:
    """
    查询特定类型花的库存数量，用于确认是否可下单。
    参数:
    - flower_type: 花的类型（如玫瑰、百合）
    返回:
    - 库存数量（整数）
    """
    print(f"\n【工具调用日志】check_inventory 开始执行")
    print(f"  参数：flower_type = {flower_type}")
    # 模拟数据库库存表
    inventory_db = {"玫瑰": 100, "百合": 80, "康乃馨": 150}
    inventory = inventory_db.get(flower_type, 0)
    print(f"  执行结果：{flower_type} 库存 = {inventory}")
    print(f"【工具调用日志】check_inventory 执行结束\n")
    return inventory

@tool
def calculate_price(flower_type: str, quantity: int, base_price: float = 5.0, markup: float = 0.2) -> float:
    """
    根据花型、数量、基础单价和加价百分比计算最终总价。
    参数:
    - flower_type: 花的类型（用于匹配基础单价）
    - quantity: 购买数量（整数）
    - base_price: 单朵花的基础价格（默认：玫瑰5元/朵）
    - markup: 加价百分比（默认20%，即0.2）
    返回:
    - 最终总价（浮点数，保留2位小数）
    """
    print(f"\n【工具调用日志】calculate_price 开始执行")
    print(f"  参数：flower_type={flower_type}, quantity={quantity}, base_price={base_price}, markup={markup}")
    # 模拟基础单价表
    base_price_db = {"玫瑰": 5.0, "百合": 8.0, "康乃馨": 3.0}
    actual_base = base_price_db.get(flower_type, base_price)
    total_price = actual_base * quantity * (1 + markup)
    total_price = round(total_price, 2)
    print(f"  执行结果：{quantity}朵{flower_type} 总价 = {total_price} 元")
    print(f"【工具调用日志】calculate_price 执行结束\n")
    return total_price

@tool
def schedule_delivery(order_id: int, flower_type: str, quantity: int, delivery_date: str) -> str:
    """
    安排鲜花订单的配送，返回配送确认信息。
    参数:
    - order_id: 订单编号（整数）
    - flower_type: 花的类型（用于配送备注）
    - quantity: 购买数量（用于配送备货）
    - delivery_date: 配送日期（格式：YYYY-MM-DD 或 "当天"）
    返回:
    - 配送状态确认信息（字符串）
    """
    print(f"\n【工具调用日志】schedule_delivery 开始执行")
    print(f"  参数：order_id={order_id}, flower_type={flower_type}, quantity={quantity}, delivery_date={delivery_date}")
    # 处理"当天"为具体日期
    import datetime
    if delivery_date == "当天":
        delivery_date = datetime.date.today().strftime("%Y-%m-%d")
    delivery_order_id = f"DEL-{order_id}-{flower_type[:2]}-{quantity}"
    result = f"订单 {order_id} 配送确认：{quantity}朵{flower_type}，配送日期{delivery_date}，配送单号{delivery_order_id}，预计当日18点前送达"
    print(f"  执行结果：{result}")
    print(f"【工具调用日志】schedule_delivery 执行结束\n")
    return result

# ==============================================
# 3. 初始化核心组件（开启所有日志开关）
# ==============================================
# 初始化大模型（GPT-3.5，temperature=0保证结果稳定）
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    verbose=True  # 打印模型调用日志
)

# 步骤1：创建规划者（Planner）- 开启详细日志
planner = load_chat_planner(
    llm=llm,
    # 自定义规划提示词（强化步骤拆解的清晰度）
    prompt=None  # 使用默认提示词，新手无需修改
)

# 步骤2：创建执行者（Executor）- 开启所有日志
tools = [check_inventory, calculate_price, schedule_delivery]
executor = load_agent_executor(
    llm=llm,
    tools=tools,
    verbose=True,  # 打印执行过程日志
    return_intermediate_steps=True,  # 返回执行中间步骤
    agent_kwargs={
        "verbose": True  # 强化Agent执行日志
    }
)

# 步骤3：初始化Plan-and-Execute Agent（核心协调器）
agent = PlanAndExecute(
    planner=planner,
    executor=executor,
    verbose=True,  # 打印整个流程的日志
    return_intermediate_steps=True,  # 返回规划+执行的所有中间步骤
    max_iterations=10  # 最大迭代次数，防止无限循环
)

# ==============================================
# 4. 运行Agent并打印详细调试日志
# ==============================================
if __name__ == "__main__":
    # 用户查询（明确、可拆解的多步骤任务）
    user_query = "查询玫瑰的库存，计算50朵玫瑰的总价（默认加价20%），并安排当天的配送，订单编号设为10086"
    
    print("="*80)
    print("【开始执行Plan-and-Execute Agent】")
    print(f"用户查询：{user_query}")
    print("="*80)
    
    # 运行Agent（捕获中间步骤和最终结果）
    try:
        # agent.run()返回最终回答，agent.return_intermediate_steps存储中间步骤
        final_result = agent.run(user_query)
        
        # 打印1：最终结果
        print("\n" + "="*80)
        print("【1. 最终回答】")
        print(final_result)
        
        # 打印2：规划的步骤（Planner生成的结构化步骤）
        print("\n" + "="*80)
        print("【2. Planner生成的执行步骤】")
        # 提取规划步骤（agent.intermediate_steps[0]是规划步骤）
        plan_steps = agent.intermediate_steps[0][0]
        pprint.pprint(plan_steps, indent=2)
        
        # 打印3：执行的中间结果（Executor执行每个步骤的结果）
        print("\n" + "="*80)
        print("【3. Executor执行的中间结果】")
        # 提取执行步骤（agent.intermediate_steps[1:]是执行步骤）
        execute_steps = agent.intermediate_steps[1:]
        for idx, step in enumerate(execute_steps):
            print(f"\n  步骤{idx+1}：")
            pprint.pprint(step, indent=4)
    
    except Exception as e:
        print(f"\n【执行异常】：{type(e).__name__} - {e}")
        print("请检查：1. API密钥是否正确 2. 网络是否正常 3. 工具参数是否完整")

2、运行前准备

1. 安装依赖：

pip install python-dotenv langchain langchain-openai langchain-experimental openai

2. 创建.env文件：在脚本同目录下新建.env文件，写入：

OPENAI_API_KEY="你的OpenAI API密钥"

3、核心日志解析（运行后会看到这些关键信息）

运行脚本后，控制台会输出分模块的详细日志，如下解读核心部分：

1. Planner 规划日志（关键）

【2. Planner生成的执行步骤】
[
  "调用check_inventory工具，参数为flower_type='玫瑰'，查询玫瑰的库存数量",
  "调用calculate_price工具，参数为flower_type='玫瑰'、quantity=50、markup=0.2，计算50朵玫瑰的总价",
  "调用schedule_delivery工具，参数为order_id=10086、flower_type='玫瑰'、quantity=50、delivery_date='当天'，安排配送",
  "汇总库存、价格、配送信息，生成最终回答"
]

👉 含义：Planner 把你的自然语言查询拆解为可执行的结构化步骤，每一步明确“调用哪个工具+传什么参数”。

2. Executor 执行日志（关键）

【工具调用日志】check_inventory 开始执行
  参数：flower_type = 玫瑰
  执行结果：玫瑰 库存 = 100
【工具调用日志】check_inventory 执行结束

【工具调用日志】calculate_price 开始执行
  参数：flower_type=玫瑰, quantity=50, base_price=5.0, markup=0.2
  执行结果：50朵玫瑰 总价 = 300.0 元
【工具调用日志】calculate_price 执行结束

【工具调用日志】schedule_delivery 开始执行
  参数：order_id=10086, flower_type=玫瑰, quantity=50, delivery_date=当天
  执行结果：订单 10086 配送确认：50朵玫瑰，配送日期2026-01-18，配送单号DEL-10086-玫瑰-50，预计当日18点前送达
【工具调用日志】schedule_delivery 执行结束

👉 含义：Executor 按 Planner 的步骤依次调用工具，每一步打印“参数→执行过程→结果”，清晰看到工具的调用细节。

3. 最终结果日志

【1. 最终回答】
玫瑰的库存数量为100朵；50朵玫瑰的总价为300.0元（基础单价5元/朵，加价20%）；订单10086已安排配送：50朵玫瑰，配送日期2026-01-18，配送单号DEL-10086-玫瑰-50，预计当日18点前送达。

👉 含义：Agent 汇总所有工具的执行结果，生成自然语言的最终回答。

这个调试脚本的核心价值在于：

1. 全链路日志：从“规划步骤生成”到“工具调用执行”再到“结果汇总”，每一步都有清晰日志；
2. 中间步骤提取：通过return_intermediate_steps=True获取规划和执行的原始数据，方便定位问题；
3. 工具级日志：在每个工具内部添加执行日志，精准看到工具的参数和返回值。

注1：

PlanAndExecute、load_chat_planner、load_agent_executor 三者可以视为langchain_experimental.plan_and_execute的三把斧，利用它们可以快速实现Plan-and-Execute Agent。
上述代码中，可以归纳出如下模式：

tools = [....]
planner = load_chat_planner(...)
executor = load_agent_executor(...)
agent = PlanAndExecute(...)
result = agent.run("....")

而自己写的函数，只需要加入装饰器@tool即可。
这种模式比前文“LangChain 中的「工具（Tool）」和「工具包（Toolkit）」漫谈”中的自己写工具的方法更简洁清晰。

注2：

langchain_experimental.plan_and_execute中含单词“experimental”，故
plan_and_execute 被归类为实验性模块，核心原因有以下4点，符合 LangChain 对“实验性”的明确定义：

1. 核心定位：新想法的“试验场”

• 探索性范式：Plan-and-Execute 是区别于传统 ReAct Agent 的全新 Agent 范式，源自 BabyAGI 和 Plan-and-Solve 论文的前沿研究成果；
• 验证抽象价值：LangChain 团队希望先让社区试用，收集反馈，确认这种“规划-执行分离”的架构是否适合长期作为稳定抽象；
• 避免过早固化：实验性标签允许团队在不破坏核心 API 兼容性的前提下，快速迭代设计（如调整 Planner/Executor 接口、优化提示词）。

2. 技术成熟度：存在已知挑战

• 性能成本高：相比 ReAct Agent，Plan-and-Execute 需要更多 LLM 调用（规划+执行），导致更高延迟与成本，需进一步优化；
• 稳定性风险：步骤拆解依赖大模型能力，复杂任务可能出现“规划不合理”“执行偏离”等问题，容错机制仍在完善；
• 边界场景适配：对简单任务效率低，对超复杂任务可能出现步骤循环，适用场景边界需更清晰的定义。

3. 安全与维护考量

• LangChain 核心轻量化策略：将非核心、探索性功能移出主库，降低核心包体积与维护负担；
• 风险隔离：实验性模块可能包含潜在安全隐患（如复杂工具调用链），单独包管理可减少对核心库的影响。

4. 版本管理规范

LangChain 明确划分了模块状态：

状态	稳定性	生产建议	变更风险
实验性（Experimental）	低	不建议直接用于生产	高（API 可能大幅变更）
Beta	中	可谨慎用于生产	中（可能小幅调整）
Stable	高	推荐生产使用	低（遵循语义化版本）

---

目前，plan_and_execute 仍处于 langchain_experimental 实验性模块中，尚未转为正式版，需等待技术成熟度、社区验证与 API 稳定化完全达标。