langgraph/tutorials/plan-and-execute/plan-and-execute/

[giscus[bot]]

langgraph/tutorials/plan-and-execute/plan-and-execute/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/tutorials/plan-and-execute/plan-and-execute/

[levbszabo]

Curious if anyone has ever built anything leveraging this in combination with 'human in the loop' components - meaning AI plans and executes the overall project and escalates to different human agents when needed.

[OmarHory]

I am not sure why I am getting an error of canot concatenate list with a tuple, I literally copied the code and pasted it

[nmquang003]

I also got this error

[LeonMing30]

+1

[NeuroTo]

The problem seems to be in the execute_step(). In the return a list of tuples should be the value of the dict instead of only a tuple. Replacing the respective line with "past_steps": [(task, agent_response["messages"][-1].content)], made it work for me.

[terriberry]

@NeuroTo thanks that does fix it

[Fasttyper]

@NeuroTo I was expecting a headache for a while, but didn't see my rescuer coming. Thanks a lot!

[592319702]

What is the disadvantages of this "plan-and-execute" style agent? If my task is simple, should I use this solution? When multiple agents collaboration, I think the messages passed between agents may cause misunderstanding. So I think when there's a simple task, it's better to use a single agent instead of multiple agents. Am I right?

[SamthinkGit]

Hi, Robotics Engineer here,

Planning is similar to say, automatize the generation of uncertain steps. In this way, if you know which steps your agent should take, then you should directly build the graph that accomplish your goal.

However, if you don't know which steps the agent must take, then using ReAct, or tool based solutions works well for easier tasks as you say (searching, RAGs, simple human-in-the-loop, interactions etc.).

But, if you want to make the agent run a set of tools/function/nodes that should reach a goal that requires complex reasoning or a long set of steps (6+), planning is the best approach, where you will have to due with failure reaction, re-planning, obtaining feedback, etc.

At last, note that simpler tasks does not require complex structures, but if the task is complex (for example: build repository in GitHub), then it is super hard to solve this without planning, multi-agents and graphs.

[xzy103]

Regarding the construction of agent_executor, the method described in the notebook does not work successfully, likely due to version updates. The issue lies with the messages_modifier parameter of the create_react_agent function. I referred to the source code and modified it as follows, which resolved the issue.

def messages_modifier(messages: list):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("placeholder", "{messages}"),
    ])
    return prompt.invoke({"messages": messages})
agent_executor = create_react_agent(llm, tools, messages_modifier=messages_modifier)

Note: The prompt placeholder is slightly different from the one on LangHub.

[DjangoPeng]

yep.

And I don't understand why there are so many broken updates caused by something like this.

[ashantanu]

Is the List type Plan step working always?

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

I get following error

pydantic.error_wrappers.ValidationError: 1 validation error for Plan steps value is not a valid list (type=type_error.list)

[Player256]

Same issue here. I think the llm is kind of giving inconsistent results. It works for some inputs for some it dosen't

[Player256]

Hey I used groq inference,got different error now
BadRequestError: Error code: 400 - {'error': {'message': "Failed to call a function. Please adjust your prompt. See 'failed_generation' for more details.", 'type': 'invalid_request_error', 'code': 'tool_use_failed', 'failed_generation': '<tool-use>\n{\n\t"tool_call": {\n\t\t"id": "pending",\n\t\t"type": "function",\n\t\t"function": {\n\t\t\t"name": "Act"\n\t\t},\n\t\t"parameters": {\n\t\t\t"action": {\n\t\t\t\t"response": {\n\t\t\t\t\t"response": "The hometown of the 2024 Australia Open winner, Jannik Sinner, is Sexten, a northern Italian town in South Tyrol."\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t}\n}\n</tool-use>'}}

[Nachoeigu]

Try with this with_structured_output(pydantic_object = Plan)

[Nachoeigu]

Also, be sure that the model you are using supports function calling. Otherwise, you will need to use the PydanticOutputParser and pass the parser.get_format_instructions() in the prompt of your system message

[MaricoHan]

This is one of the worst documents I've ever seen. The sample code is uncommented, each example is not runnable independently, and the correlation between the examples is unclear. It is a waste of time...

[gtnpromtior]

stop crying bro :=)

[DjangoPeng]

The strange plan and action always be right here nomatter what linput. Anybody knows what happenedand how to fix it?

{
  "input": "xxx",
  "plan": [
    "Identify the objective",
    "Gather necessary resources",
    "Develop a strategy",
    "Implement the strategy",
    "Monitor progress",
    "Evaluate outcomes"
  ],
  "past_steps": [
    [
      "aaa",
      "xxx"
    ],
    [
      "Identify the objective",
      "To effectively identify the objective, I need more context about the overall goal of the plan. Could you please provide details on what you are aiming to achieve with this plan?"
    ]
  ]
}

