LangChain 1.0 实战：手写 Plan-and-Execute Agent 框架

导读：LangChain 1.0 移除了实验性的 PlanAndExecute、load_chat_planner和load_agent_executor 函数，本文手把手教你如何实现自己的 Plan-and-Execute Agent 框架，让 AI 学会"先思考再行动"！

---

📋 目录

• 背景介绍
• 核心架构设计
• 完整代码实现
• 使用示例
• 关键技术点解析
• 总结与展望

---

背景介绍

LangChain 1.0 的重大变化

在 LangChain 0.x 版本中，我们可以方便地使用内置的 Plan-and-Execute Agent：

# ❌ LangChain 0.x 的写法（1.0 已移除）
from langchain_experimental.plan_and_execute import (
    PlanAndExecute, 
    load_agent_executor, 
    load_chat_planner
)

planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)

但升级到 LangChain 1.0后，这些函数被移除了！官方建议开发者根据具体需求自定义实现。

为什么需要 Plan-and-Execute？

Plan-and-Execute 是一种经典的 Agent 模式，适用于复杂任务的处理：

用户问题 → 制定计划 → 执行步骤 → 汇总答案

典型应用场景：

• “在中国，100 人民币能买几束玫瑰花？”
• “分析某公司的财务状况并给出投资建议”
• “帮我规划一次北京三日游”

这些问题都需要多步骤协作才能完成。

---

核心架构设计

我们的实现包含三个核心组件：

1. 数据模型层

• Step：表示单个步骤（ID、描述、状态）
• Plan：表示完整计划（步骤列表）

2. 核心组件层

• Planner：规划器，负责将复杂问题分解为可执行的步骤
• Executor：执行器，负责使用工具执行每个步骤

3. 协调控制层

• PlanAndExecuteAgent：主控制器，协调整个流程
• create_plan_and_execute_agent：便捷工厂函数

---

完整代码实现

第一步：定义数据模型

from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import BaseTool
from langchain.agents import create_agent
from models import get_lc_o_ali_model_client


class Step(BaseModel):
    """表示计划中的单个步骤"""
    id: int = Field(..., description="步骤的 ID")
    description: str = Field(..., description="步骤的描述")
    status: str = Field("pending", description="步骤的状态 (pending, completed, failed)")


class Plan(BaseModel):
    """表示完整的计划"""
    steps: List[Step] = Field(default_factory=list, description="计划中的步骤列表")

💡 关键点：

• 继承 BaseModel获得数据验证和序列化能力
• Field(...)表示必填字段
• Field(default_factory=list)避免可变对象共享问题

第二步：实现 Planner 规划器

class Planner:
    """负责生成计划的规划器"""
    
    def __init__(self, llm):
        self.llm = llm
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的任务规划师。你的任务是将用户的问题分解成具体的执行步骤。"
                       "请只输出 JSON 格式的计划，不要包含其他文本。\n\n"
                       "输出格式示例:\n"
                       "{{\"steps\": [{{\"id\": 1, \"description\": \"步骤 1 的描述\"}}, {{\"id\": 2, \"description\": \"步骤 2 的描述\"}}]}}"),
            ("human", "请为以下任务制定一个详细的执行计划:\n{input}")
        ])
        self.chain = self.prompt | self.llm

    def plan(self, input_str: str) -> Plan:
        """为给定的输入生成计划"""
        response = self.chain.invoke({"input": input_str})
        
        # 解析 LLM 的响应为 Plan 对象
        import json
        try:
            plan_data = json.loads(response.content)
            steps = [Step(**step) for step in plan_data.get("steps", [])]
            return Plan(steps=steps)
        except (json.JSONDecodeError, KeyError):
            # 如果解析失败，创建一个默认计划
            return Plan(steps=[Step(id=1, description=input_str)])

💡 关键点：

• 使用 ChatPromptTemplate 设计提示词
• 要求 LLM 输出纯 JSON，便于解析
• 提供容错机制，解析失败时返回默认计划

第三步：实现 Executor 执行器

