中文字幕无码av不卡一区,亚洲综合AV永久无码精品一区二区,黑人巨大白妞出浆,黄色高清无码免费看,久久久久免费精品国产,久久无码人妻丰满熟妇区毛片,欧美日韩国产综合草草,久久福利网站,亚洲一区二区三区中文字幕在线,av国内精品久久久久影院

本文深入探討了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會有所不同，“對于給定的目標，制定一個簡單的分步計劃。該計劃涉及單個任務，如果正確執行，將產生正確的答案，不要添加任何多余的步驟，最后一步的結果應該是最終答案。確保每一步都包含所需的所有信息，不要跳過步驟?！?/p>

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?”)
]
}
)

第四步 定義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(
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。