langgraph/examples/multi_agent/hierarchical_agent_teams.ipynb

Hierarchical Agent Teams

In our previous example (Agent Supervisor), we introduced the concept of a single supervisor node to route work between different worker nodes.

But what if the job for a single worker becomes too complex? What if the number of workers becomes too large?

For some applications, the system may be more effective if work is distributed hierarchically.

You can do this by composing different subgraphs and creating a top-level supervisor, along with mid-level supervisors.

To do this, let's build a simple research assistant! The graph will look something like the following:

This notebook is inspired by the paper AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation, by Wu, et. al. In the rest of this notebook, you will:

1. Define some utilities to help create the graph and their relations
2. Write the tools and agent implementations for each team
3. Compose everything together.

But before all of that, some setup:

In [1]:

# %%capture --no-stderr
# %pip install -U langgraph langchain langchain_openai langsmith

In [2]:

import getpass
import os
import uuid


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


_set_if_undefined("OPENAI_API_KEY")
_set_if_undefined("LANGCHAIN_API_KEY")
_set_if_undefined("TAVILY_API_KEY")

# Optional, add tracing in LangSmith.
# This will help you visualize and debug the control flow
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Multi-agent Collaboration"

Define Utilities

We are going to create a few utility functions to make it more concise when we want to:

1. Create a worker agent and add it to a graph.
2. Create a supervisor for the sub-graph.

These will simplify the graph compositional code at the end for us so it's easier to see what's going on.

In [3]:

from typing import Any, Callable, List, Optional, TypedDict, Union

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import Runnable
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI

from langgraph.graph import END, StateGraph


def create_worker_agent(
    graph_builder: StateGraph,
    name: str,
    llm: ChatOpenAI,
    tools: list,
    system_prompt: str,
    prelude: Optional[Union[Runnable, Callable]] = None,  # Optional required steps
) -> str:
    """Create a function-calling agent and add it to the graph."""
    system_prompt += "\nYou are one of the following team members: {team_members}"
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                system_prompt,
            ),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
    agent = create_openai_functions_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    chain = executor | (
        lambda x: {"messages": [HumanMessage(content=x["output"], name=name)]}
    )
    if prelude is not None:
        chain = prelude | chain
    graph_builder.add_node(name, chain)
    return name


def create_team_supervisor(
    graph_builder: StateGraph, llm: ChatOpenAI, system_prompt: str
) -> str:
    """An LLM-based router."""
    supervisor_id = uuid.uuid4().hex[:4]
    supervisor_name = f"supervisor - {supervisor_id}"
    members = list(graph_builder.nodes)
    options = ["FINISH"] + members
    function_def = {
        "name": "route",
        "description": "Select the next role.",
        "parameters": {
            "title": "routeSchema",
            "type": "object",
            "properties": {
                "next": {
                    "title": "Next",
                    "anyOf": [
                        {"enum": options},
                    ],
                }
            },
            "required": ["next"],
        },
    }
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system_prompt),
            MessagesPlaceholder(variable_name="messages"),
            (
                "system",
                "Given the conversation above, who should act next?"
                " Or should we FINISH? Select one of: {options}",
            ),
        ]
    ).partial(options=str(options), team_members=", ".join(members))
    chain = (
        prompt
        | llm.bind_functions(functions=[function_def], function_call="route")
        | JsonOutputFunctionsParser()
    )
    graph_builder.add_node(supervisor_name, chain)
    conditional_map = {k: k for k in members}
    conditional_map["FINISH"] = END

    for member in members:
        graph_builder.add_edge(member, supervisor_name)
    graph_builder.add_conditional_edges(
        supervisor_name, lambda x: x["next"], conditional_map
    )
    return supervisor_name

Define agents + tools

Now we can get to define our hierachical teams. "Choose your player!"

Research Team

The research team can use a search engine and url scraper to find information on the web. Feel free to add additional functionality below to boost the team performance!