class Executor:
    """负责执行计划步骤的执行器"""
    
    def __init__(self, llm, tools: List[BaseTool]):
        self.llm = llm
        self.tools = tools
        
        # 使用 LangChain 1.0 的新推荐方式 create_agent
        self.agent = create_agent(
            model=self.llm,
            tools=self.tools,
            system_prompt="你是一个执行代理人。你的任务是执行给定的单个步骤。"
                         "请使用工具来帮助你完成任务，并给出清晰的结果总结。"
        )

    def execute_step(self, step_description: str, context: str = "") -> str:
        """执行单个步骤"""
        try:
            # 构造输入消息
            input_message = f"请执行以下步骤:\n{step_description}\n\n这是之前步骤的结果:\n{context}"
            
            # 执行 agent
            result = self.agent.invoke({
                "messages": [
                    {"role": "user", "content": input_message}
                ]
            })
            
            # 提取最终的回答内容
            if isinstance(result, dict) and "messages" in result:
                messages = result["messages"]
                if messages and hasattr(messages[-1], 'content'):
                    return messages[-1].content
                elif messages:
                    return str(messages[-1])
            
            return str(result)
        except Exception as e:
            return f"执行步骤时出错：{str(e)}"

💡 关键点：

• 使用 LangChain 1.0 的 create_agent 创建执行代理
• 传递上下文信息，让后续步骤知道前面的结果
• 完善的错误处理机制

第四步：实现 PlanAndExecuteAgent 主控制器

class PlanAndExecuteAgent:
    """Plan-and-Execute Agent 主类"""
    
    def __init__(self, planner: Planner, executor: Executor):
        self.planner = planner
        self.executor = executor

    def run(self, input_str: str) -> Dict[str, Any]:
        """运行整个 Plan-and-Execute 流程"""
        # 1. 制定计划
        print("正在制定计划...")
        plan = self.planner.plan(input_str)
        
        if not plan.steps:
            return {"error": "无法生成有效的计划"}

        print(f"计划已生成，共 {len(plan.steps)} 个步骤:")
        for i, step in enumerate(plan.steps, 1):
            print(f"  {i}. {step.description}")

        # 2. 依次执行步骤
        context = ""
        results = []
        
        for i, step in enumerate(plan.steps):
            print(f"\n⚡ 正在执行步骤 {i+1}/{len(plan.steps)}: {step.description}")
            result = self.executor.execute_step(step.description, context)
            results.append(result)
            context += f"\n步骤 {i+1} 的结果：{result}"
            print(f"步骤 {i+1} 完成，结果：{result}")

        # 3. 生成最终答案
        print("\n正在生成最终答案...")
        final_prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个总结专家。你的任务是根据提供的步骤和结果，给出针对原始问题的最终答案。"),
            ("human", "原始问题：{input}\n\n执行过程和结果:\n{context}\n\n请给出最终答案:")
        ])
        
        final_chain = final_prompt | self.planner.llm
        final_response = final_chain.invoke({
            "input": input_str,
            "context": context
        })
        
        final_answer = final_response.content if hasattr(final_response, 'content') else str(final_response)
        
        return {
            "input": input_str,
            "plan": plan,
            "intermediate_steps": results,
            "final_answer": final_answer
        }

💡 关键点：

• 三步流程：制定计划 → 执行步骤 → 汇总答案
• 上下文累积：将前面步骤的结果传递给后续步骤
• 结构化返回：包含输入、计划、中间结果和最终答案

第五步：便捷工厂函数

def create_plan_and_execute_agent(tools: List[BaseTool], **kwargs) -> PlanAndExecuteAgent:
    """创建一个 Plan-and-Execute Agent 的便捷函数"""
    llm = get_lc_o_ali_model_client()
    
    planner = Planner(llm)
    executor = Executor(llm, tools)
    
    return PlanAndExecuteAgent(planner, executor)

---

使用示例

完整调用代码

# plan-and-execute_for_lcV1.py
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from dotenv import load_dotenv
from langchain_classic.chains.llm_math.base import LLMMathChain
from langchain_community.utilities import SerpAPIWrapper
from langchain_core.tools import Tool
from models import get_lc_o_ali_model_client
from cognitive.plan_and_execute_agent_for_lcV1 import create_plan_and_execute_agent

load_dotenv()

# 配置大模型
llm = get_lc_o_ali_model_client()

