LangGraph(五)——Plan-and-Execute Agents

一、Plan-and-Execute Agents

Plan-and-execute 架构：这是一种将规划（plan）和执行（execute）分离的智能代理设计模式。

LangGraph提出了三种 Plan-and-Execute 风格的 Agent，并且相对于传统ReAct风格的Agnet做了很多改进。

1. 速度提升：无需在每个动作后咨询更大的代理，子任务可以独立执行，减少对大型语言模型（LLM）的调用。
2. 成本节约：如果子任务调用LLM，通常可以使用更小的、特定领域的模型，而大型模型仅用于（重新）规划步骤和生成最终响应。
3. 性能提升：通过强制规划者明确“思考”完成整个任务所需的所有步骤，可以提高任务完成率和质量。

1.1 背景

LLM Angen通常有以下主要步骤：

1. 建议action：LLM生成文本直接响应用户或传递给函数。
2. 执行action：代码调用其他软件执行查询数据库或调用API等操作。
3. 观察：根据工具调用的响应，调用另一个函数或响应用户。

ReAct Agent 是一个典型的设计，使用重复的思考、行动、观察循环来提示语言模型:

Thought: I should call Search() to see the current score of the game.
Act: Search("What is the current score of game X?")
Observation: The current score is 24-21
... (repeat N times)

其利用了思想链(Chain-of-thought)提示，让每一步做出单一的操作选择。虽然这对于简单的任务可能有效，但它有几个主要缺点：

1. 每次工具调用都需要一个 LLM 调用。
2. LLM 一次仅计划 1 个子问题。这可能会导致不是最优过程，因为它不会强制“推理”整个任务。

克服这两个缺点的一种方法是通过明确的规划步骤。下面是我们在 LangGraph 中实现的两个这样的设计。

1.2 计划与执行（Plan-and-Execute）架构

这个想法很大程度上是受到 BabyAGI 和后来的《Plan-and-Solve Prompting》论文的启发

planner：提示LLM生成完成大型任务的多步骤计划。
Executor：接受用户查询和计划中的步骤，并调用一个或多个工具来完成任务。

执行完成后，Agent 再次被调用，使用重新规划提示，决定是完成响应还是生成后续计划。

这种Agent设计让我们不必再将整个大Plan直接调用LLM，而是分成多个小的Plan。但是它仍然受到串行工具调用的限制，并为每个任务都得LLM，因为它不支持变量分配。

▶

LangGraph代码示例

定义模型和工具

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_agent_executor

tools = [TavilySearchResults(max_results=3)]

# prompt
prompt = hub.pull("hwchase17/openai-functions-agent")

# LLM
llm = ChatOpenAI(model="gpt-4-turbo")

# Agent
agent_runnable = create_openai_functions_agent(llm, tools, prompt)

agent_executor = create_agent_executor(agent_runnable, tools)

定义state

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List, Tuple, Annotated, TypedDict
import operator


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

定义Planner

from langchain_core.pydantic_v1 import BaseModel
from langchain.chains.openai_functions import create_structured_output_runnable
from langchain_core.prompts import ChatPromptTemplate


class Plan(BaseModel):
    """Plan to follow in future"""

    steps: List[str] = Field(
        description="different steps to follow, should be in sorted order"
    )
        


planner_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.

{objective}"""
)

planner = create_structured_output_runnable(
    Plan, ChatOpenAI(model="gpt-4-turbo", temperature=0), planner_prompt
)

定义Re-Plannner

from langchain.chains.openai_functions import create_openai_fn_runnable


class Response(BaseModel):
    """Response to user."""

    response: str


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 = create_openai_fn_runnable(
    [Plan, Response],
    ChatOpenAI(model="gpt-4-turbo", temperature=0),
    replanner_prompt,
)

定义Nodes

async def execute_step(state: PlanExecute):
    task = state["plan"][0]
    agent_response = await agent_executor.ainvoke({"input": task, "chat_history": []})
    return {
        "past_steps": (task, agent_response["agent_outcome"].return_values["output"])
    }


async def plan_step(state: PlanExecute):
    plan = await planner.ainvoke({"objective": state["input"]})
    return {"plan": plan.steps}


async def replan_step(state: PlanExecute):
    output = await replanner.ainvoke(state)
    if isinstance(output, Response):
        return {"response": output.response}
    else:
        return {"plan": output.steps}


def should_end(state: PlanExecute):
    if "response" in state and state["response"]:
        return "True"
    else:
        return "False"

定义Edges

from langgraph.graph import StateGraph, END

workflow = StateGraph(PlanExecute)

# plan node
workflow.add_node("planner", plan_step)

# execution step
workflow.add_node("agent", execute_step)

# replan node
workflow.add_node("replan", replan_step)

workflow.set_entry_point("planner")

# plan  to agent
workflow.add_edge("planner", "agent")

# agent, 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,
    {
        # If `tools`, then we call the tool node.
        "True": END,
        "False": "agent",
    },
)

# 编译成runnable
app = workflow.compile()

执行

from langchain_core.messages import HumanMessage

config = {"recursion_limit": 50}
inputs = {"input": "what is the hometown of the 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 2024 Australia Open.', "Research the winner's biography to find out their hometown.", 'Confirm the hometown from multiple sources to ensure accuracy.']}
{'past_steps': ('Identify the winner of the 2024 Australia Open.', 'The winner of the 2024 Australian Open is Jannik Sinner. He won his first Grand Slam title by defeating Daniil Medvedev in a thrilling five-set final.')}
{'plan': ["Research Jannik Sinner's biography to find out his hometown.", 'Confirm the hometown from multiple sources to ensure accuracy.']}
{'past_steps': ("Research Jannik Sinner's biography to find out his hometown.", "Jannik Sinner's hometown is Innichen, a town located in the mainly German-speaking area of South Tyrol in northern Italy.")}
{'response': 'The hometown of the 2024 Australian Open winner, Jannik Sinner, is Innichen, Italy.'}

LangSmith流程展示(太长，有部分折叠)：

1.3 Reasoning WithOut Observations (ReWOO)

ReWOO论文提出了一种Agent，在这个Agent中并不是每个任务都会使用 LLM ，同时仍然允许任务依赖于先前的任务结果。他们通过允许在Planner的输出中进行变量分配来实现这一点。下图是Agent设计的示意图。

Planner生成一个由Plan和E#交错组成的列表，例如：假设用户查询“今年超级碗竞争者四分卫的统计数据是什么”，Planner可以生成以下计划：

Plan: I need to know the teams playing in the superbowl this year
E1: Search[Who is competing in the superbowl?]
Plan: I need to know the quarterbacks for each team
E2: LLM[Quarterback for the first team of #E1]
Plan: I need to know the quarterbacks for each team
E3: LLM[Quarter back for the second team of #E1]
Plan: I need to look up stats for the first quarterback
E4: Search[Stats for #E2]
Plan: I need to look up stats for the second quarterback
E5: Search[Stats for #E3]

• Planner可以使用类似#E2的语法引用之前的输出。这意味着它在执行任务列表时不必每次都重新规划。
• work节点循环遍历每个任务，并将任务输出分配给相应的变量。当调用后续调用时，它还会用变量的结果替换变量。
• 最后，slover将所有这些输出整合为最终答案。

这个设计比原生的 Plan-and-Execute Agent 更有效，每个任务只具有所需的上下文（其输入和变量值）。

• 允许在规划器的输出中进行变量赋值，从而在不需要每次都使用LLM的情况下执行任务，同时允许任务依赖于之前任务的结果。然而，它仍然依赖于顺序任务执行，这可能会产生更长的运行时间。

▶

LangGraph代码示例

定义模型

from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0)

prompt = """For the following task, make plans that can solve the problem step by step. For each plan, indicate \
which external tool together with tool input to retrieve evidence. You can store the evidence into a \
variable #E that can be called by later tools. (Plan, #E1, Plan, #E2, Plan, ...)