In [4]:

from typing import Annotated, List, Tuple, Union

import matplotlib.pyplot as plt
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langsmith import trace

tavily_tool = TavilySearchResults(max_results=5)


@tool
def scrape_webpages(urls: List[str]) -> str:
    """Use requests and bs4 to scrape the provided web pages for detailed information."""
    loader = WebBaseLoader(urls)
    docs = loader.load()
    return "\n\n".join(
        [
            f'<Document name="{doc.metadata["title"]}">\n{doc.page_content}\n</Document>'
            for doc in docs
        ]
    )

In [5]:

import functools
import operator

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai.chat_models import ChatOpenAI


# Research team graph state
class State(TypedDict):
    # A message is added after each team member finishes
    messages: Annotated[List[BaseMessage], operator.add]
    # The team members are tracked so they are aware of
    # the others' skill-sets
    team_members: List[str]
    # Used to route work. The supervisor calls a function
    # that will update this every time it makes a decision
    next: str


research_graph = StateGraph(State)
llm = ChatOpenAI(model="gpt-4-1106-preview")
create_worker_agent(
    research_graph,
    "Search",
    llm,
    [tavily_tool],
    "You are a research assistant who can search for things using a search engine.",
)
create_worker_agent(
    research_graph,
    "Web Scraper",
    llm,
    [tavily_tool],
    "You are a research assistant who can scrape specified urls for more detailed information.",
)
supervisor_node = create_team_supervisor(
    research_graph,
    llm,
    "You are a supervisor tasked with managing a conversation between the"
    " following workers:  {team_members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH.",
)

research_graph.set_entry_point(supervisor_node)


# The following functions interoperate between the top level graph state
# and the state of the research sub-graph
# this makes it so that the states of each graph don't get intermixed
def enter_chain(message: str, members: Optional[list] = None):
    results = {
        "messages": [HumanMessage(content=message)],
    }
    if members:
        results["team_members"] = "\n".join(sorted(members))
    return results


def return_final_response(state):
    return {"final_response": state["messages"][-1]}


research_chain = (
    functools.partial(enter_chain, members=research_graph.nodes)
    | research_graph.compile()
    | return_final_response
)

Document Writing Team

We will construct a graph in a similar fashion. This time using different tools.

Note that we are giving file-system access to our agent here, which is not safe in all cases.

In [10]:

from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Dict

_TEMP_DIRECTORY = TemporaryDirectory()
WORKING_DIRECTORY = Path(_TEMP_DIRECTORY.name)


@tool
def create_outline(
    points: Annotated[List[str], "List of main points or sections."],
    file_name: Annotated[str, "File path to save the outline."],
) -> Annotated[str, "Path of the saved outline file."]:
    """Create and save an outline."""
    if len(points) != len(subpoints):
        raise ValueError("Each main point must have a corresponding list of subpoints.")

    with (WORKING_DIRECTORY / file_name).open("w") as file:
        for i, point in enumerate(points):
            file.write(f"{i + 1}. {point}\n")
    return f"Outline saved to {file_name}"


@tool
def read_document(
    file_name: Annotated[str, "File path to save the document."],
    start: Annotated[Optional[int], "The start line. Default is 0"] = None,
    end: Annotated[Optional[int], "The end line. Default is None"] = None,
) -> str:
    """Read the specified document."""
    with (WORKING_DIRECTORY / file_name).open("r") as file:
        lines = file.readlines()
    if start is not None:
        start = 0
    return "\n".join(lines[start:end])


@tool
def write_document(
    content: Annotated[str, "Text content to be written into the document."],
    file_name: Annotated[str, "File path to save the document."],
) -> Annotated[str, "Path of the saved document file."]:
    """Create and save a text document."""
    with (WORKING_DIRECTORY / file_name).open("w") as file:
        file.write(content)
    return f"Document saved to {file_name}"


