Skip to content

3. LangGraph

Intermediate · 14 min read

A RAG pipeline is a straight line: retrieve, then generate. Real LLM applications need loops and branches. Retrieve again if the chunks are bad. Call tools until the task is done. Pause for a human to approve a refund. Resume a conversation tomorrow. LangGraph models an application as a graph: nodes are steps, edges decide what runs next, and a shared state flows through it. It comes from the LangChain team but doesn't require LangChain. Nodes can be any Python functions.

3.1 The three building blocks

Concept What it is Example
State A dict-like object shared by all nodes (usually a TypedDict) {"question": ..., "docs": [...], "answer": ...}
Node A function that takes the state and returns an update (only the keys it changes) retrieve(state) -> {"docs": [...]}
Edge What runs next: fixed (a → b) or conditional (a function of the state picks the next node) grade → generate if the docs are relevant, else grade → rewrite

Plus special START and END markers, and reducers, which say how an update merges into the state. For example, add_messages appends to the message list instead of replacing it.

3.2 A state graph from scratch

The core of LangGraph is a loop: run the current node, merge its update into the state, follow an edge, and stop at END. Here it is in plain Python, running a corrective RAG flow. If the retrieved documents aren't relevant, rewrite the query and retrieve again, at most twice.

START, END = "__start__", "__end__"

class StateGraph:
    def __init__(self):
        self.nodes, self.edges, self.routers = {}, {}, {}

    def add_node(self, name, fn):
        self.nodes[name] = fn

    def add_edge(self, source, target):
        self.edges[source] = target

    def add_conditional_edges(self, source, router):
        self.routers[source] = router                     # router(state) -> name of the next node

    def invoke(self, state: dict, max_steps: int = 10) -> dict:
        node = self.edges[START]
        for step in range(max_steps):
            update = self.nodes[node](state)
            state = {**state, **update}                   # merge the node's partial update
            print(f"step {step}: {node:8} -> {update}")
            node = self.routers[node](state) if node in self.routers else self.edges[node]
            if node == END:
                return state
        raise RuntimeError("Hit max_steps: a loop never reached END")   # LangGraph's recursion_limit

CORPUS = {"refunds.md": "Damaged items get a full refund including shipping.",
          "shipping.md": "Express shipping takes 1 day and costs 9 dollars."}

def retrieve(state):
    words = {w for w in state["query"].lower().split() if len(w) > 3}     # skip "a", "for" ...
    return {"docs": [name for name, text in CORPUS.items() if words & set(text.lower().rstrip(".").split())]}

def grade(state):                                         # in a real graph: an LLM judges relevance
    return {"relevant": "refunds.md" in state["docs"]}

def rewrite(state):                                       # in a real graph: an LLM rewrites the query
    better = {"money back for a broken phone": "refund for damaged items"}
    return {"query": better.get(state["query"], state["query"]), "attempts": state["attempts"] + 1}

def generate(state):
    if not state["relevant"]:
        return {"answer": "I don't know based on the documents."}
    return {"answer": f"{CORPUS[state['docs'][0]]} [{state['docs'][0]}]"}

def after_grade(state) -> str:
    if state["relevant"]:
        return "generate"
    return "rewrite" if state["attempts"] < 2 else "generate"

graph = StateGraph()
for name, fn in [("retrieve", retrieve), ("grade", grade), ("rewrite", rewrite), ("generate", generate)]:
    graph.add_node(name, fn)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "grade")
graph.add_conditional_edges("grade", after_grade)
graph.add_edge("rewrite", "retrieve")                     # the loop
graph.add_edge("generate", END)

final = graph.invoke({"query": "money back for a broken phone", "attempts": 0})
print("\nANSWER:", final["answer"])
Output
step 0: retrieve -> {'docs': []}
step 1: grade    -> {'relevant': False}
step 2: rewrite  -> {'query': 'refund for damaged items', 'attempts': 1}
step 3: retrieve -> {'docs': ['refunds.md']}
step 4: grade    -> {'relevant': True}
step 5: generate -> {'answer': 'Damaged items get a full refund including shipping. [refunds.md]'}

ANSWER: Damaged items get a full refund including shipping. [refunds.md]

The user's wording finds nothing. The grader routes to rewrite, the edge loops back to retrieve, and the second attempt, using the documents' vocabulary, succeeds. Every step returned only the keys it changed, and the graph merged them into the state.

flowchart LR
    S((START)) --> R[retrieve] --> G[grade]
    G -->|relevant| GEN[generate] --> E((END))
    G -->|not relevant, attempts < 2| RW[rewrite] --> R
    G -->|gave up| GEN

That's the whole idea. A fixed chain can't express "go back and try again". A graph can, and max_steps stops a loop that never ends.

3.3 The same graph in LangGraph

# no-run — pip install langgraph
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class RAGState(TypedDict):
    query: str
    docs: list[str]
    relevant: bool
    attempts: int
    answer: str

builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)        # the same node functions as above, or real LLM-backed ones
builder.add_node("grade", grade)
builder.add_node("rewrite", rewrite)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "grade")
builder.add_conditional_edges("grade", after_grade, ["generate", "rewrite"])
builder.add_edge("rewrite", "retrieve")
builder.add_edge("generate", END)

graph = builder.compile()
final = graph.invoke({"query": "money back for a broken phone", "attempts": 0},
                     config={"recursion_limit": 10})
print(final["answer"])
print(graph.get_graph().draw_mermaid())      # LangGraph can draw its own diagram

The code is nearly identical, and that's the point of building it by hand first. What LangGraph adds over the 25-line version is everything in 3.5–3.7: persistence, interrupts, streaming, parallel branches, and tracing.

