Build a multi-agent workflow with LangGraph

This langgraph multi-agent workflow tutorial shows how to assemble a supervisor-driven multi-agent system with explicit state transitions. You will build a graph where a router delegates to a research agent and a coding agent, then terminates when the task is done. The pattern scales to any number of specialized workers without giving up debuggability.

Step 1: Install dependencies and configure the model

Install the packages you need. LangGraph handles orchestration; LangChain’s OpenAI integration gives you a standard chat model interface.

pip install langgraph langchain-openai langchain-core pydantic

Set your API key for an OpenAI-compatible provider. If you want a single endpoint that addresses 240+ models with automatic fallback when a provider is rate-limited, point ChatOpenAI at n4n.ai’s OpenAI-compatible base URL and use your gateway key.

import os
os.environ["OPENAI_API_KEY"] = "sk-your-gateway-key"
BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, base_url=BASE_URL)

Keep the model name swappable. The rest of the graph does not care which backend serves the tokens.

Step 2: Define the shared state contract

LangGraph flows a single state object through nodes. Use a TypedDict so every agent appends to the same message list and writes a next field the supervisor reads.

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

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], "append"]
    next: str

The "append" reducer is critical. Without it, a node that returns {"messages": [msg]} overwrites prior history. With it, LangGraph concatenates, preserving the full conversation across handoffs.

Step 3: Build specialized agent nodes

A worker node takes the current state, prepends its system prompt, calls the model, and returns its response plus a signal to return control to the supervisor.

from langchain_core.messages import SystemMessage, HumanMessage

def make_agent_node(system_prompt: str):
    def node(state: AgentState):
        msgs = [SystemMessage(content=system_prompt)] + state["messages"]
        response = llm.invoke(msgs)
        return {"messages": [response], "next": "supervisor"}
    return node

researcher = make_agent_node(
    "You are a research agent. Summarize libraries or APIs needed. "
    "Return concise findings only."
)
coder = make_agent_node(
    "You are a coding agent. Write runnable Python snippets based on "
    "prior research. No explanations outside code comments."
)

This factory pattern avoids duplicating the message-assembly boilerplate. Each agent is just a pure function of state.

Step 4: Implement the supervisor router

The supervisor decides which worker runs next, or ends the loop. Use structured output to force a constrained response instead of parsing free text.

from pydantic import BaseModel

class Route(BaseModel):
    next: str  # "researcher", "coder", or "FINISH"

router_llm = llm.with_structured_output(Route)

SUPERVISOR_SYS = (
    "You coordinate a research and coding agent. "
    "If the task needs external knowledge, route to 'researcher'. "
    "If code must be written or fixed, route to 'coder'. "
    "If the user's request is fully satisfied, return 'FINISH'."
)

def supervisor_node(state: AgentState):
    msgs = [SystemMessage(content=SUPERVISOR_SYS)] + state["messages"]
    decision = router_llm.invoke(msgs)
    return {"next": decision.next}

Structured output relies on the model’s function-calling support. If you target a smaller model that lacks it, fall back to a low-temperature completion and parse the first token.

Step 5: Wire the graph with conditional edges

Create the StateGraph, register nodes, and define transitions. The supervisor loops back to itself via the workers; the FINISH string maps to END.

from langgraph.graph import StateGraph, END

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

builder.set_entry_point("supervisor")

# Workers always return control to supervisor
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")

# Supervisor routes based on state["next"]
builder.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {
        "researcher": "researcher",
        "coder": "coder",
        "FINISH": END,
    },
)

graph = builder.compile()

The lambda passed to add_conditional_edges is the routing function. It must return a key present in the mapping. If the supervisor emits an unexpected value, the graph raises at runtime—fail fast.

Step 6: Compile, run, and verify

Invoke the graph with an initial human message. LangGraph runs the loop until next == "FINISH".

initial = {
    "messages": [HumanMessage(content="Find a Python lib to parse PDFs and write a 5-line extraction script.")],
    "next": "supervisor",
}

result = graph.invoke(initial)

for m in result["messages"]:
    print(f"{m.type}: {m.content}")

How to verify success

A correct run produces at least three message entries beyond the input: a supervisor routing call (not printed, but visible in debug), a researcher response mentioning a library like pypdf or pdfplumber, and a coder response containing a Python snippet. The final next value in the returned state is "FINISH".

Add an assertion in tests:

assert result["next"] == "FINISH"
assert any("import" in m.content for m in result["messages"] if m.type == "ai")

If the loop never terminates, raise the supervisor’s temperature to zero (already set) and check that your Route schema enforces the exact strings "researcher", "coder", "FINISH". Mismatched casing is the most common bug.

Production considerations

The graph above is stateless across process restarts. For real workloads, attach a checkpointer:

from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

Now you can resume a thread after a crash. Per-token metering at the gateway layer lets you attribute cost to each agent turn without instrumenting LangGraph itself—forward the provider cache-control hints and read usage from responses.

Conditional edges are the backbone of any langgraph multi-agent workflow tutorial that claims to be production-grade. They make the control flow explicit, observable, and unit-testable, unlike prompt-only orchestration where agents silently call each other via tool loops.

When you extend this, add a max_iterations guard in the supervisor to prevent runaway loops, and give each worker a distinct model size—cheap models for routing, stronger ones for codegen. That keeps latency and spend predictable.