@tool
def edit_document(
    file_name: Annotated[str, "Path of the document to be edited."],
    inserts: Annotated[
        Dict[int, str],
        "Dictionary where key is the line number (1-indexed) and value is the text to be inserted at that line.",
    ],
) -> Annotated[str, "Path of the edited document file."]:
    """Edit a document by inserting text at specific line numbers."""
    # Read the contents of the file

    with (WORKING_DIRECTORY / file_name).open("r") as file:
        lines = file.readlines()

    # Adjust the line numbers for 0-indexing and sort
    sorted_inserts = sorted(inserts.items())

    # Perform the insertions
    for line_number, text in sorted_inserts:
        if 1 <= line_number <= len(lines) + 1:
            # Insert the text at the specified line number
            lines.insert(line_number - 1, text + "\n")
        else:
            return f"Error: Line number {line_number} is out of range."

    # Write the modified content back to the file
    with (WORKING_DIRECTORY / file_name).open("w") as file:
        file.writelines(lines)

    return f"Document edited and saved to {file_name}"


@tool
def create_plot(
    data: Annotated[
        Union[List[float], List[int]],
        "Numerical values for bar heights or line points.",
    ],
    file_name: Annotated[str, "File path to save the figure."],
    labels: Annotated[
        Union[List[str], None], "Bar or point labels, defaults to None."
    ] = None,
    title: Annotated[str, "Title of the plot."] = "Plot",
    xlabel: Annotated[str, "Label for the X-axis."] = "X",
    ylabel: Annotated[str, "Label for the Y-axis."] = "Y",
    color: Annotated[Union[str, List[str]], "Color(s) for the bars or line."] = "blue",
    plot_type: Annotated[str, "Type of plot ('bar' or 'line')."] = "bar",
) -> Annotated[str, "Path of the saved figure file."]:
    """Create a line or bar chart."""
    if plot_type not in ["bar", "line"]:
        raise ValueError("Invalid plot_type. Expected 'bar' or 'line'.")

    fig, ax = plt.subplots(figsize=(10, 6))
    x_positions = range(len(data))

    if labels and len(labels) == len(data):
        plt.xticks(x_positions, labels)

    if plot_type == "bar":
        ax.bar(x_positions, data, color=color)
    elif plot_type == "line":
        ax.plot(x_positions, data, color=color, marker="o")  # 'o' for circular markers

    ax.set_title(title)
    ax.set_xlabel(xlabel)
    ax.set_ylabel(ylabel)
    fig.savefig(str(WORKING_DIRECTORY / file_name))
    plt.close(fig)
    return f'Saved "{title}" plot to {file_name}'

In [11]:

import operator
from pathlib import Path


# Document writing team graph state
class AuthoringState(TypedDict):
    # This tracks the team's conversation internally
    messages: Annotated[List[BaseMessage], operator.add]
    # This provides each worker with context on the others' skill sets
    team_members: str
    # This is how the supervisor tells langgraph who to work next
    next: str
    # This tracks the shared directory state
    current_files: str


# This will be run before each worker agent begins work
# It makes it so they are more aware of the current state
# of the working directory.
def prelude(state):
    written_files = []
    if not WORKING_DIRECTORY.exists():
        WORKING_DIRECTORY.mkdir()
    try:
        written_files = [
            f.relative_to(WORKING_DIRECTORY) for f in WORKING_DIRECTORY.rglob("*")
        ]
    except:
        pass
    if not written_files:
        return {**state, "current_files": "No files written."}
    return {
        **state,
        "current_files": "\nBelow are files your team has written to the directory:\n"
        + "\n".join([f" - {f}" for f in written_files]),
    }


# Create the graph here:
authoring_graph = StateGraph(AuthoringState)

