AI大模型实战篇：AI Agent设计模式 – Plan & Execute

本文深入探讨了Plan-and-Execute的设计原理、实现方法及其在实际中的应用，为读者揭示了如何构建一个灵活且高效的AI系统。

在上篇文章《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

第三步 定义Planner

Planner的主要任务就是接收输入，并输出初始的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(

deion=”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?”)

]

}

)

第四步 定义Replanner

Replanner的主要任务是根据子任务的执行结果，更新计划。

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(

deion=”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}nnYou 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”,

# Next, we pass in the function that will determine which node is called next.

should_end,

)

app = workflow.compile

总结

从原理上看，Plan-and-Execute和ReAct也有一定的相似度，但是Plan-and-Execute的优点是具备明确的长期规划，这一点即使非常强大的LLM也难以做到。同时可以只使用较大的模型做规划，而使用较小的模型执行步骤，降低执行成本。

但是Plan-and-execute的局限性在于，每个任务是按顺序执行的，下一个任务都必须等上一个任务完成之后才能执行，这可能会导致总执行时间的增加。

一种有效改进的办法是将每个任务表示为有向无环图DAG，从而实现任务的并行执行。这就是下一篇文章要介绍的AI Agent设计模式，LLM Compiler。

更多精彩文章：

大佬们都在关注的AI Agent，到底是什么？用5W1H分析框架拆解AI Agent（上篇）

大佬们都在关注的AI Agent，到底是什么？用5W1H分析框架拆解AI Agent（中篇）

大佬们都在关注的AI Agent，到底是什么？用5W1H分析框架拆解AI Agent（下篇）

AI大模型实战篇：AI Agent设计模式 – ReAct

AI大模型实战篇：AI Agent设计模式 – REWOO

本文由人人都是产品经理作者【风叔】，微信公众号：【风叔云】，原创/授权 发布于人人都是产品经理，未经许可，禁止转载。

题图来自Unsplash，基于 CC0 协议。