{
  "output": {
    "action": {
      "steps": [
        "Identify the objective",
        "Gather necessary resources",
        "Assign tasks to team members",
        "Set deadlines for each task",
        "Monitor progress and make adjustments as needed",
        "Review completed steps and update the plan accordingly"
      ]
    }
  }
}

[metacav]

should I modify PlanExecute and Plan classes? or it can run without any changes?

I got the following error: replanner = replanner_prompt | self.llm.with_structured_output(Act). ValueError: Value not declarable with JSON Schema, field: name='action_Response' type=Response required=True

[kyre-99]

InvalidUpdateError: Must write to at least one of ['messages']

I am not sure where is wrong even though I copy the raw code.

[kyre-99]

I fixed this problem below, the bug seems because the missing the requiered key 'messages', and the prompt get by "prompt = hub.pull("wfh/react-agent-executor")" is '{messages}'
prompt = ChatPromptTemplate.from_messages([
("system","You are a helpful assistant."),
MessagesPlaceholder("messages")
])

Choose the LLM that will drive the agent

llm = ChatOllama(model="llama3.1")

agent_executor = create_react_agent(llm, tools, state_modifier=prompt)
agent_executor.invoke({"messages": [("user", "who is the winnner of the us open")]})

[WKX-w]

I pasted these codes,

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

but raised the erro

pydantic_core._pydantic_core.ValidationError: 1 validation error for Plan
steps
  Field required [type=missing, input_value={}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.9/v/missing

What's the reason causing this?

[kyre-99]

You need to replace {{messages}} to {messages},it is the bug in pormpt(hub.pull("wfh/react-agent-executor")" is '{messages}')

[mjunaidca]

An interesting agentic-system pattern can be to use this approach with Map Reduce Paradigm.

This will reduce the execution time but if our next step in dependent on the last one response then there will be a constraint. To overcome this, we can improve our planning phase to mark such condition and then decide on the sequential and parrel steps execution.

[curiothinker]

I got the got an error "IndexError: list index out of range" when executing the last cell :

`config = {"recursion_limit": 20}
inputs = {"input": "what is the hometown of the mens 2024 Australia open winner?"}
async for event in app.astream(inputs, config=config):

for event in app.streaputs, config=config):

for k, v in event.items():
    if k != "__end__":
        print(v)" .  `. 

The response end with [] likely during replan:
{'plan': ["Identify the men's.....']}
{'past_steps': [("Identify the men'....ps.")]}
{'plan': ['Retur... ly.']}
{'past_steps': [('Return ....Sesto_(comune))")]}
{'plan': ["Confirm th....76.")]}
{'plan': []}

What could be the problem?

[nick-galluzzo]

I'm getting the same response. Any luck fixing this?

[nick-galluzzo]

I was able to fix this by changing the replanner_prompt around to ensure that a Response is returned when a final answer has been generated:
"""
For the given objective, update your plan based on the steps already completed.

**Objective**:
{input}

**Original Plan**:
{plan}

**Completed Steps**:
{past_steps}

**Instructions**:
- If there are more steps needed to achieve the objective, return a **Plan** with the remaining steps.
- If all necessary steps have been completed, return a **Response** to the user based on the information gathered.
- Do **not** include any steps that have already been completed in the new plan.
- Do **not** return an empty plan; if no further steps are needed, you **must** return a **Response**.
- Ensure your output is in the correct structured format as per the `Act` model.

**Remember**:
- The `Act` can be either a `Plan` or a `Response`.
- A `Plan` contains a list of steps that still need to be done.
- A `Response` contains the final answer to the user.

**Provide your output below:**

"""

Hope this helps you too!

[tushjai97]

Why is it when i make an agent without using the create_react_agent i get the following error message:
ValidationError: 1 validation error for AgentState
plan
field required (type=value_error.missing)

[maciej-konczal]

Working version for me with attached Langfuse:

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")
_set_env("LANGFUSE_PUBLIC_KEY")
_set_env("LANGFUSE_SECRET_KEY")
_set_env("LANGFUSE_HOST")

from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=3)]

from langchain import hub
from langchain_openai import ChatOpenAI

from langgraph.prebuilt import create_react_agent

# Get the prompt to use - you can modify this!
prompt = hub.pull("ih/ih-react-agent-executor")
prompt.pretty_print()

# Choose the LLM that will drive the agent
llm = ChatOpenAI(model="gpt-4o-mini")
agent_executor = create_react_agent(llm, tools, state_modifier=prompt)

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

# print(res)

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

from pydantic import BaseModel, Field

from typing import Literal, Union
class Plan(BaseModel):
    """Plan to follow in future"""
    type: Literal["Plan"] = "Plan"
    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}"),
    ]
)

'''
"placeholder" is essentially a temporary holder for content that will be replaced when the template is actually used. You could also use "user" instead of "placeholder" if the messages will always be from the user's perspective.
'''

planner = planner_prompt | ChatOpenAI(
    model="gpt-4o-mini", temperature=0
).with_structured_output(Plan) #function calling because we are using structured output

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

# print(res)




class Response(BaseModel):
    """Response to user."""
    type: Literal["Response"] = "Response"
    response: str


class Act(BaseModel):
    """Action to perform."""

    action: Union[Response, Plan] = Field(
        discriminator='type',
        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."
    )

    @classmethod
    def from_response(cls, response_text: str):
        return cls(action=Response(type="Response", response=response_text))


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-mini", temperature=0
).with_structured_output(Act)

