Multi-Agent Systems with LangGraph

Introduction

30 Second Summary

When you have a complex task requiring research, analysis, and reporting, you don't assign one person to do everything. You build a team of specialists, and the coordination pattern you choose determines whether you get a reliable pipeline or an infinite loop.

In this project, you will build three distinct multi-agent architectures using LangGraph. You'll run all three against the same research question so you can directly compare how each pattern handles coordination, state, and failure.

What You'll Build

Three complete multi-agent systems implementing different architecture patterns (Supervisor, Swarm, and Plan-and-Execute), all solving the same research task for direct side-by-side comparison.

By the end of this project, you'll have:

• Three working multi-agent architectures (Supervisor, Swarm, Plan-and-Execute) built side-by-side so you can run the same question through each and compare trade-offs.
• Production-grade guardrails on every pattern including deterministic iteration caps, handoff circuit breakers, durable checkpointing, and human-in-the-loop approval gates.
• A decision framework for choosing the right architecture based on your use case, failure modes, and token efficiency requirements.
• Secret Mission: A composite Supervisor + Reflection Loop architecture showing how real production systems combine multiple patterns.

Are there any prerequisites?

This project assumes solid Python experience, familiarity with building multi-agent systems (AutoGen or similar), and an understanding of LLM tool calling. You'll also need an OpenAI API key with access to GPT-4o.

Set Up Your Environment

Before you can build multi-agent systems, you need a clean Python environment with all the LangGraph packages installed and an OpenAI API connection verified.

In this step, you will install Python 3.11+, create an isolated virtual environment, install six packages, and confirm that your API key can reach the OpenAI models.

In this step, get ready to:

• Install and verify Python 3.11+.
• Create a virtual environment and install LangGraph packages.
• Connect to the OpenAI API and verify connectivity.

Install and verify Python 3.11+

This project requires Python 3.11 or higher because the langgraph-supervisor and langgraph-swarm packages require Python 3.10+ and several type features used in the code were stabilized in 3.11.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir langgraph-agents
cd langgraph-agents

• Confirm the folder exists by running:

ls

You should see the langgraph-agents folder listed if you run ls from the Desktop. Since you already moved into it, you are now inside the empty project folder.

• Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

✔️ I see version 3.11 or higher

Your Python version is ready. Move on to the next substep.

ⓧ I see an older version

Your system has Python installed, but it is below 3.11. You need to install a newer version.

• Visit python.org/downloads and download the latest Python 3.11+ installer for your operating system.
• Run the installer. On Windows, check the Add python.exe to PATH box before clicking Install Now.
• Close and reopen your terminal after installation.
• Run the version check command again to confirm it now shows 3.11 or higher.

macOS users with Homebrew

If you installed Python via Homebrew, your system python3 may still point to an older version. After installing a newer Python, restart your terminal and run python3 --version again. If it still shows the old version, try running python3.11 --version or python3.12 --version to confirm the new install is available.

ⓧ Command not found

Python is not installed on your machine. You need to install it.

• Visit python.org/downloads and download the latest Python 3.11+ installer for your operating system.
• Run the installer. On Windows, check the Add python.exe to PATH box before clicking Install Now.
• Close and reopen your terminal after installation.
• Run the version check command again to confirm it shows 3.11 or higher.

Create your virtual environment and install packages

A virtual environment keeps your project's packages isolated from other Python projects on your machine. This prevents version conflicts between projects.

• Create a virtual environment by running this command:

macOS

python3 -m venv .venv

Windows

python -m venv .venv

• Activate the virtual environment by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (.venv) in your prompt?

• Make sure you ran the activation command from inside the langgraph-agents folder where you created the .venv directory.
• On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try activating again.
• help me activate my Python virtual environment

• Upgrade pip to the latest version by running:

pip install --upgrade pip

• Install all six project packages by running:

pip install langgraph langgraph-supervisor langgraph-swarm langchain-openai langchain-core pydantic

What are these packages?

• langgraph is the core framework for building stateful multi-agent graphs.
• langgraph-supervisor provides the create_supervisor helper for hierarchical agent orchestration.
• langgraph-swarm provides create_swarm and create_handoff_tool for peer-to-peer agent communication.
• langchain-openai connects LangChain to OpenAI's models.
• langchain-core provides base abstractions like tools and messages.
• pydantic handles structured output validation for the Plan-and-Execute pattern.

The install will take 30-60 seconds. You should see a success message ending with Successfully installed followed by a list of package names.

Seeing install errors?

• Confirm your virtual environment is active (you should see (.venv) in your prompt). If not, run the activation command again.
• If you see a permissions error, make sure you are NOT using sudo with pip inside a virtual environment.
• If a specific package fails to build, check that your Python version is 3.11+ by running python --version inside the activated environment.
• help me fix pip install errors for langgraph packages

Set your OpenAI API key and verify connectivity

The agents in this project use OpenAI's GPT-4o and GPT-4o-mini models. You need an API key to authenticate requests.

• If you do not already have an OpenAI API key, visit platform.openai.com/api-keys and create one.
• Set the OPENAI_API_KEY environment variable by running this command (replace your-api-key-here with your actual key):

macOS

export OPENAI_API_KEY=your-api-key-here

Windows

set OPENAI_API_KEY=your-api-key-here

Why not save the key in a file?

Setting the key as an environment variable keeps it out of your source code. If you accidentally commit an API key to version control, anyone who finds it can use your OpenAI credits. Environment variables are the standard approach for secrets.

• Verify that your API key works and all packages are installed correctly by running this test command:

python -c "from langchain_openai import ChatOpenAI; print(ChatOpenAI(model='gpt-4o-mini').invoke('Say hello in exactly three words').content)"

You should see a short three-word greeting printed to your terminal. The exact words will vary, but any response confirms your packages and API key are working.

Seeing an authentication error?

• Double-check that you copied the full API key with no extra spaces. Run echo $OPENAI_API_KEY (macOS) or echo %OPENAI_API_KEY% (Windows) to confirm the value is set.
• Make sure your OpenAI account has billing enabled and available credits.
• If you see an import error instead of an auth error, your packages did not install correctly. Re-run the pip install command from the previous substep.
• help me troubleshoot my OpenAI API key connection

Your environment is ready. Next up, you will define the shared state and reusable tools that all three architecture patterns will use.

Define Shared State and Tools

With your environment ready, it is time to build the foundation that all three architecture patterns share.

Every multi-agent system needs two things: typed state that flows between nodes, and tools that agents can call. In LangGraph, state is a Python TypedDict with annotated fields. The annotations control how values merge when multiple nodes write to the same key.

In this step, get ready to:

• Define typed state schemas for Supervisor, Swarm, and Plan-and-Execute patterns.
• Build four reusable tool functions that all patterns share.

Create the shared state definitions

Each architecture pattern needs its own state shape. The Supervisor tracks approval status and iteration counts. The Swarm tracks which agent is currently active and how many handoffs have occurred. Plan-and-Execute tracks a plan, past steps, and a final response.

• Create a new file called state.py in your project root.
• Add the imports, routing enum, and the main AgentState TypedDict by pasting this code:

from typing import TypedDict, Annotated, Literal
from enum import Enum
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage


class RouteDecision(str, Enum):
    RESEARCH = "research"
    ANALYSIS = "analysis"
    HUMAN_REVIEW = "human_review"
    FINISH = "finish"


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_task: str
    research_results: str
    analysis_output: str
    iteration_count: int
    approval_status: Literal["pending", "approved", "rejected"]

What does this code do?

The add_messages annotation on the messages field tells LangGraph to append new messages to the list rather than overwriting it. Every node that returns a message automatically adds it to the conversation history.

The RouteDecision enum gives the supervisor deterministic routing options enforced in Python. The LLM picks a route, but code validates that the choice is one of these four values.

The iteration_count and approval_status fields let you enforce iteration caps and human-in-the-loop gates in later steps.

• Below AgentState, add the two pattern-specific state classes:

class SwarmState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    active_agent: str
    handoff_count: int


class PlanExecuteState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    input: str
    plan: list[str]
    past_steps: list[tuple[str, str]]
    response: str
    iteration_count: int

What do these states track?

SwarmState tracks the active_agent (which peer currently holds control) and handoff_count (how many times control has transferred). You will use handoff_count as a circuit breaker to prevent infinite ping-pong loops.

PlanExecuteState holds a structured plan (list of steps), past_steps (completed work with results), and a response field for the final output. The replanner reads past_steps to decide whether to continue or deliver.

• Save state.py.
• Confirm the state definitions load correctly by running:

python -c "from state import AgentState, SwarmState, PlanExecuteState; print('All states loaded successfully')"

You should see All states loaded successfully printed in your terminal.

Seeing an ImportError?

• Make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.
• Confirm you are running the command from the same directory where state.py lives.
• Check that langgraph and langchain-core are installed by running pip list | grep langgraph.

help me debug this ImportError in state.py

✔️ Awesome, I've got everything!

Great. Double check you have saved state.py before moving on.

ⓧ I'd like to double check the full code

from typing import TypedDict, Annotated, Literal
from enum import Enum
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage


class RouteDecision(str, Enum):
    RESEARCH = "research"
    ANALYSIS = "analysis"
    HUMAN_REVIEW = "human_review"
    FINISH = "finish"


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_task: str
    research_results: str
    analysis_output: str
    iteration_count: int
    approval_status: Literal["pending", "approved", "rejected"]


class SwarmState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    active_agent: str
    handoff_count: int


class PlanExecuteState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    input: str
    plan: list[str]
    past_steps: list[tuple[str, str]]
    response: str
    iteration_count: int

Build reusable tool functions

