Index Ai Academy
Phase 10 · Agent harnesses · Scaling & shipping

Agents as graphs

State, reducers and routing, a checkpoint after every step, and pausing for a person's approval.

About 40 minutes · A free account opens the whole lesson

Create a free account Try part of it here

Already have an account? Sign in to open this lesson

Why it matters

The agent loop from earlier in this phase is a while loop: call the model, run its tools, repeat. That's the right start, and for many agents it's enough. But workflows grow: plan, research, draft, check the draft, maybe redo it, ask a person before sending, and pick up tomorrow where you left off. Written as nested ifs inside one loop, that control flow becomes hard to read, impossible to pause cleanly, and lost entirely if the process crashes. Frameworks such as LangGraph describe the workflow as a graph instead: named steps and the rules for moving between them, with the state saved after every step. This lesson builds a small graph runner from scratch, so the framework's ideas hold no mystery when you meet them.

What you'll be able to do

  • Describe an agent's workflow as nodes, edges and a shared state, and run it with a small graph runner
  • Combine node updates with reducers, and route between nodes with functions of the state
  • Save a checkpoint after every step, and resume a run after a crash without repeating finished work
  • Pause before a risky step for a person's approval, and resume with their edits

What it covers

  1. Nodes, edges and one shared state
  2. Reducers: how an update joins the state
  3. Routing a worker and an evaluator
  4. Route on the evaluator's verdict
  5. Checkpoints: save the state after every step
  6. Checkpoints in SQLite
  7. Pausing for a person
  8. The same ideas in a framework
  9. Weekend sidekick: a hike planner that asks before it texts
Quick check
An approval step waits for a manager, who might answer tomorrow. How should the agent wait?
  1. Keep a process running, waiting, until the answer comes
  2. Ask the model to wait patiently in its reply
  3. Save its state and stop; resume when the answer arrives
  4. Guess the manager's answer and carry on
Show the answer

Save its state and stop; resume when the answer arrives

A run that may pause for a day must survive restarts, so it saves a checkpoint of its state and ends. When the answer arrives, it loads the checkpoint and continues from that step.

Nodes, edges and one shared state

Three ideas make the graph:

  • The state is one dictionary that holds everything the workflow knows: the request, drafts, notes, counters.
  • A node is a function that reads the state and returns an update: a dictionary of just the keys it changes. Nodes don't call each other, so each one is small and testable on its own.
  • An edge says which node runs next. A fixed edge always goes to the same node; a router is a function of the state that picks the next node, such as "back to drafting if the check found problems". A special name, END, stops the run.

Here a two-node graph writes a reminder for a club walk and checks it. Real nodes would call a model; these are scripted so you can follow every step.

END = "END"

def run(nodes, edges, start, state, max_steps=10):
    node = start
    for step in range(max_steps):
        update = nodes[node](state)
        state = {**state, **update}
        nxt = edges[node]
        nxt = nxt(state) if callable(nxt) else nxt
        print(f"step {step}: {node:<6} set {sorted(update)} -> {nxt}")
        if nxt == END:
            return state
        node = nxt
    raise RuntimeError(f"stopped after {max_steps} steps")

def draft(state):
    tries = state.get("tries", 0) + 1
    text = "Ridge walk this Saturday, meet at the trailhead."
    if tries > 1:
        text += " Start: 8:00."
    return {"draft": text, "tries": tries}

def check(state):
    problems = [] if any(ch.isdigit() for ch in state["draft"]) else ["no start time"]
    return {"problems": problems}

def after_check(state):
    if not state["problems"]:
        return END
    return "draft" if state["tries"] < 3 else END

final = run({"draft": draft, "check": check}, {"draft": "check", "check": after_check}, "draft", {})
print(final["draft"], final["problems"])
It runs right here in your browser, with nothing to install.

The first draft has no start time, so the router sends it back; the second passes. Two limits keep it from looping forever: the router gives up after three tries, and the runner stops after max_steps whatever the routing says. An agent graph always needs both, because a model-driven check might never be satisfied.

Quick check
The check keeps finding problems. With the router above, what does this print?
END = "END"
def after_check(state):
    if not state["problems"]:
        return END
    return "draft" if state["tries"] < 3 else END

print(after_check({"problems": ["no start time"], "tries": 2}), after_check({"problems": ["no start time"], "tries": 3}))
  1. draft draft
  2. draft END
  3. END END
  4. END draft
Show the answer

draft END

With problems left and 2 tries, 2 < 3, so it goes back to draft. At 3 tries it gives up and ends, problems and all: the state still holds them, so whatever comes next can report that the check never passed.

Reducers: how an update joins the state

