Lab 06 — A graph-shaped pipeline agent with LangGraph

Lesson 06 · Building Your Own Agents · ~45 min hands-on

Goal: at the end of this lab you will have (a) run agent.py against a research question and seen a four-stage [node ...] trace, (b) inspected the structured draft output, (c) killed the process mid-graph and resumed from the last checkpoint, (d) added interrupt_before to pause at critique for manual approval, and (e) modified the critique node to always demand revision and watched the loop bailout fire. This is LangGraph’s sweet spot — a control flow with a drawable shape, checkpointed state at every transition, and a critique loop with an explicit cap.

Prereqs

• The agents/requirements.txt already installed (which pins langgraph>=0.2,<0.3 and anthropic>=0.39,<0.40). If not, run pip install -r agents/requirements.txt from the repo root.
• ANTHROPIC_API_KEY exported in your shell.
• L05 lab completed — you should be comfortable with StateGraph, conditional edges, and SqliteSaver.
• Working directory at the repo root (the paths below assume that).

Step 1 — Read agents/lesson06-graph-shaped-agent/agent.py

The full program lives in the repo at agents/lesson06-graph-shaped-agent/agent.py. Below it is reproduced inline so you can read along without flipping windows. We’ll then walk through each block.

Open the file in your editor — you don’t need to retype anything; the file is already there. The inlined copy in this lab guide is for reading convenience.

Step 2 — The full program

Here is the entire agent.py, verbatim:

"""Graph-shaped pipeline agent — Lesson 06 of the Agents course.

Built around LangGraph primitives (`langgraph>=0.2,<0.3`). The graph
has four named stages plus a critique-and-revise loop:

    START -> plan_sources -> fetch_sources -> write_draft
          -> critique_draft -> (revise -> write_draft) or END

State is persisted at every node transition via SqliteSaver, so a
crash mid-graph can be resumed without re-running successful nodes.

Run:
    export ANTHROPIC_API_KEY=...
    python agent.py "What changed in Linux kernel 6.14's networking stack?"

NOTE: this file is also inlined in lab_guide.html. If you edit it,
also update the inlined block there.
"""
from __future__ import annotations

import json
import os
import sys
from typing import Any, Annotated, TypedDict
import operator

from anthropic import Anthropic
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver

# --- Mocked data --------------------------------------------------------

PKB: dict[str, dict[str, str]] = {
    "kernel-6.14-net-notes": {
        "title": "Personal notes on Linux 6.14 networking",
        "text": (
            "Linux kernel 6.14 ships io_uring zero-copy TX for TCP sockets, "
            "enables BIG TCP for IPv4 by default (was IPv6-only), and adds a "
            "new BPF kfunc family for tc-bpf flow tracking."
        ),
    },
    "kernel-6.14-misc": {
        "title": "Kernel 6.14 misc highlights",
        "text": "Bcachefs is marked stable. sched_ext gets richer hooks.",
    },
}

WEB_INDEX: dict[str, str] = {
    "kernel 6.14 networking": (
        "LWN: io_uring zero-copy TX, BIG TCP IPv4 default, tc-bpf kfuncs."
    ),
    "big tcp ipv4": "Kernelnewbies: BIG TCP IPv4 default in 6.14.",
    "io_uring tcp": "lore.kernel.org: io_uring zero-copy TX patch series.",
}

# --- State --------------------------------------------------------------

class PipelineState(TypedDict):
    question: str
    plan: list[str]
    notes: Annotated[list[str], operator.add]
    draft: dict
    revise: bool
    revisions: int

# --- LLM client ---------------------------------------------------------

client = Anthropic()
MODEL = "claude-3-5-sonnet-latest"

def _ask_json(prompt: str, max_tokens: int = 600) -> Any:
    msg = client.messages.create(
        model=MODEL,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    )
    text = msg.content[0].text.strip()
    # Tolerate ```json ... ``` fences.
    if text.startswith("```"):
        text = text.strip("`").split("\n", 1)[1].rsplit("\n", 1)[0]
    return json.loads(text)

# --- Nodes --------------------------------------------------------------

def plan_sources(s: PipelineState) -> dict:
    prompt = (
        f"Question: {s['question']}\n\n"
        "Return a JSON list of 2-4 short search queries (strings) that "
        "would surface relevant material. JSON only, no prose."
    )
    plan = _ask_json(prompt, max_tokens=200)
    print(f"[node plan_sources]   plan={plan}")
    return {"plan": plan}


