Hierarchical Agent Pattern Langgraph

As you add more agents to your system, it might become too hard for the supervisor to manage all of them. The supervisor might start making poor decisions about which agent to call next, or the context might become too complex for a single supervisor to keep track of. In other words, you end up with the same problems that motivated the multi-agent architecture in the first place.

To address this, you can design your system hierarchically. For example, you can create separate, specialized teams of agents managed by individual supervisors, and a top-level supervisor to manage the teams.

from typing import Literal

from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, MessagesState, START, END

from langgraph.types import Command

model = ChatOpenAI()

# define team 1 (same as the single supervisor example above)

def team_1_supervisor(state: MessagesState) -> Command[Literal["team_1_agent_1", "team_1_agent_2", END]]:

    response = model.invoke(...)

    return Command(goto=response["next_agent"])

def team_1_agent_1(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:

    response = model.invoke(...)

    return Command(goto="team_1_supervisor", update={"messages": [response]})

def team_1_agent_2(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:

    response = model.invoke(...)

    return Command(goto="team_1_supervisor", update={"messages": [response]})

team_1_builder = StateGraph(Team1State)

team_1_builder.add_node(team_1_supervisor)

team_1_builder.add_node(team_1_agent_1)

team_1_builder.add_node(team_1_agent_2)

team_1_builder.add_edge(START, "team_1_supervisor")

team_1_graph = team_1_builder.compile()

# define team 2 (same as the single supervisor example above)

class Team2State(MessagesState):

    next: Literal["team_2_agent_1", "team_2_agent_2", "__end__"]

def team_2_supervisor(state: Team2State):

    ...

def team_2_agent_1(state: Team2State):

    ...

def team_2_agent_2(state: Team2State):

    ...

team_2_builder = StateGraph(Team2State)

...

team_2_graph = team_2_builder.compile()

# define top-level supervisor

builder = StateGraph(MessagesState)

def top_level_supervisor(state: MessagesState) -> Command[Literal["team_1_graph", "team_2_graph", END]]:

    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])

    # to determine which team to call next. a common pattern is to call the model

    # with a structured output (e.g. force it to return an output with a "next_team" field)

    response = model.invoke(...)

    # route to one of the teams or exit based on the supervisor's decision

    # if the supervisor returns "__end__", the graph will finish execution

    return Command(goto=response["next_team"])

builder = StateGraph(MessagesState)

builder.add_node(top_level_supervisor)

builder.add_node("team_1_graph", team_1_graph)

builder.add_node("team_2_graph", team_2_graph)

builder.add_edge(START, "top_level_supervisor")

builder.add_edge("team_1_graph", "top_level_supervisor")

builder.add_edge("team_2_graph", "top_level_supervisor")

graph = builder.compile()