LangGraph in Python: Building Stateful AI Agents

LangChain agents are powerful. But they have a fundamental limitation — they run in a straight line.

Call a tool. Get a result. Call another tool. Get another result. Respond.

That works for simple tasks. But real-world AI workflows are not straight lines — they branch, loop, pause for human approval, retry on failure, and route to different logic depending on what happened three steps ago.

LangGraph fixes this. Instead of a chain, you build a graph — nodes are functions, edges are transitions, and state flows through the whole thing. You get conditional branching, cycles, persistent checkpoints, and full control over every step.

If you built the LangChain agent in Article #1 and the FastAPI backend in Article #5, this is the natural next level — LangGraph is what you reach for when LangChain agents are not enough.

This guide covers everything from scratch: state design, graph construction, conditional routing, human-in-the-loop checkpoints, streaming, and deploying inside your FastAPI backend.

LangGraph vs LangChain Agents — The Real Difference

Before writing code, understand the architectural shift.

LangChain AgentLinear

Input

LLM decides

Tool call

Tool result

LLM responds

Output

Works great for simple tasks

vs

LangGraph AgentStateful graph

Input

Router node

Tool ATool BHuman Review

loops back to Router

Final node

Output

Built for complex, real-world workflows

The core differences:

Use LangChain when your agent runs tools and responds. Use LangGraph when your agent needs to make decisions, branch, loop, or pause.

Prerequisites

• Python 3.11+
• Completed Article #5 — the FastAPI backend setup
• OpenAI API key
• Basic understanding of LangChain

pip install langgraph langchain-openai langchain-core \
            python-dotenv pydantic fastapi uvicorn

Core Concepts Before Writing Code

LangGraph has four building blocks. Understand these and everything else follows.

• State — a typed Python dataclass that flows through every node. Every node reads from it and writes to it. Think of it as your agent's working memory.
• Nodes — Python functions that take the state, do something (call an LLM, execute a tool, make a decision), and return an updated state.
• Edges — connections between nodes. Can be unconditional (always go from A to B) or conditional (go to B or C depending on the state).
• Checkpoints — snapshots of state at any point in the graph. Used for persistence, human-in-the-loop pauses, and error recovery.

State

Typed dataclass — flows through every node

class AgentState(TypedDict): messages: list

Nodes

Python functions — read state, return updated state

def call_llm(state): -> new_state

Edges

Connections between nodes — conditional or fixed

graph.add_conditional_edges(...)

Checkpoints

State snapshots — persistence + human review

interrupt_before=["human_review"]

Project Setup

langgraph-agent/
├── app/
│   ├── main.py           # FastAPI entry point
│   ├── graph.py          # LangGraph graph definition
│   ├── state.py          # State type definitions
│   ├── nodes.py          # Node functions
│   └── tools.py          # Tool definitions
├── .env
└── requirements.txt

Step 1 — Define Your State

State is the most important design decision in LangGraph. Get this right before writing any nodes.

Create app/state.py:

from typing import TypedDict, Annotated, Literal
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    """
    The state object that flows through every node in the graph.

    Every field here is available to every node — think of it
    as your agent's shared working memory.
    """

    # Message history — add_messages is a reducer that appends
    # new messages rather than replacing the whole list
    messages: Annotated[list[BaseMessage], add_messages]

    # Current task the agent is working on
    task: str

    # How many times we have retried after an error
    retry_count: int

    # Whether the current output needs human review
    needs_review: bool

    # The final answer (set when the agent is done)
    final_answer: str | None

    # Which tools have been called in this run
    tools_used: list[str]

    # Error message if something went wrong
    error: str | None

Why Annotated[list[BaseMessage], add_messages]? By default, LangGraph replaces state values on each update. add_messages is a reducer — it tells LangGraph to append new messages to the list instead of replacing it. This is how you build up a conversation history without manually managing it.

Step 2 — Define Your Tools

# app/tools.py
from langchain_core.tools import tool
from datetime import datetime
import json


