Build a LangGraph Agent

Start here — scroll inside each diagram frame to explore, then use + / − to zoom up to 200% if needed.

Do these in order. Each step says where to work, what to install or edit, and when.

1. Step 1Check Python

   Where · Your Mac Terminal — any folder.

   Need Python 3.10+. Prefer 3.11 or 3.12.

   TerminalCopy

   1python3 --version

2. Step 2Create the folder and venv

   Where · Documents is fine. This is a new folder, not inside another project.

   venv keeps LangGraph off your system Python. Activate it before every later command in this module.

   TerminalCopy

   1mkdir -p ~/Documents/support_graph && cd ~/Documents/support_graph && python3 -m venv .venv && source .venv/bin/activate

3. Step 3Install packages

   Where · MUST be inside support_graph/ with the venv activated (your prompt should show .venv).

   LangGraph runtime, a SQLite checkpointer so resume works in a new terminal, OpenAI chat model, dotenv for the key.

   TerminalCopy

   1pip install -U langgraph langgraph-checkpoint-sqlite langchain-openai langchain-core python-dotenv

4. Step 4Add your OpenAI API key

   Where · File: support_graph/.env — never commit this file.

   No quotes. Save. The script loads it. You do not paste the key into Python.

   .envCopy

   1OPENAI_API_KEY=sk-your-real-key-here

5. Step 5Write the graph — every module in one file

   Where · File: support_graph/graph.py — create it and paste the whole file.

   Read the comments. They map to the modules you just finished: state, nodes, edges, routing, tools, checkpoints, interrupt, subgraph, stream, get_state.

   graph.pyCopy

   1import argparse2import os3import sqlite34from typing import Annotated, Literal, TypedDict5
   6from dotenv import load_dotenv7from langchain_core.messages import AnyMessage, HumanMessage8from langchain_core.tools import tool9from langchain_openai import ChatOpenAI10from langgraph.checkpoint.sqlite import SqliteSaver11from langgraph.graph import END, START, StateGraph12from langgraph.graph.message import add_messages13from langgraph.prebuilt import ToolNode14from langgraph.types import Command, interrupt15
   16load_dotenv()17
   18THREAD = {"configurable": {"thread_id": "ticket-1"}}19DB = "checkpoints.db"20
   21
   22class AgentState(TypedDict):23    messages: Annotated[list[AnyMessage], add_messages]24    ticket_type: str25    refund_approved: bool26
   27
   28@tool29def lookup_order(order_id: str) -> str:30    """Look up a fake order. Read-only."""31    return f"Order {order_id}: status=delivered, total=480, item=headphones"32
   33
   34@tool35def create_refund(order_id: str, amount: int) -> str:36    """Create a fake refund. Write — only call after a human approved."""37    return f"Refunded {amount} for order {order_id}"38
   39
   40tools = [lookup_order, create_refund]41model = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)42
   43
   44def classify(state: AgentState) -> dict:45    text = state["messages"][-1].content.lower()46    ticket_type = "billing" if "refund" in text or "charge" in text else "tech"47    return {"ticket_type": ticket_type}48
   49
   50def assistant(state: AgentState) -> dict:51    sys = (52        f"You are support. ticket_type={state.get('ticket_type')}. "53        "Use lookup_order before you promise facts. "54        "If the user wants a refund, call create_refund only after approval is mentioned in tools."55    )56    reply = model.invoke([{"role": "system", "content": sys}, *state["messages"]])57    return {"messages": [reply]}58
   59
   60def refund_gate(state: AgentState) -> dict:61    decision = interrupt(62        {63            "question": "Approve this refund?",64            "ticket_type": state.get("ticket_type"),65        }66    )67    return {"refund_approved": bool(decision)}68
   69
   70def reply(state: AgentState) -> dict:71    status = "approved" if state.get("refund_approved") else "closed"72    return {"messages": [HumanMessage(content=f"(desk) Ticket {status}.")]}73
   74
   75billing = StateGraph(AgentState)76billing.add_node("pay", ToolNode([create_refund]))77billing.add_edge(START, "pay")78billing.add_edge("pay", END)79billing_graph = billing.compile()80
   81
   82def route_after_assistant(state: AgentState) -> Literal["tools", "refund_gate", "reply"]:83    last = state["messages"][-1]84    if getattr(last, "tool_calls", None):85        names = [c["name"] for c in last.tool_calls]86        if "create_refund" in names:87            return "refund_gate"88        return "tools"89    return "reply"90
   91
   92def build():93    g = StateGraph(AgentState)94    g.add_node("classify", classify)95    g.add_node("assistant", assistant)96    g.add_node("tools", ToolNode([lookup_order]))97    g.add_node("refund_gate", refund_gate)98    g.add_node("billing", billing_graph)99    g.add_node("reply", reply)100    g.add_edge(START, "classify")101    g.add_edge("classify", "assistant")102    g.add_conditional_edges(103        "assistant",104        route_after_assistant,105        {"tools": "tools", "refund_gate": "refund_gate", "reply": "reply"},106    )107    g.add_edge("tools", "assistant")108    g.add_edge("refund_gate", "billing")109    g.add_edge("billing", "reply")110    g.add_edge("reply", END)111    conn = sqlite3.connect(DB, check_same_thread=False)112    return g.compile(checkpointer=SqliteSaver(conn))113
   114
   115def run_new(graph):116    try:117        result = graph.invoke(118            {"messages": [HumanMessage(content="Refund order 4411, it is damaged")]},119            THREAD,120        )121        print(result)122        if isinstance(result, dict) and result.get("__interrupt__"):123            print("Paused for HITL. Next: python graph.py --resume")124    except Exception as exc:125        print("Paused for HITL (this is expected):", exc)126        print("Next: python graph.py --resume")127
   128
   129def run_resume(graph):130    result = graph.invoke(Command(resume=True), THREAD)131    print(result)132
   133
   134def run_stream(graph):135    for event in graph.stream(136        {"messages": [HumanMessage(content="Where is order 4411?")]},137        {"configurable": {"thread_id": "ticket-2"}},138        stream_mode="updates",139    ):140        print(event)141
   142
   143def run_inspect(graph):144    snap = graph.get_state(THREAD)145    print(snap.values)146    print("--- history ---")147    for frame in graph.get_state_history(THREAD):148        print(frame.config, frame.next)149
   150
   151if __name__ == "__main__":152    if not os.getenv("OPENAI_API_KEY"):153        raise SystemExit("Set OPENAI_API_KEY in .env")154    p = argparse.ArgumentParser()155    p.add_argument("--resume", action="store_true")156    p.add_argument("--stream", action="store_true")157    p.add_argument("--inspect", action="store_true")158    args = p.parse_args()159    app = build()160    if args.resume:161        run_resume(app)162    elif args.stream:163        run_stream(app)164    elif args.inspect:165        run_inspect(app)166    else:167        run_new(app)

