跳到内容

计划并执行

本笔记本展示了如何创建一个“计划并执行”风格的 Agent。这在很大程度上受到了 Plan-and-Solve 论文以及 Baby-AGI 项目的启发。

核心思想是首先提出一个多步骤计划，然后一次执行计划中的一项。完成特定任务后，您可以重新审视计划并进行适当修改。

一般的计算图如下所示

plan-and-execute diagram

这与典型的 ReAct 风格的 Agent 形成对比，后者一次思考一个步骤。“计划并执行”风格的 Agent 的优势在于：

  1. 显式的长期规划（即使是非常强大的 LLM 也可能难以做到）
  2. 能够使用较小/较弱的模型进行执行步骤，仅在规划步骤中使用较大/较好的模型

以下演练演示了如何在 LangGraph 中执行此操作。生成的 Agent 将留下如下示例的跟踪记录：（链接）。

设置

首先，我们需要安装所需的软件包。

%%capture --no-stderr
%pip install --quiet -U langgraph langchain-community langchain-openai tavily-python

接下来，我们需要为 OpenAI（我们将使用的 LLM）和 Tavily（我们将使用的搜索工具）设置 API 密钥

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

设置 LangSmith 以进行 LangGraph 开发

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——阅读 此处 了解更多关于如何入门的信息。

定义工具

我们将首先定义我们要使用的工具。对于这个简单的示例，我们将使用通过 Tavily 内置的搜索工具。但是，创建您自己的工具非常容易 - 请参阅 此处 的文档，了解如何执行此操作。

from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=3)]

API 参考：TavilySearchResults

定义我们的执行 Agent

现在我们将创建我们想要用来执行任务的执行 Agent。请注意，对于此示例，我们将对每个任务使用相同的执行 Agent，但这并非必须如此。

from langchain import hub
from langchain_openai import ChatOpenAI

from langgraph.prebuilt import create_react_agent

# Choose the LLM that will drive the agent
llm = ChatOpenAI(model="gpt-4-turbo-preview")
prompt = "You are a helpful assistant."
agent_executor = create_react_agent(llm, tools, prompt=prompt)

API 参考：ChatOpenAI | create_react_agent

agent_executor.invoke({"messages": [("user", "who is the winnner of the us open")]})