All three patterns use the same specialist tools. In production, these would call real APIs. For this project, they return simulated data so you can focus on the architecture patterns without spending API credits on search or summarization calls.

• Create a new file called tools.py in your project root.
• Add the import and the first two research-focused tools by pasting this code:

from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """Search the web for information on a given topic."""
    return (
        f"Search results for '{query}':\n"
        "1. LangGraph is the production standard for multi-agent systems in 2026, "
        "adopted by Uber, LinkedIn, Klarna, Replit, and AppFolio.\n"
        "2. AutoGen has moved to maintenance mode and is not recommended for new production work.\n"
        "3. LangGraph provides durable state checkpointing (Postgres/Redis), "
        "deterministic code-level routing, and native MCP support.\n"
        "4. Key patterns: Supervisor, Hierarchical, Swarm, Reflection Loop, Plan-and-Execute.\n"
        "5. The supervisor pattern is recommended for 4-6 specialist agents with a single orchestrator."
    )


@tool
def summarize_text(text: str) -> str:
    """Summarize a given text into key bullet points."""
    return (
        "Summary of provided text:\n"
        "- LangGraph is production-standard for multi-agent systems\n"
        "- AutoGen is in maintenance mode\n"
        "- Key differentiators: checkpointing, deterministic routing, MCP support\n"
        "- Supervisor pattern recommended for small-to-medium teams\n"
        "- Hierarchical pattern for 6+ specialists"
    )

What does the @tool decorator do?

The @tool decorator from langchain_core.tools wraps a plain Python function into a LangChain Tool object. It automatically extracts the function's name, docstring, and type hints to generate the tool schema that LLMs use for tool calling.

The web_search tool simulates a web search returning structured findings. The summarize_text tool simulates condensing raw research into bullet points.

• Below the summarize_text function, add the two analysis-focused tools:

@tool
def analyze_data(data: str) -> str:
    """Analyze data and extract structured insights."""
    return (
        "Analysis Results:\n"
        "| Criterion | LangGraph | AutoGen |\n"
        "| Checkpointing | Full (Postgres/Redis) | None |\n"
        "| Routing | Deterministic (code) | String-match (LLM) |\n"
        "| MCP Support | Native | None |\n"
        "| Status | Active production | Maintenance mode |\n"
        "| Cost Control | Tiered model + iteration cap | Uncontrolled loops |"
    )


@tool
def format_report(content: str) -> str:
    """Format analysis content into a structured report."""
    return (
        "# Research Report: LangGraph vs AutoGen\n\n"
        "## Executive Summary\n"
        "LangGraph has emerged as the production standard for multi-agent systems.\n\n"
        "## Key Findings\n"
        f"{content}\n\n"
        "## Recommendation\n"
        "For all net-new production workloads, default to LangGraph. "
        "AutoGen should only be considered as a migration target for existing systems.\n\n"
        "## Methodology\n"
        "Comparative analysis across checkpointing, routing, protocol support, "
        "and production deployment patterns."
    )

What do these analysis tools do?

The analyze_data tool returns a structured comparison table. In a real system, this would run quantitative analysis over collected data.

The format_report tool accepts raw content and wraps it in a professional report structure with sections for executive summary, findings, recommendations, and methodology.

• Save tools.py.
• Confirm all four tools load correctly by running:

python -c "from tools import web_search, summarize_text, analyze_data, format_report; print('All tools loaded successfully')"

You should see All tools loaded successfully printed in your terminal.

Getting a NameError or ImportError?

• Make sure the file is named exactly tools.py (not tool.py or Tools.py).
• Check that langchain-core is installed. Run pip show langchain-core to confirm.
• Ensure each function has the @tool decorator directly above it with no blank lines between the decorator and the def line.

help me fix this tools.py import error

✔️ Awesome, I've got everything!

Great. Double check you have saved tools.py before moving on.

ⓧ I'd like to double check the full code

from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """Search the web for information on a given topic."""
    return (
        f"Search results for '{query}':\n"
        "1. LangGraph is the production standard for multi-agent systems in 2026, "
        "adopted by Uber, LinkedIn, Klarna, Replit, and AppFolio.\n"
        "2. AutoGen has moved to maintenance mode and is not recommended for new production work.\n"
        "3. LangGraph provides durable state checkpointing (Postgres/Redis), "
        "deterministic code-level routing, and native MCP support.\n"
        "4. Key patterns: Supervisor, Hierarchical, Swarm, Reflection Loop, Plan-and-Execute.\n"
        "5. The supervisor pattern is recommended for 4-6 specialist agents with a single orchestrator."
    )


@tool
def summarize_text(text: str) -> str:
    """Summarize a given text into key bullet points."""
    return (
        "Summary of provided text:\n"
        "- LangGraph is production-standard for multi-agent systems\n"
        "- AutoGen is in maintenance mode\n"
        "- Key differentiators: checkpointing, deterministic routing, MCP support\n"
        "- Supervisor pattern recommended for small-to-medium teams\n"
        "- Hierarchical pattern for 6+ specialists"
    )


@tool
def analyze_data(data: str) -> str:
    """Analyze data and extract structured insights."""
    return (
        "Analysis Results:\n"
        "| Criterion | LangGraph | AutoGen |\n"
        "| Checkpointing | Full (Postgres/Redis) | None |\n"
        "| Routing | Deterministic (code) | String-match (LLM) |\n"
        "| MCP Support | Native | None |\n"
        "| Status | Active production | Maintenance mode |\n"
        "| Cost Control | Tiered model + iteration cap | Uncontrolled loops |"
    )


@tool
def format_report(content: str) -> str:
    """Format analysis content into a structured report."""
    return (
        "# Research Report: LangGraph vs AutoGen\n\n"
        "## Executive Summary\n"
        "LangGraph has emerged as the production standard for multi-agent systems.\n\n"
        "## Key Findings\n"
        f"{content}\n\n"
        "## Recommendation\n"
        "For all net-new production workloads, default to LangGraph. "
        "AutoGen should only be considered as a migration target for existing systems.\n\n"
        "## Methodology\n"
        "Comparative analysis across checkpointing, routing, protocol support, "
        "and production deployment patterns."
    )

Your shared foundation is in place. Next up, you will wire these tools into specialist agents and build the first architecture pattern: the Supervisor.

Build the Supervisor Pattern

You have state definitions and tools ready. Now it's time to build the first architecture pattern: the Supervisor.

The Supervisor pattern uses a central orchestrator that delegates tasks to specialist agents. The LLM decides WHAT to do, but deterministic Python code decides HOW and WHEN TO STOP. This is the biggest architectural shift from AutoGen's GroupChat model, where all agents see all messages and conversation flow is LLM-driven.

In this step, get ready to:

• Create four specialist agents with focused tool assignments.
• Wire agents into a hierarchical supervisor with sub-teams and checkpointing.
• Run the supervisor system end-to-end with a research question.

Create the specialist agents

Before you can build a supervisor, you need the specialist agents it will manage. Each agent wraps a single tool and has a focused system prompt that constrains its behavior.

• Create a new file called agents.py in your project folder.
• Add the imports, model setup, and the two research team agents by pasting this code:

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from tools import web_search, summarize_text, analyze_data, format_report

model = ChatOpenAI(model="gpt-4o")

# Research team: web_researcher gathers raw data, summarizer condenses it
web_researcher = create_react_agent(
    model=model,
    tools=[web_search],
    name="web_researcher",
    prompt=(
        "You are a web research specialist. Your job is to search for information "
        "on the given topic and return comprehensive raw findings. "
        "Always use your web_search tool. Do not analyze or summarize. "
        "Defer summarization to the summarizer agent."
    ),
)

summarizer = create_react_agent(
    model=model,
    tools=[summarize_text],
    name="summarizer",
    prompt=(
        "You are a summarization specialist. Your job is to take raw research "
        "findings and produce concise, structured summaries with key bullet points. "
        "Always use your summarize_text tool. Do not do original research. "
        "Defer research to the web_researcher agent."
    ),
)

What does this code do?

Each agent is created with create_react_agent which gives it a ReAct loop: the agent reasons about what to do, calls its tool, observes the result, and decides if it's done.

The prompt constrains each agent's scope. The web_researcher is explicitly told not to summarize. The summarizer is told not to research. This separation of concerns prevents agents from stepping on each other's responsibilities.

• Below the summarizer agent, add the two analysis team agents:

# Analysis team: data_analyst extracts insights, report_writer formats output
data_analyst = create_react_agent(
    model=model,
    tools=[analyze_data],
    name="data_analyst",
    prompt=(
        "You are a data analysis specialist. Your job is to take summarized "
        "research and extract structured comparisons, metrics, and insights. "
        "Always use your analyze_data tool. Do not write final reports. "
        "Defer report writing to the report_writer agent."
    ),
)

report_writer = create_react_agent(
    model=model,
    tools=[format_report],
    name="report_writer",
    prompt=(
        "You are a report writing specialist. Your job is to take analyzed data "
        "and produce a well-structured, professional report with sections for "
        "executive summary, key findings, recommendations, and methodology. "
        "Always use your format_report tool. Do not do original research or analysis. "
        "Defer those to the appropriate specialists."
    ),
)

Why constrained prompts matter

Each prompt ends with explicit "defer to X" instructions. Without these boundaries, GPT-4o will happily do everything in one agent, defeating the purpose of a multi-agent system.

The four agents split into two natural teams: a Research Team (web_researcher + summarizer) and an Analysis Team (data_analyst + report_writer). You will wire these teams in the next substep.

• Save agents.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved agents.py before moving on.

ⓧ I'd like to double check the full code

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from tools import web_search, summarize_text, analyze_data, format_report

