Build Stateful AI Agents with LangGraph State Machines

Overview

LangGraph extends LangChain with stateful, cyclic graphs for building agentic workflows. Instead of linear chains, you define a state machine: nodes are functions that read and update a shared state, edges are transitions, and conditional edges decide the next step at runtime. That lets an agent loop (plan, call a tool, look at the result, plan again) instead of executing once and hoping for the best.

This recipe builds a research agent with tool calling, conditional routing, checkpointing, and a human-approval gate. The same skeleton works for coding assistants, data-quality agents, and any workflow where the model needs several turns to converge on an answer.

The pieces that carry the design:

• AgentState is the contract every node reads and writes. The add_messages reducer appends new messages instead of replacing the list, so conversation history survives across nodes.
• should_continue is a plain function that inspects the state and returns where to go next. The graph topology stays data; the routing stays code.
• The checkpointer (MemorySaver in dev, SQLite/Postgres in production) snapshots state after every node under a thread_id. That’s what makes conversations resumable and human-in-the-loop possible.

When to Use

• Multi-step agents that need to loop: search, evaluate, refine, repeat
• Workflows with conditional branching (if confident, answer; otherwise, search more)
• Human-in-the-loop flows where a person approves tool calls before they run
• Stateful conversations that have to persist across sessions or restarts
• Multi-agent setups where several specialized nodes share one state

When Not to Use

Reach for something simpler when:

• A single LLM call with bind_tools() answers the request, so no graph is needed for a one-shot function call.
• The pipeline is linear: for prompt → model → parser flows, LangChain chains with LCEL are shorter and easier to reason about.
• You’re running batch jobs (classify 10,000 texts): a plain loop over model.invoke() beats agent orchestration, and streaming responses covers the UX side if needed.
• The agent never needs state: checkpointers add a write per node, so don’t pay for persistence you won’t read.

Prerequisites

• Python 3.10+
• langgraph, langchain-openai, and langchain-core packages
• An OPENAI_API_KEY for the model calls in the examples

Solution

1. Install Dependencies

pip install langgraph langchain-openai langchain-core

2. Define the Agent State

import re
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    search_results: str
    attempts: int
    max_attempts: int

3. Define Tools

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Simulated search — replace with Tavily, SerpAPI, etc.
    results = {
        "redis caching": "Redis cache-aside pattern: check cache, if miss, load from DB, set cache. TTL recommended.",
        "docker compose": "Docker Compose defines multi-container apps in YAML. Use profiles for environment-specific services.",
    }
    return results.get(query.lower(), f"No results found for: {query}")

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression (digits and operators only)."""
    if not re.fullmatch(r"[\d\s+\-*/().]+", expression):
        return "Error: expression must contain only numbers and operators"
    try:
        return f"Result: {eval(expression)}"
    except Exception as e:
        return f"Error: {e}"

tools = [search_web, calculate]

The regex gate matters: eval() on raw model output is a code-injection hole. Whitelisting digits and operators keeps the tool safe enough for an example. A stricter parser (ast.literal_eval or a tiny expression evaluator) is the right call in production.

4. Define Graph Nodes

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    """Main agent node — decides what to do next."""
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """Execute tool calls from the agent's last message."""
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_id = tool_call["id"]

        # Find and execute the tool
        tool_map = {t.name: t for t in tools}
        if tool_name in tool_map:
            try:
                result = tool_map[tool_name].invoke(tool_args)
            except Exception as e:
                result = f"Tool {tool_name} failed: {e}"
            tool_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_id,
            ))

    return {"messages": tool_messages, "attempts": state.get("attempts", 0) + 1}

def should_continue(state: AgentState) -> str:
    """Conditional edge — decide whether to use tools or end."""
    last_message = state["messages"][-1]
    attempts = state.get("attempts", 0)
    max_attempts = state.get("max_attempts", 5)

    if last_message.tool_calls and attempts < max_attempts:
        return "tools"
    return END

The try/except inside tool_node is deliberate: if a tool raises and the error never reaches the model as a ToolMessage, the agent can’t recover; it just sees silence and either retries forever or gives up with no context.

5. Build the Graph

def build_agent_graph(checkpointer=None):
    """Build and compile the LangGraph agent."""
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)

    # Set entry point
    workflow.set_entry_point("agent")

    # Add conditional edge from agent
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "tools": "tools",
            END: END,
        },
    )

    # After tools, go back to agent
    workflow.add_edge("tools", "agent")

    return workflow.compile(checkpointer=checkpointer)

app = build_agent_graph(checkpointer=MemorySaver())

should_continue returns a string and the mapping dict resolves it to a node. Returning END (the __end__ constant) finishes the run; you can map it to a cleanup or summarizer node instead if the agent needs an epilogue.

6. Run the Agent

config = {"configurable": {"thread_id": "session-1"}}

result = app.invoke(
    {
        "messages": [HumanMessage(content="What is the Redis cache-aside pattern?")],
        "max_attempts": 5,
    },
    config=config,
)

# Print the final response
for msg in result["messages"]:
    if isinstance(msg, AIMessage):
        print(f"AI: {msg.content}")
    elif isinstance(msg, ToolMessage):
        print(f"Tool: {msg.content}")

7. Stream Agent Steps