from typing import Literal
from langgraph.graph import END


async def execute_step(state: PlanExecute):
    plan = state["plan"]
    if not plan:  # Check if plan is empty
        return {
            "past_steps": [],
            "response": "No steps to execute in the plan."
        }
    
    plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))
    print('=== plan str ===')
    print(plan_str)
    print('=== end of plan str ===')
    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):
    try:
        output = await replanner.ainvoke(state)
        print('=== replanner output ===')
        print(output)
        print('=== end of replanner output ===')
        
        # Handle dict response
        if isinstance(output, dict) and 'Response' in output:
            return {"response": output['Response']}
        
        # Handle proper Act instance
        if isinstance(output.action, Response):
            return {"response": output.action.response}
        else:
            return {"plan": output.action.steps}
    except Exception as e:
        print(f"Error in replan_step: {e}")
        return {"response": str(output.get('Response', "An error occurred processing the response."))}


def should_end(state: PlanExecute):
    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",
    # 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()

from langfuse.callback import CallbackHandler

langfuse_handler = CallbackHandler()

async def call_agent(input: str):

    print('=== calling agent ===')
    config = {"recursion_limit": 50, "callbacks": [langfuse_handler]}
    inputs = {"input": input}
    async for event in app.astream(inputs, config=config):
        for k, v in event.items():
            if k != "__end__":
                print(v)

if __name__ == "__main__":
    import asyncio
    # from langchain_core.messages import HumanMessage
    # inputs = {"input": "what is the hometown of the mens 2024 Australia open winner?"}
    # for s in app.stream(inputs,
    #                     config={"callbacks": [langfuse_handler]}):
    #     print(s)

    asyncio.run(call_agent("what is the hometown of the mens 2024 Australia open winner?"))

[DereMann]

You know, when learning about your framework and its principles, it would really help if you would not use "helper"-functions like create_react_agent for Agents or TavilySearchResults for Tools in your tutorials. It gives me really hard times grasping your core concepts if you apply different strategies but do not at least reference those core concepts.

For example, I cannot find a clear definition for what an Agent actually is in your docs. Coming from LangChain, I initially understood it as a Python function. To my current understanding an Agent in LangGraph is an assembly of nodes in the Graph. But still, there is this create_react_agent function for whatever reason you even recommend to not use.

So, please stick to your core concepts and repeat them as often as possible in your tutorials and guides. Seeing them multiple times in different tutorials would give a nice learning effect and the assurance of having understood them correctly. Currently, this mixture of strategies for solving the same problems is utterly confusing.

[DereMann]

.. ah and thanks for your great frameworks, by the way ;)

[hwb96]

task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}."""

I think I've found a mistake. Because step {1} is hard-coded, each LLM mistakenly thinks it is currently in the first step.

[siddhanna-janai]

Hi, I am using the Gemini model for this code and am encountering an error. The error is as follows: 'InvalidArgument: 400 * GenerateContentRequest.tools[0].function_declarations[0].parameters.properties[action].properties: should be non-empty for OBJECT type.' This happens because replanner is returning a null value.

[dungba88]

FWIW, the with_structured_output doesn't seem to work with ChatOllama and ChatCohere out of box. For ChatOllama, I needed to change to prompt to be:

        Make sure to output in the structured Plan format.
        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. \
        
        The Plan format: a JSON object with `steps` attribute defined as list of strings.

Without the last sentence, the invoke method will just return None (as the LLM only gives normal string output).

For ChatCohere, nothing seems to work.

[dungba88]

I also noticed that with_structured_output only call bind_tools with the schema, and relies on Prompt to return structured output. Cohere seems to support enforcing structured output in their API (https://docs.cohere.com/v2/docs/structured-outputs). Maybe we could also use that instead of generic prompting for a more reliable behavior? A small note I found is that Pydantic and Cohere use different JSON schema version ("json_object" in Cohere but "object" in Pydantic)