LangGraph Explained: Building Stateful, Controllable LLM Agents

Introduction

Most LLM applications start as a single prompt and a single response. That stops being enough once the model needs to call tools, check its own work, retry on failure, ask a human for approval, or remember what happened three steps ago. At that point you are no longer writing a prompt. You are writing a program with control flow, and the control flow is the hard part.

LangGraph is a library from the LangChain team for building these programs as graphs. Instead of hiding the loop inside a framework, you declare the steps (nodes), the transitions between them (edges), and the shared state that flows through them. The result is an agent you can read, test, pause, resume, and debug.

This article explains the core ideas, builds a small but realistic agent with a tool-calling loop and a human approval step, and then covers the concerns that matter once you move beyond a demo: persistence, observability, and failure handling.

Why a Graph?

Agent loops are usually written as a while loop around a model call. That works until you need any of the following:

• Branching. If the model asks for a tool, run it; if it answers, finish; if the answer looks wrong, retry with feedback.
• Cycles. The agent revisits a step until a condition is met.
• Pausing. A person must approve an action before it runs.
• Recovery. A crash in step five should not force a restart from step one.

A graph makes each of these explicit. Each node is a function that receives the state and returns an update. Each edge says where control goes next, and edges can be conditional. Because the structure is data, you can inspect it, visualize it, and reason about every path through it.

Core Concepts

LangGraph has four building blocks:

• State. A typed dictionary shared by all nodes. It is the only way nodes communicate.
• Nodes. Plain Python functions that read the state and return a partial update.
• Edges. Fixed transitions from one node to another, or conditional transitions chosen by a routing function.
• Checkpointer. An optional persistence layer that saves the state after each step, enabling resumption, time travel, and human-in-the-loop pauses.

Updates to state are merged into the existing state. A reducer decides how: by default a new value replaces the old one, but you can declare a reducer such as add_messages to append to a list instead.

Building a Tool-Calling Agent

The example below is a research assistant. It asks a model a question, lets the model call a search tool when it needs facts, loops until the model has enough information, and then drafts an answer. Before the answer is returned, a human reviewer can approve or reject it.

First, install the packages:

pip install -U langgraph langchain-openai langchain-core
export OPENAI_API_KEY=sk-...

Now the graph itself:

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


@tool
def search_docs(query: str) -> str:
    """Search the internal knowledge base and return the most relevant snippet."""
    # Replace with a real retriever (vector store, search API, etc.)
    knowledge = {
        "langgraph checkpoint": "Checkpointers persist graph state after each step.",
        "langgraph interrupt": "interrupt() pauses execution until a resume value is supplied.",
    }
    for key, value in knowledge.items():
        if key in query.lower():
            return value
    return "No relevant document found."


tools = [search_docs]
tools_by_name = {t.name: t for t in tools}
model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    approved: bool | None


def call_model(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}