model = ChatOpenAI(model="gpt-4o")

web_researcher = create_react_agent(
    model=model,
    tools=[web_search],
    name="web_researcher",
    prompt=(
        "You are a web research specialist. Your job is to search for information "
        "on the given topic and return comprehensive raw findings. "
        "Always use your web_search tool. Do not analyze or summarize. "
        "Defer summarization to the summarizer agent."
    ),
)

summarizer = create_react_agent(
    model=model,
    tools=[summarize_text],
    name="summarizer",
    prompt=(
        "You are a summarization specialist. Your job is to take raw research "
        "findings and produce concise, structured summaries with key bullet points. "
        "Always use your summarize_text tool. Do not do original research. "
        "Defer research to the web_researcher agent."
    ),
)

data_analyst = create_react_agent(
    model=model,
    tools=[analyze_data],
    name="data_analyst",
    prompt=(
        "You are a data analysis specialist. Your job is to take summarized "
        "research and extract structured comparisons, metrics, and insights. "
        "Always use your analyze_data tool. Do not write final reports. "
        "Defer report writing to the report_writer agent."
    ),
)

report_writer = create_react_agent(
    model=model,
    tools=[format_report],
    name="report_writer",
    prompt=(
        "You are a report writing specialist. Your job is to take analyzed data "
        "and produce a well-structured, professional report with sections for "
        "executive summary, key findings, recommendations, and methodology. "
        "Always use your format_report tool. Do not do original research or analysis. "
        "Defer those to the appropriate specialists."
    ),
)

Build the hierarchical supervisor with sub-teams

A flat supervisor managing four agents works, but it forces the top-level LLM to make granular routing decisions at every step. A hierarchical structure is more scalable: each sub-team has its own supervisor that handles internal coordination.

• Create a new file called pattern_supervisor.py in your project folder.
• Add the imports and build the two sub-team supervisors by pasting this code:

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.checkpoint.memory import InMemorySaver
from agents import web_researcher, summarizer, data_analyst, report_writer

model = ChatOpenAI(model="gpt-4o")

