back to knowledge base
module 086 min read

LangGraph

State graphs, nodes, edges, conditional routing, cycles, and agent loops.

LCEL chains ([[07_langchain]]) are directed acyclic — data flows one way, no loops. But agents need cycles (think → act → observe → think again), branching (route by condition), persistent state, and human-in-the-loop pauses. LangGraph models your application as a state machine / graph, giving you exactly that control.

8.1 Mental model: a graph of state transformations

  • State: a shared data object that flows through the graph and gets updated by each step.
  • Nodes: functions that read the state and return updates to it.
  • Edges: define what runs next. Normal edges are fixed; conditional edges choose the next node based on state (this enables branching and loops).
  • Compile the graph → a Runnable you .invoke() / .stream() like any LangChain component.
code
        ┌───────────────── (conditional: keep going? ) ──────────┐
START ─► agent ─► should_continue? ─► tools ─► agent ...          │
                       │ else                                     │
                       └─► END ◄──────────────────────────────────┘

8.2 State and the reducer concept

State is typically a TypedDict. Each key can have a reducer that says how updates merge. The classic example: a messages list that should append, not overwrite.

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages   # reducer that appends messages
from langchain_core.messages import BaseMessage

class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]   # append on update
    count: int                                              # default: overwrite

When a node returns {"messages": [new_msg]}, the add_messages reducer appends it to the existing list instead of replacing it. Keys without a reducer are overwritten by the returned value. Reducers are what make accumulating history clean.

8.3 Nodes and edges in code

python
def chatbot(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}        # update merged via reducer

graph = StateGraph(State)
graph.add_node("chatbot", chatbot)         # register a node
graph.add_edge(START, "chatbot")           # entry
graph.add_edge("chatbot", END)             # exit
app = graph.compile()                       # → a Runnable

app.invoke({"messages": [("user", "Hi!")]})

8.4 Conditional edges → branching and loops

A routing function inspects state and returns the name of the next node. This is how you build the agent cycle.

python
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

tools = [web_search, multiply]                  # from 07_langchain
llm_with_tools = llm.bind_tools(tools)

def agent(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def route(state: State):
    last = state["messages"][-1]
    if last.tool_calls:        # model wants to call a tool
        return "tools"
    return END                 # model gave a final answer

graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))        # prebuilt: executes tool calls
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")                # LOOP back → cycle!
app = graph.compile()

app.invoke({"messages": [HumanMessage("What's 23*17, then explain that number?")]})

The edge tools → agent creates the cycle: agent calls a tool, the tool result is appended to state, control returns to the agent, which decides whether to call another tool or finish. This is the ReAct loop ([[09_agentic_ai]]) expressed as a graph.

langgraph.prebuilt.create_react_agent(llm, tools) builds this exact graph for you in one line.

8.5 Persistence (checkpointers) & memory

LangGraph can checkpoint state after every step into a saver (memory, SQLite, Postgres). This gives:

  • Conversation memory across calls — resume a thread by id.
  • Fault tolerance — restart from the last checkpoint.
  • Time travel — inspect/replay/branch from past states.
python
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

cfg = {"configurable": {"thread_id": "user-1"}}
app.invoke({"messages": [HumanMessage("My name is Dev")]}, cfg)
app.invoke({"messages": [HumanMessage("What's my name?")]}, cfg)  # remembers → "Dev"

The thread_id separates independent conversations; the checkpointer reloads that thread's accumulated state automatically.

8.6 Human-in-the-loop

Pause the graph before a sensitive action (e.g. sending an email, executing code), get human approval, then resume. Use interrupt_before:

python
app = graph.compile(checkpointer=MemorySaver(), interrupt_before=["tools"])
# run until just before "tools"; state is saved; you can inspect/modify,
# then call app.invoke(None, cfg) to continue, or edit the state first.

Critical for safety in production agents — a human gates irreversible or costly tool calls.

8.7 Multi-agent systems

Each agent is a node (or a subgraph). A supervisor node routes work between specialist agents.

code
                 ┌──────────► researcher ──┐
START ─► supervisor ─────────► coder      ─┼─► supervisor (decide next) ─► END
                 └──────────► writer      ─┘
  • Supervisor pattern: a router LLM decides which specialist acts next based on state.
  • Network/swarm: agents hand off to each other directly.
  • Hierarchical: teams of agents, each with their own supervisor.
python
def supervisor(state):
    decision = router_llm.invoke(state["messages"])   # returns next worker name
    return {"next": decision.content}

graph.add_conditional_edges("supervisor", lambda s: s["next"],
        {"researcher": "researcher", "coder": "coder", "writer": "writer", "FINISH": END})
for worker in ["researcher", "coder", "writer"]:
    graph.add_edge(worker, "supervisor")   # report back to supervisor

8.8 Streaming graph execution

python
for event in app.stream({"messages": [HumanMessage("Plan a trip to Japan")]}, cfg):
    for node_name, update in event.items():
        print(node_name, "→", update)      # watch each node fire in real time

You can stream at node granularity (which step ran) or token granularity (LLM output) — invaluable for debugging agent reasoning.

8.9 LangGraph vs LangChain (when to use which)

LangChain (LCEL)LangGraph
FlowLinear DAG, one directionArbitrary graph with cycles & branches
StatePassed along the pipeExplicit, persistent, reducer-merged
Best forRAG, simple chains, single tool callAgents, loops, multi-step control, multi-agent, human-in-loop
MemoryBolt-on (RunnableWithMessageHistory)First-class (checkpointers)
ControlImplicitExplicit nodes/edges you design

Rule of thumb: chain for pipelines, graph for agents. They interoperate — a node in LangGraph can be an LCEL chain, and a compiled graph is itself a Runnable usable inside LCEL.

8.10 Full minimal ReAct agent (complete, runnable skeleton)

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def calculator(expr: str) -> str:
    """Evaluate a math expression like '23*17'."""
    return str(eval(expr))                       # demo only — sandbox in production

class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([calculator])

def agent(state): return {"messages": [llm.invoke(state["messages"])]}
def route(state): return "tools" if state["messages"][-1].tool_calls else END

g = StateGraph(State)
g.add_node("agent", agent)
g.add_node("tools", ToolNode([calculator]))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", route, {"tools": "tools", END: END})
g.add_edge("tools", "agent")
app = g.compile(checkpointer=MemorySaver())

cfg = {"configurable": {"thread_id": "1"}}
out = app.invoke({"messages": [HumanMessage("What is 23*17 + 5?")]}, cfg)
print(out["messages"][-1].content)

8.11 Pitfalls

  • Infinite loops: a cycle with no exit condition runs forever — always have a terminating route + a recursion/step limit (graph.compile(...) + config={"recursion_limit": 25}).
  • State overwrite vs append: forgetting a reducer means updates clobber prior state — use add_messages/custom reducers for accumulation.
  • Unbounded message growth: trim/summarize messages in a node to stay within context.
  • Non-deterministic routing: keep routing functions simple and testable; don't bury critical control flow inside an LLM call you can't inspect.

Next: [[09_agentic_ai]] — the reasoning patterns and system design that turn these graphs into capable autonomous agents.

prev · 07LangChainnext · 09Agentic AI
all modules