def run_tools(state: AgentState) -> dict:
    last: AIMessage = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        output = tools_by_name[call["name"]].invoke(call["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=call["id"]))
    return {"messages": results}


def human_review(state: AgentState) -> dict:
    # Pause here until a reviewer responds. The checkpointer keeps the state safe.
    from langgraph.types import interrupt

    answer = state["messages"][-1].content
    decision = interrupt({"question": "Approve this answer?", "draft": answer})
    return {"approved": decision == "approve"}


def route_after_model(state: AgentState) -> Literal["run_tools", "human_review"]:
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "run_tools"
    return "human_review"


def route_after_review(state: AgentState) -> Literal["call_model", "__end__"]:
    # Rejected drafts go back to the model with a note; approved ones finish.
    if state["approved"]:
        return END
    state["messages"].append(HumanMessage(content="Reviewer rejected this. Improve it."))
    return "call_model"


builder = StateGraph(AgentState)
builder.add_node("call_model", call_model)
builder.add_node("run_tools", run_tools)
builder.add_node("human_review", human_review)

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", route_after_model)
builder.add_edge("run_tools", "call_model")
builder.add_conditional_edges("human_review", route_after_review)

graph = builder.compile(checkpointer=MemorySaver())

A few details matter here:

• The tool loop is a cycle: run_tools returns to call_model, and the routing function decides when to stop.
• add_messages appends new messages instead of overwriting the history.
• interrupt() pauses the run. Nothing after it executes until the graph is resumed.
• The checkpointer is what makes the pause possible. Without it, there is nothing to resume from.

Running the graph and pausing for review looks like this:

from langgraph.types import Command

config = {"configurable": {"thread_id": "research-42"}}

result = graph.invoke(
    {"messages": [HumanMessage(content="How does LangGraph checkpointing work?")], "approved": None},
    config,
)

# The run is paused at interrupt(). Inspect the pending question:
print(result["__interrupt__"])

# A reviewer approves the draft. The same thread resumes from the saved state.
final = graph.invoke(Command(resume="approve"), config)
print(final["messages"][-1].content)

The thread_id identifies one conversation. Because state is checkpointed per thread, the reviewer can respond minutes or hours later, and a different process can resume it.

Streaming and Observability

For interactive applications, stream intermediate steps so users see progress instead of a spinner:

for update in graph.stream(
    {"messages": [HumanMessage(content="Explain interrupt()")], "approved": None},
    config,
    stream_mode="updates",
):
    print(update)

Each update names the node that produced it, which makes logs readable. In production, connect the graph to LangSmith or another tracing backend so every model call, tool call, and state transition is recorded with its inputs and outputs. When an agent gives a wrong answer, a trace shows exactly which step went wrong.

Persistence in Production

MemorySaver keeps checkpoints in process memory, which is fine for tests and nothing else. For production, use a durable checkpointer such as the PostgreSQL or Redis implementations provided by LangGraph, or write your own by implementing the checkpointer interface. Keep these rules in mind:

• Use stable thread IDs. Derive them from the user or the business object, not from a random value generated per request, so a run can be found again.
• Version your state schema. When you add a field to AgentState, old checkpoints must still load. Give new fields defaults, and avoid renaming existing ones.
• Bound the loop. A model that keeps calling tools can run forever. Add a step counter to the state and route to an end node when it exceeds a limit.
• Keep side effects idempotent. A node can be retried after a crash, so a node that sends an email or charges a card needs an idempotency key.

Error Handling and Retries

Tool calls fail. APIs time out, rate limits trigger, and models produce malformed arguments. Handle these inside the node so the graph can decide what to do next:

def run_tools_safely(state: AgentState) -> dict:
    last: AIMessage = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        try:
            output = tools_by_name[call["name"]].invoke(call["args"])
        except Exception as exc:  # narrow this to the exceptions you expect
            output = f"Tool error: {exc}. Try a different query."
        results.append(ToolMessage(content=str(output), tool_call_id=call["id"]))
    return {"messages": results}

Returning the error as a tool message, instead of raising it, lets the model read the failure and recover. Reserve raised exceptions for failures the model cannot fix, and let the retry policy on the node handle transient network errors.

Testing the Graph

Because nodes are plain functions and routing is a plain function, you can test them without calling a model:

def test_routes_to_tools_when_model_requests_a_call():
    msg = AIMessage(content="", tool_calls=[{"name": "search_docs", "args": {"query": "x"}, "id": "1"}])
    state = {"messages": [msg], "approved": None}
    assert route_after_model(state) == "run_tools"


def test_routes_to_review_when_model_answers():
    state = {"messages": [AIMessage(content="Final answer")], "approved": None}
    assert route_after_model(state) == "human_review"

Use a fake model for end-to-end tests, and keep a small set of recorded real-model runs to catch regressions in prompts.

LangGraph and Java

Most LangGraph examples are in Python and JavaScript, but the concepts map directly to Java. The project LangGraph4j provides the same model (state, nodes, conditional edges, and checkpoints) for JVM applications, which fits well in Spring Boot services that already own the business logic. The pattern is identical: a typed state object, one method per node, and routing methods that return the next node name. If your organization already runs Java services, the graph can live beside the rest of your domain code instead of in a separate Python stack.

Best Practices

1. Keep state minimal. Store the facts the graph needs to route and continue, not every intermediate string.
2. Make nodes small and single-purpose. A node that calls a model, parses output, and writes to a database is hard to test and hard to retry.
3. Put business rules in routing functions or nodes, not in the prompt. A rule the code enforces is easier to verify than one the model is asked to follow.
4. Put a human in the loop for irreversible actions. Use interrupt() before sending messages, spending money, or changing records.
5. Trace everything in staging and production. You cannot fix an agent you cannot observe.
6. Test routing and tools without the model. Most bugs live in the control flow, and that part is deterministic.

Conclusion and Key Takeaways

• LangGraph models an LLM application as a graph of nodes, edges, and shared typed state.
• Cycles, conditional routing, and checkpoints make tool-calling loops, retries, and human approval explicit instead of hidden in a while loop.
• interrupt() plus a durable checkpointer lets a run pause for a person and resume later, even in another process.
• Production concerns are the same as for any stateful service: stable IDs, schema evolution, bounded loops, idempotent side effects, and tracing.
• Keep routing logic in code where it can be tested, and let the model handle only the parts that genuinely need language understanding.

If you are building agents with LangGraph, or have moved one from prototype to production, I would like to hear what patterns worked for you and where it fought back.