{'messages': [HumanMessage(content='who is the winnner of the us open', additional_kwargs={}, response_metadata={}, id='388a14b3-f556-4f91-ad36-def0a075638e'),
  AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_5nbeRa0fgh4ZslRkjk75Kzxs', 'function': {'arguments': '{"query":"US Open 2023 winner"}', 'name': 'tavily_search_results_json'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 23, 'prompt_tokens': 97, 'total_tokens': 120, 'completion_tokens_details': {'reasoning_tokens': 0}}, 'model_name': 'gpt-4-0125-preview', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-3bb25f7a-49e5-43b7-ad53-718bd0107db1-0', tool_calls=[{'name': 'tavily_search_results_json', 'args': {'query': 'US Open 2023 winner'}, 'id': 'call_5nbeRa0fgh4ZslRkjk75Kzxs', 'type': 'tool_call'}], usage_metadata={'input_tokens': 97, 'output_tokens': 23, 'total_tokens': 120}),
  ToolMessage(content='[{"url": "https://www.youtube.com/watch?v=rZ0XQWWFIAo", "content": "The moment Coco Gauff beat Aryna Sabalenka in the final of the 2023 US Open.Don\'t miss a moment of the US Open! Subscribe now: https://bit.ly/2Pdr81iThe 2023..."}, {"url": "https://www.cbssports.com/tennis/news/us-open-2023-scores-novak-djokovic-makes-history-with-24th-grand-slam-title-while-coco-gauff-earns-her-first/", "content": "Here is all you need to know about the 2023 US Open:\\nMen\'s final\\nWomen\'s final\\nMen\'s singles seeds\\nWomen\'s singles seeds\\nOur Latest Tennis Stories\\nUS Open 2023: Schedule, scores, how to watch, seeds\\nRafael Nadal to return next month at Brisbane\\nNovak Djokovic breaks Federer\'s ATP Finals record\\nTennis bettor wins $486,000 off $28 on 10-match parlay\\nTennis player DQ\'d on match point for hitting umpire\\nRafael Nadal says Novak Djokovic is tennis\' GOAT\\nHalep suspended four years for anti-doping violations\\nDjokovic pays tribute to Kobe after winning US Open\\nDjokovic vs. Medvedev odds, US Open final picks, bets\\nAryna Sabalenka-Coco Gauff odds, US Open final picks\\n© 2004-2023 CBS Interactive. Novak Djokovic makes history with 24th Grand Slam title, while Coco Gauff earns her first\\nThe 2023 US Open is officially in the books\\nThe 2023 US open came to a close as Coco Gauff earned her first major title and Novak Djokovic made history with his 24th Grand Slam trophy. Gauff is the first woman to win the Cincinnati Masters 1000 and US Open in the same year since Williams in 2014.\\n Gauff landed in New York as the No. 6 player in the world but will be climbing to a career-best No. 3 when the next rankings get released.\\n He arrived to this competition as the world No. 2 but will improve to No. 1 in the next rankings, extending his record total of 389 weeks at the top.\\n"}, {"url": "https://www.usopen.org/en_US/news/articles/2023-09-10/novak_djokovic_wins_24th_grand_slam_singles_title_at_2023_us_open.html", "content": "Novak Djokovic defeated Daniil Medvedev in three sets to claim his 24th Grand Slam singles title and match Margaret Court\'s all-time record. The Serb saved a set point in the second set and attacked the net to win his fourth US Open crown."}]', name='tavily_search_results_json', id='3ea00623-86b3-4d6f-9978-3503a7eecf0f', tool_call_id='call_5nbeRa0fgh4ZslRkjk75Kzxs', artifact={'query': 'US Open 2023 winner', 'follow_up_questions': None, 'answer': None, 'images': [], 'results': [{'title': "Championship Point | Coco Gauff Wins Women's Singles Title | 2023 US Open", 'url': 'https://www.youtube.com/watch?v=rZ0XQWWFIAo', 'content': "The moment Coco Gauff beat Aryna Sabalenka in the final of the 2023 US Open.Don't miss a moment of the US Open! Subscribe now: https://bit.ly/2Pdr81iThe 2023...", 'score': 0.9975177, 'raw_content': None}, {'title': 'US Open 2023 scores: Novak Djokovic makes history with 24th Grand Slam ...', 'url': 'https://www.cbssports.com/tennis/news/us-open-2023-scores-novak-djokovic-makes-history-with-24th-grand-slam-title-while-coco-gauff-earns-her-first/', 'content': "Here is all you need to know about the 2023 US Open:\nMen's final\nWomen's final\nMen's singles seeds\nWomen's singles seeds\nOur Latest Tennis Stories\nUS Open 2023: Schedule, scores, how to watch, seeds\nRafael Nadal to return next month at Brisbane\nNovak Djokovic breaks Federer's ATP Finals record\nTennis bettor wins $486,000 off $28 on 10-match parlay\nTennis player DQ'd on match point for hitting umpire\nRafael Nadal says Novak Djokovic is tennis' GOAT\nHalep suspended four years for anti-doping violations\nDjokovic pays tribute to Kobe after winning US Open\nDjokovic vs. Medvedev odds, US Open final picks, bets\nAryna Sabalenka-Coco Gauff odds, US Open final picks\n© 2004-2023 CBS Interactive. Novak Djokovic makes history with 24th Grand Slam title, while Coco Gauff earns her first\nThe 2023 US Open is officially in the books\nThe 2023 US open came to a close as Coco Gauff earned her first major title and Novak Djokovic made history with his 24th Grand Slam trophy. Gauff is the first woman to win the Cincinnati Masters 1000 and US Open in the same year since Williams in 2014.\n Gauff landed in New York as the No. 6 player in the world but will be climbing to a career-best No. 3 when the next rankings get released.\n He arrived to this competition as the world No. 2 but will improve to No. 1 in the next rankings, extending his record total of 389 weeks at the top.\n", 'score': 0.9937101, 'raw_content': None}, {'title': 'Novak Djokovic wins 24th Grand Slam singles title at 2023 US Open', 'url': 'https://www.usopen.org/en_US/news/articles/2023-09-10/novak_djokovic_wins_24th_grand_slam_singles_title_at_2023_us_open.html', 'content': "Novak Djokovic defeated Daniil Medvedev in three sets to claim his 24th Grand Slam singles title and match Margaret Court's all-time record. The Serb saved a set point in the second set and attacked the net to win his fourth US Open crown.", 'score': 0.8146434, 'raw_content': None}], 'response_time': 2.24}),
  AIMessage(content="The winners of the 2023 US Open are Coco Gauff and Novak Djokovic. Coco Gauff won her first major title at the US Open, making history, while Novak Djokovic secured his 24th Grand Slam title, matching Margaret Court's all-time record and winning his fourth US Open crown. Coco Gauff defeated Aryna Sabalenka in the final, and Novak Djokovic defeated Daniil Medvedev.", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 93, 'prompt_tokens': 751, 'total_tokens': 844, 'completion_tokens_details': {'reasoning_tokens': 0}}, 'model_name': 'gpt-4-0125-preview', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-eedb1782-6120-441d-ab5d-ccf6bef75b02-0', usage_metadata={'input_tokens': 751, 'output_tokens': 93, 'total_tokens': 844})]}

定义状态

现在让我们开始定义此 Agent 要跟踪的状态。

首先，我们需要跟踪当前的计划。让我们将其表示为字符串列表。

接下来，我们应该跟踪之前执行的步骤。让我们将其表示为元组列表（这些元组将包含步骤，然后是结果）

最后，我们需要一些状态来表示最终响应以及原始输入。

import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDict


class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple], operator.add]
    response: str

