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
-
The
agents/requirements.txtalready installed (which pinslanggraph>=0.2,<0.3). If not, runpip install -r agents/requirements.txtfrom the repo root. -
No
ANTHROPIC_API_KEYneeded for this lab — the hello-world graph doesn’t call an LLM. (L06’s lab will.) - One terminal where you can run
pythonscripts. - Working directory at the repo root (the paths below assume that).
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
-
A working
hello_graph.pythat demonstrates: nodes, a linear edge, a conditional edge with a loop, aSqliteSavercheckpoint, a simulated crash, a successful resume, and state-history inspection. - The intuition that LangGraph’s primitives (graph, nodes, edges, checkpointer) are independent and compose — you can adopt them incrementally.
- A mental picture of when LangGraph is the right tool: when your control flow has a shape that’s easier to draw than to express in if-statements, and when you need checkpoint-based resume or human-in-the-loop pauses.
- Readiness for L06, which uses all five primitives in a real LLM-powered pipeline agent (planner → fetcher → critique branch → writer).
Troubleshooting
Common stumbles
-
ModuleNotFoundError: No module named 'langgraph.checkpoint.sqlite'— you have older LangGraph (<0.2). Re-install peragents/SDK-VERSIONS.md. The SQLite saver moved sub-modules across the 0.1 → 0.2 boundary. -
Graph runs forever in Step 4 — your conditional edge’s
mapping is missing one of the strings the
routefunction returns. Print the return value ofrouteand confirm every possible string is a key in the mapping dict. -
Resume in Step 6 starts over from the beginning — you
didn’t pass the same
thread_id. Each thread is an independent run; theconfig={"configurable": {"thread_id": ...}}argument is load-bearing — it’s how the checkpointer knows which run to resume. -
sqlite3.OperationalError: database is locked— you have two processes holding the same.dbfile. SQLite is fine for single-process labs and CI; use Postgres for multi-process production. -
app.invoke(None, config=config)returns the previous final state instead of resuming — the graph already reached END for that thread. Deletehello.db(or use a freshthread_id) and re-trigger the crash before invoking withNone.