@tool
def calculate(expression: str) -> str:
    """
    Evaluates a mathematical expression.
    Use for arithmetic, percentages, and unit conversions.

    Args:
        expression: Valid math expression like '(1500 * 0.15) + 200'
    """
    try:
        allowed = set('0123456789+-*/()., ')
        if not all(c in allowed for c in expression):
            return f"Error: invalid characters in '{expression}'"
        result = eval(expression, {"__builtins__": {}})
        return f"{expression} = {result}"
    except Exception as e:
        return f"Calculation error: {str(e)}"


@tool
def search_knowledge_base(query: str) -> str:
    """
    Searches the knowledge base for relevant information.
    Use when you need specific facts or documentation.

    Args:
        query: The search query
    """
    # In production: connect to Pinecone from Article #2
    return json.dumps({
        "query": query,
        "results": [f"Mock result for: {query}"],
        "note": "Connect to your Pinecone RAG pipeline here"
    })


@tool
def get_current_date() -> str:
    """Returns today's date. Use when date context is needed."""
    return datetime.now().strftime("%A, %B %d, %Y")


tools = [calculate, search_knowledge_base, get_current_date]
tool_map = {tool.name: tool for tool in tools}

Step 3 — Build Your Nodes

Nodes are where your agent logic lives. Each node is a Python function that receives the full state and returns a partial state update.

Create app/nodes.py:

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, AIMessage, ToolMessage
from app.state import AgentState
from app.tools import tools, tool_map
from app.config import get_settings
import json


def get_llm():
    settings = get_settings()
    return ChatOpenAI(
        model=settings.model_name,
        temperature=0,
        streaming=True,
    ).bind_tools(tools)


# ── Node 1: Call the LLM ─────────────────────────────────────────

def call_llm(state: AgentState) -> dict:
    """
    Sends the current state to the LLM and gets a response.
    The LLM can respond with text or request tool calls.
    """
    llm = get_llm()

    system = SystemMessage(content=f"""You are a helpful AI assistant.

Current task: {state['task']}
Tools available: calculate, search_knowledge_base, get_current_date

Think step by step. Use tools when you need real data.
When you have a complete answer, respond with your final answer clearly.""")

    messages = [system] + state["messages"]
    response = llm.invoke(messages)

    return {
        "messages": [response],
        # Track if this response needs review (e.g. high-stakes output)
        "needs_review": _should_review(response.content),
    }


def _should_review(content: str) -> bool:
    """
    Determines if the output needs human review.
    Customize this logic for your use case.
    """
    review_triggers = [
        "delete", "remove", "irreversible",
        "payment", "send email", "publish"
    ]
    return any(trigger in content.lower() for trigger in review_triggers)


# ── Node 2: Execute Tool Calls ───────────────────────────────────

def execute_tools(state: AgentState) -> dict:
    """
    Executes all tool calls requested by the LLM in the last message.
    Returns tool results as ToolMessages added to state.
    """
    last_message = state["messages"][-1]
    tool_results = []
    tools_used = state.get("tools_used", [])

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

        print(f"[Node: execute_tools] Calling {tool_name}({tool_args})")

        if tool_name in tool_map:
            try:
                result = tool_map[tool_name].invoke(tool_args)
                tools_used.append(tool_name)
            except Exception as e:
                result = f"Error executing {tool_name}: {str(e)}"
        else:
            result = f"Unknown tool: {tool_name}"

        tool_results.append(
            ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"],
                name=tool_name,
            )
        )

    return {
        "messages": tool_results,
        "tools_used": tools_used,
    }


# ── Node 3: Human Review Checkpoint ─────────────────────────────

def human_review(state: AgentState) -> dict:
    """
    Pause point for human review.
    In production: send notification, wait for approval via API.
    Graph execution pauses here until resumed externally.
    """
    print(f"[Node: human_review] Output flagged for review")
    print(f"[Node: human_review] Last message: {state['messages'][-1].content[:200]}")
    print(f"[Node: human_review] Waiting for human approval...")

    # In production: send Slack/email notification here
    # The graph is checkpointed — it will resume when you call
    # graph.invoke(None, config=thread_config) externally

    return {"needs_review": False}  # Reset flag after review