规划步骤

现在让我们考虑创建规划步骤。这将使用函数调用来创建计划。

将 Pydantic 与 LangChain 结合使用

本笔记本使用 Pydantic v2 BaseModel，这需要 langchain-core >= 0.3。使用 langchain-core < 0.3 将导致由于 Pydantic v1 和 v2 BaseModel 混合而产生的错误。

from pydantic 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)

API 参考：ChatPromptTemplate

planner.invoke(
    {
        "messages": [
            ("user", "what is the hometown of the current Australia open winner?")
        ]
    }
)

Plan(steps=['Identify the current winner of the Australia Open.', 'Find the hometown of the identified winner.'])

重新规划步骤

现在，让我们创建一个根据上一步的结果重新制定计划的步骤。

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)

创建图

我们现在可以创建图了！

from typing import Literal
from langgraph.graph import END


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}\n\nYou 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):
    if "response" in state and state["response"]:
        return END
    else:
        return "agent"

API 参考：END

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,
    ["agent", END],
)

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()

API 参考：StateGraph | START

from IPython.display import Image, display

display(Image(app.get_graph(xray=True).draw_mermaid_png()))


config = {"recursion_limit": 50}
inputs = {"input": "what is the hometown of the mens 2024 Australia open winner?"}
async for event in app.astream(inputs, config=config):
    for k, v in event.items():
        if k != "__end__":
            print(v)

{'plan': ["Identify the winner of the men's 2024 Australian Open.", 'Research the hometown of the identified winner.']}
{'past_steps': [("Identify the winner of the men's 2024 Australian Open.", "The winner of the men's singles tennis title at the 2024 Australian Open was Jannik Sinner. He defeated Daniil Medvedev in the final with scores of 3-6, 3-6, 6-4, 6-4, 6-3 to win his first major singles title.")]}
{'plan': ['Research the hometown of Jannik Sinner.']}
{'past_steps': [('Research the hometown of Jannik Sinner.', "Jannik Sinner's hometown is Sexten, which is located in northern Italy.")]}
{'response': "The hometown of the men's 2024 Australian Open winner, Jannik Sinner, is Sexten, located in northern Italy."}

结论

恭喜您制作了一个计划并执行的 Agent！上述设计的一个已知限制是，每个任务仍然按顺序执行，这意味着所有令人尴尬的并行操作都会增加总执行时间。您可以通过将每个任务表示为 DAG（类似于 LLMCompiler），而不是常规列表来改进这一点。

评论