当前位置：首页 > news >正文

于LangChain与DeepSeek实现Plan-and-Execute Agent，让AI学会“先规划后行动”

news 2026/7/7 6:02:36

、先搞懂：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. 工具定义时，要明确工具的功能和输入格式，避免执行器无法精准调用工具，导致任务执行失败。

代码实现：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

# -*- coding: utf-8 -*-

"""

Plan-and-Execute Agent using DeepSeek via LangChain

- Planner: 拆解任务为子步骤

- Executor: 逐个执行子步骤（可调用工具）

- 支持多轮对话（通过手动注入历史）

------使用默认提示词

"""

importos

fromdatetimeimportdatetime

fromdotenvimportload_dotenv

fromlangchain_openaiimportChatOpenAI

fromlangchain.agentsimportTool

fromlangchain.memoryimportConversationBufferMemory

fromlangchain_experimental.plan_and_executeimport(

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. 定义本地工具

# ========================

defcalculate(expr:str)->str:

allowed=set("0123456789+-*/(). ")

ifnotall(cinallowedforcinexpr):

return"错误：非法字符"

try:

returnstr(eval(expr, {"__builtins__": {}}, {}))

exceptException as e:

returnf"计算失败: {e}"

defget_current_time(_)->str:

returndatetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

defanswer_simple_question(query:str)->str:

knowledge={

"你是谁":"我是基于 DeepSeek 的 Plan-and-Execute 智能助手。",

"你能做什么":"我可以制定计划并执行多步任务，比如计算、查时间、逻辑推理等。",

}

returnknowledge.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. 执行函数（带记忆）

# ========================

defrun_with_memory(user_input:str)->str:

history=memory.load_memory_variables({}).get("history", "")

ifhistory:

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})

returnresponse

# ========================

# 5. 主循环

# ========================

if__name__=="__main__":

print("🧠 Plan-and-Execute Agent 启动！")

print("💡 示例：'先算 12*5，再告诉我现在几点，最后说你好'")

print("🛑 输入 'quit' 退出。\n")

whileTrue:

user_input=input("👤 你: ").strip()

ifnotuser_inputoruser_input.lower()in["quit","exit"]:

print("👋 再见！")

break

try:

response=run_with_memory(user_input)

print(f"🤖 助手: {response}\n")

exceptException 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",