llm = ChatOpenAI(model="gpt-4-1106-preview")
create_worker_agent(
    authoring_graph,
    "Author Docs",
    llm,
    [write_document, edit_document, read_document],
    "You are an expert writing a research document.\n"
    # The {current_files} value is populated automatically by the graph state
    "Below are files currently in your directory:\n{current_files}",
    prelude=prelude,
)
create_worker_agent(
    authoring_graph,
    "Outline + Notetaker",
    llm,
    [create_outline, read_document],
    "You are an expert senior researcher tasked with writing a paper outline and"
    " taking notes to craft a perfect paper.{current_files}",
    prelude=prelude,
)
create_worker_agent(
    authoring_graph,
    "Generate Charts",
    llm,
    [read_document, create_plot],
    "You are a data viz expert tasked with generating charts for a research project."
    "{current_files}",
)

supervisor_node = create_team_supervisor(
    authoring_graph,
    llm,
    "You are a supervisor tasked with managing a conversation between the"
    " following workers:  {team_members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH.",
)

authoring_graph.set_entry_point(supervisor_node)

# We re-use the enter/exit functions to wrap the graph
authoring_chain = (
    functools.partial(enter_chain, members=authoring_graph.nodes)
    | authoring_graph.compile()
    | return_final_response
)

Add Layers

In this design, we are enforcing a top-down planning policy. We've created two graphs already, but we have to decide how to route work between the two.

We'll create a third graph to orchestrate the previous two, and add some connectors to define how this top-level state is shared between the different graphs.

In [12]:

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai.chat_models import ChatOpenAI