# ── Node 4: Handle Errors ────────────────────────────────────────

def handle_error(state: AgentState) -> dict:
    """
    Error recovery node — decides whether to retry or give up.
    """
    retry_count = state.get("retry_count", 0) + 1
    print(f"[Node: handle_error] Retry attempt {retry_count}")

    return {
        "retry_count": retry_count,
        "error": None,  # Clear the error to allow retry
    }


# ── Node 5: Finalize Response ────────────────────────────────────

def finalize(state: AgentState) -> dict:
    """
    Extracts the final answer from the last AI message.
    """
    last_message = state["messages"][-1]
    return {
        "final_answer": last_message.content,
        "error": None,
    }

Step 4 — Build the Graph

This is where LangGraph comes alive — connecting nodes with conditional edges.

START

call_llmroutes every step

has tool calls

execute_tools

loop back

needs review

human_review

approved → call_llmrejected → END

error

handle_error

retry < 3 → call_llmmax retries → END

done

finalize

→ END

END

Create app/graph.py:

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage
from app.state import AgentState
from app.nodes import (
    call_llm,
    execute_tools,
    human_review,
    handle_error,
    finalize,
)
from typing import Literal


# ── Router functions (conditional edge logic) ────────────────────

def route_after_llm(
    state: AgentState,
) -> Literal["execute_tools", "human_review", "finalize", "handle_error"]:
    """
    Decides what to do after the LLM responds.
    This is the brain of your routing logic.
    """
    last_message = state["messages"][-1]

    # Error state — route to error handler
    if state.get("error"):
        return "handle_error"

    # LLM requested tool calls — execute them
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "execute_tools"

    # Output flagged for review — pause for human
    if state.get("needs_review"):
        return "human_review"

    # No tool calls, no review needed — we are done
    return "finalize"


def route_after_error(
    state: AgentState,
) -> Literal["call_llm", END]:
    """
    After an error: retry if under limit, otherwise give up.
    """
    if state.get("retry_count", 0) < 3:
        return "call_llm"
    return END


def route_after_review(
    state: AgentState,
) -> Literal["call_llm", END]:
    """
    After human review: continue if approved, stop if rejected.
    In production: read an 'approved' flag set by your review API.
    """
    # Check if the reviewer approved or rejected
    # For this example, we always continue after review
    approved = True  # Replace with: state.get("review_approved", False)

    if approved:
        return "call_llm"
    return END


# ── Build the graph ──────────────────────────────────────────────

def build_graph():
    """
    Assembles the complete agent graph.
    Called once at startup and reused across requests.
    """
    # Initialize the graph with our state type
    graph = StateGraph(AgentState)

    # Add all nodes
    graph.add_node("call_llm", call_llm)
    graph.add_node("execute_tools", execute_tools)
    graph.add_node("human_review", human_review)
    graph.add_node("handle_error", handle_error)
    graph.add_node("finalize", finalize)

    # Add edges — START to first node
    graph.add_edge(START, "call_llm")

    # Conditional routing after LLM responds
    graph.add_conditional_edges(
        "call_llm",
        route_after_llm,
        {
            "execute_tools": "execute_tools",
            "human_review": "human_review",
            "finalize": "finalize",
            "handle_error": "handle_error",
        }
    )

    # After tool execution — always loop back to LLM
    graph.add_edge("execute_tools", "call_llm")

    # After error handling — retry or stop
    graph.add_conditional_edges(
        "handle_error",
        route_after_error,
        {"call_llm": "call_llm", END: END}
    )

    # After human review — continue or stop
    graph.add_conditional_edges(
        "human_review",
        route_after_review,
        {"call_llm": "call_llm", END: END}
    )

    # Finalize always goes to END
    graph.add_edge("finalize", END)

    # MemorySaver enables checkpointing — required for human-in-the-loop
    # In production, replace with SqliteSaver or PostgresSaver
    checkpointer = MemorySaver()

    return graph.compile(
        checkpointer=checkpointer,
        # Pause execution BEFORE the human_review node
        # Graph will wait until you call .invoke() again
        interrupt_before=["human_review"],
    )


