LangGraph Tutorial: Building Stateful AI Agents

LangGraph is a Python library for building applications as graphs. Nodes are steps. Edges decide which step runs next. A shared state object carries data between them. It was designed for AI agents, but the mechanics have nothing to do with language models. That is how this tutorial teaches them: every node below is plain Python, so you can run the whole thing without an API key. The last section shows where a model call slots in.

By the end you will have a small support-ticket agent that classifies a ticket, retries a flaky lookup in a loop, drafts a reply, pauses for a person to approve it, and resumes where it stopped. Those four behaviours are why people reach for LangGraph instead of a simple chain of prompts: branching, cycles, persistence and human-in-the-loop.

The code was run with langgraph 1.2.12 on Python 3.11.

Install

python -m pip install langgraph

The LangGraph documentation is the reference for everything below. The library changes quickly, so check it when something in an older tutorial doesn't match.

Step 1: Define the state

The state is a typed dictionary that every node can read. A node doesn't change it in place. It returns a dictionary of the fields it wants to change, and LangGraph merges that update in. How a field is merged depends on its reducer. By default a new value replaces the old one. With a reducer such as operator.add, new values are appended instead:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class TicketState(TypedDict):
    ticket: str
    category: str
    order_status: str
    lookup_attempts: int
    reply: str
    log: Annotated[list[str], operator.add]   # appended to, never overwritten

Here log accumulates one line per step, and every other field is overwritten by the latest update. Getting reducers right matters more than it looks. An easy bug to write is a list that should grow but is overwritten, or the reverse.

Step 2: Write the nodes

A node is a function that takes the state and returns a partial update. The classifier below uses keywords as a stand-in for a model call. The lookup simulates an unreliable order API that times out on its first call:

def classify(state: TicketState) -> dict:
    text = state["ticket"].lower()
    category = "billing" if ("refund" in text or "charged" in text) else "general"
    return {"category": category, "log": [f"classified as {category}"]}


def lookup_order(state: TicketState) -> dict:
    attempts = state.get("lookup_attempts", 0) + 1
    # Stand-in for a flaky API: the first call times out, the second succeeds.
    status = "timeout" if attempts == 1 else "delivered"
    return {"lookup_attempts": attempts, "order_status": status,
            "log": [f"order lookup attempt {attempts}: {status}"]}


def draft_reply(state: TicketState) -> dict:
    if state["category"] == "billing":
        reply = f"Your order shows as {state['order_status']}. We have started a refund review."
    else:
        reply = "Thanks for your message. We are looking into it."
    return {"reply": reply, "log": ["reply drafted"]}

Notice state.get("lookup_attempts", 0) and the branch in draft_reply. Fields that no node has written yet are simply absent from the state. A draft_reply that read state["order_status"] for every ticket would crash with a KeyError on general tickets, which never go through the lookup. Test each path through your graph, not just the one you had in mind.

Step 3: Route with conditional edges

A normal edge always goes to the same next node. A conditional edge calls a routing function that returns the name of the next node. That is how you get branches, and also loops: a router can send control back to a node that already ran.

def route_ticket(state: TicketState) -> Literal["lookup_order", "draft_reply"]:
    return "lookup_order" if state["category"] == "billing" else "draft_reply"


def after_lookup(state: TicketState) -> Literal["lookup_order", "draft_reply"]:
    if state["order_status"] == "timeout" and state["lookup_attempts"] < 3:
        return "lookup_order"          # loop back and retry
    return "draft_reply"

Two details are worth copying:

• The Literal return types are not decoration. LangGraph reads them to learn where a router can go. With them, graph.get_graph().draw_mermaid() draws every branch. Write the same router as an untyped lambda and the generated diagram shows the classifier going straight to the end, because LangGraph cannot see the possible destinations. A type hint, or a mapping passed to add_conditional_edges, fixes this.
• Every loop needs an exit condition you control. Here it is lookup_attempts < 3. LangGraph also stops any run that exceeds its recursion limit with a GraphRecursionError, and you can set a lower limit per run with config={"recursion_limit": ...}. Treat that limit as a safety net, not as your loop logic.

Step 4: Pause for a human with interrupt

The approval step calls interrupt(). This stops the graph, saves its state, and returns the payload to whoever called it. The run resumes only when you send back a value:

def human_approval(state: TicketState) -> dict:
    decision = interrupt({"question": "Send this reply?", "reply": state["reply"]})
    return {"log": [f"human decision: {decision}"]}

One rule from the interrupts documentation will save you debugging time. On resume, the node runs again from its first line, and interrupt() then returns the value you sent. Any code placed before the interrupt() call runs twice, so keep side effects, such as sending an email, after it or in a later node.

Step 5: Assemble and compile the graph

builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("lookup_order", lookup_order)
builder.add_node("draft_reply", draft_reply)
builder.add_node("human_approval", human_approval)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_ticket)
builder.add_conditional_edges("lookup_order", after_lookup)
builder.add_edge("draft_reply", "human_approval")
builder.add_edge("human_approval", END)

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

