LangGraph Tutorial: Building Multi-Agent Systems Step by Step

LangGraph is currently the most powerful framework for building stateful, multi-actor applications with LLMs. Created by LangChain, it lets you model complex agent workflows as directed graphs—complete with state management, conditional transitions, loops, and parallel execution.

If you’ve ever tried building a multi-agent workflow with simple LangChain chains, you know how quickly things spiral. LangGraph solves exactly that problem: instead of writing spaghetti-like chains of prompts, you define a clean graph with nodes (agents), edges (transitions), and a shared state. It’s closer to real software architecture than prompt engineering.

In this tutorial, I’ll walk you through the fundamentals all the way to a fully-featured multi-agent system with human-in-the-loop feedback, memory, and error handling. Every code example is production-ready.

TL;DR — LangGraph in 90 Seconds

What Is LangGraph—and Why You Need It

The Problem with Simple Chains

Imagine you want to build a content creation pipeline: one agent researches, another writes, a third reviews. With a simple LangChain chain, it looks like this:

# Naive approach — works, but doesn't scale
research = llm.invoke("Research topic X")
draft = llm.invoke(f"Write article: {research}")
final = llm.invoke(f"Improve: {draft}")

This works for simple cases. But what if:

• The reviewer rejects the article and the writer needs to revise? → You need loops
• You want to share state between agents (like all previous messages)? → You need state management
• You want to route differently based on content? → You need conditional transitions
• You want to pause and wait for human approval? → You need human-in-the-loop
• You want multiple agents running in parallel? → You need concurrency

LangGraph handles all of this.

The LangGraph Architecture

LangGraph models agent workflows as state graphs—directed graphs where:

• Nodes (vertices) = Agent functions that receive state, process it, and return updated state
• Edges (connections) = Transitions between nodes, either sequential or conditional
• State = A shared dictionary (or Pydantic model) that flows between all nodes
• Conditional Edges = Functions that decide which node runs next based on current state
• Cycles = Loops that can execute a node multiple times (e.g., Writer → Reviewer → Writer)

The design is inspired by workflow engines like Apache Airflow or Temporal, but optimized for LLM-based agents.

Why Graphs Instead of Chains?

Feature	Simple Chain	LangGraph
Linear sequences	✅	✅
Conditional branches	❌	✅
Loops (cycles)	❌	✅
Shared state	Difficult	✅ Native
Parallel execution	❌	✅
Human-in-the-loop	❌	✅
Checkpointing / Memory	❌	✅
Graph visualization	❌	✅

Installation and Setup

pip install langgraph langchain-openai langchain-core

Additional packages recommended for this tutorial:

pip install langchain-anthropic  # For Claude models
pip install langgraph-checkpoint-sqlite  # For SQLite persistence
pip install grandalf  # For graph visualization in terminal

Set API keys:

export OPENAI_API_KEY="sk-..."
# Optional for Claude:
export ANTHROPIC_API_KEY="sk-ant-..."

Important: Never hardcode API keys. Always use environment variables or .env files with python-dotenv.

The 4 Core Concepts—In Detail

1. State—The Foundation

State is a shared data object passed between all nodes in the graph. Each node can read and update it. This is the fundamental difference from simple chains, where each step only sees the output of the previous one.

State as TypedDict (recommended for simple cases):

from typing import TypedDict, List, Optional

class ContentState(TypedDict):
    topic: str                    # The topic being worked on
    research: str                 # Research findings
    draft: str                    # Current draft
    final_article: str            # Final article
    messages: List[str]           # Full message history
    revision_count: int           # Number of revisions
    approved: bool                # Has the article been approved?
    feedback: Optional[str]       # Reviewer feedback

Why TypedDict? It gives you type safety and IDE autocompletion without sacrificing dictionary flexibility. For more complex applications, use Pydantic models:

from pydantic import BaseModel, Field

