LangGraph StateGraph Tutorial: Building Multi-Agent Workflows in Python

While simple sequential chains (like those in early LangChain) work for basic question-answering, production AI applications require loops, conditional branching, human-in-the-loop approvals, and multi-agent debate. In 2026, LangGraph has become the premier framework for developing complex, cyclic agent architectures.

In this tutorial, we will construct a complete multi-agent workflow using LangGraph’s StateGraph. We build a collaborative research pipeline where a Researcher Agent searches for information and a Synthesizer Agent reviews and formats the response, with automatic looping if details are missing.

1. Multi-Agent Graph Architecture

       ┌───────────────┐
       │     START     │
       └───────┬───────┘
               ▼
       ┌───────────────┐
       │   Researcher  │ ◄───────┐
       └───────┬───────┘         │
               ▼                 │ (Needs more data)
       ┌───────────────┐         │
       │   Evaluator   ├─────────┘
       └───────┬───────┘
               │ (Approved)
               ▼
       ┌───────────────┐
       │  Synthesizer  │
       └───────┬───────┘
               ▼
       ┌───────────────┐
       │      END      │
       └───────────────┘

2. Environment Setup

pip install langgraph>=0.2.0 langchain-core>=0.3.0 pydantic>=2.7.0

3. Implementation: StateGraph with Cyclic Feedback

Create agent_graph.py:

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, START, END

# Define shared agent state
class AgentState(TypedDict):
    query: str
    research_notes: Annotated[List[str], operator.add]
    evaluation_passed: bool
    final_report: str
    iteration_count: int

def researcher_node(state: AgentState) -> dict:
    # Simulates gathering information from web and internal docs.
    query = state["query"]
    iteration = state.get("iteration_count", 0) + 1
    new_note = f"[Iteration {iteration}] Gathered telemetry and benchmarks for query: {query}"

    return {
        "research_notes": [new_note],
        "iteration_count": iteration
    }

def evaluator_node(state: AgentState) -> dict:
    # Checks whether sufficient evidence has been gathered.
    passed = state["iteration_count"] >= 2
    return {"evaluation_passed": passed}

def synthesizer_node(state: AgentState) -> dict:
    # Compiles the final technical briefing.
    notes = "\n".join(state["research_notes"])
    report = (
        f"# Research Briefing for '{state['query']}'\n\n"
        f"## Summary of Findings:\n{notes}\n\n"
        f"**Status:** Approved for production deployment."
    )
    return {"final_report": report}

def route_evaluator(state: AgentState) -> str:
    # Conditional edge router: continue loop or finalize.
    if state["evaluation_passed"]:
        return "synthesizer"
    return "researcher"

# Build the workflow graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("researcher", researcher_node)
workflow.add_node("evaluator", evaluator_node)
workflow.add_node("synthesizer", synthesizer_node)

# Add edges
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "evaluator")
workflow.add_conditional_edges("evaluator", route_evaluator, {
    "researcher": "researcher",
    "synthesizer": "synthesizer"
})
workflow.add_edge("synthesizer", END)

# Compile the graph
app = workflow.compile()

if __name__ == "__main__":
    initial_state = {
        "query": "Model Context Protocol vs Custom REST API Benchmarks",
        "research_notes": [],
        "evaluation_passed": False,
        "final_report": "",
        "iteration_count": 0
    }

    result = app.invoke(initial_state)
    print(result["final_report"])

4. Key LangGraph Concepts Explained

• TypedDict State: All nodes read from and write to a shared typed dictionary.
• Annotated[List[str], operator.add]: Tells LangGraph to append list items rather than overwriting previous entries.
• Conditional Edges: Decides the next node at runtime based on the evaluation result.

5. Frequently Asked Questions (FAQ)

How do I prevent infinite loops in LangGraph?

LangGraph includes a built-in recursion limiter (recursion_limit=25). You should also increment an iteration_count in your state and force completion when a maximum count is reached.

Can I save conversation state to a database?

Yes! LangGraph supports persistent checkpointers like SqliteSaver and PostgresSaver.

Related posts:

Tool Calling in AI Agents: How LLMs Execute Real-World Actions in 2026 Creating Your Own PDF Chatbot : LLM Building Production RAG with PostgreSQL and pgvector in Python Build a Model Context Protocol (MCP) Server in Python: Complete Step-by-Step Guide