在上篇文章《AI大模型实战篇：AI Agent设计模式 – REWOO》中，风叔结合事理和详细源代码，详细先容了ReWOO这种非常有效的AI Agent设计模式。
ReWOO发源于ReAct，加入了方案器以减少Token的花费。但是ReWOO的方案器完成规划之后，实行器就只卖力实行，纵然操持存在缺点，实行器也只是机器式地实行命令，这就导致ReWOO这种模式非常依赖于方案器的准确性。
为了优化这个问题，我们须要为方案器增加一种Replan机制，即在操持实行的过程中，根据实际的条件和反馈，重新调度操持。这个也很符合人类做操持的模式，比如你之前操持要去自驾游，但是溘然看到新闻说前往目的地的主干道路发生了泥石流，因此你肯定会调度操持，取消自驾游或者换一个目的地。
这便是本篇文章风叔将为大家先容的AI Agent设计模式，Plan-and-Execute。
一、Plan-and-Execute的观点
Plan-and-Execute这个方法的实质是先操持再实行，即先把用户的问题分解成一个个的子任务，然后再实行各个子任务，并根据实行情形调度操持。Plan-and-Execute比较ReWOO，最大的不同便是加入了Replan机制，其架构上包含方案器、实行器和重方案器：
方案器Planner卖力让 LLM 天生一个多步操持来完成一个大任务，在书本运行中，Planner卖力第一次天生操持；实行器吸收方案中的步骤，并调用一个或多个工具来完成该任务；重方案器Replanner卖力根据实际的实行情形和信息反馈来调度操持下图是Plan-and-Execute的事理：
Planner吸收来自用户的输入，输出详细的任务清单；将任务清单给到Single-Task Agent，即实行器，实行器会在循环中逐个处理任务；实行器每处理一个任务，就将处理结果和状态同步给Replanner，Replanner一方面会输出反馈给用户，另一方面会更新任务清单；任务清单再次给到实行器进行实行。二、Plan-and-Execute的实现过程Plan-and-Execute的实现过程
下面，风叔通过实际的源码，详细先容Plan-and-Execute模式的实现方法。大家可以关注"大众号【风叔云】，回答关键词【PAE源码】，获取Plan-and-Execute设计模式的完全源代码。
第一步 构建实行器
下面，我们先创建要用来实行任务的实行器。在这个示例中，为了大略起见，我们将为每个任务利用相同的实行器，即搜索工具。但实际情形下，可以为不同的任务利用不同的实行器。
from langchain import hub
from langchain_openai import ChatOpenA
from langgraph.prebuilt import create_react_agent
from langchain_community.tools.tavily_search import TavilySearchResults
tools = [TavilySearchResults(max_results=3)]
# Get the prompt to use – you can modify this!
prompt = hub.pull(“wfh/react-agent-executor”)
prompt.pretty_print
# Choose the LLM that will drive the agent
llm = ChatOpenAI(model=”gpt-4-turbo-preview”)
agent_executor = create_react_agent(llm, tools, messages_modifier=prompt)
第二步 定义系统状态为什么要定义系统状态？由于在处理繁芜的不愿定性问题时，一个非常有效的方法是将实行阶段拆分为状态机和实行器的循环。
实行器将外部事宜输入状态机，状态机见告实行器必须采纳的操作，而原始操持则成为状态机起始状态的初始化程序。这样做的优点在于状态机不依赖于混乱的实行细节，因此我们可以对其进行详尽而彻底的测试。
首先，我们须要跟踪当前操持，将其表示为字符串列表；然后跟踪先前实行的步骤，将其表示为元组列表；末了，还须要状态来表示终极相应以及原始输入。因此，全体状态机定义如下：
import operator
from typing import Annotated, List, Tuple, TypedDict
class PlanExecute(TypedDict):
input: str
plan: List[str]
past_steps: Annotated[List[Tuple], operator.add]
response: str
第三步 定义PlannerPlanner的紧张任务便是吸收输入，并输出初始的Task List。
比较ReWOO的Planner，这里的Planner的Prompt会有所不同，“对付给定的目标，制订一个大略的分步操持。该操持涉及单个任务，如果精确实行，将产生精确的答案，不要添加任何多余的步骤，末了一步的结果该当是终极答案。确保每一步都包含所需的所有信息，不要跳过步骤。”
from langchain_core.pydantic_v1 import BaseModel, Field
class Plan(BaseModel):
“””Plan to follow in future”””
steps: List[str] = Field(
description=”different steps to follow, should be in sorted order”
)
from langchain_core.prompts import ChatPromptTemplate
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.”””,
),
(“placeholder”, “{messages}”),
]
)
planner = planner_prompt | ChatOpenAI(
model=”gpt-4o”, temperature=0
).with_structured_output(Plan)
planner.invoke(
{
“messages”: [
(“user”, “what is the hometown of the current Australia open winner?”)
]
}
)
第四步 定义ReplannerReplanner的紧张任务是根据子任务的实行结果，更新操持。
Replanner和Planner的prompt模板非常相似，但是约束了Replanner的目标任务、原始Plan、已实行的步骤、以及更新操持。
比如更新操持，我们哀求“根据实行的步骤更新操持。如果不须要更多步骤，直接可以返回给用户；否则就填写操持，并向操持中添加仍需完成的步骤，不要将之前完成的步骤作为操持的一部分返回”
from typing import Union
class Response(BaseModel):
“””Response to user.”””
response: str
class Act(BaseModel):
“””Action to perform.”””
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.
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.
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. Do not return previously done steps as part of the plan.”””
)
replanner = replanner_prompt | ChatOpenAI(
model=”gpt-4o”, temperature=0
).with_structured_output(Act)
第五步 构建流程图下面，我们构建流程图，将Planner、Replanner、实行器等节点添加进来，实行并输出结果。
from typing import Literal
async def execute_step(state: PlanExecute):
plan = state[“plan”]
plan_str = “n”.join(f”{i+1}. {step}” for i, step in enumerate(plan))
task = plan[0]
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),
}
async def plan_step(state: PlanExecute):
plan = await planner.ainvoke({“messages”: [(“user”, state[“input”])]})
return {“plan”: plan.steps}
async def replan_step(state: PlanExecute):
output = await replanner.ainvoke(state)
if isinstance(output.action, Response):
return {“response”: output.action.response}
else:
return {“plan”: output.action.steps}
def should_end(state: PlanExecute) -> Literal[“agent”, “__end__”]:
if “response” in state and state[“response”]:
return “__end__”
else:
return “agent”
from langgraph.graph import StateGraph, START
workflow = StateGraph(PlanExecute)
# Add the plan node
workflow.add_node(“planner”, plan_step)
# Add the execution step
workflow.add_node(“agent”, execute_step)
# Add a replan node
workflow.add_node(“replan”, replan_step)
workflow.add_edge(START, “planner”)
# From plan we go to agent
workflow.add_edge(“planner”, “agent”)
# From agent, we replan
workflow.add_edge(“agent”, “replan”)
workflow.add_conditional_edges(“replan”, should_end)
app = workflow.compile
总结从事理上看，Plan-and-Execute和ReAct也有一定的相似度，但是Plan-and-Execute的优点是具备明确的长期方案，这一点纵然非常强大的LLM也难以做到。同时可以只利用较大的模型做方案，而利用较小的模型实行步骤，降落实行本钱。
但是Plan-and-execute的局限性在于，每个任务是按顺序实行的，下一个任务都必须等上一个任务完成之后才能实行，这可能会导致总实行韶光的增加。
一种有效改进的办法是将每个任务表示为有向无环图DAG，从而实现任务的并行实行。这便是下一篇文章要先容的AI Agent设计模式，LLM Compiler。
本文由大家都是产品经理作者【风叔】，微信"大众年夜众号：【风叔云】，原创/授权 发布于大家都是产品经理，未经容许，禁止转载。
题图来自Unsplash，基于 CC0 协议。