6. Step 6Run until the interrupt

   Where · Terminal, inside support_graph/, venv still activated.

   First run should pause on refund_gate. That pause is success — you may see an interrupt payload or a GraphInterrupt message. Do not delete checkpoints.db. Then run --resume. If it prints a full reply and exits without pausing, the gate was skipped.

   TerminalCopy

   1cd ~/Documents/support_graph && source .venv/bin/activate && python graph.py

7. Step 7Resume the same thread

   Where · Same Terminal, same folder, same venv. Do not change thread_id.

   This is HITL + checkpoints. Command(resume=True) unblocks interrupt() using checkpoints.db. You should see create_refund run inside the billing subgraph, then the desk reply. If it starts a brand new ticket, you are in the wrong folder.

   TerminalCopy

   1python graph.py --resume

8. Step 8Stream a different ticket

   Where · Same folder. This uses ticket-2 so you do not collide with the paused/resumed thread.

   Streaming module: each printed dict is a node update. You should see classify, then assistant, then maybe tools.

   TerminalCopy

   1python graph.py --stream

9. Step 9Inspect checkpoints (time travel)

   Where · Same folder. Looks at ticket-1, the thread you refunded.

   get_state is now. get_state_history is the CCTV. If history is empty, checkpoints.db was not written — you ran from a different folder or skipped the first invoke.

   TerminalCopy

   1python graph.py --inspect

• ✕Running python graph.py without activating the venv
• ✕Starting a new thread on resume by editing THREAD
• ✕Skipping the first run and only using --resume
• ✕Committing .env
• ✕Deleting checkpoints.db and then running --resume — there is nothing to resume