for event in app.stream(
    {
        "messages": [HumanMessage(content="Calculate 15 * 23 and then search for Docker Compose")],
        "max_attempts": 5,
    },
    config=config,
):
    for node_name, node_output in event.items():
        print(f"\n--- {node_name} ---")
        if "messages" in node_output:
            for msg in node_output["messages"]:
                if isinstance(msg, AIMessage):
                    print(f"AI: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"Tool result: {msg.content[:100]}")

How It Works

The diagram shows the compiled topology: the agent loops through tools until should_continue reports END.

1. StateGraph compiles a directed graph where each node is a function that receives the shared AgentState and returns a partial update. Nodes never mutate state in place; they return dicts that LangGraph merges using each field’s reducer.
2. Reducers control the merge. add_messages appends; a plain field like attempts overwrites. Pick reducers deliberately. A list field without add_messages gets replaced on every node update and silently loses history.
3. Conditional edges route at runtime. should_continue sees the state after the agent node ran: if the last message asks for tools and the attempt budget isn’t spent, it routes to tools; otherwise the run ends.
4. The cycle is what makes this an agent rather than a chain. Tool results come back as ToolMessage entries, the agent reads them and decides again. The loop only stops when the model answers without tool calls or the budget runs out.
5. Checkpointing snapshots the full state after every node under the thread_id in config. Same thread_id means the next invoke resumes with the prior messages, which is also why each user session needs its own ID.

Variants

Human-in-the-Loop Approval

from langgraph.types import interrupt, Command

def tool_node_with_approval(state: AgentState) -> dict:
    """Execute tools with human approval for sensitive operations."""
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        if tool_call["name"] == "calculate":
            # Interrupt and wait for human approval
            approval = interrupt({
                "question": f"Approve calculation: {tool_call['args']['expression']}?",
                "tool_call": tool_call,
            })

            if not approval:
                return {"messages": [ToolMessage(
                    content="Calculation rejected by user.",
                    tool_call_id=tool_call["id"],
                )]}

    # Proceed with normal tool execution
    return tool_node(state)

# First call pauses inside the interrupt:
result = app.invoke(
    {"messages": [HumanMessage(content="Calculate 100 / 0")]},
    config=config,
)
# Resume with the human's answer:
result = app.invoke(Command(resume="approved"), config=config)

Two details worth knowing. interrupt and Command live in langgraph.types, not langgraph.graph; the older import path doesn’t exist in current releases. And resuming re-runs the whole node from the top, so keep node code idempotent (or put the interrupt early in the node).

Multi-Agent Collaboration

def researcher_node(state: AgentState) -> dict:
    """Researcher agent — searches for information."""
    response = model_with_tools.invoke([
        {"role": "system", "content": "You are a research assistant. Search for information."},
        *state["messages"],
    ])
    return {"messages": [response]}

def writer_node(state: AgentState) -> dict:
    """Writer agent — synthesizes findings into a report."""
    response = model.invoke([
        {"role": "system", "content": "You are a technical writer. Synthesize the research into a clear report."},
        *state["messages"],
    ])
    return {"messages": [response]}

def build_multi_agent_graph(checkpointer=None):
    workflow = StateGraph(AgentState)
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("writer", writer_node)

    workflow.set_entry_point("researcher")
    workflow.add_conditional_edges("researcher", should_continue, {"tools": "tools", END: "writer"})
    workflow.add_edge("tools", "researcher")
    workflow.add_edge("writer", END)

    return workflow.compile(checkpointer=checkpointer)

Here END from should_continue maps to writer instead of the real end: the conditional edge’s return value goes through the mapping, so “the run is done” can mean “hand off to the next specialist” rather than “stop”.

Persistent State with SQLite

from langgraph.checkpoint.sqlite import SqliteSaver

# Persist state to SQLite for cross-session continuity
app = build_agent_graph(checkpointer=SqliteSaver.from_conn_string("agent_state.db"))

# State persists across sessions with the same thread_id

MemorySaver dies with the process: fine for notebooks, useless for a service. SqliteSaver survives restarts on one box; PostgresSaver is the production option when several workers need the same checkpoints.

Troubleshooting

• GraphInterrupt propagates as an exception: you’re using interrupt() without a checkpointer. Interrupts persist state to resume later, so compile(checkpointer=...) is mandatory: no checkpointer, no pause.
• The agent loops forever: should_continue is returning "tools" every round. Check the max_attempts budget, then the hard stop: recursion_limit in config (default 25) kills runaway graphs.
• State doesn’t carry between invokes: different thread_id per call, or no checkpointer. Both thread ID and checkpointer must be stable.
• ToolMessage validation errors: a ToolMessage needs tool_call_id matching the call it answers. If you build tool results manually and the IDs don’t line up, validation fails. Or use the prebuilt ToolNode and skip manual plumbing.
• ImportError on interrupt or Command: they live in langgraph.types, not langgraph.graph. Update the import instead of downgrading.

Best Practices

• Set max_attempts: it prevents infinite loops where the agent keeps calling tools without converging. Treat it as a budget, not a bugfix.
• Use MemorySaver for development and SqliteSaver or PostgresSaver in production.
• Define clear system prompts: the agent needs explicit instructions about when to use tools vs. answer directly.
• Stream for real-time UX: app.stream() shows each step as it happens; LLM streaming responses covers the token-level variant.
• Put interrupt early in the node: resume re-executes the node from the start, so anything before the interrupt runs twice.
• Name nodes semantically: agent, tools, writer show up in every streamed event and checkpoint, so debugging goes faster when node names mean something.

Common Mistakes

• Forgetting the add_messages annotation: without it, each node update replaces the message list instead of appending, and conversation history vanishes mid-run.
• Letting tool exceptions escape: if a tool raises, the agent gets stuck or retries blindly. Catch errors inside the tool node and return them as ToolMessage so the model sees the failure.
• No max_attempts limit: the agent can loop indefinitely calling tools without producing a final answer.
• Reusing one thread_id across users: state leaks between sessions. One thread ID per conversation — always.
• Importing interrupt from langgraph.graph: it lives in langgraph.types in current releases; the old path raises ImportError.

See Also

• LangGraph documentation
• LangGraph on GitHub
• Complete Guide to LangChain in Production
• Human-in-the-Loop Pattern