# 创建工具
search = SerpAPIWrapper()
llm_math_chain = LLMMathChain(llm=llm, verbose=True)

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于回答关于当前事件的问题"
    ),
    Tool(
        name="Calculator",
        func=llm_math_chain.run,
        description="用于计算或解决问题，只能处理简单的数学表达式"
    )
]

# 创建并运行 Agent
agent = create_plan_and_execute_agent(tools, verbose=True)
result = agent.run("在中国，100 人民币能买几束玫瑰花？请用中文回答问题")
print(result["final_answer"])

实际运行效果

正在制定计划...
计划已生成，共 3 个步骤:
  1. 搜索中国市场上玫瑰花的平均价格
  2. 计算 100 元能购买多少束玫瑰花
  3. 用中文给出最终答案

⚡ 正在执行步骤 1/3: 搜索中国市场上玫瑰花的平均价格
步骤 1 完成，结果：根据搜索结果，中国市场上普通玫瑰花的批发价格约为 3-5 元/支...

⚡ 正在执行步骤 2/3: 计算 100 元能购买多少束玫瑰花
步骤 2 完成，结果：假设每束花包含 10 支玫瑰，加上包装费用，每束约 40-60 元...

⚡ 正在执行步骤 3/3: 用中文给出最终答案
步骤 3 完成，结果：在中国，100 元人民币大约可以购买 2-3 束玫瑰花...

正在生成最终答案...
最终答案：根据市场调查和计算，在中国 100 元人民币大约能购买 2-3 束普通玫瑰花...

---

关键技术点解析

1. Pydantic 模型的妙用

为什么继承 BaseModel？

class Step(BaseModel):
    id: int = Field(..., description="步骤的 ID")
    description: str = Field(..., description="步骤的描述")
    status: str = Field("pending", description="步骤的状态")

优势：

• ✅ 自动类型验证：确保数据格式正确
• ✅ 智能类型转换：字符串 "123" 自动转为整数 123
• ✅ JSON 序列化：轻松处理 LLM 返回的 JSON 数据
• ✅ IDE 支持：完整的代码补全和类型检查
• ✅ 自我描述：Field(description=...)让字段含义更清晰

2. Field(default_factory=list) 的作用

class Plan(BaseModel):
    steps: List[Step] = Field(default_factory=list, description="计划中的步骤列表")

避免经典陷阱：

# ❌ 错误写法
class Plan_Wrong:
    steps: List[Step] = []  # 所有实例共享同一个列表！

# ✅ 正确写法
class Plan:
    steps: List[Step] = Field(default_factory=list)  # 每个实例独立列表

3. 字典解包语法 **step

# LLM 返回的 JSON 数据
plan_data = {
    "steps": [
        {"id": 1, "description": "搜索价格"},
        {"id": 2, "description": "计算数量"}
    ]
}

# 优雅的转换方式
steps = [Step(**step) for step in plan_data.get("steps", [])]

# 等价于
steps = []
for step in plan_data.get("steps", []):
    step_obj = Step(**step)  # 字典解包为命名参数
    steps.append(step_obj)

4. LangChain 1.0 的 Agent 创建方式

# LangChain 1.0 推荐方式
self.agent = create_agent(
    model=self.llm,
    tools=self.tools,
    system_prompt="你是执行代理人..."
)

# 不再需要 AgentExecutor
# create_agent 返回的 agent 已经包含执行逻辑

5. 上下文的累积传递

context = ""
for step in plan.steps:
    result = executor.execute_step(step.description, context)
    context += f"\n步骤结果：{result}"  # 累积到上下文

这样后续步骤可以参考前面的结果，实现信息的流动。

---

总结与展望

本文亮点

✅ 完整实现：从零开始实现 Plan-and-Execute Agent 框架
✅ 最佳实践：使用 Pydantic、Type Hints 等现代 Python 特性
✅ 生产可用：包含错误处理和日志输出
✅ 易于扩展：模块化设计，方便定制

---

参考资料

• LangChain 官方文档
• Pydantic 文档
• LangChain Agents 深度解析

---

编辑时间：2026 年 3 月
版权声明：本文为 CSDN 博主原创，转载请注明出处

---

👇 互动区

如果你有任何问题或建议，欢迎在评论区留言讨论！觉得文章有用的话，别忘了点赞 + 收藏哦～ 🎉