如何使用Python-LangChain框架规划并执行AI代理任务？
摘要：基于LangChain与DeepSeek实现Plan-and-Execute Agent，让AI学会“先规划后行动” 在AI Agent开发领域，“如何让模型高效完成复杂多步任务”一直是核心痛点。传统Agent常陷入“走一步看一步”的局限，
基于LangChain与DeepSeek实现Plan-and-Execute Agent，让AI学会“先规划后行动”
在AI Agent开发领域，“如何让模型高效完成复杂多步任务”一直是核心痛点。传统Agent常陷入“走一步看一步”的局限，面对多步骤协作任务时容易偏离目标、效率低下。而Plan-and-Execute（规划-执行）模式的出现，完美解决了这一问题——它让AI先“想清楚”再“动手做”，实现了任务拆解与工具执行的高效闭环，今天就结合实际实践，聊聊如何基于LangChain与DeepSeek搭建属于自己的Plan-and-Execute Agent。
一、先搞懂：Plan-and-Execute到底是什么？
Plan-and-Execute Agent的核心的是“分工明确”：它将复杂任务拆解为两个核心模块——规划器（Planner）和执行器（Executor），二者各司其职、协同工作，彻底摆脱了传统Agent“边想边做”的混乱逻辑，让多步任务执行更高效、更可控。
二、Plan-and-Execute的核心特点：为什么值得用？
相比于传统的React风格Agent，Plan-and-Execute模式有着不可替代的优势，尤其适合复杂多步任务场景，这也是它在AI Agent开发中越来越受欢迎的原因：
1. 任务拆解更清晰，目标更聚焦
面对“先计算、再查时间、最后回复问候”这类多步骤任务，Plan-and-Execute不会盲目执行，而是先由规划器将复杂任务拆解为一个个简单、可落地的子步骤，明确每个步骤的目标和执行顺序。这种“化整为零”的思路，能有效避免Agent偏离核心目标，让每一步执行都有明确方向。
2. 工具调用更高效，资源更节省
Plan-and-Execute的执行器专门负责调用外部工具，规划器则专注于任务拆解和流程把控，分工明确让效率大幅提升。同时，它无需在每个子步骤都调用大模型，可根据需求选用轻量模型执行子任务，大幅降低API调用成本，兼顾效率与经济性。
3. 支持多轮记忆，交互更连贯
虽然Plan-and-Execute模式默认不支持对话记忆，但通过简单的手动配置，就能实现多轮对话的上下文衔接。这意味着Agent能记住之前的交互内容，后续执行任务时无需重复输入，交互体验更流畅，更贴近真实的人机对话场景。
4. 容错性更强，执行更稳定
执行过程中若出现工具调用失败（如本次实践中遇到的link fetch error链接获取失败），Agent不会直接崩溃，而是会根据错误反馈重新调整执行计划，确保任务尽可能完成。这种灵活的容错机制，让Agent在实际应用中更具实用性。
三、实操落地：基于LangChain与DeepSeek的实现思路
本次实践以DeepSeek为大模型基座，借助LangChain的相关工具，快速搭建Plan-and-Execute Agent，核心实现分为5个关键环节，全程无复杂代码，逻辑清晰且易于落地：
1. 模型初始化：搭建核心动力
首先需要完成DeepSeek大模型的初始化配置，这是Agent的“大脑”。核心是配置模型的API密钥、基础地址、模型版本以及相关参数——比如温度参数控制模型输出的随机性，最大生成 tokens 控制响应长度，合理的参数配置能让模型的规划和执行更稳定。
2. 工具定义：给Agent配备“手脚”
Plan-and-Execute依赖外部工具完成具体执行，因此需要提前定义常用工具并注册到Agent中。本次实践中，我们配置了三类基础工具：数学计算工具、当前时间获取工具、简单问答工具，每个工具都明确了功能和使用场景，让执行器能根据子步骤需求，精准调用对应的工具完成操作。
3. 核心组件搭建：规划器与执行器协同工作
这是整个Agent的核心环节，主要完成规划器和执行器的创建与组合。规划器负责接收用户任务、拆解子步骤，明确“先做什么、再做什么、用什么工具做”；执行器则按照规划器制定的步骤，逐个调用工具执行，获取执行结果并反馈给规划器。二者通过LangChain提供的工具无缝衔接，形成完整的“规划-执行”闭环。
4. 对话记忆配置：实现多轮交互
由于Plan-and-Execute模式默认不支持对话记忆，我们需要手动添加记忆模块，将每一轮的用户输入和Agent响应保存起来。这样后续用户输入新任务时，Agent能调用历史对话记录，理解上下文，避免重复提问，让交互更连贯自然。
5. 主循环封装：让Agent可直接启动使用
最后，通过主循环封装，让Agent能够接收用户输入、处理任务、返回响应，同时处理执行过程中可能出现的错误（如链接获取失败、工具调用异常等）。封装完成后，只需启动程序，就能与Agent进行交互，完成各类多步任务。
四、实践总结与避坑提示
通过本次实践，我们成功搭建了一个可直接使用的Plan-and-Execute Agent，核心收获在于理解了“规划与执行分离”的核心逻辑——Plan-and-Execute的优势不在于复杂的代码，而在于清晰的分工和高效的协同，让AI从“盲目执行”转变为“有计划地行动”。
同时也有两个避坑提示分享给大家：
1. 模型配置时，需确保API密钥和基础地址正确，否则会出现类似“link fetch error”的链接获取失败问题，影响Agent正常执行；
2. 工具定义时，要明确工具的功能和输入格式，避免执行器无法精准调用工具，导致任务执行失败。
代码实现：
# -*- coding: utf-8 -*-
"""
Plan-and-Execute Agent using DeepSeek via LangChain
- Planner: 拆解任务为子步骤
- Executor: 逐个执行子步骤（可调用工具）
- 支持多轮对话（通过手动注入历史）
------使用默认提示词
"""
import os
from datetime import datetime
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import Tool
from langchain.memory import ConversationBufferMemory
from langchain_experimental.plan_and_execute import (
PlanAndExecute,
load_agent_executor,
load_chat_planner,
)
load_dotenv()
# ========================
# 1. 初始化 DeepSeek LLM
# ========================
DEEPSEEK_API_KEY = "123" # 替换为实际的 API Key
llm = ChatOpenAI(
api_key=DEEPSEEK_API_KEY,
base_url="https://xxxxx.cn/v1", # Deepseek 的 API 基础地址
model="deepseek-v3:671b", # Deepseek 对话模型（可选：deepseek-chat-pro 等高级模型）
temperature=0.7, # 温度参数（0-1，越低越稳定）
max_tokens=1024 # 最大生成 tokens
)
# ========================
# 2. 定义本地工具
# ========================
def calculate(expr: str) -> str:
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expr):
return "错误：非法字符"
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"计算失败: {e}"
def get_current_time(_) -> str:
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
def answer_simple_question(query: str) -> str:
knowledge = {
"你是谁": "我是基于 DeepSeek 的 Plan-and-Execute 智能助手。",
"你能做什么": "我可以制定计划并执行多步任务，比如计算、查时间、逻辑推理等。",
}
return knowledge.get(query.strip("？? "), "这个问题我暂时无法回答。")
tools = [
Tool(name="Calculator", func=calculate, description="执行数学计算，输入如 '2 + 3 * 4'"),
Tool(name="CurrentTime", func=get_current_time, description="获取当前日期和时间"),
Tool(name="SimpleQA", func=answer_simple_question, description="回答简单预设问题"),
]
# ========================
# 3. 创建 Plan-and-Execute Agent
# ========================
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)
# 添加对话记忆（PlanAndExecute 默认不支持 memory，需手动处理）
memory = ConversationBufferMemory(memory_key="history", input_key="input")
# ========================
# 4. 执行函数（带记忆）
# ========================
def run_with_memory(user_input: str) -> str:
history = memory.load_memory_variables({}).get("history", "")
if history:
full_input = f"之前的对话:\n{history}\n\n现在用户问:\n{user_input}"
else:
full_input = user_input
# 使用 invoke 替代 run
response = agent.invoke({"input": full_input})["output"]
memory.save_context({"input": user_input}, {"output": response})
return response
# ========================
# 5. 主循环
# ========================
if __name__ == "__main__":
print("🧠 Plan-and-Execute Agent 启动！")
print("💡 示例：'先算 12*5，再告诉我现在几点，最后说你好'")
print("🛑 输入 'quit' 退出。\n")
while True:
user_input = input("👤 你: ").strip()
if not user_input or user_input.lower() in ["quit", "exit"]:
print("👋 再见！")
break
try:
response = run_with_memory(user_input)
print(f"🤖 助手: {response}\n")
except Exception as e:
print(f"❌ 错误: {e}\n")
执行结果：
🧠 Plan-and-Execute Agent 启动！
💡 示例：'先算 12*5，再告诉我现在几点，最后说你好'
🛑 输入 'quit' 退出。
👤 你: 先算 12*5，再告诉我现在几点，最后说你好
> Entering new PlanAndExecute chain...
steps=[Step(value='Calculate the product of 12 multiplied by 5.'), Step(value='Determine the current time.'), Step(value='Greet the user with "你好".'), Step(value="Given the above steps taken, please respond to the user's original question.\n")]
> Entering new AgentExecutor chain...
Thought: To calculate the product of 12 multiplied by 5, I will use the Calculator tool.
Action:
```
{
"action": "Calculator",
"action_input": "12 * 5"
}
```
Observation: 60
Thought:Action:
```
{
"action": "Final Answer",
"action_input": "The product of 12 multiplied by 5 is 60."
}
```
> Finished chain.
*****
Step: Calculate the product of 12 multiplied by 5.
Response: The product of 12 multiplied by 5 is 60.
> Entering new AgentExecutor chain...
Action:
```
{
"action": "CurrentTime",
"action_input": ""
}
```
Observation: 2026年04月14日 18:43:00
Thought:{
"action": "Final Answer",
"action_input": "The current time is 2026年04月14日 18:43:00."
}
> Finished chain.
*****
Step: Determine the current time.
Response: {
"action": "Final Answer",
"action_input": "The current time is 2026年04月14日 18:43:00."
}
> Entering new AgentExecutor chain...
```
{
"action": "Final Answer",
"action_input": "你好"
}
```
> Finished chain.
*****
Step: Greet the user with "你好".
Response: 你好
> Entering new AgentExecutor chain...
{
"action": "Final Answer",
"action_input": "你好\n\nThe product of 12 multiplied by 5 is 60.\n\nThe current time is 2026年04月14日 18:43:00."
}
> Finished chain.
*****
Step: Given the above steps taken, please respond to the user's original question.
Response: {
"action": "Final Answer",
"action_input": "你好\n\nThe product of 12 multiplied by 5 is 60.\n\nThe current time is 2026年04月14日 18:43:00."
}
> Finished chain.
🤖 助手: {
"action": "Final Answer",
"action_input": "你好\n\nThe product of 12 multiplied by 5 is 60.\n\nThe current time is 2026年04月14日 18:43:00."
}
更多学习资料尽在
老虎网盘资源