本文深入探討了Plan-and-Execute的設(shè)計原理、實現(xiàn)方法及其在實際中的應(yīng)用，為讀者揭示了如何構(gòu)建一個靈活且高效的AI系統(tǒng)。
在上篇文章《AI大模型實戰(zhàn)篇：AI Agent設(shè)計模式 REWOO》中，風(fēng)叔結(jié)合原理和具體源代碼，詳細介紹了ReWOO這種非常有效的AI Agent設(shè)計模式。
ReWOO發(fā)源于ReAct，加入了規(guī)劃器以減少Token的消耗。但是ReWOO的規(guī)劃器完成規(guī)劃之后，執(zhí)行器就只負責(zé)執(zhí)行，即使計劃存在錯誤，執(zhí)行器也只是機械式地執(zhí)行命令，這就導(dǎo)致ReWOO這種模式非常依賴于規(guī)劃器的準確性。
為了優(yōu)化這個問題，我們需要為規(guī)劃器增加一種Replan機制，即在計劃執(zhí)行的過程中，根據(jù)實際的條件和反饋，重新調(diào)整計劃。這個也很符合人類做計劃的模式，比如你之前計劃要去自駕游，但是突然看到新聞?wù)f前往目的地的主干道路發(fā)生了泥石流，因此你肯定會調(diào)整計劃，取消自駕游或者換一個目的地。
這就是本篇文章風(fēng)叔將為大家介紹的AI Agent設(shè)計模式，Plan-and-Execute。
一、Plan-and-Execute的概念
Plan-and-Execute這個方法的本質(zhì)是先計劃再執(zhí)行，即先把用戶的問題分解成一個個的子任務(wù)，然后再執(zhí)行各個子任務(wù)，并根據(jù)執(zhí)行情況調(diào)整計劃。Plan-and-Execute相比ReWOO，最大的不同就是加入了Replan機制，其架構(gòu)上包含規(guī)劃器、執(zhí)行器和重規(guī)劃器：
規(guī)劃器Planner負責(zé)讓 LLM 生成一個多步計劃來完成一個大任務(wù)，在書籍運行中，Planner負責(zé)第一次生成計劃；
執(zhí)行器接收規(guī)劃中的步驟，并調(diào)用一個或多個工具來完成該任務(wù)；
重規(guī)劃器Replanner負責(zé)根據(jù)實際的執(zhí)行情況和信息反饋來調(diào)整計劃
下圖是Plan-and-Execute的原理：
Planner接收來自用戶的輸入，輸出具體的任務(wù)清單；
將任務(wù)清單給到Single-Task Agent，即執(zhí)行器，執(zhí)行器會在循環(huán)中逐個處理任務(wù)；
執(zhí)行器每處理一個任務(wù)，就將處理結(jié)果和狀態(tài)同步給Replanner，Replanner一方面會輸出反饋給用戶，另一方面會更新任務(wù)清單；
任務(wù)清單再次給到執(zhí)行器進行執(zhí)行。
二、Plan-and-Execute的實現(xiàn)過程
Plan-and-Execute的實現(xiàn)過程
第一步 構(gòu)建執(zhí)行器
下面，我們先創(chuàng)建要用來執(zhí)行任務(wù)的執(zhí)行器。在這個示例中，為了簡單起見，我們將為每個任務(wù)使用相同的執(zhí)行器，即搜索工具。但實際情況下，可以為不同的任務(wù)使用不同的執(zhí)行器。
from langchain import hubfrom langchain_openai import ChatOpenAfrom langgraph.prebuilt import create_react_agentfrom langchain_community.tools.tavily_search import TavilySearchResultstools = [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 agentllm = ChatOpenAI(model=”gpt-4-turbo-preview”)agent_executor = create_react_agent(llm, tools, messages_modifier=prompt)
第二步 定義系統(tǒng)狀態(tài)
為什么要定義系統(tǒng)狀態(tài)？因為在處理復(fù)雜的不確定性問題時，一個非常有效的方法是將執(zhí)行階段拆分為狀態(tài)機和執(zhí)行器的循環(huán)。
執(zhí)行器將外部事件輸入狀態(tài)機，狀態(tài)機告訴執(zhí)行器必須采取的操作，而原始計劃則成為狀態(tài)機起始狀態(tài)的初始化程序。這樣做的優(yōu)點在于狀態(tài)機不依賴于混亂的執(zhí)行細節(jié)，因此我們可以對其進行詳盡而徹底的測試。
首先，我們需要跟蹤當前計劃，將其表示為字符串列表；然后跟蹤先前執(zhí)行的步驟，將其表示為元組列表；最后，還需要狀態(tài)來表示最終響應(yīng)以及原始輸入。因此，整個狀態(tài)機定義如下：
import operatorfrom typing import Annotated, List, Tuple, TypedDictclass PlanExecute(TypedDict): input: str plan: List[str] past_steps: Annotated[List[Tuple], operator.add] response: str
第三步 定義Planner
Planner的主要任務(wù)就是接收輸入，并輸出初始的Task List。
相比ReWOO的Planner，這里的Planner的Prompt會有所不同，“對于給定的目標，制定一個簡單的分步計劃。該計劃涉及單個任務(wù)，如果正確執(zhí)行，將產(chǎn)生正確的答案，不要添加任何多余的步驟，最后一步的結(jié)果應(yīng)該是最終答案。確保每一步都包含所需的所有信息，不要跳過步驟。”
第四步 定義Replanner
Replanner的主要任務(wù)是根據(jù)子任務(wù)的執(zhí)行結(jié)果，更新計劃。
Replanner和Planner的prompt模板非常相似，但是約束了Replanner的目標任務(wù)、原始Plan、已執(zhí)行的步驟、以及更新計劃。
比如更新計劃，我們要求“根據(jù)執(zhí)行的步驟更新計劃。如果不需要更多步驟，直接可以返回給用戶；否則就填寫計劃，并向計劃中添加仍需完成的步驟，不要將之前完成的步驟作為計劃的一部分返回”
from typing import Unionclass 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)
下面，我們構(gòu)建流程圖，將Planner、Replanner、執(zhí)行器等節(jié)點添加進來，執(zhí)行并輸出結(jié)果。
from typing import Literalasync 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, STARTworkflow = StateGraph(PlanExecute)# Add the plan nodeworkflow.add_node(“planner”, plan_step)# Add the execution stepworkflow.add_node(“agent”, execute_step)# Add a replan nodeworkflow.add_node(“replan”, replan_step)workflow.add_edge(START, “planner”)# From plan we go to agentworkflow.add_edge(“planner”, “agent”)# From agent, we replanworkflow.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()
總結(jié)
從原理上看，Plan-and-Execute和ReAct也有一定的相似度，但是Plan-and-Execute的優(yōu)點是具備明確的長期規(guī)劃，這一點即使非常強大的LLM也難以做到。同時可以只使用較大的模型做規(guī)劃，而使用較小的模型執(zhí)行步驟，降低執(zhí)行成本。
但是Plan-and-execute的局限性在于，每個任務(wù)是按順序執(zhí)行的，下一個任務(wù)都必須等上一個任務(wù)完成之后才能執(zhí)行，這可能會導(dǎo)致總執(zhí)行時間的增加。
一種有效改進的辦法是將每個任務(wù)表示為有向無環(huán)圖DAG，從而實現(xiàn)任務(wù)的并行執(zhí)行。這就是下一篇文章要介紹的AI Agent設(shè)計模式，LLM Compiler。