def fetch_sources(s: PipelineState) -> dict:
    new_notes: list[str] = []
    for q in s["plan"]:
        q_lower = q.lower()
        for doc_id, doc in PKB.items():
            if any(w in doc["text"].lower() for w in q_lower.split()):
                new_notes.append(f"[{doc_id}] {doc['text']}")
        for key, snippet in WEB_INDEX.items():
            if any(w in key for w in q_lower.split()):
                new_notes.append(f"[web:{key}] {snippet}")
    # Deduplicate while preserving order.
    seen: set[str] = set()
    deduped = [n for n in new_notes if not (n in seen or seen.add(n))]
    print(f"[node fetch_sources]  notes_count={len(deduped)}")
    return {"notes": deduped}


def write_draft(s: PipelineState) -> dict:
    notes_blob = "\n".join(s["notes"]) or "(no notes)"
    prior = json.dumps(s.get("draft", {}), indent=2) if s.get("draft") else "(none)"
    prompt = (
        f"Question: {s['question']}\n\nNotes:\n{notes_blob}\n\n"
        f"Prior draft (if any):\n{prior}\n\n"
        "Return a JSON object with keys 'intro' (string, 1-2 sentences), "
        "'bullets' (list of exactly 3 short strings), 'sources' (list of "
        "strings — doc_ids or 'web:...' keys from the notes). JSON only."
    )
    draft = _ask_json(prompt, max_tokens=600)
    revisions = s.get("revisions", 0) + 1
    print(
        f"[node write_draft]    "
        f"bullets={len(draft.get('bullets', []))} revisions={revisions}"
    )
    return {"draft": draft, "revisions": revisions}


def critique_draft(s: PipelineState) -> dict:
    prompt = (
        f"Question: {s['question']}\n\n"
        f"Draft: {json.dumps(s['draft'], indent=2)}\n\n"
        f"Notes:\n{chr(10).join(s['notes'])}\n\n"
        "Does the draft need revision? Consider: are all 3 bullets grounded "
        "in the notes? Is the intro accurate? Return JSON: "
        '{"revise": true/false, "reason": "..."}'
    )
    verdict = _ask_json(prompt, max_tokens=200)
    print(f"[node critique_draft] revise={verdict.get('revise')}")
    return {"revise": bool(verdict.get("revise"))}


def route_after_critique(s: PipelineState) -> str:
    if s["revise"] and s["revisions"] < 2:
        return "revise"
    return "finish"

# --- Graph assembly -----------------------------------------------------

def build_graph() -> StateGraph:
    g = StateGraph(PipelineState)
    g.add_node("plan_sources", plan_sources)
    g.add_node("fetch_sources", fetch_sources)
    g.add_node("write_draft", write_draft)
    g.add_node("critique_draft", critique_draft)
    g.add_edge(START, "plan_sources")
    g.add_edge("plan_sources", "fetch_sources")
    g.add_edge("fetch_sources", "write_draft")
    g.add_edge("write_draft", "critique_draft")
    g.add_conditional_edges(
        "critique_draft",
        route_after_critique,
        {"revise": "write_draft", "finish": END},
    )
    return g


def run_pipeline_agent(question: str, db_path: str = "pipeline.db") -> dict:
    """Run the pipeline agent and return the final state.

    Imported by Plan F's Lesson 11 scorecard runner — keep the signature
    stable (one positional `question` arg, optional `db_path` kwarg).
    """
    g = build_graph()
    memory = SqliteSaver.from_conn_string(db_path)
    app = g.compile(checkpointer=memory)
    config = {"configurable": {"thread_id": f"pipeline-{question[:40]}"}}
    initial: PipelineState = {
        "question": question,
        "plan": [],
        "notes": [],
        "draft": {},
        "revise": False,
        "revisions": 0,
    }
    return app.invoke(initial, config=config)

# --- CLI ----------------------------------------------------------------

if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("usage: python agent.py 'your research question'")
    if not os.environ.get("ANTHROPIC_API_KEY"):
        sys.exit("ANTHROPIC_API_KEY not set")
    out = run_pipeline_agent(sys.argv[1])
    print("\n=== final draft ===")
    print(json.dumps(out["draft"], indent=2))
    print(f"\n[revisions: {out['revisions']}, notes: {len(out['notes'])}]")

