LangGraph Tutorial: Build a Production-Ready AI Agent in Python

TL;DR

This tutorial builds a production-shaped LangGraph agent — not a toy chat loop. You will define typed state, register real tools, wire conditional edges, persist checkpoints with MemorySaver / PostgresSaver, pause for human approval with interrupt_before, and expose the graph behind FastAPI. If you already followed the OpenAI Agents SDK path on the create-an-agent guide, this page is the next level: durable state, resume-after-crash, and human-in-the-loop — the pieces you need before charging a card or writing to production systems.

Stack for this walkthrough: Python 3.11+, LangGraph 1.2+, langchain-openai, and a checkpointer. Budget one focused evening for the graph; another for Postgres persistence and the API wrapper.

Why LangGraph

Most "agent frameworks" hide the loop. LangGraph makes the loop explicit: nodes mutate state, edges decide what runs next, and a checkpointer freezes that state after every step. That model maps cleanly onto production requirements — retries, approvals, multi-day workflows — because the runtime already knows how to pause and resume.

Choose LangGraph when you need branching control flow, durable execution across process restarts, or a human gate before irreversible tools. Prefer a higher-level SDK when you want a fast demo on one model provider and do not yet need graph-level persistence.

In practice, production agents fail between steps: the search API times out, a reviewer is offline, a deploy restarts mid-run. A plain while-loop loses that context. A LangGraph thread keyed by thread_id keeps messages, custom fields, and the next node identity in a checkpointer so you can resume exactly where you stopped. That is the difference between a notebook demo and a system you can operate.

The OpenAI SDK tutorial teaches the reason → act → observe loop with minimal ceremony. This LangGraph tutorial assumes that mental model and focuses on what demos usually skip: checkpoints, interruptible tool execution, and a deployable stateful API.

Prerequisites & Setup

You need basic Python (type hints, virtualenvs), an OpenAI-compatible API key, and comfort with a terminal. Prior LangChain experience helps but is not required — LangGraph can use LangChain chat models and tools without adopting the full chain abstraction. If you have not built a tool-calling loop before, skim the how to create an AI agent guide first, then return here for durable graphs.

Create the project and install LangGraph 1.2+ style dependencies:

mkdir langgraph-prod-agent && cd langgraph-prod-agent
python3 -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install "langgraph>=1.2.0" langchain-openai langchain-core \
  langgraph-checkpoint-postgres python-dotenv requests fastapi uvicorn

Store secrets in .env (never commit this file):

OPENAI_API_KEY=sk-your-key-here
# Optional: LangSmith tracing
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_your_key_here
LANGSMITH_PROJECT=langgraph-prod-agent
# Production checkpointer
DATABASE_URL=postgresql://postgres:postgres@localhost:5432/langgraph

Suggested layout:

langgraph-prod-agent/
├── .env
├── agent/
│   ├── __init__.py
│   ├── state.py      # AgentState
│   ├── tools.py      # web_search, calculator, run_python
│   ├── nodes.py      # agent node
│   ├── graph.py      # StateGraph compile + checkpointer
│   └── run.py        # CLI entrypoint
└── api.py            # FastAPI wrapper

Load env vars at process start:

from dotenv import load_dotenv
import os

load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "Set OPENAI_API_KEY in .env"

Step 1: Define AgentState

LangGraph state is a typed dictionary. Messages use the add_messages reducer so each node can append without clobbering history. Extra fields (like a step counter) let you enforce production limits inside the graph instead of only in a while-loop.

# agent/state.py
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage


class AgentState(TypedDict):
    """Shared state that flows through every node."""

    messages: Annotated[list[AnyMessage], add_messages]
    # Optional production knobs — mutate from nodes as needed
    step_count: int

Why this matters: frameworks that only keep a chat list make it hard to checkpoint custom fields (approval flags, customer IDs, cart totals). Typed state is how LangGraph stays durable and inspectable.

Step 2: Register Real Tools

Tools are ordinary Python functions decorated with @tool. LangGraph's ToolNode will call them from model-emitted tool calls. Keep implementations defensive — return error strings instead of crashing the graph.

# agent/tools.py
import ast
import operator
import requests
from langchain_core.tools import tool

OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
}


def _eval(node):
    if isinstance(node, ast.Expression):
        return _eval(node.body)
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp):
        return OPS[type(node.op)](_eval(node.left), _eval(node.right))
    if isinstance(node, ast.UnaryOp):
        return OPS[type(node.op)](_eval(node.operand))
    raise ValueError("Unsupported expression")