# Research team graph
class State(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str


def get_last_message(state: State) -> str:
    return state["messages"][-1].content


def join_graph(response: dict):
    return {"messages": [response["final_response"]]}


super_graph = StateGraph(State)
super_graph.add_node("Research team", get_last_message | research_chain | join_graph)
super_graph.add_node(
    "Paper writing team", get_last_message | authoring_chain | join_graph
)
llm = ChatOpenAI(model="gpt-4-1106-preview")
supervisor_node = create_team_supervisor(
    super_graph,
    llm,
    "You are a supervisor tasked with managing a conversation between the"
    " following teams: {team_members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH.",
)

super_graph.set_entry_point(supervisor_node)
super_graph = enter_chain | super_graph.compile()

In [13]:

results = super_graph.invoke(
    "Research and write a report about the climate impacts"
    " on crop yields in Bangladesh in 2023. Write the paper and include plots.",
    {"recursion_limit": 150},
)
results["messages"][-1]

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[13], line 1
----> 1 results = super_graph.invoke(
      2     "Research and write a report about the climate impacts"
      3     " on crop yields in Bangladesh in 2023. Write the paper and include plots.",
      4     {"recursion_limit": 150},
      5 )
      6 results["messages"][-1]

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:2034, in RunnableSequence.invoke(self, input, config)
   2032 try:
   2033     for i, step in enumerate(self.steps):
-> 2034         input = step.invoke(
   2035             input,
   2036             # mark each step as a child run
   2037             patch_config(
   2038                 config, callbacks=run_manager.get_child(f"seq:step:{i+1}")
   2039             ),
   2040         )
   2041 # finish the root run
   2042 except BaseException as e:

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:531, in Pregel.invoke(self, input, config, output_keys, input_keys, **kwargs)
    521 def invoke(
    522     self,
    523     input: Union[dict[str, Any], Any],
   (...)
    528     **kwargs: Any,
    529 ) -> Union[dict[str, Any], Any]:
    530     latest: Union[dict[str, Any], Any] = None
--> 531     for chunk in self.stream(
    532         input,
    533         config,
    534         output_keys=output_keys if output_keys is not None else self.output,
    535         input_keys=input_keys,
    536         **kwargs,
    537     ):
    538         latest = chunk
    539     return latest

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:567, in Pregel.transform(self, input, config, output_keys, input_keys, **kwargs)
    558 def transform(
    559     self,
    560     input: Iterator[Union[dict[str, Any], Any]],
   (...)
    565     **kwargs: Any,
    566 ) -> Iterator[Union[dict[str, Any], Any]]:
--> 567     for chunk in self._transform_stream_with_config(
    568         input,
    569         self._transform,
    570         config,
    571         output_keys=output_keys,
    572         input_keys=input_keys,
    573         **kwargs,
    574     ):
    575         yield chunk

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:1486, in Runnable._transform_stream_with_config(self, input, transformer, config, run_type, **kwargs)
   1484 try:
   1485     while True:
-> 1486         chunk: Output = context.run(next, iterator)  # type: ignore
   1487         yield chunk
   1488         if final_output_supported:

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:342, in Pregel._transform(self, input, run_manager, config, input_keys, output_keys)
    332 done, inflight = concurrent.futures.wait(
    333     [
    334         executor.submit(proc.invoke, input, config)
   (...)
    338     timeout=self.step_timeout,
    339 )
    341 # interrupt on failure or timeout
--> 342 _interrupt_or_proceed(done, inflight, step)
    344 # apply writes to channels
    345 _apply_writes(checkpoint, channels, pending_writes, config, step + 1)

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:650, in _interrupt_or_proceed(done, inflight, step)
    648             inflight.pop().cancel()
    649         # raise the exception
--> 650         raise exc
    651         # TODO this is where retry of an entire step would happen
    653 if inflight:
    654     # if we got here means we timed out

File ~/.pyenv/versions/3.11.2/lib/python3.11/concurrent/futures/thread.py:58, in _WorkItem.run(self)
     55     return
     57 try:
---> 58     result = self.fn(*self.args, **self.kwargs)
     59 except BaseException as exc:
     60     self.future.set_exception(exc)

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:3868, in RunnableBindingBase.invoke(self, input, config, **kwargs)
   3862 def invoke(
   3863     self,
   3864     input: Input,
   3865     config: Optional[RunnableConfig] = None,
   3866     **kwargs: Optional[Any],
   3867 ) -> Output:
-> 3868     return self.bound.invoke(
   3869         input,
   3870         self._merge_configs(config),
   3871         **{**self.kwargs, **kwargs},
   3872     )

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:2034, in RunnableSequence.invoke(self, input, config)
   2032 try:
   2033     for i, step in enumerate(self.steps):
-> 2034         input = step.invoke(
   2035             input,
   2036             # mark each step as a child run
   2037             patch_config(
   2038                 config, callbacks=run_manager.get_child(f"seq:step:{i+1}")
   2039             ),
   2040         )
   2041 # finish the root run
   2042 except BaseException as e:

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:531, in Pregel.invoke(self, input, config, output_keys, input_keys, **kwargs)
    521 def invoke(
    522     self,
    523     input: Union[dict[str, Any], Any],
   (...)
    528     **kwargs: Any,
    529 ) -> Union[dict[str, Any], Any]:
    530     latest: Union[dict[str, Any], Any] = None
--> 531     for chunk in self.stream(
    532         input,
    533         config,
    534         output_keys=output_keys if output_keys is not None else self.output,
    535         input_keys=input_keys,
    536         **kwargs,
    537     ):
    538         latest = chunk
    539     return latest

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:567, in Pregel.transform(self, input, config, output_keys, input_keys, **kwargs)
    558 def transform(
    559     self,
    560     input: Iterator[Union[dict[str, Any], Any]],
   (...)
    565     **kwargs: Any,
    566 ) -> Iterator[Union[dict[str, Any], Any]]:
--> 567     for chunk in self._transform_stream_with_config(
    568         input,
    569         self._transform,
    570         config,
    571         output_keys=output_keys,
    572         input_keys=input_keys,
    573         **kwargs,
    574     ):
    575         yield chunk

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:1486, in Runnable._transform_stream_with_config(self, input, transformer, config, run_type, **kwargs)
   1484 try:
   1485     while True:
-> 1486         chunk: Output = context.run(next, iterator)  # type: ignore
   1487         yield chunk
   1488         if final_output_supported:

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:342, in Pregel._transform(self, input, run_manager, config, input_keys, output_keys)
    332 done, inflight = concurrent.futures.wait(
    333     [
    334         executor.submit(proc.invoke, input, config)
   (...)
    338     timeout=self.step_timeout,
    339 )
    341 # interrupt on failure or timeout
--> 342 _interrupt_or_proceed(done, inflight, step)
    344 # apply writes to channels
    345 _apply_writes(checkpoint, channels, pending_writes, config, step + 1)

File ~/code/lc/langgraph/langgraph/pregel/__init__.py:650, in _interrupt_or_proceed(done, inflight, step)
    648             inflight.pop().cancel()
    649         # raise the exception
--> 650         raise exc
    651         # TODO this is where retry of an entire step would happen
    653 if inflight:
    654     # if we got here means we timed out

File ~/.pyenv/versions/3.11.2/lib/python3.11/concurrent/futures/thread.py:58, in _WorkItem.run(self)
     55     return
     57 try:
---> 58     result = self.fn(*self.args, **self.kwargs)
     59 except BaseException as exc:
     60     self.future.set_exception(exc)

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:3868, in RunnableBindingBase.invoke(self, input, config, **kwargs)
   3862 def invoke(
   3863     self,
   3864     input: Input,
   3865     config: Optional[RunnableConfig] = None,
   3866     **kwargs: Optional[Any],
   3867 ) -> Output:
-> 3868     return self.bound.invoke(
   3869         input,
   3870         self._merge_configs(config),
   3871         **{**self.kwargs, **kwargs},
   3872     )

File ~/code/lc/langchain/libs/core/langchain_core/runnables/base.py:2034, in RunnableSequence.invoke(self, input, config)
   2032 try:
   2033     for i, step in enumerate(self.steps):
-> 2034         input = step.invoke(
   2035             input,
   2036             # mark each step as a child run
   2037             patch_config(
   2038                 config, callbacks=run_manager.get_child(f"seq:step:{i+1}")
   2039             ),
   2040         )
   2041 # finish the root run
   2042 except BaseException as e:

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/chains/base.py:162, in Chain.invoke(self, input, config, **kwargs)
    160 except BaseException as e:
    161     run_manager.on_chain_error(e)
--> 162     raise e
    163 run_manager.on_chain_end(outputs)
    164 final_outputs: Dict[str, Any] = self.prep_outputs(
    165     inputs, outputs, return_only_outputs
    166 )

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/chains/base.py:156, in Chain.invoke(self, input, config, **kwargs)
    149 run_manager = callback_manager.on_chain_start(
    150     dumpd(self),
    151     inputs,
    152     name=run_name,
    153 )
    154 try:
    155     outputs = (
--> 156         self._call(inputs, run_manager=run_manager)
    157         if new_arg_supported
    158         else self._call(inputs)
    159     )
    160 except BaseException as e:
    161     run_manager.on_chain_error(e)

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/agents/agent.py:1376, in AgentExecutor._call(self, inputs, run_manager)
   1374 # We now enter the agent loop (until it returns something).
   1375 while self._should_continue(iterations, time_elapsed):
-> 1376     next_step_output = self._take_next_step(
   1377         name_to_tool_map,
   1378         color_mapping,
   1379         inputs,
   1380         intermediate_steps,
   1381         run_manager=run_manager,
   1382     )
   1383     if isinstance(next_step_output, AgentFinish):
   1384         return self._return(
   1385             next_step_output, intermediate_steps, run_manager=run_manager
   1386         )

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/agents/agent.py:1102, in AgentExecutor._take_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps, run_manager)
   1093 def _take_next_step(
   1094     self,
   1095     name_to_tool_map: Dict[str, BaseTool],
   (...)
   1099     run_manager: Optional[CallbackManagerForChainRun] = None,
   1100 ) -> Union[AgentFinish, List[Tuple[AgentAction, str]]]:
   1101     return self._consume_next_step(
-> 1102         [
   1103             a
   1104             for a in self._iter_next_step(
   1105                 name_to_tool_map,
   1106                 color_mapping,
   1107                 inputs,
   1108                 intermediate_steps,
   1109                 run_manager,
   1110             )
   1111         ]
   1112     )

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/agents/agent.py:1102, in <listcomp>(.0)
   1093 def _take_next_step(
   1094     self,
   1095     name_to_tool_map: Dict[str, BaseTool],
   (...)
   1099     run_manager: Optional[CallbackManagerForChainRun] = None,
   1100 ) -> Union[AgentFinish, List[Tuple[AgentAction, str]]]:
   1101     return self._consume_next_step(
-> 1102         [
   1103             a
   1104             for a in self._iter_next_step(
   1105                 name_to_tool_map,
   1106                 color_mapping,
   1107                 inputs,
   1108                 intermediate_steps,
   1109                 run_manager,
   1110             )
   1111         ]
   1112     )

File ~/.pyenv/versions/3.11.2/lib/python3.11/site-packages/langchain/agents/agent.py:1198, in AgentExecutor._iter_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps, run_manager)
   1196         tool_run_kwargs["llm_prefix"] = ""
   1197     # We then call the tool on the tool input to get an observation
-> 1198     observation = tool.run(
   1199         agent_action.tool_input,
   1200         verbose=self.verbose,
   1201         color=color,
   1202         callbacks=run_manager.get_child() if run_manager else None,
   1203         **tool_run_kwargs,
   1204     )
   1205 else:
   1206     tool_run_kwargs = self.agent.tool_run_logging_kwargs()

File ~/code/lc/langchain/libs/core/langchain_core/tools.py:374, in BaseTool.run(self, tool_input, verbose, start_color, color, callbacks, tags, metadata, run_name, **kwargs)
    372 except (Exception, KeyboardInterrupt) as e:
    373     run_manager.on_tool_error(e)
--> 374     raise e
    375 else:
    376     run_manager.on_tool_end(
    377         str(observation), color=color, name=self.name, **kwargs
    378     )

File ~/code/lc/langchain/libs/core/langchain_core/tools.py:346, in BaseTool.run(self, tool_input, verbose, start_color, color, callbacks, tags, metadata, run_name, **kwargs)
    343 try:
    344     tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
    345     observation = (
--> 346         self._run(*tool_args, run_manager=run_manager, **tool_kwargs)
    347         if new_arg_supported
    348         else self._run(*tool_args, **tool_kwargs)
    349     )
    350 except ToolException as e:
    351     if not self.handle_tool_error:

File ~/code/lc/langchain/libs/core/langchain_core/tools.py:641, in StructuredTool._run(self, run_manager, *args, **kwargs)
    632 if self.func:
    633     new_argument_supported = signature(self.func).parameters.get("callbacks")
    634     return (
    635         self.func(
    636             *args,
    637             callbacks=run_manager.get_child() if run_manager else None,
    638             **kwargs,
    639         )
    640         if new_argument_supported
--> 641         else self.func(*args, **kwargs)
    642     )
    643 raise NotImplementedError("Tool does not support sync")

Cell In[10], line 15, in create_outline(points, file_name)
      9 @tool
     10 def create_outline(
     11     points: Annotated[List[str], "List of main points or sections."],
     12     file_name: Annotated[str, "File path to save the outline."],
     13 ) -> Annotated[str, "Path of the saved outline file."]:
     14     """Create and save an outline."""
---> 15     if len(points) != len(subpoints):
     16         raise ValueError("Each main point must have a corresponding list of subpoints.")
     18     with (WORKING_DIRECTORY / file_name).open("w") as file:

NameError: name 'subpoints' is not defined

In [ ]: