LangGraph — State-Machine Framework for Production Agents

Quick Start — Multi-Agent Supervisor Graph

Supervisor pattern: a routing node inspects state and picks the next worker, workers return control to the supervisor, supervisor eventually returns END. The InMemorySaver checkpoints state; swap for SqliteSaver / PostgresSaver in production for durable resume. Thread IDs isolate concurrent conversations.

# pip install langgraph langchain-openai
from typing import Literal, TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.tools import tool

llm = ChatOpenAI(model="gpt-4o-mini")

class State(TypedDict):
    messages: Annotated[list, lambda a, b: a + b]
    next: str

# Two specialized workers
def researcher(state: State) -> State:
    resp = llm.invoke([HumanMessage(f"Research briefly: {state['messages'][-1].content}")])
    return {"messages": [AIMessage(content=f"[researcher] {resp.content}")], "next": ""}

def writer(state: State) -> State:
    resp = llm.invoke([HumanMessage(f"Polish into 2 sentences: {state['messages'][-1].content}")])
    return {"messages": [AIMessage(content=f"[writer] {resp.content}")], "next": ""}

# Supervisor routes to the right worker
def supervisor(state: State) -> State:
    last = state["messages"][-1].content if state["messages"] else ""
    resp = llm.invoke([HumanMessage(
        f"Pick the next worker (researcher, writer, or FINISH): {last}")]).content.upper()
    next_ = "researcher" if "RESEARCH" in resp else ("writer" if "WRITE" in resp else "FINISH")
    return {"messages": state["messages"], "next": next_}

def route(state: State) -> Literal["researcher", "writer", "__end__"]:
    return state["next"] if state["next"] != "FINISH" else END

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route,
                            {"researcher": "researcher", "writer": "writer", END: END})
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "demo"}}
for event in app.stream({"messages": [HumanMessage("The state of AI agents in 2026")], "next": ""},
                        config=cfg):
    print(event)

Key Features

Comparison

Use Cases

Pricing & License

Related Assets on TokRepo

Frequently Asked Questions