{**state, **update} replaces each key a node returns. That's right for a draft, where the new one should win, but wrong for a list of notes or a running cost: the second node's notes would wipe out the first's. A reducer says how to combine the old value with the new one for a particular key: for a list, add them end to end; for a cost, sum them. Keys without a reducer are replaced as before.

import operator

def merge(state, update, reducers):
    new = dict(state)
    for key, value in update.items():
        if key in reducers and key in state:
            new[key] = reducers[key](state[key], value)
        else:
            new[key] = value
    return new

reducers = {"notes": operator.add, "cost": operator.add}
state = {"notes": ["wants a ridge walk"], "cost": 0.0, "draft": "v1"}
for update in [{"notes": ["forecast: dry"], "cost": 0.002}, {"notes": ["trail open"], "cost": 0.003, "draft": "v2"}]:
    state = merge(state, update, reducers)
print(state)
print({**state, **{"notes": ["forecast: rain"]}})    # no reducer: one update wipes the history

operator.add joins two lists or adds two numbers. Conversation messages are the classic case: every node appends to the transcript, and nothing should ever overwrite it. Frameworks let you attach a reducer to a key in the state's type; LangGraph, for one, writes it as Annotated[list, operator.add], and offers a message reducer, add_messages, that appends new messages but replaces one with the same id, so a person can correct a message in place.

Routing a worker and an evaluator

A common agent shape pairs two model calls. A worker tries to do the task, using tools. An evaluator reads the result against success criteria the user gave ("a trail under 10 km, dry on Saturday, and the group texted") and returns a verdict: done, not yet (with feedback for the worker), or stuck because it needs the user. A router turns the verdict into the next node. Writing the evaluator as a separate step is what makes the loop honest: the worker doesn't get to grade its own homework.

Try it now

The router reads the evaluator's verdict from the state; the round counter guarantees the loop ends.

Drill

Route on the evaluator's verdict

Write route(state, max_rounds=3) for the graph above. state["verdict"] is a dictionary with "done", "needs_user" (both booleans) and "feedback"; state["rounds"] counts worker rounds so far. Return END if the verdict is done; otherwise "ask_user" if it needs the user; otherwise "give_up" once rounds has reached max_rounds; otherwise "worker". A missing verdict (the evaluator's reply couldn't be parsed) counts as not done, so the worker tries again, within the round limit.

END = "END"


def route(state, max_rounds=3):
    return "worker"


print(route({"verdict": {"done": True, "needs_user": False, "feedback": ""}, "rounds": 1}))
In the lesson, your code is checked as you write it.

Checkpoints: save the state after every step