The checkpointer is what makes the graph stateful across calls. After every step LangGraph saves a checkpoint of the state under a thread ID. Interrupts need a checkpointer, because there is nothing to resume without saved state. InMemorySaver keeps checkpoints in memory, which is fine for learning. The persistence documentation describes the SQLite and Postgres savers to use when state has to survive a restart.

Step 6: Run it, pause it, resume it

config = {"configurable": {"thread_id": "ticket-1042"}}
result = graph.invoke({"ticket": "I was charged twice, please refund one payment."}, config)

print(result["__interrupt__"][0].value)
snapshot = graph.get_state(config)
print("paused before:", snapshot.next)

final = graph.invoke(Command(resume="approve"), config)
for line in final["log"]:
    print("-", line)
print("checkpoints saved:", len(list(graph.get_state_history(config))))

Output:

{'question': 'Send this reply?', 'reply': 'Your order shows as delivered. We have started a refund review.'}
paused before: ('human_approval',)
- classified as billing
- order lookup attempt 1: timeout
- order lookup attempt 2: delivered
- reply drafted
- human decision: approve
checkpoints saved: 7

Read it line by line and the whole model becomes concrete:

• The first invoke ran until the interrupt and returned its payload under __interrupt__. The graph is now paused, not finished.
• get_state shows the next node waiting to run. In a real application, this is when your UI shows the draft to a support agent. That could be seconds or days later, because the state is saved.
• The second invoke passes Command(resume="approve") with the same thread ID. LangGraph loads the checkpoint, and interrupt() returns "approve".
• The log shows the retry loop: one timeout, then success. The operator.add reducer kept every line.
• get_state_history returns every saved checkpoint for the thread, newest first. You can inspect any of them, or resume from an earlier one to replay a run, which the docs call time travel.

Run the graph again with a different thread ID and a ticket such as "How do I change my delivery address?". It is classified as general, skips the lookup entirely, and pauses at the same approval step. Each thread's state is independent, which is how one compiled graph serves many conversations at once.

Where the language model goes

Everything above is the agent's control flow, and it is deterministic, testable Python. A model replaces the parts that need judgement. The classifier is the obvious candidate. Write the node so it receives the model instead of creating one. The graph code then never depends on a provider:

from pydantic import BaseModel


class Triage(BaseModel):
    category: Literal["billing", "general"]


def make_classifier(model):
    """model: anything whose .invoke(prompt) returns a Triage."""
    def classify(state: TicketState) -> dict:
        result = model.invoke(f"Classify this support ticket: {state['ticket']}")
        return {"category": result.category, "log": [f"classified as {result.category}"]}
    return classify

In tests, a stub stands in for the model. It returns whatever category you give it, so you can drive each branch of the graph without an API key or a network call:

class StubTriageModel:
    def __init__(self, answer: str):
        self.answer = answer

    def invoke(self, prompt: str) -> Triage:
        return Triage(category=self.answer)


classify = make_classifier(StubTriageModel("billing"))
print(classify({"ticket": "Where is my parcel? It is two days late."}))

{'category': 'billing', 'log': ['classified as billing']}

The stub filed a delivery question under billing because it was told to. The node trusts the model's answer, so the type is the safety net: an answer outside the Literal, such as "shipping", fails Pydantic validation when Triage is built instead of reaching the router. In production you pass a real chat model configured for structured output against the same Triage class. LangChain's init_chat_model("provider:model-name").with_structured_output(Triage) does this; it needs the provider's integration package and an API key (LangChain models guide). Pass make_classifier(...) to add_node("classify", ...) in place of the keyword version, and the rest of the graph is unchanged.

The design principle carries over to larger agents. Let the model make decisions, such as which category, which tool, or what to write. Let the graph own the control flow: retries, limits, approvals and persistence. When an agent misbehaves, you can then tell whether the model made a bad call or the graph routed it badly.

Common mistakes

• Forgetting the checkpointer or the thread ID. Without them there is no memory between calls and no way to resume an interrupt.
• Side effects before interrupt(). They run again on resume.
• Unbounded loops. Put an explicit counter in the state, as lookup_attempts does here.
• Returning the whole state from a node. Return only the fields you changed; with an append reducer, returning everything duplicates the list.
• Testing only the happy path. Missing keys on untaken branches are the classic crash.

Where to go next

Add a real tool call in lookup_order, replace InMemorySaver with the SQLite saver, and build a small review screen for the interrupt. For how agents choose and call tools, see function calling and tool use in LLMs. For systems where several agents hand work to each other, see building a multi-agent system. The agentic AI roadmap places LangGraph within the wider set of agent skills. The GloryTecks Agentic AI course covers LangGraph alongside LangChain and RAG, with projects such as a multi-agent task solver.