Building a multi-agent content pipeline with LangGraph

This tutorial builds a three-agent pipeline — researcher, writer, reviewer — with a revision loop, modeled as a LangGraph state graph.

1. Define shared state

from typing import TypedDict

class PipelineState(TypedDict):
    topic: str
    research: str
    draft: str
    needs_revision: bool
    revision_notes: str

2. Write the nodes

def researcher_node(state: PipelineState) -> PipelineState:
    state["research"] = run_research(state["topic"])
    return state

def writer_node(state: PipelineState) -> PipelineState:
    state["draft"] = write_draft(state["research"], state.get("revision_notes"))
    return state

def reviewer_node(state: PipelineState) -> PipelineState:
    result = review(state["draft"])
    state["needs_revision"] = result.needs_revision
    state["revision_notes"] = result.notes
    return state

3. Wire the graph with a revision loop

from langgraph.graph import StateGraph, END

graph = StateGraph(PipelineState)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges(
    "reviewer",
    lambda state: "writer" if state["needs_revision"] else END,
)

app = graph.compile()

4. Run it

result = app.invoke({"topic": "agentic RAG architectures", "needs_revision": False, "revision_notes": ""})
print(result["draft"])

Next steps

• Add a max-revision counter to avoid infinite review loops
• Persist intermediate state so a failed run can resume instead of restarting
• Swap the reviewer’s rubric for a structured, checkable schema instead of free text