An agent run can take minutes and dozens of model calls, and processes die: a deploy restarts the server, a tool times out, the laptop closes. If the state lives only in memory, everything is lost and the run starts over, paying for every model call again. So the runner saves a checkpoint after each node: which node comes next, and the full state, under a thread id that names this run. To resume, load the latest checkpoint and carry on from its next node. Because the state is saved as JSON, it can live in a file or a database table and survive a restart (keep the state to JSON types: a tuple comes back as a list and a set can't be saved at all). These checkpoints save the workflow's state; the memory and planning lesson's checkpoints were snapshots of files. Same word, same idea of a save point, different things saved.

import json

END = "END"
store = {}                     # thread id -> latest checkpoint (a database table in real use)
calls = {"plan": 0, "search": 0, "write": 0}

def run(nodes, edges, node, state, thread):
    while node != END:
        state = {**state, **nodes[node](state)}
        node = edges[node]
        store[thread] = json.dumps({"next": node, "state": state})    # after every step
    return state

def resume(nodes, edges, thread):
    saved = json.loads(store[thread])
    return run(nodes, edges, saved["next"], saved["state"], thread)

def plan(state):
    calls["plan"] += 1
    return {"plan": "find a dry trail"}

def search(state):
    calls["search"] += 1
    if calls["search"] == 1:
        raise ConnectionError("forecast service timed out")
    return {"trail": "Fern Hollow loop"}

def write(state):
    calls["write"] += 1
    return {"message": f"Saturday: {state['trail']}"}

nodes, edges = {"plan": plan, "search": search, "write": write}, {"plan": "search", "search": "write", "write": END}
try:
    run(nodes, edges, "plan", {}, "trip-42")
except ConnectionError as e:
    print("crashed:", e, "| saved:", store["trip-42"])
print(resume(nodes, edges, "trip-42"))
print(calls)

The crash happened inside search, so the last checkpoint is the one saved after plan: next is search. The resume runs search and write, and plan ran exactly once. One caution: a node that crashes halfway may already have done something outside the program, such as charging a card. Make such nodes idempotent (the reliability lesson's idempotency keys), or put them behind a pause, which is next.

Drill

Checkpoints in SQLite

Write save(db, thread, step, next_node, state) and load(db, thread, step=None) over the table checkpoints(thread TEXT, step INTEGER, next TEXT, state TEXT). save stores the state as JSON. load returns (next_node, state) from the thread's checkpoint with the highest step, or from exactly step if one is given (rewinding to an earlier point, which frameworks call time travel); it returns None if there's no such checkpoint. Other threads' checkpoints must never mix in.

import json
import sqlite3


def save(db, thread, step, next_node, state):
    pass


def load(db, thread, step=None):
    return None


db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE checkpoints (thread TEXT, step INTEGER, next TEXT, state TEXT)")
save(db, "trip-42", 0, "search", {"plan": "find a dry trail"})
print(load(db, "trip-42"))
In the lesson, your code is checked as you write it.

Pausing for a person

Some steps shouldn't happen without a person's yes: sending a message to the whole club, booking a hut, spending money. The permissions lesson asked for approval inside a running loop. A graph can do better, because it has checkpoints: mark the node as pause before, and when the run reaches it, the runner saves a checkpoint and returns. Nothing waits in memory. The person might answer in a minute or on Monday; when they do, the program loads the checkpoint, applies any edits they made to the state (a fixed time, a shorter message), and resumes, running the approved node this time without pausing again.

import json

END = "END"
store = {}

def run(nodes, edges, node, state, thread, pause_before=(), approved=None):
    while node != END:
        if node in pause_before and node != approved:
            store[thread] = json.dumps({"next": node, "state": state})
            return {"status": "paused", "next": node, "state": state}
        approved = None
        state = {**state, **nodes[node](state)}
        node = edges[node]
        store[thread] = json.dumps({"next": node, "state": state})
    return {"status": "done", "state": state}

def resume(nodes, edges, thread, edits=None, pause_before=()):
    saved = json.loads(store[thread])
    state = {**saved["state"], **(edits or {})}
    return run(nodes, edges, saved["next"], state, thread, pause_before, approved=saved["next"])

nodes = {"draft": lambda s: {"text": "Ridge walk Saturday, 7:00 at Fern Hollow."},
         "send": lambda s: {"sent": f"texted the club: {s['text']}"}}
edges = {"draft": "send", "send": END}
first = run(nodes, edges, "draft", {}, "walk-7", pause_before={"send"})
print(first["status"], "before", first["next"], "|", first["state"]["text"])
later = resume(nodes, edges, "walk-7", edits={"text": "Ridge walk Saturday, 8:00 at Fern Hollow."}, pause_before={"send"})
print(later["status"], "|", later["state"]["sent"])

The person read the draft, moved the start to 8:00, and approved. The text that went out is the edited one, because the edit changed the saved state before send ran.

Quick check
Why does the runner save a checkpoint and return at a pause, instead of calling input() and waiting for the answer?
  1. Returning is faster than waiting, so the user gets an answer sooner
  2. The answer may take hours or days; the run must survive restarts and not hold a process open
  3. input() doesn't work inside functions, so the runner can't call it from a node
  4. Checkpoints make the model's answers better, since it can reread its earlier work
Show the answer

The answer may take hours or days; the run must survive restarts and not hold a process open

A paused run is just a saved state and a next node. Any process can resume it later, after a deploy or a restart, and nothing sits blocked in memory meanwhile. On a web server, input() isn't even possible: the approval arrives as a separate request.

The same ideas in a framework

You've now built the core of what agent-graph frameworks provide. When you meet one, map the names:

ideathis lessonLangGraph, for example
state and its typea dictionaryStateGraph(State), with State a TypedDict
reducerreducers = {"notes": operator.add}notes: Annotated[list, operator.add] in the type
nodea function returning an updategraph.add_node("search", search)
where a run startsthe start argumentadd_edge(START, "plan")
fixed edge and routeredges[node], a name or a functionadd_edge and add_conditional_edges
checkpoints and threadsstore[thread]a checkpointer passed to compile, and config={"configurable": {"thread_id": "trip-42"}} on each call
pause for approvalpause_beforean interrupt() call inside the node, resumed with Command(resume=...) (interrupt_before at compile time exists too, mainly for debugging)
a person's editsedits replace keysupdate_state, which passes edits through the reducers

One difference to know: when LangGraph resumes after an interrupt(), it reruns the interrupted node from its start, so any code before the interrupt() call runs twice; keep side effects after it. Frameworks add parallel branches, streaming, visual graph drawings and ready-made database checkpointers, and they change their APIs often, so check the current docs. The ideas underneath (an explicit state, small nodes, routing by function, a checkpoint after every step and pauses that return) stay the same, and they work fine in plain Python when a framework would be more than you need.

Portfolio brief

Weekend sidekick: a hike planner that asks before it texts

Build an assistant that plans the walking club's Saturday hike and texts the group, as a graph like the one in this lesson. A worker model call picks a trailhead and a start time using two tools (the club's trailheads and Saturday's hourly forecast, from the practice API). An evaluator model call checks the plan against the success criteria in the starter and returns a verdict. When the plan passes, a node writes the text, and the graph pauses for a person to approve or edit it before a last node sends it through the practice SMS route.

Two things make this a real system rather than a script: the run survives a crash because the state is checkpointed after every step, and the text goes out exactly once, only after a yes. Keep runs small: lesson code gets 300 model calls a day. With simulated replies (no API key) the model's choices are made up and the evaluator's JSON may not parse, so handle that: the code must still run end to end.

It must

  • The state is one dictionary with a messages list (an append reducer), the criteria, the latest verdict, a round counter and the draft text
  • Nodes for the worker (model call with the two tools), running tools (tool_result messages, errors as results), the evaluator (a second model call returning JSON with done, needs_user and feedback; a reply that won't parse counts as not done), writing the text, and sending it
  • A router after the evaluator with a round limit, and a step limit on the whole run
  • A checkpoint in the SQLite table after every step under a thread id; show a resume after a deliberate crash that repeats no finished model call (count the calls)
  • The run pauses before sending: it prints the draft, a person can edit it, and the resumed run sends the edited text; resuming twice still sends one text
  • Tool results stay small (only the hours and fields the plan needs), and a run log prints each step with the next node and the tokens used

Go further

  • Add the needs_user path: the evaluator asks the person a question (such as how far they'll drive), and the run resumes with the answer
  • Write the approval as a separate function that loads the run by thread id, the way a web request would
  • Port the graph to LangGraph and compare: what did the framework save you, and what did it hide?
import json
import sqlite3
import requests

API = "https://api.gradient.test/v1"
SMS_HEADERS = {"Authorization": "Bearer sk_demo_sms_42_not_a_real_secret"}
GROUP_PHONE = "+15035550134"   # 555-01xx numbers are reserved for fiction


def list_trailheads():
    return requests.get(f"{API}/trailheads").json()["data"]


def forecast(trailhead):
    data = requests.get(f"{API}/forecast", params={"trailhead": trailhead}).json()
    return [{k: hour[k] for k in ("time", "temp_f", "precip_chance", "code")} for hour in data["hours"][:24]]


def send_sms(body):
    """A side effect: the graph calls this only after a person approves the text."""
    response = requests.post(f"{API}/sms", headers=SMS_HEADERS, json={"to": GROUP_PHONE, "body": body})
    response.raise_for_status()
    return response.json()


TOOLS = [
    {"name": "list_trailheads", "description": "List the club's trailheads with their ids.",
     "input_schema": {"type": "object", "properties": {}}},
    {"name": "forecast", "description": "Saturday's hourly forecast for one trailhead: time, temp_f, precip_chance (%), code.",
     "input_schema": {"type": "object", "properties": {"trailhead": {"type": "string"}}, "required": ["trailhead"]}},
]


def call_claude(system, messages, tools=None):
    body = {"model": "claude-opus-5-5", "max_tokens": 800, "system": system, "messages": messages}
    if tools:
        body["tools"] = tools
    response = requests.post("https://api.anthropic.com/v1/messages",
        headers={"x-api-key": "added-by-the-academy-server", "anthropic-version": "2026-10-21"},
        json=body, timeout=90)
    response.raise_for_status()
    return response.json()


REQUEST = "Find a dry morning hike for Saturday, June 14, and text the group the plan."
CRITERIA = [
    "names exactly one trailhead from the club's list",
    "a start time between 7:00 and 10:00 when the precipitation chance is under 30%",
    "a text under 300 characters with the trailhead, the time and what to bring",
]

db = sqlite3.connect(":memory:")   # on your own machine, use a file so a run survives a restart
db.execute("CREATE TABLE checkpoints (thread TEXT, step INTEGER, next TEXT, state TEXT)")

print(list_trailheads())
print(forecast("fern-hollow")[7:10])
reply = call_claude("You plan hikes for a walking club. Use the tools; never guess the weather.",
                    [{"role": "user", "content": REQUEST}], tools=TOOLS)
print("stop reason:", reply["stop_reason"])
# your code: the state and its reducers, the nodes, the router, checkpoints and the pause before sending
In the lesson, you build it here and the tutor can review it.

Practice what you just read

  • 4 code examples you can edit and run
  • a practice exercise that checks your code
  • a 5-question quiz
  • 5 review cards for spaced repetition

Create a free account

Try a quiz question

Two nodes each return {"messages": [...]} and the state has no reducer for messages. What happens?

  1. The second update replaces the first node's messages
  2. Python raises a KeyError when the second update arrives
  3. The two lists are joined automatically, end to end
  4. The graph refuses to run until a reducer is added

Answer it at the end of the lesson to see if you're right, and why.

Before this lesson