The control-first way to build agents and multi-step workflows: an explicit state graph you own, with durable memory, human-in-the-loop, and streaming built in. This is the long version, from first install to production.
The mental model
LangGraph models an LLM application as a graph. Nodes are plain functions (call a model, run a tool, check a rule). Edges are the control flow you write. A shared state object flows through the graph, and each node returns a partial update to it. On top of that it adds the production plumbing a raw while loop lacks: durable state via checkpointing, streaming, branching, retries, and human-in-the-loop pauses.
LangChain (same team) is the higher-level library of model wrappers, tools, and integrations. LangGraph is the lower-level runtime you reach for when you need real control over the flow, and it is what powers most serious agent stacks in 2026.
flowchart LR START(["START"]) --> CL["classify
node"] CL -->|"simple"| ANS["answer
node"] CL -->|"needs tools"| AG["agent
node"] AG --> TN["tool
node"] TN --> AG AG --> ANS ANS --> E(["END"]) classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B; classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B; class CL,TN code; class AG,ANS model;
A flowchart you can actually run. Every box is code you can read, test, and set a breakpoint in. The model only acts inside the boxes you allow it to, and the arrows between boxes are yours.
Install & setup
LangGraph is Python-first (a full TypeScript port exists as @langchain/langgraph). We use Python throughout. You also install a model provider adapter, here Anthropic.
# core runtime + a model provider
pip install langgraph langchain "langchain[anthropic]"
# set your key (or OPENAI_API_KEY, etc.)
export ANTHROPIC_API_KEY="sk-ant-..."Hello world
Before the low-level graph API, meet the shortcut. create_react_agent builds a complete tool-using (ReAct) agent: it wires the model-call → tool-call → loop for you. Start here, then drop to StateGraph when you outgrow it.
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city.""" # the docstring IS the tool description
return f"It's 22C and sunny in {city}."
agent = create_react_agent(
model="anthropic:claude-sonnet-4-5",
tools=[get_weather],
prompt="You are a concise weather assistant.",
)
result = agent.invoke({"messages": [{"role": "user", "content": "weather in Oslo?"}]})
print(result["messages"][-1].content)this one call is a full StateGraph under the hood, you just don't see it yet
The core · state and graph
The real power is the low-level API. Three pieces: a state schema (a TypedDict), nodes (functions that take state and return a partial update), and edges (how control moves). A reducer like add_messages tells LangGraph how to merge each update, appending to the message list instead of overwriting it.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
model = init_chat_model("anthropic:claude-sonnet-4-5")
class State(TypedDict):
messages: Annotated[list, add_messages] # reducer: append, don't replace
def chatbot(state: State) -> dict:
return {"messages": [model.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile()
graph.invoke({"messages": [{"role": "user", "content": "hi"}]})A node returns a dict with only the keys it wants to change. Everything else in state is left alone. That partial-update model is what makes graphs composable.
Conditional routing
The whole point of a graph over a chain is branching. A routing function reads the state and returns the name of the next node. This is Session 2's code/model boundary made literal: a cheap classifier decides the path, and code owns the branch.
def route(state: State) -> str:
last = state["messages"][-1].content.lower()
if "refund" in last:
return "billing" # -> node named "billing"
return "general"
builder.add_conditional_edges(
"classify", # from this node
route, # run this to pick the next
{"billing": "billing", "general": "general"}, # map return -> node
)Prefer returning a Command when a node wants to update state and jump in one move: return Command(update={"messages": [...]}, goto="billing"). It keeps routing logic next to the state change that motivates it.
Tools
A tool is a Python function with a docstring and typed args. Bind tools to the model so it can request a call; a ToolNode executes the requested calls and feeds results back. The prebuilt tools_condition loops back to the model until it stops asking for tools.
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
@tool
def lookup_order(order_id: str) -> dict:
"""Look up an order by its ID."""
return {"id": order_id, "status": "shipped"}
tools = [lookup_order]
model_with_tools = model.bind_tools(tools)
builder.add_node("agent", lambda s: {"messages": [model_with_tools.invoke(s["messages"])]})
builder.add_node("tools", ToolNode(tools))
builder.add_conditional_edges("agent", tools_condition) # tool call? -> "tools", else END
builder.add_edge("tools", "agent") # feed results back, loopA bound tool means the model proposes the call, LangGraph runs it. For anything with a side effect (refunds, deletes, emails) validate the arguments and check authorization inside the tool function, or gate it behind a human-in-the-loop interrupt. Binding a tool is not the same as authorizing every call.
Memory · checkpointing
Compile the graph with a checkpointer and every step is saved. Pass a thread_id and the graph remembers that conversation across calls, survives a crash, and can be replayed. Swap InMemorySaver for the Postgres saver in production, nothing else changes.
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [{"role": "user", "content": "my name is Sam"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "what's my name?"}]}, config)
# -> "Your name is Sam." State was reloaded from the checkpoint by thread_id.
# production: from langgraph.checkpoint.postgres import PostgresSaverCheckpointing is short-term memory (one thread). For long-term memory shared across threads, LangGraph adds a Store (a key-value store with optional vector search) you read and write inside nodes.
Human-in-the-loop
Because state is checkpointed, the graph can stop mid-run, wait for a human, and pick up exactly where it left off, even days later. Call interrupt() inside a node to pause; resume by invoking with Command(resume=...). This is how you gate risky actions.
from langgraph.types import interrupt, Command
def approve_refund(state: State) -> dict:
decision = interrupt({"ask": "Approve this refund?", "amount": state["amount"]})
if decision != "yes":
return {"messages": [{"role": "assistant", "content": "Refund declined."}]}
return {"messages": [{"role": "assistant", "content": "Refund issued."}]}
# first call pauses at the interrupt and returns control to you
graph.invoke(inputs, config)
# a human decides, then you resume the SAME thread
graph.invoke(Command(resume="yes"), config)Streaming
Use .stream() (or .astream()) with a stream_mode to get progressive output instead of waiting for the whole run. This is what makes an agent feel responsive.
# "updates": what each node changed, as it finishes
for chunk in graph.stream(inputs, config, stream_mode="updates"):
print(chunk)
# "messages": token-by-token LLM output, for a typing effect
for token, meta in graph.stream(inputs, config, stream_mode="messages"):
print(token.content, end="")
# "values": the full state snapshot after every step| stream_mode | You get | Use it for |
|---|---|---|
values | full state after each step | debugging, audit trails |
updates | only what a node changed | progress UI (“searching…”) |
messages | LLM tokens as they generate | the typing effect |
custom | whatever you emit from a node | tool progress, custom events |
Multi-agent
A compiled graph is itself a node. That is how you build multi-agent systems: each agent is a subgraph, and a supervisor node routes between them. The two common shapes:
flowchart TB
U(["user"]) --> S{"supervisor
routes"}
S -->|"research"| R["researcher
subgraph"]
S -->|"write"| W["writer
subgraph"]
S -->|"done"| E(["END"])
R --> S
W --> S
classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
class S code;
class R,W model;
from langgraph.graph import StateGraph, MessagesState, START
researcher = create_react_agent(model, tools=[web_search], prompt="Research facts.")
writer = create_react_agent(model, tools=[], prompt="Write the final answer.")
def supervisor(state: MessagesState) -> Command:
# a routing LLM (or rules) picks the next worker, or ends
nxt = pick_next(state) # "researcher" | "writer" | END
return Command(goto=nxt)
team = StateGraph(MessagesState)
team.add_node("supervisor", supervisor)
team.add_node("researcher", researcher) # a subgraph IS a node
team.add_node("writer", writer)
team.add_edge(START, "supervisor")
app = team.compile()For the common supervisor pattern there is a prebuilt: pip install langgraph-supervisor gives you create_supervisor(...) so you don't hand-write the routing.
Putting it together
PII stripped in code, the model classifies and can look up an order, a code rule pages on-call for high urgency, everything checkpointed per user. One fuzzy step, code on both sides.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.tools import tool
from langchain.chat_models import init_chat_model
model = init_chat_model("anthropic:claude-sonnet-4-5")
class State(TypedDict):
messages: Annotated[list, add_messages]
urgency: str
@tool
def lookup_order(order_id: str) -> dict:
"""Look up an order status by ID."""
return {"id": order_id, "status": "delayed"}
tools = [lookup_order]
agent_model = model.bind_tools(tools)
def strip_pii(state: State) -> dict:
# deterministic redaction BEFORE the model sees anything
return {"messages": [redact(m) for m in state["messages"]]}
def agent(state: State) -> dict:
return {"messages": [agent_model.invoke(state["messages"])]}
def route_urgency(state: State) -> str:
return "page" if state.get("urgency") == "high" else "done"
def page_oncall(state: State) -> dict:
notify_oncall() # code owns the side effect
return {}
b = StateGraph(State)
b.add_node("strip_pii", strip_pii)
b.add_node("agent", agent)
b.add_node("tools", ToolNode(tools))
b.add_node("page", page_oncall)
b.add_edge(START, "strip_pii")
b.add_edge("strip_pii", "agent")
b.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: "__route__"})
b.add_edge("tools", "agent")
b.add_conditional_edges("__route__", route_urgency, {"page": "page", "done": END})
b.add_edge("page", END)
app = b.compile(checkpointer=InMemorySaver())Production
InMemorySaver for PostgresSaver (or the async variant) so state survives restarts and scales across workers.langgraph-cli (a langgraph.json declares your graph), or just wrap graph.invoke in your own FastAPI route.LANGSMITH_TRACING=true and you get a step-by-step timeline, token counts, and latencies for free.graph.invoke(..., {"recursion_limit": 25})) so a looping agent can't run forever.Gotchas
If messages is a plain list without Annotated[list, add_messages], each node overwrites the history instead of appending. Symptom: the agent forgets everything mid-run. The reducer is not optional for message state.
Memory across calls needs both a checkpointer at compile time and a thread_id in the config on every invoke. Miss either and each call starts blank.
An agent that keeps calling tools hits the default recursion limit and raises. That's a feature, but tune the limit and add an explicit stop condition rather than relying on the crash.
Reach for it when
Skip it when
llm helpers is simpler and cheaper to reason about.The principle it teaches
The code / model boundary should be explicit. LangGraph makes the boundary a diagram: the graph is deterministic code you own, the nodes are where the model is allowed to act, and the checkpointer plus interrupts are where humans stay in control. That is Sessions 2 and 3, in a library.
A Week 1 tool. The router you build in Lab 1 is a LangGraph graph in miniature: classify, then branch. When your orchestrator-workers or evaluator-optimizer pattern outgrows plain functions, this is what you graduate to.