# Singleton — build once at module load
agent_graph = build_graph()

Step 5 — Running the Graph

# app/run.py
from app.graph import agent_graph
from app.state import AgentState
from langchain_core.messages import HumanMessage
import uuid


async def run_agent(task: str, thread_id: str | None = None) -> dict:
    """
    Runs the agent graph for a given task.
    thread_id enables conversation persistence across calls.
    """
    # Each thread_id is an isolated conversation
    # Same thread_id = same conversation history loaded from checkpoint
    thread_id = thread_id or str(uuid.uuid4())

    config = {
        "configurable": {
            "thread_id": thread_id,
        }
    }

    # Initial state
    initial_state: AgentState = {
        "messages": [HumanMessage(content=task)],
        "task": task,
        "retry_count": 0,
        "needs_review": False,
        "final_answer": None,
        "tools_used": [],
        "error": None,
    }

    result = await agent_graph.ainvoke(initial_state, config=config)

    return {
        "answer": result.get("final_answer"),
        "tools_used": result.get("tools_used", []),
        "thread_id": thread_id,
    }

Run it:

python -m app.run

Step 6 — Streaming Graph Execution

One of LangGraph's best features — you can stream individual node outputs as the graph runs, giving users real-time visibility into what the agent is doing.

# app/streaming.py
from app.graph import agent_graph
from app.state import AgentState
from langchain_core.messages import HumanMessage
from typing import AsyncIterator
import json


async def stream_agent(
    task: str,
    thread_id: str
) -> AsyncIterator[str]:
    """
    Streams graph execution events as Server-Sent Events.
    Your FastAPI endpoint yields these to the browser.
    """
    config = {"configurable": {"thread_id": thread_id}}

    initial_state: AgentState = {
        "messages": [HumanMessage(content=task)],
        "task": task,
        "retry_count": 0,
        "needs_review": False,
        "final_answer": None,
        "tools_used": [],
        "error": None,
    }

    # stream_mode="updates" yields the state delta after each node runs
    async for event in agent_graph.astream(
        initial_state,
        config=config,
        stream_mode="updates"
    ):
        for node_name, node_output in event.items():
            # Yield which node just ran
            yield json.dumps({
                "type": "node_complete",
                "node": node_name,
                "data": _serialize_output(node_output)
            })

    yield json.dumps({"type": "done"})


def _serialize_output(output: dict) -> dict:
    """
    Converts LangChain message objects to JSON-serializable dicts.
    """
    serialized = {}
    for key, value in output.items():
        if key == "messages":
            serialized[key] = [
                {
                    "type": type(msg).__name__,
                    "content": msg.content if hasattr(msg, 'content') else str(msg)
                }
                for msg in (value if isinstance(value, list) else [value])
            ]
        else:
            serialized[key] = value
    return serialized

Step 7 — Integrating with FastAPI

Wire the graph into your FastAPI backend from Article #5:

# app/main.py — add to your existing FastAPI app
from fastapi import FastAPI, HTTPException
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from app.run import run_agent
from app.streaming import stream_agent
import uuid


class AgentTaskRequest(BaseModel):
    task: str
    thread_id: str | None = None


class HumanReviewDecision(BaseModel):
    thread_id: str
    approved: bool
    feedback: str | None = None


# Standard agent endpoint
@app.post("/api/langgraph/agent")
async def run_langgraph_agent(request: AgentTaskRequest):
    try:
        thread_id = request.thread_id or str(uuid.uuid4())
        result = await run_agent(request.task, thread_id)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# Streaming agent endpoint
@app.post("/api/langgraph/agent/stream")
async def stream_langgraph_agent(request: AgentTaskRequest):
    thread_id = request.thread_id or str(uuid.uuid4())

    async def event_generator():
        async for chunk in stream_agent(request.task, thread_id):
            yield {"data": chunk, "event": "update"}

    return EventSourceResponse(event_generator())


