Building a multi-agent system with LangGraph

Building a LangGraph multi-agent system forces you to confront state ownership, handoffs, and failure isolation early. This tutorial walks through a supervised architecture where a planner delegates to specialized workers and a supervisor reconciles results before returning to the user.

Prerequisites

• Python 3.10 or newer
• Familiarity with LangChain expressions and Python typing
• Installed packages:

pip install langgraph==0.2.0 langchain-openai==0.1.0 langchain-core==0.2.0 python-dotenv

You need an API key for an OpenAI-compatible chat model. If you point ChatOpenAI at n4n.ai, an OpenAI-compatible gateway, you get automatic fallback when a provider is rate-limited without changing agent code.

Architecture: supervisor pattern

A LangGraph multi-agent system does not need a complex framework. The supervisor pattern keeps a single decision node that routes to workers. Each worker runs as its own ReAct loop, then returns control. This contains tool errors and prevents one agent’s malformed output from corrupting the whole graph.

We will build three nodes: supervisor, researcher, coder. The supervisor reads the message list and outputs the name of the next node or FINISH.

Define the shared state

LangGraph requires an explicit state schema. Use a TypedDict with an annotated messages list. The reducer appends messages so concurrent writes do not clobber.

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

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], lambda x, y: x + y]
    next: str

The next field drives conditional edges. Keep it a plain string; do not store rich objects in state unless you write a custom reducer.

Build worker agents

We use create_react_agent from langgraph.prebuilt. Each worker gets a narrow toolset. Narrow tools reduce prompt injection surface and make debugging tractable.

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def web_search(query: str) -> str:
    """Return a mocked search snippet for the query."""
    return f"Result for '{query}': LangGraph supports cyclic graphs and human-in-the-loop."

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

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    base_url="https://api.n4n.ai/v1",
    api_key="sk-your-key"
)

researcher = create_react_agent(llm, [web_search])
coder = create_react_agent(llm, [calculator])

Wrap them as graph nodes. The node returns new messages and sends control back to supervisor.

def researcher_node(state: AgentState) -> AgentState:
    out = researcher.invoke({"messages": state["messages"]})
    return {"messages": out["messages"], "next": "supervisor"}

def coder_node(state: AgentState) -> AgentState:
    out = coder.invoke({"messages": state["messages"]})
    return {"messages": out["messages"], "next": "supervisor"}

Supervisor node

The supervisor is a plain function that calls the LLM with a strict instruction. Do not let it use tools; its only job is routing.

SUPERVISOR_SYS = (
    "You route tasks. Reply with exactly one word: "
    "'researcher', 'coder', or 'FINISH'."
)

def supervisor_node(state: AgentState) -> AgentState:
    resp = llm.invoke(
        [{"role": "system", "content": SUPERVISOR_SYS}] + state["messages"]
    )
    decision = resp.content.strip().lower()
    if "finish" in decision:
        return {"next": END}
    if "researcher" in decision:
        return {"next": "researcher"}
    return {"next": "coder"}

Compile the graph

Add nodes, edges, and the entry point. Conditional edges map the next string to the correct node.

from langgraph.graph import StateGraph, END

builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("coder", coder_node)

builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
builder.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {"researcher": "researcher", "coder": "coder", END: END}
)
builder.set_entry_point("supervisor")
graph = builder.compile()

Run and inspect

Invoke with an initial user message. Set a recursion limit to prevent infinite loops.

inputs = {
    "messages": [{"role": "user", "content": "Compute 3*7+2. Then summarize what LangGraph is."}],
    "next": ""
}

for step in graph.stream(inputs, {"recursion_limit": 8}):
    print(step.keys(), step.get("next"))

Expected output shows the supervisor cycling to coder, then researcher, then FINISH:

dict_keys(['supervisor']) supervisor
dict_keys(['coder']) supervisor
dict_keys(['supervisor']) researcher
dict_keys(['researcher']) supervisor
dict_keys(['supervisor']) __end__

The final state contains the aggregated message list with tool calls and natural language answers.

Failure isolation

If calculator throws, the coder node catches it and returns a string error. The supervisor still receives a clean message list. This is why a LangGraph multi-agent system beats a single monolithic agent: a broken tool does not abort the process. You can add a try/except around invoke to route to a fallback node.

Extending the pattern

Add a human_review node by inserting a interrupt before FINISH. Or replace the string router with a structured output model:

from langchain_core.pydantic_v1 import BaseModel, Field

class Route(BaseModel):
    next: str = Field(description="researcher, coder, or FINISH")

Bind that schema to the supervisor LLM for stricter parsing. The LangGraph multi-agent system scales to dozens of workers because the supervisor logic stays constant.

Keep worker graphs isolated; share only the state contract. That discipline keeps your codebase debuggable when a model drifts.