3.4 A tool-calling agent as a graph

An agent is a small graph with a loop. The model node decides whether to call tools. If it does, the tool node runs them and loops back. If it doesn't, the graph ends. This is the tool loop drawn as a graph:

# no-run
from typing import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def get_order_status(order_id: str) -> str:
    """Look up the shipping status of an order by its ID."""
    return {"A100": "Shipped, arriving Friday"}.get(order_id, "Order not found")

tools = [get_order_status]
model = init_chat_model("openai:gpt-4o-mini", temperature=0).bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list, add_messages]        # reducer: append new messages instead of overwriting

def call_model(state: State):
    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))         # runs every tool call in the last AI message
builder.add_edge(START, "model")
builder.add_conditional_edges("model", tools_condition)   # tool calls? -> "tools", else -> END
builder.add_edge("tools", "model")
agent = builder.compile()

For a standard agent you don't need to write this yourself. LangChain's create_agent (page 1) builds this graph for you. Write the graph by hand when you need custom steps: a guardrail node before tools, a planning step, or different models for different nodes.

3.5 Checkpoints: memory and resuming

Compile with a checkpointer and LangGraph saves the state after every step, keyed by a thread_id. That one feature gives you:

  • Conversation memory: call again with the same thread_id, and the earlier messages are already in the state.
  • Fault tolerance: a crashed run resumes from the last completed step.
  • Time travel: inspect or replay the state at any earlier step, which helps a lot with debugging.
# no-run
from langgraph.checkpoint.memory import InMemorySaver      # Postgres / SQLite / Redis savers for production

agent = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "customer-42"}}

agent.invoke({"messages": [{"role": "user", "content": "Hi, I'm Priya. Where is order A100?"}]}, config)
reply = agent.invoke({"messages": [{"role": "user", "content": "What was my name again?"}]}, config)
print(reply["messages"][-1].content)                       # remembers "Priya": same thread, saved state

print(agent.get_state(config).values["messages"][-1].content)   # inspect the saved state at any time

3.6 Human-in-the-loop with interrupts

Some actions need a person's approval: refunds, emails, deleting data. interrupt() pauses the graph inside a node and saves the state. Your app shows the request to a human. Later, perhaps hours later, it resumes with their answer:

# no-run — needs a checkpointer, because the paused state must be saved
from langgraph.types import Command, interrupt

def issue_refund(state):
    decision = interrupt({"action": "refund", "order": state["order_id"], "amount": state["amount"]})
    if decision != "approve":
        return {"result": "Refund rejected by a human reviewer."}
    return {"result": process_refund(state["order_id"], state["amount"])}

result = graph.invoke({"order_id": "A100", "amount": 1200}, config)
print(result["__interrupt__"])                     # the payload to show the reviewer
graph.invoke(Command(resume="approve"), config)    # continue from exactly where it paused

This is the safe pattern for agents with real-world actions: the LLM can propose a risky action, but code and a human decide whether it happens.

3.7 Streaming and multi-agent patterns

Streaming: graph.stream(inputs, config, stream_mode="updates") yields each node's update as it finishes, which you can show as "Searching documents... Checking relevance... Writing answer...". stream_mode="messages" streams the LLM tokens from inside nodes.

Multi-agent systems are graphs whose nodes are agents:

Pattern How it works Use when
Supervisor A router agent decides which specialist agent runs next, then reviews the result Clear specialist roles (research, SQL, writing)
Handoffs / swarm Agents transfer control to each other directly through tool calls Conversational flows like triage → billing → technical support
Hierarchical Supervisors of supervisors. Each team is a subgraph Large systems with team-level separation
Parallel fan-out Several nodes run at once on the same state, and a node merges their outputs Independent sub-tasks: multi-query search, several reviewers

Start with one agent and good tools. Multi-agent graphs multiply cost, latency and the ways things can fail. Add agents only when a single agent's prompt and tool list become unmanageable.

Interview questions

What is LangGraph, and why use it instead of a simple chain?

LangGraph models an LLM application as a state graph: nodes are functions that update a shared state, and edges, which can be conditional, pick the next node. Unlike a linear chain, it supports loops (retry, agent tool loops), branches and parallel steps. Checkpointing gives memory, fault tolerance and time travel, and interrupts enable human-in-the-loop. Use it when control flow goes beyond a straight line.

How does memory work in LangGraph?

You compile the graph with a checkpointer (in-memory, SQLite, Postgres, Redis). After each step the state is saved under a thread_id from the config. Invoking again with the same thread_id loads that state, so earlier messages are present. That's short-term, per-conversation memory. Long-term memory across threads uses a separate store.

How would you add human approval before an agent issues a refund?

Put the refund in its own node and call interrupt() with the details before executing. With a checkpointer, the graph pauses and saves its state. The app shows the request to a reviewer, then resumes with Command(resume=decision). The node runs the refund only on approval. The LLM proposes, and code plus a human authorise.

Supervisor versus swarm multi-agent architectures?

In a supervisor architecture, a central agent routes work to specialist agents and collects results, so control is centralised and easy to reason about. In a swarm or handoff architecture, agents pass control to each other directly, so control is decentralised, which suits conversational transfers. Both add cost and complexity over a single agent, so justify them with evals.

Practice

  • In the from-scratch graph, set the rewrite lookup to return the same query and watch it give up after 2 attempts.
  • Add a checkpoints list to the from-scratch StateGraph.invoke that stores the state after each step. Print it.
  • Draw (on paper or in Mermaid) a supervisor graph with a "policy search" agent and an "order lookup" agent.

Next: 4. When plain SDK code is better Back to: Frameworks overview · Notes overview