# Human review decision endpoint
# Call this from your review UI to resume a paused graph
@app.post("/api/langgraph/review")
async def submit_review(decision: HumanReviewDecision):
    """
    Resume a paused graph after human review.
    The graph was paused at the human_review node —
    calling this resumes it from that checkpoint.
    """
    from app.graph import agent_graph

    config = {"configurable": {"thread_id": decision.thread_id}}

    # Update state with review decision, then resume by passing
    # None as input — it picks up from the checkpoint
    result = await agent_graph.ainvoke(
        None,
        config=config,
    )

    return {
        "resumed": True,
        "answer": result.get("final_answer"),
        "thread_id": decision.thread_id,
    }

Human-in-the-Loop: How the Pause Works

This is the most powerful LangGraph feature and the hardest to find good documentation on.

When you set interrupt_before=["human_review"] in the graph compilation:

• Graph runs normally until it is about to enter human_review.
• Graph saves its full state as a checkpoint and stops.
• Your API returns control to the caller.
• Later — minutes, hours, or days — you call graph.ainvoke(None, config) with the same thread_id.
• Graph loads the checkpoint and continues from exactly where it stopped.

This means a human can review AI output asynchronously — through a Slack message, an admin dashboard, or an email link — and the agent resumes when they decide.

# Example: Check if a thread is paused
async def get_graph_status(thread_id: str) -> dict:
    config = {"configurable": {"thread_id": thread_id}}

    # Get the current state snapshot
    snapshot = await agent_graph.aget_state(config)

    return {
        "thread_id": thread_id,
        "next_nodes": snapshot.next,  # Empty list if done
        "is_paused": len(snapshot.next) > 0,
        "paused_at": snapshot.next[0] if snapshot.next else None,
    }

Production: Replacing MemorySaver with Postgres

MemorySaver stores checkpoints in memory — they vanish when your server restarts. For production, use PostgresSaver:

pip install langgraph-checkpoint-postgres psycopg

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import os

async def build_production_graph():
    DB_URI = os.getenv("DATABASE_URL")

    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()  # Creates checkpoint tables

        graph = StateGraph(AgentState)
        # ... add nodes and edges ...

        return graph.compile(
            checkpointer=checkpointer,
            interrupt_before=["human_review"],
        )

With Postgres checkpointing, every graph execution is persisted. Users can pause a task, close their browser, come back the next day, and the agent resumes exactly where it left off — using the same thread_id.

LangGraph in Python vs Node.js — Key Differences

If you explored LangGraph in Node.js, here is what changes in Python:

Python is significantly ahead on LangGraph — more examples, better tooling, deeper documentation. If you are serious about LangGraph in production, Python is the right choice today.

What to Build Next

You now have a production-ready LangGraph agent — stateful, branching, streaming, with human-in-the-loop review and Postgres persistence. Natural next steps:

• Add RAG to the graph — create a retrieval node that queries your Pinecone index from Article #2 before the LLM node runs.
• Build a multi-agent graph — add a supervisor node that routes tasks to specialized sub-agents (research agent, writing agent, code agent).
• Connect to the Next.js frontend — stream the node execution events to the browser so users see "Searching knowledge base... Calculating... Done" in real time.
• Advanced RAG techniques — improve your retrieval quality with reranking and HyDE.

The full source code is on GitHub.

Frequently asked questions

Where to go next

The graph pattern in this guide plugs directly into the rest of the series:

• How to Build an AI Agent with LangChain and OpenAIThe linear agent this graph replaces — compare the chain to the stateful graph side by side.→
• Building an AI Agent Backend with Python, FastAPI, and LangChainThe FastAPI backend this graph slots directly into.→
• Building a RAG Application with Next.js, OpenAI, and PineconeGive call_llm a retrieval node backed by this exact Pinecone pipeline.→
• How to Build an AI SaaS Product with Next.js, Stripe, and OpenAIPut this graph behind auth, billing, and usage limits as a real product.→

A chain can only run forward. The moment your agent needs to branch, retry, loop, or wait on a human, you are not writing a bigger chain — you need a graph. Get the state right, keep every routing decision explicit, and checkpoint everything: that discipline scales to any workflow you can sketch on a whiteboard.