# Build sub-team supervisors
research_team = create_supervisor(
    [web_researcher, summarizer],
    model=model,
    prompt=(
        "You are the Research Team supervisor. You manage a web researcher and a summarizer. "
        "For any research task: first delegate to web_researcher to gather raw information, "
        "then delegate to summarizer to produce a structured summary. "
        "Return the final summary to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="research_team")

analysis_team = create_supervisor(
    [data_analyst, report_writer],
    model=model,
    prompt=(
        "You are the Analysis Team supervisor. You manage a data analyst and a report writer. "
        "For any analysis task: first delegate to data_analyst to extract structured insights, "
        "then delegate to report_writer to produce a formatted report. "
        "Return the final report to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="analysis_team")

What does this code do?

Each sub-team is itself a supervisor created with create_supervisor. The output_mode="last_message" setting means only the final response from each sub-team flows up to the parent. Without this, the parent supervisor would see every internal message, wasting tokens and context.

The .compile(name="research_team") call converts the StateGraph into a runnable graph. Compiled sub-teams can then be passed as "agents" to a parent supervisor to create multi-level hierarchies.

• Below the sub-teams, add the top-level supervisor and compile it with a checkpointer:

# Build top-level supervisor
top_supervisor = create_supervisor(
    [research_team, analysis_team],
    model=model,
    prompt=(
        "You are the top-level supervisor managing two teams: "
        "research_team and analysis_team. "
        "For any research question:\n"
        "1. First delegate to research_team to gather and summarize information.\n"
        "2. Then delegate to analysis_team to analyze findings and produce a report.\n"
        "3. Return the final report to the user.\n"
        "Always follow this sequence. Do not skip steps."
    ),
    supervisor_name="top_supervisor",
)

checkpointer = InMemorySaver()
app = top_supervisor.compile(
    name="hierarchical_research_system",
    checkpointer=checkpointer,
)

Why use a checkpointer?

The InMemorySaver checkpointer stores every state snapshot in memory. This enables features you will use later: human-in-the-loop interrupts, pause-resume flows, and state inspection. In production, you would swap this for a Postgres or Redis checkpointer for durability across server restarts.

• Below the compiled app, add the run_supervisor function and the __main__ block:

def run_supervisor(question: str) -> dict:
    """Run the supervisor pattern."""
    config = {"configurable": {"thread_id": "supervisor-session-1"}}

    print("\n[SUPERVISOR PATTERN]")
    print(f"Question: {question}\n")

    result = app.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=config,
        recursion_limit=25,
    )

    final_msg = result["messages"][-1].content
    print(f"Result: {final_msg[:300]}...")
    print(f"Total messages: {len(result['messages'])}")
    return result


if __name__ == "__main__":
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    run_supervisor(question)

What does this code do?

The thread_id in the config lets the checkpointer track this conversation session. If you invoke the graph again with the same thread_id, it resumes from the last saved state.

The recursion_limit=25 is a deterministic Python-level cap on how many graph nodes can execute. This is the key difference from AutoGen: iteration limits are enforced in code, not in prompts. A prompt saying "stop after 5 turns" will be ignored. A Python recursion_limit will not.

• Save pattern_supervisor.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved pattern_supervisor.py before running it.

ⓧ I'd like to double check the full code

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.checkpoint.memory import InMemorySaver
from agents import web_researcher, summarizer, data_analyst, report_writer

model = ChatOpenAI(model="gpt-4o")

# Build sub-team supervisors
research_team = create_supervisor(
    [web_researcher, summarizer],
    model=model,
    prompt=(
        "You are the Research Team supervisor. You manage a web researcher and a summarizer. "
        "For any research task: first delegate to web_researcher to gather raw information, "
        "then delegate to summarizer to produce a structured summary. "
        "Return the final summary to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="research_team")

analysis_team = create_supervisor(
    [data_analyst, report_writer],
    model=model,
    prompt=(
        "You are the Analysis Team supervisor. You manage a data analyst and a report writer. "
        "For any analysis task: first delegate to data_analyst to extract structured insights, "
        "then delegate to report_writer to produce a formatted report. "
        "Return the final report to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="analysis_team")

# Build top-level supervisor
top_supervisor = create_supervisor(
    [research_team, analysis_team],
    model=model,
    prompt=(
        "You are the top-level supervisor managing two teams: "
        "research_team and analysis_team. "
        "For any research question:\n"
        "1. First delegate to research_team to gather and summarize information.\n"
        "2. Then delegate to analysis_team to analyze findings and produce a report.\n"
        "3. Return the final report to the user.\n"
        "Always follow this sequence. Do not skip steps."
    ),
    supervisor_name="top_supervisor",
)

checkpointer = InMemorySaver()
app = top_supervisor.compile(
    name="hierarchical_research_system",
    checkpointer=checkpointer,
)


def run_supervisor(question: str) -> dict:
    """Run the supervisor pattern."""
    config = {"configurable": {"thread_id": "supervisor-session-1"}}

    print("\n[SUPERVISOR PATTERN]")
    print(f"Question: {question}\n")

    result = app.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=config,
        recursion_limit=25,
    )

    final_msg = result["messages"][-1].content
    print(f"Result: {final_msg[:300]}...")
    print(f"Total messages: {len(result['messages'])}")
    return result


if __name__ == "__main__":
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    run_supervisor(question)

Run the supervisor system end-to-end

Time to see the hierarchical supervisor in action. The top-level supervisor will delegate to the research team first, then the analysis team, producing a structured report.

• Run the supervisor pattern by executing this command:

python pattern_supervisor.py

• You should see output that starts with [SUPERVISOR PATTERN] followed by the question, then a result containing a research report and a total message count.

What should I see?

The output shows the final message from the graph (truncated to 300 characters) and the total number of messages exchanged. The message count reflects every handoff between supervisors and agents. A typical run produces 10-20 messages as the top supervisor delegates to research_team, which delegates to web_researcher then summarizer, then back up to the top supervisor which delegates to analysis_team.

Seeing an error?

If you see an authentication error, verify your OPENAI_API_KEY is set correctly in your terminal session. Run echo $OPENAI_API_KEY to confirm it has a value.

If you see a GraphRecursionError, the model may be looping. This can happen if the prompt is not explicit enough. Double check that your supervisor prompts match the code above exactly.

If you see an ImportError for agents, make sure agents.py is saved in the same directory as pattern_supervisor.py.

help me debug my supervisor pattern error

Why this approach over a flat supervisor

A flat supervisor managing all four agents directly requires 4-way routing decisions at every step. The hierarchical approach reduces this to 2-way decisions at the top level, with 2-way decisions within each sub-team. This scales better as you add more agents.

The output_mode="last_message" on sub-teams means the top supervisor only sees final results, not internal chatter. This keeps the top supervisor's context window lean and its routing decisions accurate.

Your first architecture pattern is running. Next up, you will build the Swarm pattern where agents coordinate without any central supervisor at all.

Build the Swarm Pattern

Your Supervisor pattern works by funneling every decision through a central orchestrator. That creates auditability, but it also creates a bottleneck.

The Swarm pattern removes the orchestrator entirely. Agents hand off directly to peers using tool calls, and the system tracks which agent is currently active. No supervisor, no bottleneck.

In this step, get ready to:

• Build four swarm agents with explicit peer handoff tools.
• Compile the swarm with a checkpointer and recursion guard against ping-pong loops.
• Run the swarm on the same research question and compare behavior to the Supervisor.

Create the swarm agents with handoff tools

In a swarm, each agent decides who to hand off to next. You express this by giving each agent a create_handoff_tool for every peer it can reach.

Why Swarm instead of Supervisor?

The Supervisor pattern routes every message through a central node. That node sees everything, which is great for audit trails, but adds latency and token cost on every turn.

The Swarm pattern eliminates the orchestrator. Agents talk directly to peers, which reduces total tokens by roughly 40% in multi-turn scenarios. The trade-off is observability: debugging emergent handoff chains is harder.

• Create a new file called pattern_swarm.py in your project folder.
• Add the imports, model setup, and the first agent (researcher_agent) by pasting this code:

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_handoff_tool, create_swarm
from langgraph.checkpoint.memory import InMemorySaver
from tools import web_search, summarize_text, analyze_data, format_report

model = ChatOpenAI(model="gpt-4o")

# Create agents with handoff tools for peer-to-peer communication
researcher_agent = create_react_agent(
    model=model,
    tools=[
        web_search,
        create_handoff_tool(
            agent_name="summarizer_agent",
            description="Hand off to summarizer when you have raw research findings that need summarization.",
        ),
    ],
    name="researcher_agent",
    prompt=(
        "You are a web research specialist in a swarm. Search for information "
        "using your web_search tool. Once you have findings, hand off to "
        "summarizer_agent for summarization. Do not summarize yourself."
    ),
)

What does this code do?

• create_handoff_tool generates a tool called transfer_to_summarizer_agent that the LLM can call. When called, it issues a Command that moves control to the named agent.
• The full message history is passed along with the handoff, so the next agent has complete context without needing a supervisor to relay it.
• Unlike the Supervisor pattern where agents were imported from agents.py, each swarm agent is defined fresh here because they need different tool sets (handoff tools specific to their peers).

• Now add the remaining three agents below the researcher_agent definition:

summarizer_agent = create_react_agent(
    model=model,
    tools=[
        summarize_text,
        create_handoff_tool(
            agent_name="analyst_agent",
            description="Hand off to analyst when summary is complete and needs structured analysis.",
        ),
        create_handoff_tool(
            agent_name="researcher_agent",
            description="Hand off back to researcher if more raw data is needed.",
        ),
    ],
    name="summarizer_agent",
    prompt=(
        "You are a summarization specialist in a swarm. Summarize the research "
        "findings using your summarize_text tool. Once done, hand off to "
        "analyst_agent for structured analysis."
    ),
)

analyst_agent = create_react_agent(
    model=model,
    tools=[
        analyze_data,
        create_handoff_tool(
            agent_name="writer_agent",
            description="Hand off to writer when analysis is complete and a report needs to be written.",
        ),
        create_handoff_tool(
            agent_name="summarizer_agent",
            description="Hand off back to summarizer if the data needs re-summarization.",
        ),
    ],
    name="analyst_agent",
    prompt=(
        "You are a data analysis specialist in a swarm. Analyze the summarized "
        "findings using your analyze_data tool. Once done, hand off to "
        "writer_agent for report formatting."
    ),
)

writer_agent = create_react_agent(
    model=model,
    tools=[format_report],
    name="writer_agent",
    prompt=(
        "You are a report writing specialist in a swarm. Format the analyzed "
        "data into a professional report using your format_report tool. "
        "You are the final agent in the chain. Produce the final output directly."
    ),
)

What does the handoff topology look like?

Each agent can only hand off to specific peers. This creates a directed graph of possible transitions:

• researcher_agent can hand off to summarizer_agent (forward only).
• summarizer_agent can hand off to analyst_agent (forward) or back to researcher_agent (backward, for more data).
• analyst_agent can hand off to writer_agent (forward) or back to summarizer_agent (backward).
• writer_agent has no handoff tools. It is the terminal agent that produces the final output.

The backward handoffs are what make ping-pong loops possible. You will guard against them with a recursion_limit in the next substep.

• Save pattern_swarm.py.

Seeing a TypeError about 'description'?

This happens when langchain-core is outdated. Make sure you installed the latest version in Step 1. You can upgrade it by running:

pip install --upgrade langchain-core

help me fix a TypeError from create_handoff_tool in my LangGraph swarm

Compile the swarm with a recursion guard

Backward handoffs can cause Agent A to hand off to Agent B, which hands back to Agent A indefinitely. In production you would track handoff_count in state (you already defined this in SwarmState). For this project, LangGraph's built-in recursion_limit acts as the circuit breaker.

• Add the swarm build and compile logic below your agent definitions in pattern_swarm.py:

# Build the swarm
workflow = create_swarm(
    [researcher_agent, summarizer_agent, analyst_agent, writer_agent],
    default_active_agent="researcher_agent",
)

checkpointer = InMemorySaver()
swarm_app = workflow.compile(checkpointer=checkpointer)

What does this code do?

• create_swarm takes the list of agents and builds a StateGraph with a built-in router that tracks the active_agent in state. When a handoff tool fires, the router updates active_agent and sends the next message to that agent's node.
• default_active_agent="researcher_agent" means new conversations always start at the researcher.
• The InMemorySaver checkpointer ensures the swarm remembers which agent was last active, so multi-turn conversations resume at the correct agent.

• Save pattern_swarm.py.

Run the swarm and observe autonomous routing

• Add the run_swarm function and __main__ block below the compile line in pattern_swarm.py:

def run_swarm(question: str) -> dict:
    """Run the swarm pattern."""
    config = {"configurable": {"thread_id": "swarm-session-1"}}

    print("\n[SWARM PATTERN]")
    print(f"Question: {question}\n")

    result = swarm_app.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=config,
        recursion_limit=30,
    )

    final_msg = result["messages"][-1].content
    print(f"Result: {final_msg[:300]}...")
    print(f"Total messages: {len(result['messages'])}")
    return result


if __name__ == "__main__":
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    run_swarm(question)

What does this code do?

• recursion_limit=30 is the circuit breaker. If agents bounce back and forth more than 30 times, LangGraph raises a GraphRecursionError and stops execution. This prevents runaway loops from burning through your API budget.
• The Supervisor used recursion_limit=25. The Swarm gets a slightly higher limit (30) because peer-to-peer handoffs are inherently more chatty than centralized routing.
• thread_id ties into the checkpointer. If you ran the same thread ID again, the swarm would resume from the last active agent rather than starting fresh at researcher_agent.

• Save pattern_swarm.py.
• Run the swarm by executing this command:

python pattern_swarm.py

• You should see [SWARM PATTERN] printed, followed by a research result and a total message count. Compare the message count to your Supervisor run. The swarm typically produces fewer total messages because there is no supervisor overhead on each turn.

Hitting a GraphRecursionError?

This means agents bounced between each other more than 30 times. This is the guard working as intended. If it happens consistently, check that writer_agent has no handoff tools. It must be the terminal node that produces output without handing off.

help me debug a GraphRecursionError in my LangGraph swarm

Extra for Experts: how does the swarm track active_agent?

Under the hood, create_swarm uses langgraph_swarm.SwarmState as its default state schema. This schema has an active_agent field that the handoff tool updates via a Command object. The router reads this field on every step to decide which agent node to invoke next.

✔️ Awesome, I've got everything!

Great. Double check you have saved pattern_swarm.py and that it ran without errors.

ⓧ I'd like to double check the full code

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_handoff_tool, create_swarm
from langgraph.checkpoint.memory import InMemorySaver
from tools import web_search, summarize_text, analyze_data, format_report

model = ChatOpenAI(model="gpt-4o")

# Create agents with handoff tools for peer-to-peer communication
researcher_agent = create_react_agent(
    model=model,
    tools=[
        web_search,
        create_handoff_tool(
            agent_name="summarizer_agent",
            description="Hand off to summarizer when you have raw research findings that need summarization.",
        ),
    ],
    name="researcher_agent",
    prompt=(
        "You are a web research specialist in a swarm. Search for information "
        "using your web_search tool. Once you have findings, hand off to "
        "summarizer_agent for summarization. Do not summarize yourself."
    ),
)

summarizer_agent = create_react_agent(
    model=model,
    tools=[
        summarize_text,
        create_handoff_tool(
            agent_name="analyst_agent",
            description="Hand off to analyst when summary is complete and needs structured analysis.",
        ),
        create_handoff_tool(
            agent_name="researcher_agent",
            description="Hand off back to researcher if more raw data is needed.",
        ),
    ],
    name="summarizer_agent",
    prompt=(
        "You are a summarization specialist in a swarm. Summarize the research "
        "findings using your summarize_text tool. Once done, hand off to "
        "analyst_agent for structured analysis."
    ),
)

analyst_agent = create_react_agent(
    model=model,
    tools=[
        analyze_data,
        create_handoff_tool(
            agent_name="writer_agent",
            description="Hand off to writer when analysis is complete and a report needs to be written.",
        ),
        create_handoff_tool(
            agent_name="summarizer_agent",
            description="Hand off back to summarizer if the data needs re-summarization.",
        ),
    ],
    name="analyst_agent",
    prompt=(
        "You are a data analysis specialist in a swarm. Analyze the summarized "
        "findings using your analyze_data tool. Once done, hand off to "
        "writer_agent for report formatting."
    ),
)

writer_agent = create_react_agent(
    model=model,
    tools=[format_report],
    name="writer_agent",
    prompt=(
        "You are a report writing specialist in a swarm. Format the analyzed "
        "data into a professional report using your format_report tool. "
        "You are the final agent in the chain. Produce the final output directly."
    ),
)

# Build the swarm
workflow = create_swarm(
    [researcher_agent, summarizer_agent, analyst_agent, writer_agent],
    default_active_agent="researcher_agent",
)

checkpointer = InMemorySaver()
swarm_app = workflow.compile(checkpointer=checkpointer)


def run_swarm(question: str) -> dict:
    """Run the swarm pattern."""
    config = {"configurable": {"thread_id": "swarm-session-1"}}

    print("\n[SWARM PATTERN]")
    print(f"Question: {question}\n")

    result = swarm_app.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=config,
        recursion_limit=30,
    )

    final_msg = result["messages"][-1].content
    print(f"Result: {final_msg[:300]}...")
    print(f"Total messages: {len(result['messages'])}")
    return result


if __name__ == "__main__":
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    run_swarm(question)

You now have two complete patterns running side-by-side. Next up, you will build the Plan-and-Execute pattern, which takes yet another approach: formulate the entire plan first, then execute it step by step.

Build the Plan-and-Execute Pattern

The Supervisor and Swarm patterns both decide the next action in real time. The Supervisor routes after every agent response. The Swarm lets agents hand off on the fly. Neither formulates a full plan before starting work.

The Plan-and-Execute pattern takes a fundamentally different approach. A Planner generates a structured task list up front, an Executor runs one step at a time, and a Replanner adapts based on results. This two-phase architecture shines for complex multi-step tasks where you need the full plan visible before execution begins.

In this step, get ready to:

• Build a custom StateGraph with three async node functions (Planner, Executor, Replanner).
• Wire conditional edges that loop execution until the task is complete or the iteration cap is hit.
• Run the pattern against the same research question and compare the structured output.

Define the Pydantic models and node functions

Unlike the Supervisor and Swarm patterns which use pre-built helpers, Plan-and-Execute is a custom StateGraph you wire by hand. You need Pydantic models to enforce structured output from the LLM, ensuring the planner always returns a typed plan rather than freeform text.

• Create a new file called pattern_plan_execute.py in your project folder.
• Add the imports and Pydantic models by pasting this code:

from typing import Union
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field
from state import PlanExecuteState
from tools import web_search, analyze_data

planner_model = ChatOpenAI(model="gpt-4o", temperature=0)
executor_model = ChatOpenAI(model="gpt-4o-mini")


class Plan(BaseModel):
    steps: list[str] = Field(description="Numbered list of steps to complete the task")


class Response(BaseModel):
    response: str = Field(description="Final response to deliver to the user")


class Act(BaseModel):
    action: Union[Response, Plan] = Field(
        description=(
            "Action to perform. Use Response if the task is complete. "
            "Use Plan if more steps are needed."
        )
    )

What does this code do?

Two models are initialized with different roles. planner_model uses gpt-4o with temperature=0 for deterministic planning. executor_model uses the cheaper gpt-4o-mini for speed and cost savings on execution. This tiered approach cuts per-request cost significantly.

The three Pydantic models enforce typed output from the LLM. Plan holds a list of step strings. Response holds the final answer. Act is a union that forces the replanner to either deliver a final response OR request more steps.

Now add the first node function. The Planner takes the user's question and generates a structured list of research steps.

• Add the plan_step function below the Act class:

async def plan_step(state: PlanExecuteState) -> dict:
    """Generate a structured plan from the user input."""
    planner = planner_model.with_structured_output(Plan)
    plan = await planner.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a research planner. Given a research question, "
                    "produce a numbered list of 3-5 concrete steps to answer it. "
                    "Each step should be actionable and self-contained."
                ),
            },
            {"role": "user", "content": state["input"]},
        ]
    )
    return {"plan": plan.steps}

What does this code do?

The .with_structured_output(Plan) call wraps the LLM so its output is always parsed into a Plan object. If the LLM returns malformed JSON, the structured output parser retries automatically. The function returns only the plan field, which updates that key in the shared state.

Next is the Executor. It picks up the first step in the plan, runs it, and records the result.

• Add the execute_step function below plan_step:

async def execute_step(state: PlanExecuteState) -> dict:
    """Execute the next step in the plan."""
    plan = state["plan"]
    past_steps = state.get("past_steps", [])

    task = plan[0]
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    past_str = "\n".join(f"- {step}: {result}" for step, result in past_steps)

    prompt = (
        f"Plan:\n{plan_str}\n\n"
        f"Past steps completed:\n{past_str}\n\n"
        f"Execute step: {task}\n"
        "Provide a concise result."
    )

    result = await executor_model.ainvoke(
        [
            {
                "role": "system",
                "content": "You are a research executor. Complete the assigned step concisely.",
            },
            {"role": "user", "content": prompt},
        ]
    )

    return {
        "past_steps": [(task, result.content)],
        "iteration_count": state.get("iteration_count", 0) + 1,
    }

What does this code do?

The executor always processes plan[0] (the first remaining step). It builds a prompt containing the full plan for context, any past results, and the specific step to execute. The lighter gpt-4o-mini model handles this since execution is simpler than planning.

The return value appends to past_steps and increments iteration_count. The iteration count is what you will use later to enforce a hard stop.

The third node is the Replanner. After each execution step, it decides whether the task is fully answered or more work is needed.

• Add the replan_step function below execute_step:

async def replan_step(state: PlanExecuteState) -> dict:
    """Decide whether to continue executing or deliver final response."""
    replanner = planner_model.with_structured_output(Act)

    past_str = "\n".join(f"- {step}: {result}" for step, result in state["past_steps"])

    output = await replanner.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a research replanner. Based on completed steps, "
                    "decide: if the task is fully answered, return a Response with "
                    "the final answer. If more work is needed, return a Plan with "
                    "remaining steps."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Original task: {state['input']}\n\n"
                    f"Steps completed:\n{past_str}\n\n"
                    f"Remaining plan: {state['plan'][1:]}\n\n"
                    "Should we continue or deliver the final answer?"
                ),
            },
        ]
    )

    if isinstance(output.action, Response):
        return {"response": output.action.response}
    else:
        return {"plan": output.action.steps}

What does this code do?

The replanner uses .with_structured_output(Act) which forces the LLM to return either a Response (task is done) or a Plan (more steps needed). This binary decision is what drives the execute-replan loop.

If it returns a Response, the function sets response in state (which signals completion). If it returns a Plan, it overwrites the plan field with the updated steps. The graph's routing function will check which case happened.

• Save pattern_plan_execute.py.

You won't see output yet. The next substep wires these functions into a runnable graph.

Seeing a syntax error on save?

Check that each function starts with async def (not just def). All three node functions use await internally, so they must be async.

Also verify that the f-strings inside execute_step and replan_step use matching quotes. The outer parentheses use double quotes, so the f-string expressions use single quotes for dictionary access like state['input'].

Help me debug a syntax error in my plan_step, execute_step, or replan_step functions

Wire the StateGraph with conditional edges

With the three node functions defined, you now build the graph topology. The key difference from Supervisor and Swarm: you are writing the routing logic yourself as a plain Python function.

• Add the routing function and graph construction below replan_step:

def should_end(state: PlanExecuteState) -> str:
    """Route to END if response is ready or iteration cap hit."""
    if state.get("response"):
        return "end"
    if state.get("iteration_count", 0) >= 5:
        return "end"
    return "execute"


# Build the Plan-and-Execute graph
builder = StateGraph(PlanExecuteState)

builder.add_node("planner", plan_step)
builder.add_node("execute", execute_step)
builder.add_node("replan", replan_step)

builder.add_edge(START, "planner")
builder.add_edge("planner", "execute")
builder.add_edge("execute", "replan")
builder.add_conditional_edges(
    "replan",
    should_end,
    {"execute": "execute", "end": END},
)

checkpointer = InMemorySaver()
plan_execute_app = builder.compile(checkpointer=checkpointer)

What does this code do?

The should_end function is the iteration cap. It checks two conditions: is the response field set (replanner decided the task is done), or has iteration_count hit 5 (hard stop regardless of quality). This is the "LLM decides WHAT, Python decides WHEN TO STOP" principle.

The graph flows: START to planner, planner to execute, execute to replan. From replan, the conditional edge calls should_end to decide whether to loop back to execute or route to END. This creates the execute-replan loop with a guaranteed exit.

Why enforce iteration caps in Python, not in prompts?

A prompt that says "stop after 5 turns" will be ignored when the LLM is mid-reasoning. The check if state.get("iteration_count", 0) >= 5 in Python code cannot be ignored. Deterministic guards are one of LangGraph's key advantages over frameworks that rely on LLM-driven flow control.

• Save pattern_plan_execute.py.

You won't see output yet. The next chunk adds the run function that invokes this graph.

Seeing a NameError for START or END?

Make sure your import at the top of the file includes both: from langgraph.graph import StateGraph, START, END. If you only imported StateGraph, Python won't recognize START or END.

Help me debug a NameError in my Plan-and-Execute graph construction

Run the Plan-and-Execute pattern

• Add the run_plan_execute function and the __main__ block at the bottom of pattern_plan_execute.py:

async def run_plan_execute(question: str) -> dict:
    """Run the plan-and-execute pattern."""
    config = {"configurable": {"thread_id": "plan-exec-session-1"}}

    print("\n[PLAN-AND-EXECUTE PATTERN]")
    print(f"Question: {question}\n")

    result = await plan_execute_app.ainvoke(
        {
            "input": question,
            "messages": [],
            "plan": [],
            "past_steps": [],
            "response": "",
            "iteration_count": 0,
        },
        config=config,
    )

    print(f"Final response: {result.get('response', 'No response')[:300]}...")
    print(f"Steps executed: {len(result.get('past_steps', []))}")
    print(f"Iterations: {result.get('iteration_count', 0)}")
    return result


if __name__ == "__main__":
    import asyncio

    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    asyncio.run(run_plan_execute(question))

What does this code do?

The run_plan_execute function is async because the graph nodes use await. It passes a fully initialized state dict with all fields set to their zero values. The thread_id enables checkpointing so you can inspect or resume the graph mid-execution.

The __main__ block uses asyncio.run() to execute the async function from the command line. This is the same research question used in the Supervisor and Swarm patterns, so you can directly compare outputs.

• Save pattern_plan_execute.py.
• Run the pattern by executing this command:

python pattern_plan_execute.py

• You should see [PLAN-AND-EXECUTE PATTERN] followed by the question, a final response summary, the number of steps executed, and the iteration count.

What should I see?

The output prints three metrics. Final response shows the first 300 characters of the replanner's synthesized answer. Steps executed shows how many plan steps were completed (typically 3-5). Iterations shows the total count, which should be 5 or fewer thanks to your cap.

Seeing a RuntimeError about event loops?

If you see an error about the event loop already running, you may be in a Jupyter notebook or IPython session. The asyncio.run() call requires a fresh event loop. Run the script directly from your terminal with python pattern_plan_execute.py instead.

If you see an API authentication error, verify your OPENAI_API_KEY is still set in your current terminal session. Virtual environment activation does not persist environment variables across new terminal windows.

Help me debug my Plan-and-Execute pattern run

✔️ Awesome, I've got everything!

Great. Double check you have saved pattern_plan_execute.py and confirmed output in your terminal.

ⓧ I'd like to double check the full code

Here is the complete pattern_plan_execute.py file:

from typing import Union
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field
from state import PlanExecuteState
from tools import web_search, analyze_data

planner_model = ChatOpenAI(model="gpt-4o", temperature=0)
executor_model = ChatOpenAI(model="gpt-4o-mini")


class Plan(BaseModel):
    steps: list[str] = Field(description="Numbered list of steps to complete the task")


class Response(BaseModel):
    response: str = Field(description="Final response to deliver to the user")


class Act(BaseModel):
    action: Union[Response, Plan] = Field(
        description=(
            "Action to perform. Use Response if the task is complete. "
            "Use Plan if more steps are needed."
        )
    )


async def plan_step(state: PlanExecuteState) -> dict:
    """Generate a structured plan from the user input."""
    planner = planner_model.with_structured_output(Plan)
    plan = await planner.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a research planner. Given a research question, "
                    "produce a numbered list of 3-5 concrete steps to answer it. "
                    "Each step should be actionable and self-contained."
                ),
            },
            {"role": "user", "content": state["input"]},
        ]
    )
    return {"plan": plan.steps}


async def execute_step(state: PlanExecuteState) -> dict:
    """Execute the next step in the plan."""
    plan = state["plan"]
    past_steps = state.get("past_steps", [])

    task = plan[0]
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    past_str = "\n".join(f"- {step}: {result}" for step, result in past_steps)

    prompt = (
        f"Plan:\n{plan_str}\n\n"
        f"Past steps completed:\n{past_str}\n\n"
        f"Execute step: {task}\n"
        "Provide a concise result."
    )

    result = await executor_model.ainvoke(
        [
            {
                "role": "system",
                "content": "You are a research executor. Complete the assigned step concisely.",
            },
            {"role": "user", "content": prompt},
        ]
    )

    return {
        "past_steps": [(task, result.content)],
        "iteration_count": state.get("iteration_count", 0) + 1,
    }


async def replan_step(state: PlanExecuteState) -> dict:
    """Decide whether to continue executing or deliver final response."""
    replanner = planner_model.with_structured_output(Act)

    past_str = "\n".join(f"- {step}: {result}" for step, result in state["past_steps"])

    output = await replanner.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a research replanner. Based on completed steps, "
                    "decide: if the task is fully answered, return a Response with "
                    "the final answer. If more work is needed, return a Plan with "
                    "remaining steps."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Original task: {state['input']}\n\n"
                    f"Steps completed:\n{past_str}\n\n"
                    f"Remaining plan: {state['plan'][1:]}\n\n"
                    "Should we continue or deliver the final answer?"
                ),
            },
        ]
    )

    if isinstance(output.action, Response):
        return {"response": output.action.response}
    else:
        return {"plan": output.action.steps}


def should_end(state: PlanExecuteState) -> str:
    """Route to END if response is ready or iteration cap hit."""
    if state.get("response"):
        return "end"
    if state.get("iteration_count", 0) >= 5:
        return "end"
    return "execute"


# Build the Plan-and-Execute graph
builder = StateGraph(PlanExecuteState)

builder.add_node("planner", plan_step)
builder.add_node("execute", execute_step)
builder.add_node("replan", replan_step)

builder.add_edge(START, "planner")
builder.add_edge("planner", "execute")
builder.add_edge("execute", "replan")
builder.add_conditional_edges(
    "replan",
    should_end,
    {"execute": "execute", "end": END},
)

checkpointer = InMemorySaver()
plan_execute_app = builder.compile(checkpointer=checkpointer)


async def run_plan_execute(question: str) -> dict:
    """Run the plan-and-execute pattern."""
    config = {"configurable": {"thread_id": "plan-exec-session-1"}}

    print("\n[PLAN-AND-EXECUTE PATTERN]")
    print(f"Question: {question}\n")

    result = await plan_execute_app.ainvoke(
        {
            "input": question,
            "messages": [],
            "plan": [],
            "past_steps": [],
            "response": "",
            "iteration_count": 0,
        },
        config=config,
    )

    print(f"Final response: {result.get('response', 'No response')[:300]}...")
    print(f"Steps executed: {len(result.get('past_steps', []))}")
    print(f"Iterations: {result.get('iteration_count', 0)}")
    return result


if __name__ == "__main__":
    import asyncio

    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    asyncio.run(run_plan_execute(question))

You now have all three architecture patterns running. Next up, you will add human-in-the-loop approval to the Supervisor pattern so a human can review and approve (or reject) the final output before delivery.

Add Human-in-the-Loop and Checkpointing

All three patterns you built already use InMemorySaver for checkpointing. This means every graph invocation saves state at each superstep, giving you durability and the ability to resume from any point.

But production systems need a second guardrail: a human approval gate that pauses execution, lets a person inspect the output, and resumes only when approved. In this step, you will use LangGraph's interrupt() primitive and Command to build this flow on top of your Supervisor graph.

In this step, get ready to:

• Add an interrupt gate to the Supervisor graph for human approval before final output.
• Demonstrate the full pause-resume flow with frozen state inspection.
• Handle rejected output by routing back for revision with a maximum of 2 revision cycles.

Understand how interrupt() and Command work together

The interrupt() function pauses graph execution at the exact point where it is called. The graph state is saved via the checkpointer and execution waits indefinitely until you resume.

To resume, you invoke the graph again with a Command(resume=value) instead of new input. The resume value becomes the return value of the interrupt() call inside the paused node.

What is interrupt()?

Think of interrupt() as a programmatic breakpoint you insert into any node. When the graph hits it, three things happen:

• Execution freezes at that exact line.
• The checkpointer persists the full graph state (messages, intermediate results, everything).
• The payload you pass to interrupt() surfaces to the caller so they know what the graph is waiting on.

The thread_id in your config is the pointer. Reusing the same thread_id when you pass Command(resume=...) tells LangGraph which frozen checkpoint to load and continue from.

• Open a Python interactive session by running:

python

• Import the interrupt and Command primitives along with your existing supervisor graph by entering:

from langgraph.types import interrupt, Command
from pattern_supervisor import app

print("Imports ready. The supervisor graph 'app' is compiled with InMemorySaver.")
print(f"Graph name: {app.name}")

Why does this work without modifying pattern_supervisor.py?

Your app graph was already compiled with InMemorySaver as the checkpointer. That is the only prerequisite for interrupt() to function. You can explore the interrupt/resume flow interactively without changing your source files.

Demonstrate the full pause-resume flow

You will now build a minimal approval-gated graph that demonstrates the full interrupt lifecycle: invoke, pause, inspect frozen state, and resume with approval or rejection.

• Define the state and node functions for the approval-gated graph by entering this code in your Python session:

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage
from typing import TypedDict, Annotated

# Define a minimal state for the approval demo
class ApprovalState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    report: str
    approval_status: str
    revision_count: int

def generate_report(state: ApprovalState) -> dict:
    """Simulate report generation."""
    return {
        "report": "LangGraph is production-standard. AutoGen is in maintenance mode.",
        "approval_status": "pending",
    }

def human_approval_gate(state: ApprovalState) -> dict:
    """Pause for human review before delivering the report."""
    # interrupt() freezes execution and surfaces the payload to the caller
    decision = interrupt({
        "question": "Do you approve this report for delivery?",
        "report_preview": state["report"][:200],
    })
    # When resumed, decision becomes the value passed to Command(resume=...)
    return {"approval_status": decision}

def deliver_report(state: ApprovalState) -> dict:
    """Deliver the approved report."""
    return {"messages": [{"role": "assistant", "content": f"DELIVERED: {state['report']}"}]}

def revise_report(state: ApprovalState) -> dict:
    """Revise the report after rejection."""
    count = state.get("revision_count", 0) + 1
    return {
        "report": f"[REVISED v{count}] {state['report']} (enhanced with additional data)",
        "revision_count": count,
    }

def route_after_approval(state: ApprovalState) -> str:
    """Route based on approval status with max 2 revision cycles."""
    if state["approval_status"] == "approved":
        return "deliver"
    # Force delivery after hitting the revision cap
    if state.get("revision_count", 0) >= 2:
        return "deliver"
    return "revise"

print("Approval node functions defined.")

What does this code do?

human_approval_gate calls interrupt() with a JSON payload describing what the graph is waiting for. When the graph hits this node, execution freezes and the payload surfaces to the caller.

route_after_approval checks both the decision AND a revision counter. If the human rejects but 2 revisions have already happened, it forces delivery anyway. This prevents infinite rejection loops.

• Build and compile the approval graph by entering:

# Build the approval-gated graph
builder = StateGraph(ApprovalState)

builder.add_node("generate", generate_report)
builder.add_node("approve", human_approval_gate)
builder.add_node("deliver", deliver_report)
builder.add_node("revise", revise_report)

# Wire edges: generate -> approve -> (deliver or revise)
builder.add_edge(START, "generate")
builder.add_edge("generate", "approve")
builder.add_conditional_edges(
    "approve",
    route_after_approval,
    {"deliver": "deliver", "revise": "revise"},
)
# After revision, loop back to approval gate
builder.add_edge("revise", "approve")
builder.add_edge("deliver", END)

# Compile with checkpointer (required for interrupt to work)
hitl_checkpointer = InMemorySaver()
hitl_app = builder.compile(checkpointer=hitl_checkpointer)

print("Approval graph compiled with InMemorySaver.")
print(f"Nodes: {list(hitl_app.get_graph().nodes.keys())}")
print("Flow: START -> generate -> approve -> (deliver | revise -> approve)...")

Why is InMemorySaver required?

interrupt() needs somewhere to persist the frozen state. Without a checkpointer, calling interrupt() would raise an error because there is no persistence layer to save to.

In production, you would swap InMemorySaver for AsyncPostgresSaver or Redis so that frozen state survives server restarts.

• Invoke the graph to trigger the interrupt at the approval gate:

# Invoke the graph - it will pause at the interrupt inside the approve node
hitl_config = {"configurable": {"thread_id": "approval-session-1"}}

result = hitl_app.invoke(
    {
        "messages": [{"role": "user", "content": "Generate a comparison report."}],
        "report": "",
        "approval_status": "pending",
        "revision_count": 0,
    },
    config=hitl_config,
)

# The graph paused at interrupt - inspect the frozen state
print("Graph paused at interrupt!")
frozen_state = hitl_app.get_state(hitl_config)
print(f"Next nodes: {frozen_state.next}")
print(f"Report in state: {frozen_state.values.get('report', 'N/A')[:100]}")
print(f"Approval status: {frozen_state.values.get('approval_status', 'N/A')}")

You should see the graph pause and report that it is waiting at the approve node. The next field shows which node would execute next when resumed.

What does get_state() return?

The get_state() method returns a StateSnapshot object with these key fields:

• values: the current channel values (your messages, report, approval status).
• next: which nodes would execute next (empty tuple means the graph finished).
• config: the checkpoint config including thread_id and checkpoint_id.
• tasks: pending tasks with their interrupts if the graph is paused.

• Resume with approval to see the happy path complete:

# Resume with approval - the value becomes the return of interrupt() in the node
print("--- RESUMING WITH APPROVAL ---")
approved_result = hitl_app.invoke(
    Command(resume="approved"),
    config=hitl_config,
)

print(f"Approval status: {approved_result.get('approval_status', 'N/A')}")
print(f"Final message: {approved_result['messages'][-1].content[:200]}")
print("Report delivered successfully!")

You should see the report delivered with the DELIVERED: prefix, confirming the graph completed end-to-end after human approval.

Seeing an error about missing thread state?

Make sure you used the exact same hitl_config (same thread_id) for both the initial invoke and the resume. The thread_id is how LangGraph finds the frozen checkpoint.

If you restarted your Python session between the invoke and resume, the InMemorySaver lost its data. Run both the invoke and resume in the same session.

help me debug my interrupt/resume flow

Handle rejected output with revision cycles

The real power of human-in-the-loop shows when the human rejects output and the graph routes back for revision. You want this loop to have a hard cap enforced in Python code, not in a prompt.

• Start a fresh thread and demonstrate the rejection flow with a maximum of 2 revision cycles by entering:

# Start a fresh thread for the rejection demo
reject_config = {"configurable": {"thread_id": "approval-session-2"}}

# First invocation - hits interrupt at approve node
result = hitl_app.invoke(
    {
        "messages": [{"role": "user", "content": "Generate a comparison report."}],
        "report": "",
        "approval_status": "pending",
        "revision_count": 0,
    },
    config=reject_config,
)
print("Round 1: Graph paused for approval.")
print(f"Report: {hitl_app.get_state(reject_config).values['report'][:80]}...")

# Reject - routes to revise node, then back to approve (hits interrupt again)
print("\n--- REJECTING (round 1) ---")
result = hitl_app.invoke(Command(resume="rejected"), config=reject_config)
print("Round 2: Graph revised and paused again for approval.")
state = hitl_app.get_state(reject_config)
print(f"Revision count: {state.values.get('revision_count', 0)}")
print(f"Revised report: {state.values['report'][:80]}...")

# Reject again - routes to revise, then back to approve (hits interrupt again)
print("\n--- REJECTING (round 2) ---")
result = hitl_app.invoke(Command(resume="rejected"), config=reject_config)
print("Round 3: Graph revised and paused again for approval.")
state = hitl_app.get_state(reject_config)
print(f"Revision count: {state.values.get('revision_count', 0)}")
print(f"Revised report: {state.values['report'][:80]}...")

# Third rejection - revision cap hit, forces delivery regardless
print("\n--- REJECTING (round 3 - cap forces delivery) ---")
result = hitl_app.invoke(Command(resume="rejected"), config=reject_config)
print(f"\nFinal status: {result.get('approval_status', 'N/A')}")
print(f"Final message: {result['messages'][-1].content[:200]}")
print("\nReport force-delivered after hitting the 2-revision cap!")

You should see three rejection cycles. After the second revision, the route_after_approval function detects revision_count >= 2 and forces delivery regardless of the human decision.

Why enforce a revision cap in code?

A prompt telling an LLM to stop after 2 revisions will be ignored. The LLM has no concept of enforcing limits on itself. The route_after_approval function checks revision_count >= 2 in Python. This is deterministic and cannot be bypassed.

This is the core mental model shift from AutoGen to LangGraph: the LLM decides WHAT to do, but Python code decides HOW and WHEN TO STOP.

• Print a summary and exit the Python interactive session:

print("\n" + "=" * 50)
print("HUMAN-IN-THE-LOOP DEMO COMPLETE")
print("=" * 50)
print("\nKey takeaways:")
print("1. interrupt() pauses execution and saves state via checkpointer")
print("2. Command(resume=value) continues from the frozen checkpoint")
print("3. Same thread_id = same checkpoint = correct resume")
print("4. Revision caps enforced in Python, never in prompts")
print("5. Production: swap InMemorySaver for AsyncPostgresSaver")

exit()

Why not modify pattern_supervisor.py permanently?

Your pattern_supervisor.py needs to run cleanly for the comparison step coming next. The interrupt/resume pattern you just explored works with ANY LangGraph graph that has a checkpointer.

In production, you would add interrupt() calls directly into your node functions and use AsyncPostgresSaver or Redis for durable state that survives server restarts.

✔️ Awesome, I've got everything!

Great. Your project files are unchanged and ready for the comparison step.

ⓧ I'd like to double check the full code

Your pattern_supervisor.py should still look exactly like this (no changes from the previous step):

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.checkpoint.memory import InMemorySaver
from agents import web_researcher, summarizer, data_analyst, report_writer

model = ChatOpenAI(model="gpt-4o")

# Build sub-team supervisors
research_team = create_supervisor(
    [web_researcher, summarizer],
    model=model,
    prompt=(
        "You are the Research Team supervisor. You manage a web researcher and a summarizer. "
        "For any research task: first delegate to web_researcher to gather raw information, "
        "then delegate to summarizer to produce a structured summary. "
        "Return the final summary to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="research_team")

analysis_team = create_supervisor(
    [data_analyst, report_writer],
    model=model,
    prompt=(
        "You are the Analysis Team supervisor. You manage a data analyst and a report writer. "
        "For any analysis task: first delegate to data_analyst to extract structured insights, "
        "then delegate to report_writer to produce a formatted report. "
        "Return the final report to your parent supervisor when complete."
    ),
    output_mode="last_message",
).compile(name="analysis_team")

# Build top-level supervisor
top_supervisor = create_supervisor(
    [research_team, analysis_team],
    model=model,
    prompt=(
        "You are the top-level supervisor managing two teams: "
        "research_team and analysis_team. "
        "For any research question:\n"
        "1. First delegate to research_team to gather and summarize information.\n"
        "2. Then delegate to analysis_team to analyze findings and produce a report.\n"
        "3. Return the final report to the user.\n"
        "Always follow this sequence. Do not skip steps."
    ),
    supervisor_name="top_supervisor",
)

checkpointer = InMemorySaver()
app = top_supervisor.compile(
    name="hierarchical_research_system",
    checkpointer=checkpointer,
)


def run_supervisor(question: str) -> dict:
    """Run the supervisor pattern."""
    config = {"configurable": {"thread_id": "supervisor-session-1"}}

    print("\n[SUPERVISOR PATTERN]")
    print(f"Question: {question}\n")

    result = app.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=config,
        recursion_limit=25,
    )

    final_msg = result["messages"][-1].content
    print(f"Result: {final_msg[:300]}...")
    print(f"Total messages: {len(result['messages'])}")
    return result


if __name__ == "__main__":
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )
    run_supervisor(question)

You now have a working mental model of human-in-the-loop approval gates with deterministic revision caps. Next up, you will run all three patterns side-by-side and compare their trade-offs directly.

Compare All Three Architectures

You now have three complete multi-agent architectures, each with its own topology, routing strategy, and failure mode. The natural question is: which one should I use?

The answer depends on the task shape. In this step, you will run all three patterns against the same research question and compare their metrics side-by-side so you can build a concrete decision framework.

