PythonAI算法
AI Agent 架构解析：构建 Plan-and-Execute 智能体
本文深入探讨了 AI Agent 的 Plan-and-Execute 架构。对比传统 ReAct 架构在效率与记忆上的不足，该方案通过规划器生成多步计划并由执行器分步实施，显著提升了任务完成率和成本效益。文章基于 LangGraph 提供了完整的 Python 实现示例，涵盖状态定义、节点编排及重规划机制，为开发者构建复杂智能体系统提供了实践参考。
本文深入探讨了 AI Agent 的 Plan-and-Execute 架构。对比传统 ReAct 架构在效率与记忆上的不足，该方案通过规划器生成多步计划并由执行器分步实施，显著提升了任务完成率和成本效益。文章基于 LangGraph 提供了完整的 Python 实现示例，涵盖状态定义、节点编排及重规划机制，为开发者构建复杂智能体系统提供了实践参考。
随着大语言模型（LLM）能力的不断提升，AI Agent（智能体）在解决复杂任务方面展现出巨大潜力。然而，传统的交互模式在面对多步骤、长链条任务时往往显得力不从心。本文深入探讨一种新兴的架构模式——Plan-and-Execute（规划与执行），并通过 Python 和 LangGraph 框架提供完整的实践指南。
ReAct（Reasoning + Acting）架构曾是 AI Agent 的主流范式，它要求模型在每一步行动前都进行推理。尽管灵活，但在实际生产环境中暴露出显著缺陷：
Plan-and-Execute 架构通过分离'思考'与'执行'，从根本上优化了上述问题：
该架构主要包含两个关键模块：
LangGraph 提供了构建有状态、多节点工作流的强大能力，非常适合实现 Plan-and-Execute 模式。
首先定义工作流的状态结构，用于存储输入、计划、历史步骤及最终响应。
from typing import TypedDict, List, Annotated
import operator
class PlanExecute(TypedDict):
input: str
plan: List[str]
past_steps: List[tuple]
response: str
规划节点利用 LLM 的结构化输出功能生成计划。需确保 Prompt 明确约束步骤数量和信息完整性。
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
class Plan(BaseModel):
steps: List[str] = Field(description="Step by step plan")
planner_prompt = ChatPromptTemplate.from_messages([
("system", """For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. \
Do not add any superfluous steps. The result of the final step should be the final answer. \
Make sure that each step has all the information needed - do not skip steps.""")
])
async def plan_step(state: PlanExecute):
# 假设 model 已初始化且支持结构化输出
plan = await planner_prompt | model.with_structured_output(Plan).ainvoke({"messages": [("user", state["input"])]})
return {"plan": plan.steps}
执行节点负责处理当前步骤。若步骤涉及外部工具，此处应集成相应的 ToolNode。
async def execute_step(state: PlanExecute):
plan = state["plan"]
task = plan[0]
# 格式化计划上下文，帮助执行器理解全局目标
plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))
task_formatted = f"""For the following plan:
{plan_str}
You are tasked with executing step 1: {task}."""
agent_response = await agent_executor.ainvoke(
{"messages": [("user", task_formatted)]}
)
return {
"past_steps": [(task, agent_response["messages"][-1].content)],
}
当执行失败或环境变化时，系统需具备动态调整计划的能力。这通常通过条件边实现。
from typing import Union
class Response(BaseModel):
response: str
class Act(BaseModel):
action: Union[Response, Plan] = Field(
description="Action to perform. If you want to respond to user, use Response. \
If you need to further use tools to get the answer, use Plan."
)
replanner_prompt = ChatPromptTemplate.from_template(
"""For the given objective, come up with a simple step by step plan. \
Your objective was this: {input}
Your original plan was this: {plan}
You have currently done the follow steps: {past_steps}
Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. \
Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done."""
)
async def replan_step(state: PlanExecute):
output = await replanner_prompt | model.with_structured_output(Act).ainvoke(state)
if isinstance(output.action, Response):
return {"response": output.action.response}
else:
return {"plan": output.action.steps}
最后，将所有节点连接成图，定义流转逻辑。
from langgraph.graph import StateGraph, START, END
def should_end(state: PlanExecute):
# 判断是否已完成所有步骤或得到最终回答
if state.get("response"):
return "END"
elif len(state.get("plan", [])) == 0:
return "END"
else:
return "agent"
workflow = StateGraph(PlanExecute)
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")
workflow.add_conditional_edges(
"replan",
should_end,
["agent", END]
)
app = workflow.compile()
在实际部署中，建议关注以下几点以提升稳定性：
Plan-and-Execute 架构通过解耦规划与执行，有效解决了传统 Agent 在处理复杂任务时的效率与记忆瓶颈。结合 LangGraph 等现代开发框架，开发者可以构建出更加稳健、可控的智能体系统。随着相关生态的成熟，这一模式有望成为企业级 AI 应用的标准架构之一。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online