@tool
def web_search(query: str) -> str:
    """Search the web via DuckDuckGo Instant Answer API (demo-friendly)."""
    try:
        resp = requests.get(
            "https://api.duckduckgo.com/",
            params={"q": query, "format": "json", "no_html": 1},
            timeout=20,
        )
        resp.raise_for_status()
        data = resp.json()
        abstract = data.get("AbstractText") or data.get("Answer") or ""
        related = [
            f"- {t.get('Text')}"
            for t in data.get("RelatedTopics", [])[:3]
            if isinstance(t, dict) and t.get("Text")
        ]
        return abstract or "\n".join(related) or "No results found."
    except Exception as exc:
        return f"ERROR: web_search failed: {exc}"


@tool
def calculator(expression: str) -> str:
    """Safely evaluate a basic arithmetic expression like '(12.5 * 4) + 3'."""
    try:
        tree = ast.parse(expression, mode="eval")
        return str(_eval(tree))
    except Exception as exc:
        return f"ERROR: calculator failed: {exc}"


@tool
def run_python(code: str) -> str:
    """Run a tiny Python snippet; only print/range/len/sum are available."""
    output: list[str] = []

    def _print(*args, **kwargs):
        output.append(" ".join(str(a) for a in args))

    safe_builtins = {"print": _print, "range": range, "len": len, "sum": sum}
    try:
        exec(code, {"__builtins__": safe_builtins}, {})
    except Exception as exc:
        return f"ERROR: run_python failed: {exc}"
    return "\n".join(output) if output else "(no output)"


TOOLS = [web_search, calculator, run_python]

Swap DuckDuckGo for a paid search API in real deployments, and move run_python into a container sandbox before exposing it publicly. The LangGraph wiring stays the same.

Step 3: Build the Agent Node

The agent node binds tools to a chat model and appends the model response to state. Increment step_count so you can fail closed when the agent starts thrashing.

# agent/nodes.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from agent.state import AgentState
from agent.tools import TOOLS

SYSTEM = """You are a production research agent.
Use tools when they improve accuracy. Prefer calculator for math and
run_python for formatting. When the goal is done, answer in plain text
with no further tool calls. If a tool returns ERROR, recover or explain."""

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(TOOLS)
MAX_STEPS = 8


def agent_node(state: AgentState) -> dict:
    messages = state["messages"]
    if not messages or messages[0].type != "system":
        messages = [SystemMessage(content=SYSTEM), *messages]

    steps = state.get("step_count", 0) + 1
    if steps > MAX_STEPS:
        return {
            "messages": [
                {
                    "role": "assistant",
                    "content": "Stopped: max steps reached without a final answer.",
                }
            ],
            "step_count": steps,
        }

    response = llm.invoke(messages)
    return {"messages": [response], "step_count": steps}

Step 4: Wire Graph Edges

Conditional edges turn the agent into a graph: after the agent node, route to tools if the model requested tool calls; otherwise end. After tools run, always return to the agent so it can observe results and continue. This is the same reason → act → observe loop as a hand-rolled SDK agent — except the routing decision is data in the graph, so checkpoints and interrupts can attach to named nodes instead of living only inside a Python for loop.

# agent/graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

from agent.state import AgentState
from agent.nodes import agent_node
from agent.tools import TOOLS


def build_graph(checkpointer=None, interrupt_before=None):
    graph = StateGraph(AgentState)
    graph.add_node("agent", agent_node)
    graph.add_node("tools", ToolNode(TOOLS))

    graph.add_edge(START, "agent")
    # tools_condition → "tools" if last message has tool_calls, else END
    graph.add_conditional_edges(
        "agent",
        tools_condition,  # equivalent to a should_continue router
        {"tools": "tools", END: END},
    )
    graph.add_edge("tools", "agent")

    return graph.compile(
        checkpointer=checkpointer or MemorySaver(),
        interrupt_before=interrupt_before,
    )


# Default compiled app for local use
app = build_graph()

Prefer an explicit should_continue when you want custom routing (for example, skip tools after MAX_STEPS):

from typing import Literal
from langgraph.graph import END


def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return END


# graph.add_conditional_edges("agent", should_continue)

Step 5: Checkpointing with MemorySaver & PostgresSaver

Checkpointing is the production differentiator. After each node, LangGraph serializes state under a thread_id. Restart the process, reuse the same thread, and the graph resumes instead of starting over. That is how you unlock human-in-the-loop, crash recovery, and time-travel debugging. Without checkpoints you only have an in-process conversation list — fine for a script, fragile for an API that must survive restarts and horizontal scale.

Use MemorySaver for local development. It is single-process and disappears on restart — perfect for tests, wrong for multi-worker production.

from langgraph.checkpoint.memory import MemorySaver
from agent.graph import build_graph

memory = MemorySaver()
app = build_graph(checkpointer=memory)

config = {"configurable": {"thread_id": "demo-user-1"}}
# Every invoke with this config reads/writes the same durable thread

