Building Stateful Agents with LangGraph

The Complete ReAct Agent with LangGraph

This lesson builds a production-ready stateful agent from scratch. We’ll use LangGraph’s prebuilt components where they help and custom nodes where we need control.

Step 1: Define State and Tools

# src/agents/react_agent.py
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    # add_messages reducer appends new messages rather than replacing the list
    messages: Annotated[list[BaseMessage], add_messages]
    # simple int counter: no reducer, so each update overwrites the previous value
    iterations: int
    final_answer: str | None

# TavilySearchResults wraps the Tavily search API as a LangChain tool
search_tool = TavilySearchResults(max_results=5)
tools = [search_tool]

# bind_tools() attaches the JSON schema of each tool to the LLM so it can
# generate structured tool_call objects in its responses
llm = ChatOpenAI(model="gpt-5.6-sol", temperature=0)
llm_with_tools = llm.bind_tools(tools)

Step 2: Define Nodes

MAX_ITERATIONS = 15  # hard cap to prevent runaway agent loops

def call_llm(state: AgentState) -> dict:
    """Call the LLM with the current conversation state."""
    # Guard: prevent infinite loops by returning a graceful degradation message
    if state["iterations"] >= MAX_ITERATIONS:
        return {
            "messages": [AIMessage(content="I've reached the maximum number of steps. Here's what I've found so far: " + str(state["messages"][-1].content))],
            "final_answer": "max_iterations_reached"
        }
    
    # llm_with_tools sends the tool schemas alongside the messages in each API call
    response = llm_with_tools.invoke(state["messages"])
    
    return {
        "messages": [response],             # add_messages appends this to history
        "iterations": state["iterations"] + 1,  # increment the loop counter
    }

# ToolNode is prebuilt: it reads tool_calls from the last AIMessage, executes
# each tool function by name, and returns ToolMessage results automatically
tool_node = ToolNode(tools)

Step 3: Assemble and Compile the Graph

# Build the graph
builder = StateGraph(AgentState)

builder.add_node("llm", call_llm)
builder.add_node("tools", tool_node)

builder.set_entry_point("llm")

# tools_condition is prebuilt: it returns "tools" if the last AIMessage has
# tool_calls, END otherwise: no need to write your own routing function
builder.add_conditional_edges(
    "llm",
    tools_condition,   # returns "tools" or END
)
# after tools run, always loop back to the LLM with the tool results in context
builder.add_edge("tools", "llm")

# MemorySaver stores checkpoints in-process; swap for PostgresSaver in production
memory = MemorySaver()
agent = builder.compile(checkpointer=memory)

Step 4: Run the Agent

import asyncio

async def run(user_query: str, thread_id: str = "default") -> str:
    # thread_id scopes the checkpoint so each conversation is isolated
    config = {"configurable": {"thread_id": thread_id}}
    
    initial_state: AgentState = {
        "messages": [HumanMessage(content=user_query)],
        "iterations": 0,
        "final_answer": None,
    }
    
    # ainvoke runs the full graph asynchronously and returns when it reaches END
    result = await agent.ainvoke(initial_state, config=config)
    
    # Extract the last AIMessage: this is the agent's final response
    ai_messages = [m for m in result["messages"] if isinstance(m, AIMessage)]
    return ai_messages[-1].content

# Run it
async def main():
    answer = await run(
        "What are the latest developments in Model Context Protocol (MCP)?",
        thread_id="research-session-1"
    )
    print(answer)

asyncio.run(main())

Streaming Agent Execution

Streaming lets you show progress in real time: essential for agents that take 10-30 seconds:

async def run_streaming(user_query: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    
    initial_state: AgentState = {
        "messages": [HumanMessage(content=user_query)],
        "iterations": 0,
        "final_answer": None,
    }
    
    print(f"Starting agent: {user_query[:50]}...")
    
    # astream yields one event dict per node completion: {node_name: node_output}
    async for event in agent.astream(initial_state, config=config):
        # event is a dict mapping node_name → node_output
        for node_name, output in event.items():
            if node_name == "llm" and "messages" in output:
                msg = output["messages"][0]
                if msg.tool_calls:
                    # tool_calls is a list of {name, args, id} dicts
                    for tc in msg.tool_calls:
                        print(f"  🔧 Tool: {tc['name']}({tc['args']})")
                else:
                    print(f"  💬 Final answer: {msg.content[:100]}...")
            elif node_name == "tools" and "messages" in output:
                # ToolMessages contain the raw return value from each tool function
                for tm in output["messages"]:
                    print(f"  ✅ Tool result: {str(tm.content)[:80]}...")

Human-in-the-Loop with interrupt()

For agents that take consequential actions (sending emails, making purchases, modifying databases), require human approval:

from langgraph.types import interrupt, Command

def check_and_approve_email(state: AgentState) -> dict:
    """Node that drafts an email and waits for human approval."""
    last_message = state["messages"][-1]
    draft = extract_email_draft(last_message.content)
    
    # interrupt() pauses graph execution and serializes state to the checkpointer
    # the dict passed here is surfaced to the caller so the UI can show it to the user
    human_decision = interrupt({
        "action": "approve_email",
        "draft": draft,
        "prompt": "Do you approve sending this email? (yes/no/edit)"
    })
    
    # Code after interrupt() only runs when the graph is resumed via Command(resume=...)
    if human_decision["approved"]:
        send_email(draft)
        return {"messages": [AIMessage(content=f"Email sent to {draft['to']}")]}
    else:
        return {"messages": [AIMessage(content=f"Email cancelled. Reason: {human_decision.get('reason', 'User declined')}")]}

# Add to graph
builder.add_node("approve_email", check_and_approve_email)

# In your API handler (FastAPI example):
async def handle_agent_run(request: AgentRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    
    result = await agent.ainvoke(request.initial_state, config=config)
    
    # get_state() reads the persisted checkpoint: check if graph is waiting at interrupt
    state = agent.get_state(config)
    if state.next == ("approve_email",):
        # interrupts[0].value is the dict passed to interrupt() above
        interrupt_value = state.tasks[0].interrupts[0].value
        return {"status": "waiting_approval", "data": interrupt_value}
    
    return {"status": "complete", "result": result}

# Resume endpoint
async def handle_approval(thread_id: str, decision: dict):
    config = {"configurable": {"thread_id": thread_id}}
    
    # Command(resume=decision) injects the human's decision as the return value
    # of interrupt() and continues graph execution from where it paused
    result = await agent.ainvoke(
        Command(resume=decision),
        config=config
    )
    return {"status": "complete", "result": result}

Inspecting State at Any Point

# Get the current state of a run
config = {"configurable": {"thread_id": "my-thread"}}
current_state = agent.get_state(config)

# next is a tuple of node names the graph will execute next (empty if at END)
print("Current node:", current_state.next)
print("Message count:", len(current_state.values["messages"]))
print("Iterations:", current_state.values["iterations"])

# get_state_history yields a snapshot for every node that has completed
# ordered from most recent to oldest: useful for debugging wrong answers
for state_snapshot in agent.get_state_history(config):
    print(f"Step: {state_snapshot.metadata['step']}")
    print(f"Next: {state_snapshot.next}")
    print(f"Messages: {len(state_snapshot.values['messages'])}")

# Time travel: replay from an earlier state
# state_snapshot.config contains the checkpoint ID needed to resume from that point
target_config = state_snapshot.config  # config of a historical state
result = await agent.ainvoke(None, config=target_config)

State inspection is invaluable during debugging. When an agent produces a wrong answer, you can trace exactly which tool call returned bad data, which model decision was incorrect, and replay from that point with a different input.