Lesson 05 LangGraph Intro
Theme:

Lab 05 — First graph with LangGraph

Lesson 05 · Building Your Own Agents · ~30 min hands-on

Goal: at the end of the lab you will have (a) verified langgraph is installed at the pinned version, (b) built a 3-node hello-world graph that runs successfully, (c) added a conditional edge that branches on state and loops, (d) attached a SqliteSaver checkpointer and simulated a crash-and-resume, and (e) inspected the state history to see LangGraph’s time-travel surface. No LLM calls in this lab — just the graph primitives in your hands. L06 will wire all of this into a real multi-stage agent.

Prereqs

Step 1 — Confirm langgraph is installed

python -c "import langgraph; print(langgraph.__version__)"

Expected: a version string in the 0.2.x range. If you get ModuleNotFoundError, re-run pip install -r agents/requirements.txt and confirm the version against agents/SDK-VERSIONS.md.

✓ Self-check: a version like 0.2.43 prints. No traceback.

Step 2 — The 20-line hello world

Create hello_graph.py anywhere convenient (the repo root works fine):

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    n: int
    log: list[str]

def add_one(s: S) -> dict:
    return {"n": s["n"] + 1, "log": s["log"] + ["add_one"]}

def double(s: S) -> dict:
    return {"n": s["n"] * 2, "log": s["log"] + ["double"]}

g = StateGraph(S)
g.add_node("add_one", add_one)
g.add_node("double", double)
g.add_edge(START, "add_one")
g.add_edge("add_one", "double")
g.add_edge("double", END)
app = g.compile()
print(app.invoke({"n": 3, "log": []}))

Run it:

python hello_graph.py

Expected: {'n': 8, 'log': ['add_one', 'double']} — 3 + 1 = 4, then 4 × 2 = 8; the log records the two node names in execution order.

✓ Self-check: the printed dict matches exactly. n is 8, log has two entries in the right order.

Step 3 — Inspect the graph

Add this at the bottom of the file:

print(app.get_graph().draw_mermaid())

Re-run. Expected: a Mermaid diagram description (text) showing START → add_one → double → END. Paste it into any Mermaid viewer (e.g., mermaid.live) for a picture. This is the inspectability win — your graph is a queryable artefact, not just runtime behaviour. A reviewer can see the shape without reading code.

Step 4 — Add a conditional edge (with a loop)

Replace the linear g.add_edge("add_one", "double") with a conditional edge that loops on add_one until n > 10, then jumps to double:

def route(s: S) -> str:
    return "double" if s["n"] > 10 else "add_one"

g.add_conditional_edges(
    "add_one",
    route,
    {"add_one": "add_one", "double": "double"},
)
g.add_edge("double", END)

Re-run with the same input {"n": 3, "log": []}. Expected: the graph loops through add_one until n > 10, then runs double once and ends. The final state should be roughly {'n': 22, 'log': ['add_one', 'add_one', ..., 'double']} — eight add_one entries (3→4→5→...→11) followed by one double (11 × 2 = 22).

✓ Self-check: log has 9 entries; the last is double; n == 22.

Step 5 — Add a SqliteSaver checkpointer

Add this import at the top of the file:

from langgraph.checkpoint.sqlite import SqliteSaver

Replace app = g.compile() with:

memory = SqliteSaver.from_conn_string("hello.db")
app = g.compile(checkpointer=memory)

Replace the print(app.invoke(...)) line with:

config = {"configurable": {"thread_id": "run-1"}}
print(app.invoke({"n": 3, "log": []}, config=config))

Run the script twice in a row. Expected: the first run prints the final state and creates hello.db in the working directory. The second run — with the same thread_id — returns immediately with the same state. The graph has nothing left to do because the END state is already persisted for that thread.

✓ Self-check: hello.db exists after run 1; run 2 prints the same dict and returns near-instantly.

Step 6 — Simulate a crash mid-graph

Modify double to raise on its first call:

calls = {"double": 0}

def double(s: S) -> dict:
    calls["double"] += 1
    if calls["double"] == 1:
        raise RuntimeError("simulated crash")
    return {"n": s["n"] * 2, "log": s["log"] + ["double"]}

Delete hello.db first (so you start from a clean checkpoint store):

rm hello.db

Now run the script once. Expected: it crashes inside double with RuntimeError: simulated crash. The add_one loop’s work was already checkpointed before the crash — you can prove it by running again with the same thread_id and an empty input:

print(app.invoke(None, config=config))

Expected: the graph resumes at double instead of starting over. In this same process, calls["double"] is now 2, so the second call succeeds and the graph completes. The log list contains the full history from the original add_one loop — that work was not re-run.

Process-local state caveat

The calls dict is in-process — when you restart Python, it resets to {"double": 0} and double raises again. To make the resume actually succeed across process restarts, change the bail-out from in-memory calls to a file flag: e.g., create /tmp/double_crashed on the first call, check for it on subsequent calls, and succeed if present. The lesson here is that checkpoints persist LangGraph state — your node functions’ own side-effects need their own persistence.

Step 7 — Inspect state history

Add this after the resume invoke:

for snap in app.get_state_history(config):
    print(snap.values, snap.next)

Expected: a list of snapshots from oldest to newest, each showing the merged state at that point and the next node the graph would run. This is the “time-travel” surface — you can pick any snapshot and call app.update_state(config, ..., as_of=snap.config) to fork a new branch from that point. The pattern is invaluable for debugging long agent runs and for building human-in-the-loop UIs where a reviewer rewinds a few steps and tries again.

What you should leave with

Troubleshooting

Common stumbles