For production, swap in PostgresSaver from langgraph-checkpoint-postgres. Call .setup() once to create tables:

import os
from langgraph.checkpoint.postgres import PostgresSaver
from agent.graph import build_graph

DB_URI = os.environ["DATABASE_URL"]

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # idempotent schema bootstrap
    app = build_graph(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "prod-user-42"}}
    result = app.invoke(
        {
            "messages": [{"role": "user", "content": "What is 17 * 19?"}],
            "step_count": 0,
        },
        config=config,
    )
    print(result["messages"][-1].content)

Keep thread_id stable per conversation or workflow run. Reusing it accumulates history; minting a new ID starts a fresh thread. In FastAPI, map authenticated user + session → thread_id.

Step 6: Human-in-the-Loop with interrupt_before / interrupt_after

Agents that can spend money or mutate data need a pause point. Compile with interrupt_before=["tools"] so the graph stops after the model proposes tool calls and before ToolNode executes them. Inspect state, approve or edit, then resume with invoke(None, config).

from langgraph.checkpoint.memory import MemorySaver
from agent.graph import build_graph

app = build_graph(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # pause before side effects
)

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

# 1) Run until the interrupt
app.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Search for LangGraph checkpointing docs, then summarize.",
            }
        ],
        "step_count": 0,
    },
    config=config,
)

# 2) Inspect pending tool calls
snapshot = app.get_state(config)
print("next nodes:", snapshot.next)  # ('tools',) when paused
last = snapshot.values["messages"][-1]
print("pending tool_calls:", getattr(last, "tool_calls", None))

# 3) Human approves → resume from the checkpoint (do NOT resend the user msg)
final = app.invoke(None, config=config)
print(final["messages"][-1].content)

# Optional: interrupt_after=["agent"] pauses after each model turn for audit

Rejecting a tool call is an update_state problem: replace or append a message that cancels the action, then resume or end the thread. The point of HITL in LangGraph is not a UI widget — it is a durable pause with a resumable checkpoint. Pair static interrupts (interrupt_before / interrupt_after) with policy: for example, auto-run calculator but always pause before web_search or any write tool. As graphs grow, you can also call the dynamic interrupt() helper inside a node or tool when only some paths need a human.

Step 7: Run the Agent

A thin CLI entrypoint is enough to validate tools, routing, and checkpoints before you wrap HTTP around them.

# agent/run.py
from agent.graph import build_graph
from langgraph.checkpoint.memory import MemorySaver

app = build_graph(checkpointer=MemorySaver())


def run(goal: str, thread_id: str = "cli") -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = app.invoke(
        {
            "messages": [{"role": "user", "content": goal}],
            "step_count": 0,
        },
        config=config,
    )
    return result["messages"][-1].content


if __name__ == "__main__":
    print(
        run(
            "Find a short fact about Tokyo's population, multiply it by 1.05 "
            "with the calculator, and format the result with commas using run_python."
        )
    )

Run with python -m agent.run. Stream intermediate events during development with app.stream(...) and print each (node, update) pair so you can see agent ↔ tools oscillation.

config = {"configurable": {"thread_id": "stream-1"}}
for event in app.stream(
    {
        "messages": [{"role": "user", "content": "What is 21 * 2?"}],
        "step_count": 0,
    },
    config=config,
    stream_mode="updates",
):
    print(event)

Step 8: Deploy with FastAPI

Production agents are usually services. Expose invoke and resume endpoints so a frontend (or ops tool) can start a thread, approve tool calls, and continue. Hold the Postgres checkpointer for the process lifetime — do not create a new saver per request.

# api.py
import os
from contextlib import asynccontextmanager
from typing import Optional

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from langgraph.checkpoint.postgres import PostgresSaver

from agent.graph import build_graph

load_dotenv()

checkpointer: PostgresSaver | None = None
graph = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global checkpointer, graph
    db = os.environ["DATABASE_URL"]
    # Keep the context open for the app lifetime
    with PostgresSaver.from_conn_string(db) as cp:
        cp.setup()
        checkpointer = cp
        graph = build_graph(
            checkpointer=cp,
            interrupt_before=["tools"],  # require approval for tools
        )
        yield


app = FastAPI(title="LangGraph Production Agent", lifespan=lifespan)


class InvokeRequest(BaseModel):
    goal: str
    thread_id: str = Field(..., min_length=1)


class ResumeRequest(BaseModel):
    thread_id: str
    approve: bool = True


class AgentResponse(BaseModel):
    thread_id: str
    status: str
    next: list[str] = []
    message: Optional[str] = None
    pending_tool_calls: Optional[list] = None


