AI大模型實(shí)戰(zhàn)篇:AI Agent設(shè)計(jì)模式 – Plan and Execute

0 評(píng)論 2388 瀏覽 2 收藏 14 分鐘
🔗 B端产品经理需要更多地关注客户的商业需求、痛点、预算、决策流程等,而C端产品经理需要更多地关注用户的个人需求

本文深入探討了Plan-and-Execute的設(shè)計(jì)原理、實(shí)現(xiàn)方法及其在實(shí)際中的應(yīng)用,為讀者揭示了如何構(gòu)建一個(gè)靈活且高效的AI系統(tǒng)。

在上篇文章《AI大模型實(shí)戰(zhàn)篇:AI Agent設(shè)計(jì)模式 – REWOO》中,風(fēng)叔結(jié)合原理和具體源代碼,詳細(xì)介紹了ReWOO這種非常有效的AI Agent設(shè)計(jì)模式。

ReWOO發(fā)源于ReAct,加入了規(guī)劃器以減少Token的消耗。但是ReWOO的規(guī)劃器完成規(guī)劃之后,執(zhí)行器就只負(fù)責(zé)執(zhí)行,即使計(jì)劃存在錯(cuò)誤,執(zhí)行器也只是機(jī)械式地執(zhí)行命令,這就導(dǎo)致ReWOO這種模式非常依賴于規(guī)劃器的準(zhǔn)確性。

為了優(yōu)化這個(gè)問題,我們需要為規(guī)劃器增加一種Replan機(jī)制,即在計(jì)劃執(zhí)行的過程中,根據(jù)實(shí)際的條件和反饋,重新調(diào)整計(jì)劃。這個(gè)也很符合人類做計(jì)劃的模式,比如你之前計(jì)劃要去自駕游,但是突然看到新聞?wù)f前往目的地的主干道路發(fā)生了泥石流,因此你肯定會(huì)調(diào)整計(jì)劃,取消自駕游或者換一個(gè)目的地。

這就是本篇文章風(fēng)叔將為大家介紹的AI Agent設(shè)計(jì)模式,Plan-and-Execute。

一、Plan-and-Execute的概念

Plan-and-Execute這個(gè)方法的本質(zhì)是先計(jì)劃再執(zhí)行,即先把用戶的問題分解成一個(gè)個(gè)的子任務(wù),然后再執(zhí)行各個(gè)子任務(wù),并根據(jù)執(zhí)行情況調(diào)整計(jì)劃。Plan-and-Execute相比ReWOO,最大的不同就是加入了Replan機(jī)制,其架構(gòu)上包含規(guī)劃器、執(zhí)行器和重規(guī)劃器:

  • 規(guī)劃器Planner負(fù)責(zé)讓 LLM 生成一個(gè)多步計(jì)劃來(lái)完成一個(gè)大任務(wù),在書籍運(yùn)行中,Planner負(fù)責(zé)第一次生成計(jì)劃;
  • 執(zhí)行器接收規(guī)劃中的步驟,并調(diào)用一個(gè)或多個(gè)工具來(lái)完成該任務(wù);
  • 重規(guī)劃器Replanner負(fù)責(zé)根據(jù)實(shí)際的執(zhí)行情況和信息反饋來(lái)調(diào)整計(jì)劃

下圖是Plan-and-Execute的原理:

  • Planner接收來(lái)自用戶的輸入,輸出具體的任務(wù)清單;
  • 將任務(wù)清單給到Single-Task Agent,即執(zhí)行器,執(zhí)行器會(huì)在循環(huán)中逐個(gè)處理任務(wù);
  • 執(zhí)行器每處理一個(gè)任務(wù),就將處理結(jié)果和狀態(tài)同步給Replanner,Replanner一方面會(huì)輸出反饋給用戶,另一方面會(huì)更新任務(wù)清單;
  • 任務(wù)清單再次給到執(zhí)行器進(jìn)行執(zhí)行。

二、Plan-and-Execute的實(shí)現(xiàn)過程

Plan-and-Execute的實(shí)現(xiàn)過程

下面,風(fēng)叔通過實(shí)際的源碼,詳細(xì)介紹Plan-and-Execute模式的實(shí)現(xiàn)方法。大家可以關(guān)注公眾號(hào)【風(fēng)叔云】,回復(fù)關(guān)鍵詞【PAE源碼】,獲取Plan-and-Execute設(shè)計(jì)模式的完整源代碼。

第一步 構(gòu)建執(zhí)行器

下面,我們先創(chuàng)建要用來(lái)執(zhí)行任務(wù)的執(zhí)行器。在這個(gè)示例中,為了簡(jiǎn)單起見,我們將為每個(gè)任務(wù)使用相同的執(zhí)行器,即搜索工具。但實(shí)際情況下,可以為不同的任務(wù)使用不同的執(zhí)行器。

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)

第二步 定義系統(tǒng)狀態(tài)

