Lesson 06 — A Graph-Shaped Pipeline Agent

LangGraph at its sweet spot: planner → fetcher → critique branch → writer, with checkpoint resume.

Agents Module 3 LangGraph Pipeline

Why this is LangGraph's sweet spot

• The control flow has a shape — 4 named stages plus a branch and a possible loop. You can draw it on a whiteboard.
• Each stage is checkpointable — LLM calls aren't cheap to re-run, so per-node persistence pays for itself.
• Critique is a natural human-in-the-loop seam — pause before it, hand the draft to a person, resume.
• The writer's output should be revisable without re-fetching sources. The graph lets you loop back to write_draft only.

Hits every condition on L05 Slide 11's “Pick it when” list.

The agent we're building

Real example question: “What changed in Linux kernel 6.14's networking stack?”

Same question as L04 — so the reader can compare the two agents' shapes, not their outputs.

The four pipeline stages

Four nodes, each a plain Python function. No subclassing, no decorators — just functions of state.

The graph topology

START
  -> plan_sources
  -> fetch_sources
  -> write_draft
  -> critique_draft
       |-- revise == True  -> write_draft  (loop, max 2 revisions)
       |-- revise == False -> END

Note the critique-then-loop-back-to-write edge. This is the shape you cannot draw in a chain.

The state schema

from typing import TypedDict, Annotated
import operator

class PipelineState(TypedDict):
    question: str
    plan: list[str]                            # search queries
    notes: Annotated[list[str], operator.add]  # accumulating snippets
    draft: dict                                # {intro, bullets, sources}
    revise: bool                               # critique sets this
    revisions: int                             # bailout counter

notes uses the operator.add reducer so each fetch appends; revisions is the loop bailout.

The conditional edge after critique

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

g.add_conditional_edges(
    "critique_draft",
    route_after_critique,
    {"revise": "write_draft", "finish": END},
)

The revisions < 2 bailout matters: even if the critique keeps demanding revisions, the loop stops after two passes.

What's in a node — the writer

def write_draft(s: PipelineState) -> dict:
    notes_blob = "\n".join(s["notes"])
    prompt = (
        f"Question: {s['question']}\n\nNotes:\n{notes_blob}\n\n"
        "Write a JSON object with keys 'intro' (string), "
        "'bullets' (list of 3 short strings), 'sources' (list of strings)."
    )
    msg = client.messages.create(
        model="claude-3-5-sonnet-latest",
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}],
    )
    draft = json.loads(msg.content[0].text)
    return {"draft": draft, "revisions": s["revisions"] + 1}

Node calls anthropic directly, parses JSON, returns a partial state update. No SDK abstraction. Increments the revision counter on every write.

Compile + checkpoint setup

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

config = {"configurable": {"thread_id": f"pipeline-{question[:30]}"}}
result = app.invoke(
    {"question": question, "plan": [], "notes": [], "draft": {},
     "revise": False, "revisions": 0},
    config=config,
)

thread_id is per-question: reruns of the same question resume; reruns of new questions start fresh.

A real run trace

[node plan_sources]      plan=['kernel 6.14 networking', 'BIG TCP IPv4']
[node fetch_sources]     notes_count=4
[node write_draft]       draft.bullets=3  revisions=1
[node critique_draft]    revise=True
[node write_draft]       draft.bullets=3  revisions=2
[node critique_draft]    revise=False
[END]                    final report ready

Two write passes; the second clears critique. The bailout counter would have stopped a third even if critique stayed unhappy.

Crash resume — the killer demo

• Kill the process after fetch_sources succeeds but before write_draft finishes.
• Restart and call app.invoke(None, config=config) with the same thread_id.
• The graph picks up at write_draft with notes already populated — no re-fetch, no re-plan. Just the work you hadn't done yet.
• In L04, a crash means re-running the entire tool-call sequence (and re-paying for it). Here the cost is bounded to the node that crashed.

Human-in-the-loop with interrupt_before

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

• The graph now stops before critique.
• A human (or another agent) reads state.values["draft"] and decides what to do next.
• Keep going: app.invoke(None, config=config).
• Skip critique by tweaking state directly: app.update_state(config, {"revise": False}).

What's left to wire (Module 6 hook)

Recap & what's next

• Graph-shaped = LangGraph sweet spot. Named stages + state-dependent transitions + LLMs inside the nodes.
• The critique → revise loop needs a bailout counter — otherwise “always revise” is a free pass to infinite spend.
• Checkpoints make crash resume free. Same thread_id, app.invoke(None, config), done.