class ContentState(BaseModel):
    topic: str = Field(description="The topic being worked on")
    research: str = Field(default="", description="Research findings")
    draft: str = Field(default="", description="Current draft")
    revision_count: int = Field(default=0)
    approved: bool = Field(default=False)

State Reduction—How Updates Work:

By default, a node’s return value overwrites the state. But you can define reducer functions to accumulate state fields:

from typing import Annotated
from operator import add

class ContentState(TypedDict):
    # messages gets appended, not overwritten
    messages: Annotated[List[str], add]
    # research gets overwritten (default behavior)
    research: str

When two nodes return messages, the lists combine instead of replacing each other. This is invaluable for maintaining conversation history.

2. Nodes — The Agents

Nodes are Python functions that receive the state, process it, and return an updated state (or a partial update). Each node is an agent or a processing unit.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def researcher(state: ContentState) -> dict:
    """Research agent: gathers information on the topic."""
    topic = state["topic"]
    
    prompt = f"""You are a research expert. Gather the key 
    information on the topic: {topic}.
    
    Structure your response as:
    1. Definition and fundamentals
    2. Current developments
    3. Practical examples
    4. Common misconceptions
    
    Return only the research, no introduction."""
    
    response = llm.invoke(prompt)
    
    # Partial update: return only the fields you change
    return {
        "research": response.content,
        "messages": [f"Researcher: {response.content[:200]}..."]
    }

Key points about nodes:

1. Partial updates: You don’t need to return the entire state, just the fields you modify. LangGraph merges it with the existing state.
2. No side effects: Nodes should ideally be pure functions (input → output). Side effects (file writing, API calls) should be handled explicitly.
3. Error handling: If a node throws an exception, the entire graph stops. Use try/except for operations prone to failure.

def researcher(state: ContentState) -> dict:
    try:
        response = llm.invoke(prompt)
        return {"research": response.content}
    except Exception as e:
        # Write error to state instead of breaking the graph
        return {
            "research": f"Research error: {str(e)}",
            "messages": [f"Researcher ERROR: {str(e)}"]
        }

3. Edges — The Transitions

Edges define which node runs after the current one. There are three types:

Simple edges (fixed transitions):

# researcher is always followed by writer
workflow.add_edge("researcher", "writer")

Conditional edges (branching logic): This is the most powerful feature of LangGraph. A routing function decides which node comes next based on the state:

def route_after_review(state: ContentState) -> str:
    """Decides what happens after the review."""
    if state.get("approved"):
        return "publish"           # Approved → publish
    elif state.get("revision_count", 0) >= 3:
        return "human_review"      # Too many revisions → human check
    else:
        return "writer"            # Not approved → writer revises

# Add conditional edge
workflow.add_conditional_edges(
    "reviewer",           # Source node
    route_after_review,   # Routing function
    {
        "publish": "publisher",
        "human_review": "human_node",
        "writer": "writer"
    }
)

Entry point (start node):

workflow.set_entry_point("researcher")

4. Cycles — Loops for Iterative Improvement

Cycles are the key difference between LangGraph and simple chains. A cycle allows an agent team to work iteratively:

researcher → writer → reviewer → (not good?) → writer → reviewer → (good!) → publish

In LangGraph, you achieve this with conditional edges that loop back to an earlier node:

# Reviewer can send Writer back (cycle!)
workflow.add_conditional_edges(
    "reviewer",
    route_after_review,  # Can return "writer"
)

# Writer always goes to reviewer
workflow.add_edge("writer", "reviewer")

Preventing infinite loops: Always implement a counter or maximum iteration limit:

def route_after_review(state: ContentState) -> str:
    revision_count = state.get("revision_count", 0)
    
    if state.get("approved"):
        return "publish"
    
    if revision_count >= 5:
        return "human_review"  # Safety valve
    
    return "writer"  # Try again

A Complete Example — Step by Step

Now we’ll build a full multi-agent system: Researcher + Writer + Reviewer + Publisher with human-in-the-loop, revision limits, and memory.

Step 1: Define the state

from typing import TypedDict, List, Optional, Annotated
from operator import add

class ContentState(TypedDict):
    topic: str
    research: str
    draft: str
    final_article: str
    messages: Annotated[List[str], add]  # Appended, not overwritten
    revision_count: int
    approved: bool
    feedback: Optional[str]

Step 2: Define nodes (agents)

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)  # Some creativity for the writer

def researcher(state: ContentState) -> dict:
    """Research agent: gathers structured information."""
    topic = state["topic"]
    
    prompt = f"""You are a research expert. Gather the key 
    information on the topic: {topic}.
    
    Consider:
    - Definition and fundamentals
    - Current developments (2025-2026)
    - Practical examples and use cases
    - Common misconceptions and pitfalls
    
    Return only the structured research."""
    
    response = llm.invoke(prompt)
    return {
        "research": response.content,
        "messages": [f"Researcher: Research complete ({len(response.content)} characters)"]
    }

def writer(state: ContentState) -> dict:
    """Writer agent: writes article based on research and feedback."""
    research = state["research"]
    feedback = state.get("feedback", "")
    revision_count = state.get("revision_count", 0)
    
    if feedback and revision_count > 0:
        prompt = f"""You are a professional writer. Revise the 
        following article based on the reviewer's feedback.
        
        Research: {research}
        
        Current draft: {state.get('draft', '')}
        
        Reviewer feedback: {feedback}
        
        Write the improved article. Address all points 
        from the feedback."""
    else:
        prompt = f"""You are a professional writer. Write a 
        comprehensive, well-structured article on the topic: {state['topic']}
        
        Based on this research: {research}
        
        The article should:
        - Have a clear introduction
        - Use subheadings
        - Include practical examples
        - End with a summary"""
    
    response = llm.invoke(prompt)
    return {
        "draft": response.content,
        "revision_count": revision_count + 1,
        "messages": [f"Writer: Draft v{revision_count + 1} created"]
    }

def reviewer(state: ContentState) -> dict:
    """Reviewer agent: checks quality and provides feedback."""
    draft = state["draft"]
    revision_count = state.get("revision_count", 0)
    
    prompt = f"""You are a critical reviewer. Evaluate the following article 
    based on these criteria:
    
    1. Structure and readability (1-10)
    2. Technical accuracy (1-10)
    3. Practical relevance (1-10)
    4. Completeness (1-10)
    
    Article: {draft}
    
    If all criteria reach at least 7/10, respond with "APPROVED".
    Otherwise respond with "REJECTED" and provide concrete feedback for improvement."""
    
    response = llm.invoke(prompt)
    content = response.content
    
    approved = "APPROVED" in content.upper()
    
    return {
        "approved": approved,
        "feedback": content if not approved else None,
        "final_article": draft if approved else "",
        "messages": [f"Reviewer: {'Approved' if approved else 'Rejected (Revision ' + str(revision_count) + ')'}"]
    }

def publisher(state: ContentState) -> dict:
    """Publisher agent: formats the final article."""
    article = state["final_article"]
    
    prompt = f"""Format this article as Markdown with:
    - Title as H1
    - Metadata (author, date, reading time)
    - Clean section headings
    
    Article: {article}"""
    
    response = llm.invoke(prompt)
    return {
        "final_article": response.content,
        "messages": [f"Publisher: Article published"]
    }

def human_review(state: ContentState) -> dict:
    """Human-in-the-loop: human decides when too many revisions occur."""
    print(f"\n=== HUMAN REVIEW ===")
    print(f"Topic: {state['topic']}")
    print(f"Revisions: {state.get('revision_count', 0)}")
    print(f"Reviewer feedback: {state.get('feedback', 'None')}")
    print(f"\nLatest draft:\n{state.get('draft', '')[:500]}...")
    
    approval = input("\nApprove article? (y/n): ")
    return {
        "approved": approval.lower() == "y",
        "messages": [f"Human: {'Approved' if approval.lower() == 'y' else 'Rejected'}"]
    }

Step 3: Define Routing Functions

from langgraph.graph import END

def route_after_review(state: ContentState) -> str:
    """Route after review with an emergency stop."""
    if state.get("approved"):
        return "publisher"
    
    if state.get("revision_count", 0) >= 3:
        return "human_review"
    
    return "writer"  # Back to the writer

def route_after_human(state: ContentState) -> str:
    """Route after human review."""
    if state.get("approved"):
        return "publisher"
    return END  # Stop if human rejects

Step 4: Assemble the Graph

from langgraph.graph import StateGraph

# Create the graph
workflow = StateGraph(ContentState)

# Add all nodes
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
workflow.add_node("publisher", publisher)
workflow.add_node("human_review", human_review)

# Define edges
workflow.set_entry_point("researcher")

# researcher → writer (always)
workflow.add_edge("researcher", "writer")

# writer → reviewer (always)
workflow.add_edge("writer", "reviewer")

# reviewer → conditional (writer, publisher, or human_review)
workflow.add_conditional_edges(
    "reviewer",
    route_after_review,
    {
        "writer": "writer",
        "publisher": "publisher",
        "human_review": "human_review"
    }
)

# human_review → conditional (publisher or END)
workflow.add_conditional_edges(
    "human_review",
    route_after_human,
    {
        "publisher": "publisher",
        END: END
    }
)

# publisher → END (always)
workflow.add_edge("publisher", END)

# Compile the graph
app = workflow.compile()

Step 5: Execute

# Initial state
initial_state = {
    "topic": "AI Programming: Best Practices 2026",
    "research": "",
    "draft": "",
    "final_article": "",
    "messages": [],
    "revision_count": 0,
    "approved": False,
    "feedback": None
}

# Run the graph
result = app.invoke(initial_state)

# Display the result
print("\n" + "=" * 60)
print("FINAL ARTICLE:")
print("=" * 60)
print(result["final_article"])
print("\n" + "=" * 60)
print(f"Revisions: {result['revision_count']}")
print(f"Message history:")
for msg in result["messages"]:
    print(f"  - {msg}")

What’s happening here?

1. Researcher gathers information on the topic
2. Writer drafts an initial version based on the research
3. Reviewer evaluates the draft against 4 criteria
4. If not approved: loop back to Writer with feedback
5. After 3 failed revisions: escalate to Human Review (the safety net)
6. If approved: Publisher formats the final article
7. END — the article is ready

The entire workflow is a graph with a cycle (writer ↔ reviewer) and two conditional edges (reviewer and human_review).

Advanced Patterns

Memory and Persistence with Checkpointing

LangGraph can persist the state after each node executes (checkpointing). This enables:

• Recovery from crashes
• Resuming paused workflows
• Conversation history across multiple sessions

from langgraph.checkpoint.memory import MemorySaver
# For persistence across restarts:
# from langgraph.checkpoint.sqlite import SqliteSaver

# In-memory checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Execute with a thread ID (essential for memory)
result = app.invoke(
    initial_state,
    config={"configurable": {"thread_id": "article-ai-programming"}}
)

# Resume later (e.g., after human approval)
result2 = app.invoke(
    {"approved": True},  # Just send the update
    config={"configurable": {"thread_id": "article-ai-programming"}}
)

How it works: The checkpointer saves the state after each node executes. With the thread_id, you can resume the same thread later. This is especially useful for human-in-the-loop scenarios, where the workflow pauses waiting for a human decision.

Parallel Execution

You can run multiple nodes in parallel and then merge their results:

def researcher_general(state: ContentState) -> dict:
    """Research general information."""
    response = llm.invoke(f"General info about: {state['topic']}")
    return {"research_general": response.content}

def researcher_technical(state: ContentState) -> dict:
    """Research technical details."""
    response = llm.invoke(f"Technical details about: {state['topic']}")
    return {"research_technical": response.content}

def merge_research(state: ContentState) -> dict:
    """Merge both research results."""
    combined = f"General:\n{state['research_general']}\n\nTechnical:\n{state['research_technical']}"
    return {"research": combined}

# In the graph:
workflow.add_node("researcher_general", researcher_general)
workflow.add_node("researcher_technical", researcher_technical)
workflow.add_node("merge", merge_research)

# Both researchers in parallel from the start
workflow.set_entry_point("researcher_general")
workflow.add_edge("researcher_general", "researcher_technical")
# Actually parallel: both from entry, both → merge
# LangGraph automatically merges parallel nodes

Tool Integration — Agents with Tools

Agents become powerful when they can use tools. LangGraph integrates seamlessly with LangChain tools:

from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun

# Web search tool
search = DuckDuckGoSearchRun()

def researcher_with_tools(state: ContentState) -> dict:
    """Research agent with web search."""
    topic = state["topic"]
    
    # Search first
    search_results = search.run(f"{topic} 2026 latest developments")
    
    # Feed results to the LLM
    prompt = f"""Based on these search results, create a 
    structured research document about {topic}:
    
    Search results: {search_results}"""
    
    response = llm.invoke(prompt)
    return {"research": response.content}

Sub-Graphs — Graphs Within Graphs

You can use one LangGraph as a node in another LangGraph. This is useful for modular architectures:

# Sub-graph for research
research_graph = StateGraph(ContentState)
research_graph.add_node("web_search", web_searcher)
research_graph.add_node("summarize", summarizer)
research_graph.set_entry_point("web_search")
research_graph.add_edge("web_search", "summarize")
research_graph.add_edge("summarize", END)
research_app = research_graph.compile()

# Main graph uses sub-graph as a node
main_workflow = StateGraph(ContentState)
main_workflow.add_node("research", research_app)  # Sub-graph as a node!
main_workflow.add_node("write", writer)
main_workflow.set_entry_point("research")
main_workflow.add_edge("research", "write")
main_workflow.add_edge("write", END)

Streaming — Real-Time Results

LangGraph supports streaming so you see intermediate results as the graph executes:

# Stream all node updates
for event in app.stream(initial_state):
    for node_name, node_output in event.items():
        print(f"[{node_name}] → {list(node_output.keys())} updated")

# Stream only the messages
for event in app.stream(
    initial_state,
    stream_mode="values"
):
    messages = event.get("messages", [])
    if messages:
        print(f"Latest: {messages[-1]}")

Visualizing the Graph

LangGraph can visualize your graph — invaluable for debugging:

# In a Jupyter Notebook:
from IPython.display import Image, display

# Render graph as an image
display(Image(app.get_graph().draw_mermaid_png()))

# In the terminal:
app.get_graph().print_ascii()

This gives you a visual representation of all nodes, edges, and conditional routes — essential when working with complex graphs.

LangGraph vs. CrewAI — Which Framework for What?

Aspect	LangGraph	CrewAI
Paradigm	State Graph (graph theory)	Role-Based Agents (team metaphor)
Flexibility	Very high — complete control over each step	Moderate — abstraction hides details
Learning Curve	Steep — requires understanding graph concepts	Shallow — “define agents, run tasks”
Control	Fine-grained — conditional edges, state management	Abstracted — framework makes many decisions
Debugging	Graph visualization, streaming, tracing with LangSmith	Logs, simple print statements
Cycles/Loops	Native support	Limited (max_iterations)
Memory/State	Native (checkpointing, state reduction)	Limited (memory parameters)
Human-in-the-Loop	Native (interrupt_before, interrupt_after)	Manually implemented
Best For	Complex workflows, production systems	Quick prototypes, simple agent teams
Community	LangChain ecosystem (very large)	Growing, but smaller

My recommendation:

• CrewAI for prototypes and simple agent teams (2-3 agents, linear workflows)
• LangGraph for production systems, complex workflows with loops, conditions, and human-in-the-loop
• Both for different projects — they’re not mutually exclusive