Tools can be one of the following:
(1) Google[input]: Worker that searches results from Google. Useful when you need to find short
and succinct answers about a specific topic. The input should be a search query.
(2) LLM[input]: A pretrained LLM like yourself. Useful when you need to act with general
world knowledge and common sense. Prioritize it when you are confident in solving the problem
yourself. Input can be any instruction.

For example,
Task: Thomas, Toby, and Rebecca worked a total of 157 hours in one week. Thomas worked x
hours. Toby worked 10 hours less than twice what Thomas worked, and Rebecca worked 8 hours
less than Toby. How many hours did Rebecca work?
Plan: Given Thomas worked x hours, translate the problem into algebraic expressions and solve
with Wolfram Alpha. #E1 = WolframAlpha[Solve x + (2x − 10) + ((2x − 10) − 8) = 157]
Plan: Find out the number of hours Thomas worked. #E2 = LLM[What is x, given #E1]
Plan: Calculate the number of hours Rebecca worked. #E3 = Calculator[(2 ∗ #E2 − 10) − 8]

Begin! 
Describe your plans with rich details. Each Plan should be followed by only one #E.

Task: {task}"""

task = "what is the hometown of the 2024 australian open winner"

Graph State

from typing import TypedDict, List


class ReWOO(TypedDict):
    task: str
    plan_string: str
    steps: List
    results: dict
    result: str

Planner Node

import re
from langchain_core.prompts import ChatPromptTemplate

# Regex to match expressions of the form E#... = ...[...]
regex_pattern = r"Plan:\s*(.+)\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"
prompt_template = ChatPromptTemplate.from_messages([("user", prompt)])
planner = prompt_template | model


def get_plan(state: ReWOO):
    task = state["task"]
    result = planner.invoke({"task": task})
    # Find all matches in the sample text
    matches = re.findall(regex_pattern, result.content)
    return {"steps": matches, "plan_string": result.content}

Executor

from langchain_community.tools.tavily_search import TavilySearchResults

search = TavilySearchResults()

def _get_current_task(state: ReWOO):
    if state["results"] is None:
        return 1
    if len(state["results"]) == len(state["steps"]):
        return None
    else:
        return len(state["results"]) + 1


def tool_execution(state: ReWOO):
    """Worker node that executes the tools of a given plan."""
    _step = _get_current_task(state)
    _, step_name, tool, tool_input = state["steps"][_step - 1]
    _results = state["results"] or {}
    for k, v in _results.items():
        tool_input = tool_input.replace(k, v)
    if tool == "Google":
        result = search.invoke(tool_input)
    elif tool == "LLM":
        result = model.invoke(tool_input)
    else:
        raise ValueError
    _results[step_name] = str(result)
    return {"results": _results}

Solver

solve_prompt = """Solve the following task or problem. To solve the problem, we have made step-by-step Plan and \
retrieved corresponding Evidence to each Plan. Use them with caution since long evidence might \
contain irrelevant information.

{plan}

Now solve the question or task according to provided Evidence above. Respond with the answer
directly with no extra words.