In this step, get ready to:

• Create a comparison script that runs all three architectures against the same question.
• Collect and compare metrics (message counts, step counts) across patterns.
• Build a decision framework for choosing the right pattern for a given task.

Write the comparison script

Each of your three pattern files already exposes a runner function. The comparison script imports all three, runs them sequentially with the same input, and prints a formatted table of metrics.

• Create a new file called compare.py in your project root and add the following code:

import asyncio
from pattern_supervisor import run_supervisor
from pattern_swarm import run_swarm
from pattern_plan_execute import run_plan_execute


async def compare_all():
    """Run all three patterns and compare results."""
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )

    print("=" * 70)
    print("MULTI-AGENT ARCHITECTURE COMPARISON")
    print("=" * 70)

    # Pattern 1: Supervisor
    sup_result = run_supervisor(question)
    sup_msgs = len(sup_result["messages"])

    # Pattern 2: Swarm
    swarm_result = run_swarm(question)
    swarm_msgs = len(swarm_result["messages"])

    # Pattern 3: Plan-and-Execute
    pe_result = await run_plan_execute(question)
    pe_steps = len(pe_result.get("past_steps", []))

    # Print comparison
    print("\n" + "=" * 70)
    print("COMPARISON SUMMARY")
    print("=" * 70)
    print(f"\n{'Pattern':<20} {'Messages':<12} {'Best For'}")
    print("-" * 70)
    print(f"{'Supervisor':<20} {sup_msgs:<12} Deterministic pipelines, auditability, regulated industries")
    print(f"{'Swarm':<20} {swarm_msgs:<12} Open-ended workflows, token efficiency, dynamic agent counts")
    print(f"{'Plan-Execute':<20} {pe_steps:<12} Complex multi-step tasks, structured decomposition")
    print()
    print("DECISION FRAMEWORK:")
    print("  - Linear task? -> Use Pipeline (not shown)")
    print("  - Dynamic routing needed? -> Supervisor")
    print("  - Agents are peers, open-ended? -> Swarm (with observability)")
    print("  - Complex planning before execution? -> Plan-and-Execute")
    print("  - Quality needs iterative refinement? -> Add Reflection Loop")
    print("  - 6+ workers with sub-teams? -> Hierarchical Supervisor")


if __name__ == "__main__":
    asyncio.run(compare_all())

What does this code do?

The compare_all() function runs each pattern sequentially against the exact same research question. This gives you a fair comparison since the input is identical.

• run_supervisor() and run_swarm() are synchronous, so we call them directly. run_plan_execute() is async, so we await it.
• For Supervisor and Swarm, the metric is total message count (how many messages the agents exchanged). For Plan-and-Execute, the metric is the number of steps executed.
• The decision framework at the bottom maps task shapes to recommended patterns. This is the mental model you carry to production.

• Save compare.py.

Run the comparison and interpret results

Now run all three patterns back-to-back. This will make several OpenAI API calls, so expect it to take 30-60 seconds.

• Run the comparison script by executing this command:

python compare.py

• You should see each pattern run in sequence (with its own header and output), followed by a formatted comparison table and the decision framework.

What should I see?

The output begins with each pattern running individually (you will see the [SUPERVISOR PATTERN], [SWARM PATTERN], and [PLAN-AND-EXECUTE PATTERN] headers you built in previous steps). After all three finish, the COMPARISON SUMMARY table prints.

The exact message counts will vary between runs because LLM responses are non-deterministic. The Swarm will typically have fewer messages than the Supervisor for the same task. Plan-and-Execute reports steps executed (typically 3-5) rather than raw message count.

Seeing a recursion limit error?

If one pattern hits its recursion limit, it means the agents are looping beyond the cap you set. The Supervisor uses recursion_limit=25 and the Swarm uses recursion_limit=30. If you see this error, verify those values in pattern_supervisor.py and pattern_swarm.py.

help me debug a recursion limit error in my LangGraph comparison script

When should I choose each pattern?

The decision framework in the output maps directly to production scenarios. Supervisor is best when you need auditability and deterministic task ordering (common in regulated industries). Swarm is best when agents are true peers and the workflow is open-ended with no fixed sequence.

Plan-and-Execute shines when the full plan must be formulated before any execution begins. This is common for complex multi-step research where later steps depend on earlier results.

What are the failure modes of each pattern?

• Supervisor bottleneck: At high agent counts (6+), the supervisor becomes a single point of contention. Every routing decision goes through one LLM call. The fix is to go hierarchical (sub-teams), which you already built.
• Swarm ping-pong loops: Agent A hands off to Agent B, which immediately hands back. Without a circuit breaker (like the handoff_count you defined in state), this loops forever.
• Plan-and-Execute rigidity: If the environment changes mid-execution (API goes down, new data arrives), the plan becomes stale. The replanner mitigates this, but it cannot recover from fundamental assumption violations.

✔️ Awesome, I've got everything!

Great. Double check you have saved compare.py and that your comparison ran successfully.

ⓧ I'd like to double check the full code

import asyncio
from pattern_supervisor import run_supervisor
from pattern_swarm import run_swarm
from pattern_plan_execute import run_plan_execute


async def compare_all():
    """Run all three patterns and compare results."""
    question = (
        "What are the key differences between LangGraph and AutoGen "
        "for production multi-agent systems?"
    )

    print("=" * 70)
    print("MULTI-AGENT ARCHITECTURE COMPARISON")
    print("=" * 70)

    # Pattern 1: Supervisor
    sup_result = run_supervisor(question)
    sup_msgs = len(sup_result["messages"])

    # Pattern 2: Swarm
    swarm_result = run_swarm(question)
    swarm_msgs = len(swarm_result["messages"])

    # Pattern 3: Plan-and-Execute
    pe_result = await run_plan_execute(question)
    pe_steps = len(pe_result.get("past_steps", []))

    # Print comparison
    print("\n" + "=" * 70)
    print("COMPARISON SUMMARY")
    print("=" * 70)
    print(f"\n{'Pattern':<20} {'Messages':<12} {'Best For'}")
    print("-" * 70)
    print(f"{'Supervisor':<20} {sup_msgs:<12} Deterministic pipelines, auditability, regulated industries")
    print(f"{'Swarm':<20} {swarm_msgs:<12} Open-ended workflows, token efficiency, dynamic agent counts")
    print(f"{'Plan-Execute':<20} {pe_steps:<12} Complex multi-step tasks, structured decomposition")
    print()
    print("DECISION FRAMEWORK:")
    print("  - Linear task? -> Use Pipeline (not shown)")
    print("  - Dynamic routing needed? -> Supervisor")
    print("  - Agents are peers, open-ended? -> Swarm (with observability)")
    print("  - Complex planning before execution? -> Plan-and-Execute")
    print("  - Quality needs iterative refinement? -> Add Reflection Loop")
    print("  - 6+ workers with sub-teams? -> Hierarchical Supervisor")


if __name__ == "__main__":
    asyncio.run(compare_all())

Secret mission

Add a Reflection Loop to the Supervisor

Most production multi-agent systems are not a single pattern. They are composites of two or three patterns layered together. In this secret mission, you will add a Reflection Loop quality gate to your Supervisor pattern. A reviewer agent scores the final report and routes it back for revision if the quality is below threshold. This is how teams at Uber and LinkedIn actually ship LangGraph systems.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with pay-per-use OpenAI API calls. There are no ongoing costs once you stop running the scripts.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Project folder (contains state.py, tools.py, agents.py, pattern_supervisor.py, pattern_swarm.py, pattern_plan_execute.py, compare.py, and the .venv virtual environment).
• OPENAI_API_KEY environment variable (set in your terminal session).
• OpenAI API key (on the OpenAI platform).

Keep everything running

No action needed. Choose this if you plan to continue experimenting with the multi-agent patterns or want to extend them further.

• Your project folder stays on your Desktop.
• Your OPENAI_API_KEY only persists for the current terminal session unless you added it to your shell profile.
• The OpenAI API key remains active but incurs no cost unless you make API calls.

Pause - I'll come back to this later

Shut down your virtual environment to free up your terminal. Your files stay intact for when you return.

• Deactivate the virtual environment by running:

deactivate

• Close your terminal window. The OPENAI_API_KEY environment variable only exists in that session.
• Your project folder and .venv remain on disk. When you come back, reactivate with source .venv/bin/activate and re-export your API key.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your code and virtual environment.

• Delete the project folder by running this command (macOS/Linux):

rm -rf ~/Desktop/langgraph-multi-agent

• On Windows, run this instead:

rmdir /s /q %USERPROFILE%\Desktop\langgraph-multi-agent

• Unset the environment variable in your current session:

unset OPENAI_API_KEY

• If you added the key to your shell profile (.bashrc, .zshrc, or similar), open that file and remove the export OPENAI_API_KEY=... line.
• If you created this API key specifically for this project and want to revoke it, go to platform.openai.com/settings/organization/api-keys and delete the key.

That's a Wrap!

That's a Wrap!

Nice work. You have built and compared three production-grade multi-agent architectures with LangGraph, giving you a hands-on decision framework for choosing the right topology for any agent system.

You've learned how to:

• Build three distinct multi-agent architectures (Supervisor, Swarm, and Plan-and-Execute) and compare their trade-offs in routing, token usage, and failure modes.
• Implement production guardrails on all three patterns including deterministic iteration caps in Python, circuit breakers, and recursion limits.
• Add human-in-the-loop approval gates with the interrupt/resume flow and durable checkpointing for pause-and-resume across sessions.
• Created a composite Supervisor + Reflection Loop architecture showing how real production systems combine multiple patterns for quality iteration.

Ready to quiz yourself?