LangGraph Fundamentals

The core primitives

• State — a typed dict (Pydantic / TypedDict) passed to every node.
• Nodes — pure-ish functions that take state and return a partial update.
• Edges — directed transitions; can be static or conditional.
• Reducers — how updates merge into state (replace, append, custom).
• Checkpointer — persists state per thread so runs can pause/resume.

A minimal graph

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # reducer = append

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

app = graph.compile()
app.invoke({"messages": [("user", "hello")]})

State & reducers

Each key in state can declare a reducer via Annotated[T, fn]. Without one, the new value replaces the old. Withadd_messages or operator.add, updates are appended — essential for chat history or accumulating tool calls.

Conditional edges & cycles

Conditional edges let a router function decide the next node — this is how ReAct loops, tool-calling agents, and retry logic are expressed.

def route(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_conditional_edges("agent", route, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")   # cycle back

Checkpointing & threads

A checkpointer (in-memory, SQLite, Postgres, Redis) snapshots state after every super-step. Each conversation/run is keyed by athread_id, so you get durable memory, resumability, and time-travel for free.

from langgraph.checkpoint.postgres import PostgresSaver

app = graph.compile(checkpointer=PostgresSaver.from_conn_string(DB_URL))
config = {"configurable": {"thread_id": "user-42"}}

app.invoke({"messages": [("user", "hi")]}, config)
app.invoke({"messages": [("user", "what did I just say?")]}, config)  # remembers

Human-in-the-loop

Pause before sensitive nodes (tool execution, writes, payments) withinterrupt_before or theinterrupt() primitive. The graph suspends, state is checkpointed, and a UI can review/edit before resuming with app.invoke(None, config).

Streaming

Three stream modes cover most UI needs:

• values — full state after each step.
• updates — only the diff a node produced.
• messages — token-by-token LLM output.

for chunk in app.stream(input, config, stream_mode="updates"):
    print(chunk)

Multi-agent topologies

• Supervisor — a router agent dispatches to worker agents.
• Hierarchical — supervisors of supervisors for complex teams.
• Network — peer agents handing off via Command(goto=...).
• Swarm — dynamic handoffs with shared scratchpad.

Subgraphs

Compile a graph and use it as a node in another graph. Great for encapsulating a research agent, a RAG pipeline, or a writer/critic loop as a reusable unit.

Execution model

super-step N
  ├─ collect all nodes whose inputs are ready
  ├─ run them in parallel
  ├─ merge their updates via reducers
  └─ checkpoint state
super-step N+1 …

Inspired by Pregel: deterministic, parallel where possible, and replayable from any checkpoint.

When to choose LangGraph over LCEL

Production checklist

• Use a durable checkpointer (Postgres/Redis) in production.
• Cap recursion with recursion_limit to prevent runaway loops.
• Wire LangSmith tracing for step-by-step debugging.
• Interrupt before any destructive tool call.
• Deploy long-running graphs on LangGraph Platform / Cloud.