Building a hierarchical agent team in LangGraph

A langgraph hierarchical agent team keeps a coordinator in charge while specialized workers handle narrow subtasks. This pattern stops a single prompt from ballooning with unrelated tools and makes each step inspectable. The following build uses LangGraph’s StateGraph and prebuilt react agents to ship a working supervisor–worker loop.

Step 1: Define the shared team state

Every node in the graph reads and writes the same state. Use a TypedDict so LangGraph can track message history and the supervisor’s routing decision.

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage

class TeamState(TypedDict):
    messages: Annotated[list[BaseMessage], "full chat history"]
    next: str  # route chosen by supervisor: "search", "calc", or "FINISH"

The next field is the control signal. Workers reset it to "supervisor" after they finish, forcing the loop back to the coordinator.

Step 2: Implement specialized worker agents

Workers should stay narrow. Each gets its own tool set and a cheap model. create_react_agent from langgraph.prebuilt gives you a ReAct loop without hand-rolling tool calling.

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

def search_tool(query: str) -> str:
    """Return a hardcoded snippet to simulate retrieval."""
    return f"Result for '{query}': LangGraph is built by LangChain Inc."

def calc_tool(expr: str) -> str:
    """Evaluate a basic arithmetic expression."""
    try:
        return str(eval(expr, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Math error: {e}"

model = ChatOpenAI(model="gpt-4o-mini")
search_agent = create_react_agent(model, [search_tool])
calc_agent = create_react_agent(model, [calc_tool])

def search_node(state: TeamState):
    result = search_agent.invoke({"messages": state["messages"]})
    return {"messages": result["messages"], "next": "supervisor"}

def calc_node(state: TeamState):
    result = calc_agent.invoke({"messages": state["messages"]})
    return {"messages": result["messages"], "next": "supervisor"}

Keep worker prompts implicit in the tool docstrings. The supervisor, not the worker, decides when to call them.

Step 3: Build the supervisor router

The supervisor is a language model forced to return structured output. Define a Pydantic model so the graph can branch deterministically.

from pydantic import BaseModel, Field

class Route(BaseModel):
    next: str = Field(description="One of: search, calc, FINISH")

supervisor_model = ChatOpenAI(model="gpt-4o").with_structured_output(Route)

SYSTEM = (
    "You coordinate a team. Use 'search' for factual lookups, "
    "'calc' for arithmetic, and 'FINISH' when the user's question is fully answered."
)

def supervisor_node(state: TeamState):
    response = supervisor_model.invoke(
        [{"role": "system", "content": SYSTEM}] + state["messages"]
    )
    return {"next": response.next}

Structured output turns the model’s reasoning into a string the conditional edge can match. No fuzzy parsing required.

Step 4: Compose the hierarchical graph

Wire the nodes so the supervisor always sits between worker calls. This is the core of a langgraph hierarchical agent team: a cyclic graph with one decision node and multiple leaf nodes.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("search", search_node)
builder.add_node("calc", calc_node)

def route(state: TeamState) -> str:
    if state["next"] == "FINISH":
        return END
    return state["next"]

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route)
builder.add_edge("search", "supervisor")
builder.add_edge("calc", "supervisor")

graph = builder.compile()

The route function maps the supervisor’s string to a node name or END. Workers always edge back to supervisor, creating the hierarchy rather than a flat pipeline.

Step 5: Run and verify behavior

Invoke the graph with a mixed question. Streaming the state after each step shows the loop in action.

from langchain_core.messages import HumanMessage

if __name__ == "__main__":
    initial = {
        "messages": [HumanMessage("What is 24 * 7 and who maintains LangGraph?")],
        "next": "",
    }
    final = graph.invoke(initial)
    for msg in final["messages"]:
        msg.pretty_print()

Verification: You should see the supervisor emit calc, the calc worker return 168, then the supervisor emit search, the search worker return the LangChain snippet, and finally the supervisor emit FINISH with a synthesized answer. If the loop never terminates, raise the recursion_limit on compile() or tighten the supervisor system prompt.

Step 6: Route model calls through a resilient gateway

When workers use different models or you fear rate limits, point ChatOpenAI at a single OpenAI-compatible endpoint instead of hardcoding providers. n4n.ai is an OpenRouter-class gateway that exposes 240+ models behind one URL and automatically fails over when a provider is degraded, which suits a langgraph hierarchical agent team that may fan out across many model types.

model = ChatOpenAI(
    model="openai/gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key="YOUR_N4N_KEY",
)

Swap this model into both supervisor and workers and you get per-token metering and client-side routing hints without changing graph logic.

Production caveats

Set a MemorySaver checkpointer if you need to resume runs or inspect state across calls. Add a recursion_limit (default 25) to prevent runaway supervisor loops. For real tools, wrap side effects behind human approval by inserting an interrupt before the worker edge. The hierarchy stays clean as long as the supervisor owns routing and workers stay stateless beyond the message list.