Best Practices from the Field

1. Keep state as minimal as possible

Not every piece of information belongs in state. Local variables within nodes are often sufficient. State should only hold data that truly needs to be shared between nodes.

2. Always set revision limits

Without a limit, a cycle can run indefinitely. Always add a counter and a circuit breaker:

MAX_REVISIONS = 5

def route_after_review(state: ContentState) -> str:
    if state.get("revision_count", 0) >= MAX_REVISIONS:
        return "human_review"  # or END
    # ...

3. Enable tracing with LangSmith

LangSmith (by LangChain) gives you detailed insights into every step:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

You’ll then see every LLM call, every state transition, and token calculations in the LangSmith dashboard.

4. Keep costs in mind

Each node makes at least one LLM call. With 3 agents and 3 revisions, that’s 9+ LLM calls. With GPT-4o, a single run can easily cost $5-10. Use cheaper models for simple tasks (e.g., GPT-4o-mini for the reviewer).

5. Use local models for development

For development and testing, use local models (Ollama) to reduce costs:

from langchain_community.chat_models import ChatOllama

llm = ChatOllama(model="llama3.1:8b", temperature=0)

Common Problems and Solutions

Problem: “Graph doesn’t have an entry point”

Solution: Call workflow.set_entry_point("node_name") before compiling.

Problem: “Node ‘xyz’ not found”

Solution: Every node referenced in edges must be added with add_node before edges are defined.

Problem: Infinite loop

Solution: Conditional edge always returns the same node. Check your routing function — it must return different values based on state.

Problem: State not updating

Solution: Node returns a partial update, but the field uses an Annotated reducer. Verify the reducer is defined correctly. With Annotated[List[str], add], the node must return a list to be added.

Key Exam Points

• LangGraph: Framework for stateful multi-agent applications based on graph theory
• 4 Core Concepts: State (shared state), Nodes (agent functions), Edges (transitions), Conditional Edges (branching logic)
• Cycles: Loops enable iterative refinement (Writer ↔ Reviewer)
• State Reduction: Annotated reducers (e.g., add) for accumulating state fields
• Human-in-the-Loop: Pause workflows for human decisions
• Checkpointing: Save state after each node, restore on failure
• Parallel Execution: Run multiple nodes simultaneously, auto-merge results
• Sub-Graphs: Embed one graph as a node in another — modular architecture
• Streaming: Real-time updates during graph execution
• LangSmith: Tracing and debugging for LangGraph workflows
• Comparison: LangGraph (flexible, complex, production-ready) vs. CrewAI (simple, fast, prototyping)

FAQ

Do I need LangChain experience to use LangGraph? Yes, basic LangChain knowledge is helpful since LangGraph builds on LangChain (same LLM integration, same tools, same runnables). If you’re new to LangChain, start with LangChain fundamentals first.

Is LangGraph free? LangGraph itself is open source (MIT license). LLM costs (OpenAI, Anthropic) are separate. LangSmith (tracing) has a free tier but requires payment for heavy use.

Can LangGraph work with local models? Yes, via Ollama, llama.cpp, or LM Studio. This is especially recommended for development and testing to avoid API costs. For production, cloud models typically offer better quality.

How do I debug a LangGraph workflow? Three tools: 1) app.get_graph().draw_mermaid_png() for visual graph representation. 2) Streaming with app.stream() for real-time updates. 3) LangSmith for detailed tracing of every LLM call.

What’s the difference between LangGraph and LangChain? LangChain is a general framework for LLM applications (chains, agents, tools). LangGraph specializes in stateful, multi-actor workflows with graph structure. LangGraph builds on LangChain but doesn’t replace it.

Can I use LangGraph in production? Yes, LangGraph is production-ready. With checkpointing (persistence), error handling, and LangSmith tracing, you have everything needed for reliable production systems. LangGraph Cloud also offers managed hosting for LangGraph applications.

Recommended Reading