Task: {task}
Response:"""


def solve(state: ReWOO):
    plan = ""
    for _plan, step_name, tool, tool_input in state["steps"]:
        _results = state["results"] or {}
        for k, v in _results.items():
            tool_input = tool_input.replace(k, v)
            step_name = step_name.replace(k, v)
        plan += f"Plan: {_plan}\n{step_name} = {tool}[{tool_input}]"
    prompt = solve_prompt.format(plan=plan, task=state["task"])
    result = model.invoke(prompt)
    return {"result": result.content}

Graph

from langgraph.graph import StateGraph, END

def _route(state):
    _step = _get_current_task(state)
    if _step is None:
        # We have executed all tasks
        return "solve"
    else:
        # We are still executing tasks, loop back to the "tool" node
        return "tool"

graph = StateGraph(ReWOO)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.set_entry_point("plan")


app = graph.compile()

执行

for s in app.stream({"task": task}):
    print(s)
    print("---")

输出

{'plan': {'plan_string': 'Plan: Use Google to search for the 2024 Australian Open winner. #E1 = Google[2024 Australian Open winner]\nPlan: Once the winner is identified, search for their hometown using Google. #E2 = Google[Hometown of 2024 Australian Open winner]', 'steps': [('Use Google to search for the 2024 Australian Open winner. ', '#E1', 'Google', '2024 Australian Open winner'), ('Once the winner is identified, search for their hometown using Google. ', '#E2', 'Google', 'Hometown of 2024 Australian Open winner')]}}
---
{'tool': {'results': {'#E1': '[{\'url\': \'https://ausopen.com/articles/news/sinner-winner-italian-takes-first-major-ao-2024\', \'content\': \'The agile right-hander, who had claimed victory from a two-set deficit only once previously in his young career, is the second Italian man to achieve singles glory at a major, following Adriano Panatta in1976.With victories over Andrey Rublev, 10-time AO champion Novak Djokovic, and Medvedev, the Italian is the youngest player to defeat top 5 opponents in the final three matches of a major since Michael Stich did it at Wimbledon in 1991 – just weeks before Sinner was born.\\n He saved the only break he faced with an ace down the tee, and helped by scoreboard pressure, broke Medvedev by slamming a huge forehand to force an error from his more experienced rival, sealing the fourth set to take the final to a decider.\\n Sensing a shift in momentum as Medvedev served to close out the second at 5-3, Sinner set the RLA crowd alight with a pair of brilliant passing shots en route to creating a break point opportunity, which Medvedev snuffed out with trademark patience, drawing a forehand error from his opponent. “We are trying to get better every day, even during the tournament we try to get stronger, trying to understand every situation a little bit better, and I’m so glad to have you there supporting me, understanding me, which sometimes it’s not easy because I am a little bit young sometimes,” he said with a smile.\\n Medvedev, who held to love in his first three service games of the second set, piled pressure on the Italian, forcing the right-hander to produce his best tennis to save four break points in a nearly 12-minute second game.\\n\'}, {\'url\': \'https://m.youtube.com/watch?v=frRVq6FI_5c\', \'content\': \'Watch the highlights of Jannik Sinner v Daniil Medvedev in the final of the Australian Open 2024.Subscribe to keep up with the latest from the Australian Ope...\'}, {\'url\': \'https://www.bbc.com/sport/tennis/68120937\', \'content\': \'Live scores, results and order of play\\nAlerts: Get tennis news sent to your phone\\nRelated Topics\\nTop Stories\\nFA Cup: Blackburn Rovers v Wrexham - live text commentary\\nRussian skater Valieva given four-year ban for doping\\nLinks to Barcelona are \\\'totally untrue\\\' - Arteta\\nElsewhere on the BBC\\nThe truth behind the fake grooming scandal\\nFeaturing unseen police footage and interviews with the officers at the heart of the case\\nDid their father and uncle kill Nazi war criminals?\\n A real-life murder mystery following three brothers in their quest for the truth\\nWhat was it like to travel on the fastest plane?\\nTake a behind-the-scenes look at the supersonic story of the Concorde\\nToxic love, ruthless ambition and shocking betrayal\\nTell Me Lies follows a passionate college relationship with unimaginable consequences...\\n "\\nMarathon man Medvedev runs out of steam\\nMedvedev is the first player to lose two Grand Slam finals after winning the opening two sets\\nSo many players with the experience of a Grand Slam final have talked about how different the occasion can be, particularly if it is the first time, and potentially overwhelming.\\n Jannik Sinner beats Daniil Medvedev in Melbourne final\\nJannik Sinner is the youngest player to win the Australian Open men\\\'s title since Novak Djokovic in 2008\\nJannik Sinner landed the Grand Slam title he has long promised with an extraordinary fightback to beat Daniil Medvedev in the Australian Open final.\\n "\\nSinner starts 2024 in inspired form\\nSinner won the first Australian Open men\\\'s final since 2005 which did not feature Roger Federer, Rafael Nadal or Novak Djokovic\\nSinner was brought to the forefront of conversation when discussing Grand Slam champions in 2024 following a stunning end to last season.\\n\'}, {\'url\': \'https://www.cbssports.com/tennis/news/australian-open-2024-jannik-sinner-claims-first-grand-slam-title-in-epic-comeback-win-over-daniil-medvedev/\', \'content\': \'"\\nOur Latest Tennis Stories\\nSinner makes epic comeback to win Australian Open\\nSinner, Sabalenka win Australian Open singles titles\\n2024 Australian Open odds, Sinner vs. Medvedev picks\\nSabalenka defeats Zheng to win 2024 Australian Open\\n2024 Australian Open odds, Sabalenka vs. Zheng picks\\n2024 Australian Open odds, Medvedev vs. Zverev picks\\nAustralian Open odds: Djokovic vs. Sinner picks, bets\\nAustralian Open odds: Gauff vs. Sabalenka picks, bets\\nAustralian Open odds: Zheng vs. Yastremska picks, bets\\nNick Kyrgios reveals he\\\'s contemplating retirement\\n© 2004-2024 CBS Interactive. Jannik Sinner claims first Grand Slam title in epic comeback win over Daniil Medvedev\\nSinner, 22, rallied back from a two-set deficit to become the third ever Italian Grand Slam men\\\'s singles champion\\nAfter almost four hours, Jannik Sinner climbed back from a two-set deficit to win his first ever Grand Slam title with an epic 3-6, 3-6, 6-4, 6-4, 6-3 comeback victory against Daniil Medvedev. Sinner became the first Italian man to win the Australian Open since 1976, and just the eighth man to successfully come back from two sets down in a major final.\\n He did not drop a single set until his meeting with Djokovic, and that win in itself was an accomplishment as Djokovic was riding a 33-match winning streak at the Australian Open and had never lost a semifinal in Melbourne.\\n @janniksin • @wwos • @espn • @eurosport • @wowowtennis pic.twitter.com/DTCIqWoUoR\\n"We are trying to get better everyday, and even during the tournament, trying to get stronger and understand the situation a little bit better," Sinner said.\'}, {\'url\': \'https://www.cnn.com/2024/01/28/sport/jannik-sinner-daniil-medvedev-2024-australian-open-spt-intl/index.html\', \'content\': \'Related article\\nMeet the Carota Boys, the carrot-clad fan group rooting for Jannik Sinner, Italy’s rising tennis star\\nSuch drama provided a fitting finale for a tournament which has featured an incredible 35 five-set thrillers, equaling the Open Era record for any grand slam, as Medvedev raced to a two-set lead, initially heaping pressure on the young Italian who was left defending from every corner of the court in the wake of the No. 3 seed’s precise, aggressive approach.\\n CNN values your feedback\\nJannik Sinner rallies from two sets down to win men’s Australian Open final over Daniil Medvedev, his first grand slam title\\nFollow:\\nJannik Sinner overcame a two-set deficit to defeat Russia’s Daniil Medvedev 3-6 3-6 6-4 6-4 6-3 in a thrilling five-set final and claim the men’s Australian Open title on Sunday, becoming the first Italian man to win a grand slam since 1976.\\n While Sinner was appearing in his first ever grand slam final, Medvedev is already a grand slam champion who was playing in his fourth major final and that disparity in experience initally showed as the Russian controlled the pace of the match with 14 winners and five aces in the first set.\\n As Medvedev tired, Sinner took control of the match and sealed a crucial break point early in the fifth set with a crosscourt forehand winner before completing his remarkable comeback with a forehand down the line, falling to the floor immediately afterwards to celebrate a stunning maiden grand slam title.\\n That momentum carried Sinner through the fourth and fifth sets as he overturned the deficit and became the youngest male player to win the Australian Open since 2008, as well as just the third Italian man to ever win a grand slam.\\n\'}]'}}}
---
{'tool': {'results': {'#E1': '[{\'url\': \'https://ausopen.com/articles/news/sinner-winner-italian-takes-first-major-ao-2024\', \'content\': \'The agile right-hander, who had claimed victory from a two-set deficit only once previously in his young career, is the second Italian man to achieve singles glory at a major, following Adriano Panatta in1976.With victories over Andrey Rublev, 10-time AO champion Novak Djokovic, and Medvedev, the Italian is the youngest player to defeat top 5 opponents in the final three matches of a major since Michael Stich did it at Wimbledon in 1991 – just weeks before Sinner was born.\\n He saved the only break he faced with an ace down the tee, and helped by scoreboard pressure, broke Medvedev by slamming a huge forehand to force an error from his more experienced rival, sealing the fourth set to take the final to a decider.\\n Sensing a shift in momentum as Medvedev served to close out the second at 5-3, Sinner set the RLA crowd alight with a pair of brilliant passing shots en route to creating a break point opportunity, which Medvedev snuffed out with trademark patience, drawing a forehand error from his opponent. “We are trying to get better every day, even during the tournament we try to get stronger, trying to understand every situation a little bit better, and I’m so glad to have you there supporting me, understanding me, which sometimes it’s not easy because I am a little bit young sometimes,” he said with a smile.\\n Medvedev, who held to love in his first three service games of the second set, piled pressure on the Italian, forcing the right-hander to produce his best tennis to save four break points in a nearly 12-minute second game.\\n\'}, {\'url\': \'https://m.youtube.com/watch?v=frRVq6FI_5c\', \'content\': \'Watch the highlights of Jannik Sinner v Daniil Medvedev in the final of the Australian Open 2024.Subscribe to keep up with the latest from the Australian Ope...\'}, {\'url\': \'https://www.bbc.com/sport/tennis/68120937\', \'content\': \'Live scores, results and order of play\\nAlerts: Get tennis news sent to your phone\\nRelated Topics\\nTop Stories\\nFA Cup: Blackburn Rovers v Wrexham - live text commentary\\nRussian skater Valieva given four-year ban for doping\\nLinks to Barcelona are \\\'totally untrue\\\' - Arteta\\nElsewhere on the BBC\\nThe truth behind the fake grooming scandal\\nFeaturing unseen police footage and interviews with the officers at the heart of the case\\nDid their father and uncle kill Nazi war criminals?\\n A real-life murder mystery following three brothers in their quest for the truth\\nWhat was it like to travel on the fastest plane?\\nTake a behind-the-scenes look at the supersonic story of the Concorde\\nToxic love, ruthless ambition and shocking betrayal\\nTell Me Lies follows a passionate college relationship with unimaginable consequences...\\n "\\nMarathon man Medvedev runs out of steam\\nMedvedev is the first player to lose two Grand Slam finals after winning the opening two sets\\nSo many players with the experience of a Grand Slam final have talked about how different the occasion can be, particularly if it is the first time, and potentially overwhelming.\\n Jannik Sinner beats Daniil Medvedev in Melbourne final\\nJannik Sinner is the youngest player to win the Australian Open men\\\'s title since Novak Djokovic in 2008\\nJannik Sinner landed the Grand Slam title he has long promised with an extraordinary fightback to beat Daniil Medvedev in the Australian Open final.\\n "\\nSinner starts 2024 in inspired form\\nSinner won the first Australian Open men\\\'s final since 2005 which did not feature Roger Federer, Rafael Nadal or Novak Djokovic\\nSinner was brought to the forefront of conversation when discussing Grand Slam champions in 2024 following a stunning end to last season.\\n\'}, {\'url\': \'https://www.cbssports.com/tennis/news/australian-open-2024-jannik-sinner-claims-first-grand-slam-title-in-epic-comeback-win-over-daniil-medvedev/\', \'content\': \'"\\nOur Latest Tennis Stories\\nSinner makes epic comeback to win Australian Open\\nSinner, Sabalenka win Australian Open singles titles\\n2024 Australian Open odds, Sinner vs. Medvedev picks\\nSabalenka defeats Zheng to win 2024 Australian Open\\n2024 Australian Open odds, Sabalenka vs. Zheng picks\\n2024 Australian Open odds, Medvedev vs. Zverev picks\\nAustralian Open odds: Djokovic vs. Sinner picks, bets\\nAustralian Open odds: Gauff vs. Sabalenka picks, bets\\nAustralian Open odds: Zheng vs. Yastremska picks, bets\\nNick Kyrgios reveals he\\\'s contemplating retirement\\n© 2004-2024 CBS Interactive. Jannik Sinner claims first Grand Slam title in epic comeback win over Daniil Medvedev\\nSinner, 22, rallied back from a two-set deficit to become the third ever Italian Grand Slam men\\\'s singles champion\\nAfter almost four hours, Jannik Sinner climbed back from a two-set deficit to win his first ever Grand Slam title with an epic 3-6, 3-6, 6-4, 6-4, 6-3 comeback victory against Daniil Medvedev. Sinner became the first Italian man to win the Australian Open since 1976, and just the eighth man to successfully come back from two sets down in a major final.\\n He did not drop a single set until his meeting with Djokovic, and that win in itself was an accomplishment as Djokovic was riding a 33-match winning streak at the Australian Open and had never lost a semifinal in Melbourne.\\n @janniksin • @wwos • @espn • @eurosport • @wowowtennis pic.twitter.com/DTCIqWoUoR\\n"We are trying to get better everyday, and even during the tournament, trying to get stronger and understand the situation a little bit better," Sinner said.\'}, {\'url\': \'https://www.cnn.com/2024/01/28/sport/jannik-sinner-daniil-medvedev-2024-australian-open-spt-intl/index.html\', \'content\': \'Related article\\nMeet the Carota Boys, the carrot-clad fan group rooting for Jannik Sinner, Italy’s rising tennis star\\nSuch drama provided a fitting finale for a tournament which has featured an incredible 35 five-set thrillers, equaling the Open Era record for any grand slam, as Medvedev raced to a two-set lead, initially heaping pressure on the young Italian who was left defending from every corner of the court in the wake of the No. 3 seed’s precise, aggressive approach.\\n CNN values your feedback\\nJannik Sinner rallies from two sets down to win men’s Australian Open final over Daniil Medvedev, his first grand slam title\\nFollow:\\nJannik Sinner overcame a two-set deficit to defeat Russia’s Daniil Medvedev 3-6 3-6 6-4 6-4 6-3 in a thrilling five-set final and claim the men’s Australian Open title on Sunday, becoming the first Italian man to win a grand slam since 1976.\\n While Sinner was appearing in his first ever grand slam final, Medvedev is already a grand slam champion who was playing in his fourth major final and that disparity in experience initally showed as the Russian controlled the pace of the match with 14 winners and five aces in the first set.\\n As Medvedev tired, Sinner took control of the match and sealed a crucial break point early in the fifth set with a crosscourt forehand winner before completing his remarkable comeback with a forehand down the line, falling to the floor immediately afterwards to celebrate a stunning maiden grand slam title.\\n That momentum carried Sinner through the fourth and fifth sets as he overturned the deficit and became the youngest male player to win the Australian Open since 2008, as well as just the third Italian man to ever win a grand slam.\\n\'}]', '#E2': '[{\'url\': "https://en.wikipedia.org/wiki/2024_Australian_Open_–_Men\'s_singles_final", \'content\': "This was the first Australian Open final since 2005 not to feature any of the Big Three members.[5]\\nMedvedev set an Open Era record for the most time spent playing at a singles major, at 24 hours and 17 minutes.[6] Medvedev also became the first player in the Open Era to lose two major finals after having a two-set lead.[7]\\nBackground[edit]\\nEntering the final, Medvedev lead the career head-to-head 6–3. He became the second Italian man in the Open Era to win a singles major, after Adriano Panatta at the 1976 French Open,[2] and the first new Australian Open champion in ten years, since Stan Wawrinka in 2014.[3] At 22, Sinner was the youngest Australian Open men\'s singles champion and finalist since Novak Djokovic in 2008.[4] Contents\\n2024 Australian Open – Men\'s singles final\\nThe 2024 Australian Open Men\'s Singles final was the championship tennis match of the men\'s singles tournament at the 2024 Australian Open, contested by fourth-seed Jannik Sinner and third-seed Daniil Medvedev. Also in the semifinals, Medvedev came back from two-sets-to-love down against Alexander Zverev to reach a third Australian Open final.[9] Medvedev had already played two other five-set matches, against Emil Ruusuvuori in the second round (when he came back from two-sets-to-love down as well) and against Hubert Hurkacz in the quarterfinals.\\n Novak Djokovic, ending his 33-match winning streak at the Australian Open (dating back from the 2019 tournament), as well as marking the Serbian\'s first-ever defeat in an Australian Open semifinal and his first defeat in any major semifinal since the 2019 French Open."}, {\'url\': \'https://ausopen.com/articles/news/sinner-winner-italian-takes-first-major-ao-2024\', \'content\': \'The agile right-hander, who had claimed victory from a two-set deficit only once previously in his young career, is the second Italian man to achieve singles glory at a major, following Adriano Panatta in1976.With victories over Andrey Rublev, 10-time AO champion Novak Djokovic, and Medvedev, the Italian is the youngest player to defeat top 5 opponents in the final three matches of a major since Michael Stich did it at Wimbledon in 1991 – just weeks before Sinner was born.\\n He saved the only break he faced with an ace down the tee, and helped by scoreboard pressure, broke Medvedev by slamming a huge forehand to force an error from his more experienced rival, sealing the fourth set to take the final to a decider.\\n Sensing a shift in momentum as Medvedev served to close out the second at 5-3, Sinner set the RLA crowd alight with a pair of brilliant passing shots en route to creating a break point opportunity, which Medvedev snuffed out with trademark patience, drawing a forehand error from his opponent. “We are trying to get better every day, even during the tournament we try to get stronger, trying to understand every situation a little bit better, and I’m so glad to have you there supporting me, understanding me, which sometimes it’s not easy because I am a little bit young sometimes,” he said with a smile.\\n Medvedev, who held to love in his first three service games of the second set, piled pressure on the Italian, forcing the right-hander to produce his best tennis to save four break points in a nearly 12-minute second game.\\n\'}, {\'url\': \'https://www.bbc.com/sport/tennis/68120937\', \'content\': \'Live scores, results and order of play\\nAlerts: Get tennis news sent to your phone\\nRelated Topics\\nTop Stories\\nFA Cup: Blackburn Rovers v Wrexham - live text commentary\\nRussian skater Valieva given four-year ban for doping\\nLinks to Barcelona are \\\'totally untrue\\\' - Arteta\\nElsewhere on the BBC\\nThe truth behind the fake grooming scandal\\nFeaturing unseen police footage and interviews with the officers at the heart of the case\\nDid their father and uncle kill Nazi war criminals?\\n A real-life murder mystery following three brothers in their quest for the truth\\nWhat was it like to travel on the fastest plane?\\nTake a behind-the-scenes look at the supersonic story of the Concorde\\nToxic love, ruthless ambition and shocking betrayal\\nTell Me Lies follows a passionate college relationship with unimaginable consequences...\\n "\\nMarathon man Medvedev runs out of steam\\nMedvedev is the first player to lose two Grand Slam finals after winning the opening two sets\\nSo many players with the experience of a Grand Slam final have talked about how different the occasion can be, particularly if it is the first time, and potentially overwhelming.\\n Jannik Sinner beats Daniil Medvedev in Melbourne final\\nJannik Sinner is the youngest player to win the Australian Open men\\\'s title since Novak Djokovic in 2008\\nJannik Sinner landed the Grand Slam title he has long promised with an extraordinary fightback to beat Daniil Medvedev in the Australian Open final.\\n "\\nSinner starts 2024 in inspired form\\nSinner won the first Australian Open men\\\'s final since 2005 which did not feature Roger Federer, Rafael Nadal or Novak Djokovic\\nSinner was brought to the forefront of conversation when discussing Grand Slam champions in 2024 following a stunning end to last season.\\n\'}, {\'url\': \'https://en.wikipedia.org/wiki/2024_Australian_Open\', \'content\': "The tournament featured the following changes from previous tournaments:[9]\\nSingles players[edit]\\nEvents[edit]\\nMen\'s singles[edit]\\nWomen\'s singles[edit]\\nMen\'s doubles[edit]\\nWomen\'s doubles[edit]\\nMixed doubles[edit]\\nWheelchair men\'s singles[edit]\\nWheelchair women\'s singles[edit]\\nWheelchair quad singles[edit]\\nWheelchair men\'s doubles[edit]\\nWheelchair women\'s doubles[edit]\\nWheelchair quad doubles[edit]\\nBoys\' singles[edit]\\nGirls\' singles[edit]\\nBoys\' doubles[edit]\\nGirls\' doubles[edit]\\nPoints and prize money[edit]\\nPoint distribution[edit]\\nBelow is a series of tables for each competition showing the ranking points offered for each event.[12][13][14]\\nPrize money[edit]\\nThe Australian Open total prize money for 2024 increased by 13.07% year on year to a tournament record A$86,500,000. Aryna Sabalenka successfully defended the women\'s singles title as she claimed her second major singles title, defeating Zheng Qinwen without losing a set during the tournament.[6][7]\\nIn the tournament\'s 119-year history, this was the first Australian Open Tennis Championships to be held on an opening Sunday.[8]\\n It was the 112th edition of the Australian Open, the 56th in the Open Era, and the first major of the year. The total represented a 162% increase in prize money over the last ten years, from the A$33 million on offer in 2014.\\n Contents\\n2024 Australian Open\\nThe 2024 Australian Open was a Grand Slam level tennis tournament held at Melbourne Park, from 14–28 January 2024.[1]"}, {\'url\': \'https://ausopen.com/articles/news/sinner-v-medvedev-how-ao-2024-mens-final-was-decided\', \'content\': "MORE: All the stats from the AO 2024 men’s final\\nDaniil Medvedev could not maintain his scintillating start on Sunday night, while first-time major finalist Jannik Sinner simply grew in strength as the five-set contest progressed.\\nSinner, winner: Italian takes first major at AO 2024\\nNEWS\\nSinner, the morning after: “It\'s great emotions, slowly realising what I\'ve done”\\nNEWS\\n Sinner v Medvedev: How the AO 2024 men\'s final was decided\\nJannik Sinner weathered an early onslaught to reel in Daniil Medvedev, growing in potency to win the Australian Open 2024 final – his first Grand Slam singles title.\\n Early in the second set Medvedev was averaging 77 per cent of first serves in for the match, and towards the end of it this figure had dipped to 65. Medvedev had spent more than 20 hours on court in his previous six rounds, the equivalent of almost two more matches than Sinner (who had spent less than 15 hours on court in advancing to the final).\\n During the shift, Sinner’s forehand was gathering speed, having increased from an average of 122.3 km/h in the first set to 128.7 km/h by the third.\\n"}]'}}}
---
{'solve': {'result': 'The hometown of the 2024 Australian Open winner, Jannik Sinner, is Italy.'}}
---

LangSmith流程展示：

1.4 LLMCompiler

LLMCompiler是一种Agent架构，旨在进一步提高任务执行速度，超越了Plan-and-Execute和ReWOO Agent，甚至超越了OpenAI的并行工具调用。

LLMCompiler有以下主要组件:

• Planner：流式传输任务的 DAG。每个任务都包含一个工具、参数和依赖项列表。
• 任务获取单元(Task Fetching Unit)：调度和执行任务，接受任务流。一旦满足依赖关系就安排任务。由于许多工具涉及对搜索引擎或 LLM 的其他调用，因此额外的并行性可以显着提高速度（论文声称提高了 3.6 倍）
• Joiner：一个LLM步骤，根据整个图历史（包括任务执行结果）动态重新规划或结束，决定是否用最终答案响应，或者将进度传回规划代理继续工作。

关键提升点：

• Planner的输出是流式的，输出解析器不停的解析任务参数及其依赖关系。
• 任务获取单元(Task Fetching Unit)接收解析的任务流，并在满足所有任务的依赖关系后调度任务。
• 任务参数可以是变量，它们是 DAG 中先前任务的输出。 例如，模型可以调用 search("${1}") 来搜索任务 1 的输出生成的查询。这使得代理的工作速度比 OpenAI 中调用的并行工具更快。

通过将任务格式化为 DAG，代理可以在调用工具时节省宝贵的时间，从而带来更好的整体用户体验。

▶

LangGraph代码示例

定义模型和工具

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults

from langchain_community.tools.google_serper import GoogleSerperResults
from langchain_community.utilities import GoogleSerperAPIWrapper

# Imported from the https://github.com/langchain-ai/langgraph/tree/main/examples/plan-and-execute repo
from math_tools import get_math_tool

calculate = get_math_tool(ChatOpenAI(model="gpt-4-turbo"))
search = GoogleSerperResults(api_wrapper=GoogleSerperAPIWrapper(k=1),
                             description='google_serper_results_json(query="the search query") - a search engine.',)

tools = [search, calculate]

Planner^[1]

下面的代码构建了Planner的提示模板，并将其与 LLM 和输出解析器（在output_parser.py 中定义）组合在一起。输出解析器处理以下形式的任务列表：

1. tool_1(arg1="arg1", arg2=3.5, ...)
Thought: I then want to find out Y by using tool_2
2. tool_2(arg1="", arg2="${1}")'
3. join()<END_OF_PLAN>"

Thought行是可选的。 ${1} 占位符是变量。它们用于将工具（任务）输出路由到其他工具。

from typing import Sequence

from langchain_core.language_models import BaseChatModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain_core.tools import BaseTool
from langchain_core.messages import (
    BaseMessage,
    FunctionMessage,
    HumanMessage,
    SystemMessage,
)

from output_parser import LLMCompilerPlanParser, Task
from langchain import hub
from langchain_openai import ChatOpenAI


prompt = hub.pull("wfh/llm-compiler")
print(prompt.pretty_print())

输出

================================ System Message ================================

Given a user query, create a plan to solve it with the utmost parallelizability. Each plan should comprise an action from the following {num_tools} types:
{tool_descriptions}
{num_tools}. join(): Collects and combines results from prior actions.

 - An LLM agent is called upon invoking join() to either finalize the user query or wait until the plans are executed.
 - join should always be the last action in the plan, and will be called in two scenarios:
   (a) if the answer can be determined by gathering the outputs from tasks to generate the final response.
   (b) if the answer cannot be determined in the planning phase before you execute the plans. Guidelines:
 - Each action described above contains input/output types and description.
    - You must strictly adhere to the input and output types for each action.
    - The action descriptions contain the guidelines. You MUST strictly follow those guidelines when you use the actions.
 - Each action in the plan should strictly be one of the above types. Follow the Python conventions for each action.
 - Each action MUST have a unique ID, which is strictly increasing.
 - Inputs for actions can either be constants or outputs from preceding actions. In the latter case, use the format $id to denote the ID of the previous action whose output will be the input.
 - Always call join as the last action in the plan. Say '<END_OF_PLAN>' after you call join
 - Ensure the plan maximizes parallelizability.
 - Only use the provided action types. If a query cannot be addressed using these, invoke the join action for the next steps.
 - Never introduce new actions other than the ones provided.

============================= Messages Placeholder =============================

{messages}

================================ System Message ================================

Remember, ONLY respond with the task list in the correct format! E.g.:
idx. tool(arg_name=args)
None

def create_planner(
    llm: BaseChatModel, tools: Sequence[BaseTool], base_prompt: ChatPromptTemplate
):
    tool_descriptions = "\n".join(
        f"{i+1}. {tool.description}\n" for i, tool in enumerate(tools) # +1 to offset the 0 starting index, we want it count normally from 1.
    )
    planner_prompt = base_prompt.partial(
        replan="",
        num_tools=len(tools)+1, # Add one because we're adding the join() tool at the end.
        tool_descriptions=tool_descriptions,
    )
    replanner_prompt = base_prompt.partial(
        replan=' - You are given "Previous Plan" which is the plan that the previous agent created along with the execution results '
        "(given as Observation) of each plan and a general thought (given as Thought) about the executed results."
        'You MUST use these information to create the next plan under "Current Plan".\n'
        ' - When starting the Current Plan, you should start with "Thought" that outlines the strategy for the next plan.\n'
        " - In the Current Plan, you should NEVER repeat the actions that are already executed in the Previous Plan.\n"
        " - You must continue the task index from the end of the previous one. Do not repeat task indices.",
        num_tools=len(tools)+1,
        tool_descriptions=tool_descriptions,
    )

    def should_replan(state: list):
        # Context is passed as a system message
        return isinstance(state[-1], SystemMessage)

    def wrap_messages(state: list):
        return {"messages": state}

    def wrap_and_get_last_index(state: list):
        next_task = 0
        for message in state[::-1]:
            if isinstance(message, FunctionMessage):
                next_task = message.additional_kwargs["idx"] + 1
                break
        state[-1].content = state[-1].content + f" - Begin counting at : {next_task}"
        return {"messages": state}

    return (
        RunnableBranch(
            (should_replan, wrap_and_get_last_index | replanner_prompt),
            wrap_messages | planner_prompt,
        )
        | llm
        | LLMCompilerPlanParser(tools=tools)
    )

llm = ChatOpenAI(model="gpt-4-turbo")
# This is the primary "agent" in our application
planner = create_planner(llm, tools, prompt)

Task Fetching Unit

{
    tool: BaseTool,
    dependencies: number[],
}

基本思想是一旦满足工具的依赖性就开始执行工具。 这是通过多线程完成的。我们将下面的任务获取单元和执行器结合起来：

from typing import Any, Union, Iterable, List, Tuple, Dict
from typing_extensions import TypedDict
import re

from langchain_core.runnables import (
    chain as as_runnable,
)

from concurrent.futures import ThreadPoolExecutor, wait
import time


def _get_observations(messages: List[BaseMessage]) -> Dict[int, Any]:
    # Get all previous tool responses
    results = {}
    for message in messages[::-1]:
        if isinstance(message, FunctionMessage):
            results[int(message.additional_kwargs["idx"])] = message.content
    return results


class SchedulerInput(TypedDict):
    messages: List[BaseMessage]
    tasks: Iterable[Task]


def _execute_task(task, observations, config):
    tool_to_use = task["tool"]
    if isinstance(tool_to_use, str):
        return tool_to_use
    args = task["args"]
    try:
        if isinstance(args, str):
            resolved_args = _resolve_arg(args, observations)
        elif isinstance(args, dict):
            resolved_args = {
                key: _resolve_arg(val, observations) for key, val in args.items()
            }
        else:
            # This will likely fail
            resolved_args = args
    except Exception as e:
        return (
            f"ERROR(Failed to call {tool_to_use.name} with args {args}.)"
            f" Args could not be resolved. Error: {repr(e)}"
        )
    try:
        return tool_to_use.invoke(resolved_args, config)
    except Exception as e:
        return (
            f"ERROR(Failed to call {tool_to_use.name} with args {args}."
            + f" Args resolved to {resolved_args}. Error: {repr(e)})"
        )


def _resolve_arg(arg: Union[str, Any], observations: Dict[int, Any]):
    # $1 or ${1} -> 1
    ID_PATTERN = r"\$\{?(\d+)\}?"

    def replace_match(match):
        # If the string is ${123}, match.group(0) is ${123}, and match.group(1) is 123.

        # Return the match group, in this case the index, from the string. This is the index
        # number we get back.
        idx = int(match.group(1))
        return str(observations.get(idx, match.group(0)))

    # For dependencies on other tasks
    if isinstance(arg, str):
        return re.sub(ID_PATTERN, replace_match, arg)
    elif isinstance(arg, list):
        return [_resolve_arg(a, observations) for a in arg]
    else:
        return str(arg)


@as_runnable
def schedule_task(task_inputs, config):
    task: Task = task_inputs["task"]
    observations: Dict[int, Any] = task_inputs["observations"]
    try:
        observation = _execute_task(task, observations, config)
    except Exception:
        import traceback

        observation = traceback.format_exception()  # repr(e) +
    observations[task["idx"]] = observation


def schedule_pending_task(
    task: Task, observations: Dict[int, Any], retry_after: float = 0.2
):
    while True:
        deps = task["dependencies"]
        if deps and (any([dep not in observations for dep in deps])):
            # Dependencies not yet satisfied
            time.sleep(retry_after)
            continue
        schedule_task.invoke({"task": task, "observations": observations})
        break


@as_runnable
def schedule_tasks(scheduler_input: SchedulerInput) -> List[FunctionMessage]:
    """Group the tasks into a DAG schedule."""
    # For streaming, we are making a few simplifying assumption:
    # 1. The LLM does not create cyclic dependencies
    # 2. That the LLM will not generate tasks with future deps
    # If this ceases to be a good assumption, you can either
    # adjust to do a proper topological sort (not-stream)
    # or use a more complicated data structure
    tasks = scheduler_input["tasks"]
    args_for_tasks = {}
    messages = scheduler_input["messages"]
    # If we are re-planning, we may have calls that depend on previous
    # plans. Start with those.
    observations = _get_observations(messages)
    task_names = {}
    originals = set(observations)
    # ^^ We assume each task inserts a different key above to
    # avoid race conditions...
    futures = []
    retry_after = 0.25  # Retry every quarter second
    with ThreadPoolExecutor() as executor:
        for task in tasks:
            deps = task["dependencies"]
            task_names[task["idx"]] = (
                task["tool"] if isinstance(task["tool"], str) else task["tool"].name
            )
            args_for_tasks[task["idx"]] = (task["args"])
            if (
                # Depends on other tasks
                deps
                and (any([dep not in observations for dep in deps]))
            ):
                futures.append(
                    executor.submit(
                        schedule_pending_task, task, observations, retry_after
                    )
                )
            else:
                # No deps or all deps satisfied
                # can schedule now
                schedule_task.invoke(dict(task=task, observations=observations))
                # futures.append(executor.submit(schedule_task.invoke dict(task=task, observations=observations)))

        # All tasks have been submitted or enqueued
        # Wait for them to complete
        wait(futures)
    # Convert observations to new tool messages to add to the state
    new_observations = {
        k: (task_names[k], args_for_tasks[k], observations[k])
        for k in sorted(observations.keys() - originals)
    }
    tool_messages = [
        FunctionMessage(name=name, content=str(obs), additional_kwargs={"idx": k, 'args':task_args})
        for k, (name, task_args, obs) in new_observations.items()
    ]
    return tool_messages

import itertools


@as_runnable
def plan_and_schedule(messages: List[BaseMessage], config):
    tasks = planner.stream(messages, config)
    # Begin executing the planner immediately
    try:
        tasks = itertools.chain([next(tasks)], tasks)
    except StopIteration:
        # Handle the case where tasks is empty.
        tasks = iter([])
    scheduled_tasks = schedule_tasks.invoke(
        {
            "messages": messages,
            "tasks": tasks,
        },
        config,
    )
    return scheduled_tasks

Joiner

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.chains.openai_functions import create_structured_output_runnable
from langchain_core.messages import AIMessage


class FinalResponse(BaseModel):
    """The final response/answer."""

    response: str


class Replan(BaseModel):
    feedback: str = Field(
        description="Analysis of the previous attempts and recommendations on what needs to be fixed."
    )


class JoinOutputs(BaseModel):
    """Decide whether to replan or whether you can return the final response."""

    thought: str = Field(
        description="The chain of thought reasoning for the selected action"
    )
    action: Union[FinalResponse, Replan]


joiner_prompt = hub.pull("wfh/llm-compiler-joiner").partial(
    examples=""
)  # You can optionally add examples
llm = ChatOpenAI(model="gpt-4-turbo")

runnable = create_structured_output_runnable(JoinOutputs, llm, joiner_prompt)


def _parse_joiner_output(decision: JoinOutputs) -> List[BaseMessage]:
    response = [AIMessage(content=f"Thought: {decision.thought}")]
    if isinstance(decision.action, Replan):
        return response + [
            SystemMessage(
                content=f"Context from last attempt: {decision.action.feedback}"
            )
        ]
    else:
        return response + [AIMessage(content=decision.action.response)]


def select_recent_messages(messages: list) -> dict:
    selected = []
    for msg in messages[::-1]:
        selected.append(msg)
        if isinstance(msg, HumanMessage):
            break
    return {"messages": selected[::-1]}


joiner = select_recent_messages | runnable | _parse_joiner_output

Graph

from langgraph.graph import MessageGraph, END
from typing import Dict

graph_builder = MessageGraph()

# 1.  Define vertices
# We defined plan_and_schedule above already
# Assign each node to a state variable to update
graph_builder.add_node("plan_and_schedule", plan_and_schedule)
graph_builder.add_node("join", joiner)


## Define edges
graph_builder.add_edge("plan_and_schedule", "join")

### This condition determines looping logic


def should_continue(state: List[BaseMessage]):
    if isinstance(state[-1], AIMessage):
        return END
    return "plan_and_schedule"


graph_builder.add_conditional_edges(
    "join",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
)
graph_builder.set_entry_point("plan_and_schedule")
chain = graph_builder.compile()

简单问题

for step in chain.stream([HumanMessage(content="What's the GDP of New York?")]):
    print(step)
    print("---")

输出

{'plan_and_schedule': [FunctionMessage(content="{'searchParameters': {'q': 'GDP of New York', 'gl': 'us', 'hl': 'en', 'type': 'search', 'num': 1, 'engine': 'google'}, 'organic': [{'title': 'Economy of New York (state) - Wikipedia', 'link': 'https://en.wikipedia.org/wiki/Economy_of_New_York_(state)', 'snippet': 'The economy of the State of New York is reflected in its gross state product in 2022 of $2.053 trillion, ranking third in size behind the larger states of ...', 'sitelinks': [{'title': 'New York City', 'link': 'https://en.wikipedia.org/wiki/Economy_of_New_York_(state)#New_York_City'}, {'title': 'Agriculture', 'link': 'https://en.wikipedia.org/wiki/Economy_of_New_York_(state)#Agriculture'}, {'title': 'Energy', 'link': 'https://en.wikipedia.org/wiki/Economy_of_New_York_(state)#Energy'}], 'position': 1}], 'relatedSearches': [{'query': 'GDP of New York 2023'}, {'query': 'New York City GDP per capita'}, {'query': 'GDP of California'}, {'query': 'Gdp of new york 2021'}, {'query': 'GDP of Texas'}, {'query': 'New York City GDP 2022'}, {'query': 'Economy of New York City'}, {'query': 'New York GDP 2024'}]}", additional_kwargs={'idx': 1, 'args': {'query': 'GDP of New York'}}, name='google_serper_results_json', id='4bcdf0bf-be99-4e51-9159-4c9a52485129'), FunctionMessage(content='join', additional_kwargs={'idx': 2, 'args': ()}, name='join', id='491ca30e-5813-4f98-8ece-31052fc1b02f')]}
---
{'join': [AIMessage(content="Thought: The search results clearly state that the GDP of New York in 2022 was $2.053 trillion. This information is sufficient to provide a direct answer to the user's query.", id='32f5f56a-b717-4dda-b49a-1ded66c43681'), AIMessage(content='The GDP of New York in 2022 was $2.053 trillion.', id='b154c632-f38b-4439-8dbb-5ab8b45b26b6')]}

# Final answer
print(step['join'][-1].content)

输出

The GDP of New York in 2022 was $2.053 trillion.

LangSmith流程展示：

多跳问题

steps = chain.stream(
    [
        HumanMessage(
            content="What's the oldest parrot alive, and how much longer is that than the average?"
        )
    ],
    {
        "recursion_limit": 100,
    },
)
for step in steps:
    print(step)
    print("---")

输出

{'plan_and_schedule': [FunctionMessage(content='{\'searchParameters\': {\'q\': \'oldest parrot alive\', \'gl\': \'us\', \'hl\': \'en\', \'type\': \'search\', \'num\': 1, \'engine\': \'google\'}, \'organic\': [{\'title\': \'3 Oldest Parrots in The World\', \'link\': \'https://www.oldest.org/animals/parrots/\', \'snippet\': "Most birds of Poncho\'s breed live to be around 50 or 60, though some can reach 80 years of age with a good, healthy diet. Poncho, however, is 92, and is ...", \'position\': 1}], \'relatedSearches\': [{\'query\': \'Top 10 oldest parrot alive\'}, {\'query\': \'Parrot lifespan 140 years\'}, {\'query\': \'Longest living parrot in captivity\'}, {\'query\': \'Is Charlie the parrot still alive\'}, {\'query\': \'Oldest African Grey parrot\'}, {\'query\': \'Oldest bird\'}, {\'query\': \'100 year old parrot\'}, {\'query\': \'Cocky Bennett the cockatoo\'}]}', additional_kwargs={'idx': 1, 'args': {'query': 'oldest parrot alive'}}, name='google_serper_results_json', id='abab82fe-4477-4029-aff6-832bdbc6fea4'), FunctionMessage(content="{'searchParameters': {'q': 'average lifespan of a parrot', 'gl': 'us', 'hl': 'en', 'type': 'search', 'num': 1, 'engine': 'google'}, 'knowledgeGraph': {'title': 'Parrots', 'type': 'Birds', 'imageUrl': 'https://encrypted-tbn0.gstatic.com/licensed-image?q=tbn:ANd9GcRldZn14udjcIcuAOvalFdHTcLCTmyYQWq6faVtdOvqc7Zo5cYp1YAQRB2DlNeWDbCRLEd0&s=19', 'description': 'Parrots, also known as psittacines, are birds with a strong curved beak, upright stance, and clawed feet. They are conformed by four families that contain roughly 410 species in 101 genera, found mostly in tropical and subtropical regions.', 'descriptionSource': 'Wikipedia', 'descriptionLink': 'https://en.wikipedia.org/wiki/Parrot', 'attributes': {'Lifespan': 'Cockatoos: 40 – 60\\xa0years, Kākāpō: 60\\xa0years, and Hyacinth macaw: 50\\xa0years', 'Scientific name': 'Psittaciformes', 'Clade': 'Psittacopasseres', 'Class': 'Aves', 'Domain': 'Eukaryota', 'Kingdom': 'Animalia'}}, 'organic': [{'title': 'How Long Do Parrots Live? - PetMD', 'link': 'https://www.petmd.com/bird/how-long-do-parrots-live', 'snippet': 'Some pets, such as tortoises and parrots, may live for over 50 years. Because they are a lifelong commitment, lawyers often urge pet parents to ...', 'date': 'Jul 19, 2023', 'sitelinks': [{'title': 'Parakeet Care Sheet', 'link': 'https://www.petmd.com/bird/parakeet-care-sheet'}, {'title': 'Cockatiel Care Sheet', 'link': 'https://www.petmd.com/bird/cockatiel-care-sheet'}, {'title': 'Budgie Care Sheet', 'link': 'https://www.petmd.com/bird/budgie-care-sheet'}], 'position': 1}], 'relatedSearches': [{'query': 'Parrot lifespan 140 years'}, {'query': 'Average lifespan of a parrot in the wild'}, {'query': 'Average lifespan of a parrot in captivity'}, {'query': 'Green parrot lifespan'}, {'query': 'Indian parrot lifespan 140 years'}, {'query': 'Talking parrot lifespan'}, {'query': 'African Grey parrot lifespan'}, {'query': 'Macaw parrot lifespan'}]}", additional_kwargs={'idx': 2, 'args': {'query': 'average lifespan of a parrot'}}, name='google_serper_results_json', id='b4a0cdb9-284c-4738-8043-f482aab3fd5f'), FunctionMessage(content='join', additional_kwargs={'idx': 3, 'args': ()}, name='join', id='47c9e6e0-93e2-448b-bb9f-b810e8e835a6')]}
---
{'join': [AIMessage(content="Thought: The search results indicate that the oldest parrot alive is named Poncho and is 92 years old. The average lifespan of parrots varies by type, but for many common types it ranges between 40 to 60 years. Using the upper range of 60 years for comparison, Poncho's age of 92 is 32 years longer than the average lifespan of his breed.", id='8e65ac41-4b35-465d-af40-f97158e90fd5'), AIMessage(content='The oldest parrot alive is named Poncho, who is 92 years old. This is 32 years longer than the upper average lifespan of his breed, which is about 60 years.', id='c02330b3-b772-4a99-bce3-4596963c07f2')]}
---

# Final answer
print(step['join'][-1].content)

输出

The oldest parrot alive is named Poncho, who is 92 years old. This is 32 years longer than the upper average lifespan of his breed, which is about 60 years.

LangSmith流程展示：

多步骤数学问题

for step in chain.stream(
    [
        HumanMessage(
            content="What's ((3*(4+5)/0.5)+3245) + 8? What's 32/4.23? What's the sum of those two values?"
        )
    ]
):
    print(step)

输出

{'plan_and_schedule': [FunctionMessage(content='3307.0', additional_kwargs={'idx': 1, 'args': {'problem': '((3*(4+5)/0.5)+3245) + 8'}}, name='math', id='a8e6d377-d660-42d2-8cd4-85d9f98ffe9a'), FunctionMessage(content='7.565011820330969', additional_kwargs={'idx': 2, 'args': {'problem': '32/4.23'}}, name='math', id='101c6d4e-4f40-49d6-91d9-d926e05b2825'), FunctionMessage(content='3314.565011820331', additional_kwargs={'idx': 3, 'args': {'problem': '$1 + $2', 'context': ['$1', '$2']}}, name='math', id='9eee847c-9be2-461d-9210-8612a34cff95'), FunctionMessage(content='join', additional_kwargs={'idx': 4, 'args': ()}, name='join', id='611383f1-05de-40db-acf5-e996153ce6ad')]}
{'join': [AIMessage(content="Thought: The calculations were performed correctly. The first value calculated was 3307.0, and the second was 7.565011820330969. The sum of these two values is 3314.565011820331, which answers the user's question about the sum.", id='c7d48b7a-5033-47d4-b722-b2527a994c92'), AIMessage(content='The first value is 3307.0, the second value is approximately 7.57, and their sum is approximately 3314.57.', id='3f20fd0a-3e1b-4a0a-8164-f48d350fda0d')]}
---

# Final answer
print(step['join'][-1].content)

输出

The first value is 3307.0, the second value is approximately 7.57, and their sum is approximately 3314.57.

LangSmith流程展示：

结论

• 这三种Agent架构是Plan-and-Execute设计模式的典型代表，将LLM驱动的Planner与工具执行运行时分离。
• 如果应用程序需要多次工具调用或API调用，这些方法可以减少返回最终结果所需的时间，并通过减少对更强大的LLM的调用频率来帮助节省成本。

官方资源

代码示例更详细说明：

• Plan-and-execute
• ReWOO
• LLMCompiler
• 官方YouTube
• 官方Blog
• B站视频
  
  1. https://github.com/SqueezeAILab/LLMCompiler/blob/main/src/llm_compiler/output_parser.py ↩︎