為什么要定義系統(tǒng)狀態(tài)?因?yàn)樵谔幚韽?fù)雜的不確定性問題時(shí),一個(gè)非常有效的方法是將執(zhí)行階段拆分為狀態(tài)機(jī)和執(zhí)行器的循環(huán)。

執(zhí)行器將外部事件輸入狀態(tài)機(jī),狀態(tài)機(jī)告訴執(zhí)行器必須采取的操作,而原始計(jì)劃則成為狀態(tài)機(jī)起始狀態(tài)的初始化程序。這樣做的優(yōu)點(diǎn)在于狀態(tài)機(jī)不依賴于混亂的執(zhí)行細(xì)節(jié),因此我們可以對(duì)其進(jìn)行詳盡而徹底的測(cè)試。

首先,我們需要跟蹤當(dāng)前計(jì)劃,將其表示為字符串列表;然后跟蹤先前執(zhí)行的步驟,將其表示為元組列表;最后,還需要狀態(tài)來(lái)表示最終響應(yīng)以及原始輸入。因此,整個(gè)狀態(tài)機(jī)定義如下:

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的主要任務(wù)就是接收輸入,并輸出初始的Task List。

相比ReWOO的Planner,這里的Planner的Prompt會(huì)有所不同,“對(duì)于給定的目標(biāo),制定一個(gè)簡(jiǎn)單的分步計(jì)劃。該計(jì)劃涉及單個(gè)任務(wù),如果正確執(zhí)行,將產(chǎn)生正確的答案,不要添加任何多余的步驟,最后一步的結(jié)果應(yīng)該是最終答案。確保每一步都包含所需的所有信息,不要跳過步驟?!?/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的主要任務(wù)是根據(jù)子任務(wù)的執(zhí)行結(jié)果,更新計(jì)劃。

Replanner和Planner的prompt模板非常相似,但是約束了Replanner的目標(biāo)任務(wù)、原始Plan、已執(zhí)行的步驟、以及更新計(jì)劃。

比如更新計(jì)劃,我們要求“根據(jù)執(zhí)行的步驟更新計(jì)劃。如果不需要更多步驟,直接可以返回給用戶;否則就填寫計(jì)劃,并向計(jì)劃中添加仍需完成的步驟,不要將之前完成的步驟作為計(jì)劃的一部分返回”

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)

第五步 構(gòu)建流程圖

下面,我們構(gòu)建流程圖,將Planner、Replanner、執(zhí)行器等節(jié)點(diǎn)添加進(jìn)來(lái),執(zhí)行并輸出結(jié)果。

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()

總結(jié)

從原理上看,Plan-and-Execute和ReAct也有一定的相似度,但是Plan-and-Execute的優(yōu)點(diǎn)是具備明確的長(zhǎng)期規(guī)劃,這一點(diǎn)即使非常強(qiáng)大的LLM也難以做到。同時(shí)可以只使用較大的模型做規(guī)劃,而使用較小的模型執(zhí)行步驟,降低執(zhí)行成本。

但是Plan-and-execute的局限性在于,每個(gè)任務(wù)是按順序執(zhí)行的,下一個(gè)任務(wù)都必須等上一個(gè)任務(wù)完成之后才能執(zhí)行,這可能會(huì)導(dǎo)致總執(zhí)行時(shí)間的增加。

一種有效改進(jìn)的辦法是將每個(gè)任務(wù)表示為有向無(wú)環(huán)圖DAG,從而實(shí)現(xiàn)任務(wù)的并行執(zhí)行。這就是下一篇文章要介紹的AI Agent設(shè)計(jì)模式,LLM Compiler。

本文由人人都是產(chǎn)品經(jīng)理作者【風(fēng)叔】,微信公眾號(hào):【風(fēng)叔云】,原創(chuàng)/授權(quán) 發(fā)布于人人都是產(chǎn)品經(jīng)理,未經(jīng)許可,禁止轉(zhuǎn)載。

題圖來(lái)自Unsplash,基于 CC0 協(xié)議。

更多精彩內(nèi)容,請(qǐng)關(guān)注人人都是產(chǎn)品經(jīng)理微信公眾號(hào)或下載App
評(píng)論
評(píng)論請(qǐng)登錄
  1. 目前還沒評(píng)論,等你發(fā)揮!
专题
11812人已学习10篇文章
对于产品、运营人,在不同的职业发展阶段,所需要关注的重点也不同。本专题的文章分享了运营人如何规划职业生涯。
专题
12178人已学习13篇文章
通过仪表盘,用户可以查看并分析产品的数据和图表,还可以通过控件来控制数据的显示、过滤等功能。本专题的文章分享了仪表盘设计指南。
专题
13356人已学习13篇文章
本专题的文章分享了WMS系统设计指南。
专题
30199人已学习19篇文章
2018年过去了,你都收获了什么?新的一年,你需要如何前行?
专题
68727人已学习26篇文章
学会数据化运营能够提升效率,让你的工作事半功倍。
专题
96926人已学习11篇文章
不管你是产品、运营、设计、还是技术,流程图都是基础技能。