Step 3 — Run it

From the repo root:

python agents/lesson06-graph-shaped-agent/agent.py "What changed in Linux kernel 6.14's networking stack?"

Expected output: a 4–6-line [node ...] trace — one log line per node fire (plan_sources once, fetch_sources once, write_draft once or twice, critique_draft once or twice) — ending in a === final draft === block showing the structured intro, bullets, and sources, and a small footer reporting [revisions: N, notes: M]. With Sonnet, expect roughly $0.008–$0.020 per run depending on whether a revise pass fires.

✓ Self-check: you saw at least four [node ...] lines, a final-draft JSON block, and no traceback. A pipeline.db file appeared in your working directory.

Step 4 — Walk through the file

A guided tour, top to bottom:

• PKB and WEB_INDEX — the mocked data the agent searches. Smaller than L04’s mock because LangGraph doesn’t need the tool-call shape; the fetch node just iterates these dicts directly, no @tool wrapper, no LLM tool-decision round-trip.
• PipelineState — six fields. Notice notes is Annotated[list[str], operator.add] so fetch appends rather than replaces. Without the operator.add reducer, returning a new list from fetch_sources would overwrite any prior notes — LangGraph would treat the new value as a wholesale replacement.
• _ask_json — a tiny helper that wraps client.messages.create(...) and parses the result as JSON. Tolerates ```json fences because Claude sometimes adds them despite “JSON only” in the prompt.
• plan_sources — LLM call, returns a JSON list of search queries (typically 2–4). This is the first node and it sets up the work for the rest of the graph.
• fetch_sources — pure Python, no LLM call. Iterates plan, queries the mocks, dedupes. This is the “intersperse cheap deterministic steps” pattern from theory §4 — most pipelines benefit from non-LLM nodes between the expensive ones.
• write_draft — LLM call, returns the structured draft and increments revisions. Sees the prior draft on revise passes so it can improve on its own work rather than starting from scratch.
• critique_draft — LLM call, returns {"revise": bool, "reason": str}. The route function reads revise; reason is logged but not used for routing — it’s there for the human reading the trace.
• route_after_critique — the bailout-aware decision function. revisions < 2 caps the loop. Without that guard, an over-strict critic would loop forever. This guard is your responsibility — LangGraph won’t time out a critique loop for you.
• build_graph — wires the four nodes, three unconditional edges (START → plan_sources → fetch_sources → write_draft → critique_draft), and one conditional edge (back to write_draft or out to END).
• run_pipeline_agent — the entry point. Stable signature for Plan F: one positional question, optional db_path. Don’t break this — the L11 scorecard imports it.
• __main__ block — CLI wrapper. Prints the final draft JSON and a small footer with revision count and note count.

Step 5 — Inspect the checkpoint database

After the run completes, peek at what SqliteSaver wrote to disk:

sqlite3 pipeline.db "SELECT thread_id, checkpoint_id FROM checkpoints LIMIT 20;"

Expected: a list of rows, one per node transition for your run. The thread_id matches the prefix of your question (pipeline-What changed in Linux kernel 6.14…). Each checkpoint_id is a UUID identifying a saved point in the graph’s execution. This is how LangGraph stores resume points — one row per “the graph just finished a node, here is its full state right now.”

✓ Self-check: at least 4 checkpoint rows appeared (one per node), all with the same thread_id.

Step 6 — Simulate a crash mid-graph

Now the load-bearing demo: a graph that crashes can be resumed without re-running the successful nodes. Edit write_draft in agent.py to raise on the first call:

_calls = {"write": 0}

def write_draft(s: PipelineState) -> dict:
    _calls["write"] += 1
    if _calls["write"] == 1:
        raise RuntimeError("simulated crash before LLM call")
    # ... rest of the original implementation

Delete pipeline.db (so we start clean), then run the script:

rm pipeline.db
python agents/lesson06-graph-shaped-agent/agent.py "What changed in Linux kernel 6.14's networking stack?"

Expected: the script crashes inside write_draft with the simulated RuntimeError. The previous nodes’ results (plan_sources and fetch_sources) are checkpointed in pipeline.db.

Now restart Python (so the in-memory _calls counter resets) and call run_pipeline_agent(...) with the same question — same thread_id:

from agents.lesson06_graph_shaped_agent.agent import run_pipeline_agent
out = run_pipeline_agent("What changed in Linux kernel 6.14's networking stack?")

Expected: the run resumes at write_draft — but _calls["write"] is back to 0 in the new process, so it raises again. Fix: swap the in-memory _calls for a file flag (e.g., /tmp/write_crashed) so the side-effect-of-having-crashed persists across processes:

def write_draft(s: PipelineState) -> dict:
    flag = "/tmp/write_crashed"
    if not os.path.exists(flag):
        open(flag, "w").close()
        raise RuntimeError("simulated crash before LLM call")
    # ... rest of the original implementation

Re-run. Expected: this time write_draft succeeds on the second call, and the graph finishes — without re-running plan_sources or fetch_sources. Confirm by counting the [node plan_sources] log lines across both processes — there’s exactly one (from the first, crashed run).

Clean up: rm /tmp/write_crashed, then restore write_draft to its original implementation before moving on.

Step 7 — Pause at critique with interrupt_before

Now wire human-in-the-loop. Edit run_pipeline_agent to pass interrupt_before=["critique_draft"] when compiling the graph:

app = g.compile(checkpointer=memory, interrupt_before=["critique_draft"])

Delete pipeline.db and re-run. Expected: the script prints [node plan_sources], [node fetch_sources], [node write_draft], and then returns — the graph paused before critique_draft. The final-draft block at the bottom of the output shows the draft that was just written; critique_draft never ran.

Inspect the paused state from a Python REPL:

from agents.lesson06_graph_shaped_agent.agent import build_graph
from langgraph.checkpoint.sqlite import SqliteSaver

g = build_graph()
memory = SqliteSaver.from_conn_string("pipeline.db")
app = g.compile(checkpointer=memory, interrupt_before=["critique_draft"])
question = "What changed in Linux kernel 6.14's networking stack?"
config = {"configurable": {"thread_id": f"pipeline-{question[:40]}"}}

state = app.get_state(config)
print(state.values["draft"])
print("next:", state.next)
# next: ('critique_draft',)

Resume by calling app.invoke(None, config=config). The None is important — it means “no new input, just continue.” If you pass a fresh initial dict, LangGraph will treat it as a brand-new run and clobber the resume.

out = app.invoke(None, config=config)
print(out["draft"])

Expected: the graph continues from critique_draft and ends at END. In production, this gap is where a human (or another system) reviews the draft before the critique LLM call is made — an approval gate, a policy filter, a moderation queue.

✓ Self-check: state.next == ('critique_draft',) while paused, and the resume call produced a final state with a populated draft.

Remove the interrupt_before=["critique_draft"] kwarg before moving on.

Step 8 — Force the revise loop bailout

Finally, prove that the revisions < 2 guard in route_after_critique is the only thing standing between you and an infinite loop. Edit critique_draft to always return {"revise": True}:

def critique_draft(s: PipelineState) -> dict:
    print("[node critique_draft] revise=True (forced)")
    return {"revise": True}

Delete pipeline.db and re-run. Expected trace:

• [node plan_sources] — fires once.
• [node fetch_sources] — fires once.
• [node write_draft] — revisions=1.
• [node critique_draft] — revise=True; route returns "revise" because 1 < 2.
• [node write_draft] — revisions=2.
• [node critique_draft] — revise=True; route returns "finish" because 2 < 2 is false. Graph ends.

write_draft fires twice, critique_draft fires twice (both demanding revise), and then the route function bails to END on the third decision because the cap is hit. The graph ends at END despite the critic still being unhappy — that’s the bailout you wrote. Without it the loop never terminates.

✓ Self-check: write_draft fired exactly twice, the process exited normally, and the final state has revisions == 2 and revise == True (the critic’s last verdict survives in state even though the route ignored it).

Restore critique_draft to its original implementation before moving on.

What you should leave with

• agent.py running end-to-end against a research question.
• First-hand observation of crash-and-resume — and the side-effect caveat that LangGraph persists state, not in-process counters.
• A working interrupt_before pause for human-in-the-loop, plus the app.invoke(None, config) resume idiom.
• Confidence that loop bailouts are your responsibility — the graph runtime won’t save you from an over-strict critic.
• The instinct that LangGraph’s “you draw the graph, it runs the loop” division of labour fits a pipeline shape like a glove — planner, fetcher, critique, writer is a control flow you can stare at on a whiteboard.

Troubleshooting