@app.post("/agent/invoke", response_model=AgentResponse)
def invoke_agent(payload: InvokeRequest):
    assert graph is not None
    config = {"configurable": {"thread_id": payload.thread_id}}
    graph.invoke(
        {
            "messages": [{"role": "user", "content": payload.goal}],
            "step_count": 0,
        },
        config=config,
    )
    snap = graph.get_state(config)
    last = snap.values["messages"][-1]
    pending = getattr(last, "tool_calls", None)
    if snap.next:
        return AgentResponse(
            thread_id=payload.thread_id,
            status="interrupted",
            next=list(snap.next),
            pending_tool_calls=pending,
        )
    return AgentResponse(
        thread_id=payload.thread_id,
        status="completed",
        message=getattr(last, "content", None),
    )


@app.post("/agent/resume", response_model=AgentResponse)
def resume_agent(payload: ResumeRequest):
    assert graph is not None
    config = {"configurable": {"thread_id": payload.thread_id}}
    snap = graph.get_state(config)
    if not snap.next:
        raise HTTPException(400, "Thread is not waiting on an interrupt")

    if not payload.approve:
        graph.update_state(
            config,
            {
                "messages": [
                    {
                        "role": "assistant",
                        "content": "Action cancelled by human reviewer.",
                    }
                ]
            },
        )
        return AgentResponse(
            thread_id=payload.thread_id,
            status="cancelled",
            message="Action cancelled by human reviewer.",
        )

    graph.invoke(None, config=config)
    snap = graph.get_state(config)
    last = snap.values["messages"][-1]
    if snap.next:
        return AgentResponse(
            thread_id=payload.thread_id,
            status="interrupted",
            next=list(snap.next),
            pending_tool_calls=getattr(last, "tool_calls", None),
        )
    return AgentResponse(
        thread_id=payload.thread_id,
        status="completed",
        message=getattr(last, "content", None),
    )


# uvicorn api:app --host 0.0.0.0 --port 8000

Smoke-test the interrupt / resume flow:

curl -s -X POST http://localhost:8000/agent/invoke \
  -H 'Content-Type: application/json' \
  -d '{"goal":"Search for LangGraph PostgresSaver","thread_id":"u1"}'

curl -s -X POST http://localhost:8000/agent/resume \
  -H 'Content-Type: application/json' \
  -d '{"thread_id":"u1","approve":true}'

Put authentication, rate limits, and network egress controls in front of this API before it faces the public internet. The graph is only as safe as the tools you allow it to call.

Debugging with LangSmith & Time-Travel

Set LANGSMITH_TRACING=true and a project name to capture every node, tool call, and token usage. When a production thread misbehaves, open the trace, then use checkpoint history to replay from an earlier snapshot — LangGraph's time-travel — instead of guessing which prompt change broke routing.

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

# After a run (or interrupt), walk history newest → oldest
history = list(app.get_state_history(config))
for i, snap in enumerate(history[:5]):
    print(i, snap.config["configurable"].get("checkpoint_id"), snap.next)

# Replay from an older checkpoint (time-travel)
if len(history) >= 2:
    past = history[1].config
    # Fork: continue from that checkpoint with optional state edits
    app.invoke(None, config=past)

Practical loop: reproduce with the same thread_id, inspect get_state / history, patch the system prompt or tool schema, then re-run a golden goal suite (must call calculator, must interrupt before tools, must finish under N steps). Treat traces as part of the product: if you cannot explain why the agent called a tool, you are not ready to widen permissions. LangSmith Engine can flag looping or empty-tool patterns; still keep your own assertions around step counts and interrupt status so CI catches regressions before a deploy.

Common Pitfalls

• No checkpointer + interrupt — interrupt_before requires a checkpointer. Without one, pause/resume cannot work.
• Resending the user message on resume — call invoke(None, config) (or Command(resume=...) for dynamic interrupts). Resending input starts a conflicting update.
• MemorySaver in multi-worker prod — in-memory state is not shared across Uvicorn workers or deploys. Use Postgres (or another durable saver).
• Forgetting add_messages — assigning a bare list overwrites history; the reducer appends correctly.
• Unbounded tool output — truncate or summarize before it blows the context window; checkpoint size grows with messages.
• Treating LangGraph like a chatbot SDK — if you only need a single tool loop with no durability, the OpenAI Agents SDK may be enough. Reach for LangGraph when state and control flow are the product.

FAQ

Next Steps

You now have a LangGraph agent with typed state, real tools, conditional edges, durable checkpoints, human-in-the-loop interrupts, and a FastAPI surface. From here: harden the Python sandbox, add auth in front of resume endpoints, and expand the golden eval set before you widen tool permissions.

Compare frameworks side by side, or revisit the architecture fundamentals if you want the reason → act → observe model without framework specifics.