Build a Multi-Agent Blog Writing Team

Orchestrate AI agents with LangGraph to research and write blog posts.

Introduction

30 Second Summary

When you ask an AI to write something complex, a single prompt rarely gets you what you want. The best results come from breaking the task into stages, with different specialists handling research, writing, and quality control.

In this project, you will build a multi-agent blog writing pipeline where a Supervisor agent coordinates a Researcher and Writer to produce researched blog posts from any topic you give it.

What You'll Build

A Python-based multi-agent orchestration system using LangGraph and Gemini that accepts a topic, searches the web for research, and produces a polished blog post through coordinated AI agents.

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

  • A working multi-agent pipeline with a Supervisor, Researcher, and Writer that collaborate to produce blog posts.
  • An LLM-based dynamic routing system where the Supervisor decides which agent works next at runtime.
  • A human-in-the-loop checkpoint where you review and modify research before the Writer proceeds.
  • Secret Mission: Add a Fact-Checker agent that verifies claims from the research before writing begins.

Are there any prerequisites?

You should be comfortable writing basic Python (variables, functions, dictionaries). A basic understanding of LLM APIs (making calls to a model and getting responses) is helpful but not strictly required.

Before We Start

Before We Start...

Set Up Your Environment

A multi-agent system needs several moving parts working together: an orchestration framework, an LLM connection, and tools for searching the web.

In this step, you'll install the Python packages that power each part. You'll also connect to Gemini and confirm everything works with a quick test script.

In this step, get ready to:
  • Install the Python packages for LangGraph, Gemini, and web search.
  • Set your Gemini API key as an environment variable.
  • Verify the setup by running a test script that calls Gemini.
Install packages

Your multi-agent system relies on four packages. LangGraph handles the agent orchestration graph. langchain-google-genai connects to the Gemini LLM. langchain-community provides community tools. duckduckgo-search powers free web searching.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal 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 blog-writing-team
cd blog-writing-team
  • Confirm the folder was created by running:
ls

You should see the blog-writing-team folder listed in the output (it will appear empty for now).

  • Install the project dependencies by running:
pip install langgraph langchain-google-genai langchain-community duckduckgo-search

What does this command do?

  • langgraph provides the StateGraph framework for wiring agents into an orchestration graph.
  • langchain-google-genai is the LangChain integration for calling Gemini models.
  • langchain-community includes community-maintained tools like DuckDuckGoSearchRun.
  • duckduckgo-search is the underlying search library. It's free and requires no API key.
  • Wait for the installation to complete. You should see Successfully installed followed by a list of package names at the end of the output.

Seeing an error during install?

  • If you see a "command not found" error for pip, try using pip3 instead. Some systems use pip3 for Python 3 packages.
  • If you see a permission error, add --user to the end of the command to install packages for your user only.
  • If you see a network timeout, check your internet connection and try running the command again.

Help me troubleshoot my pip install error.

Set up your Gemini API key

The langchain-google-genai package looks for your API key in an environment variable called GOOGLE_API_KEY. You need to set this in your terminal session so the LLM can authenticate your requests.

macOS

  • Set your API key by running this command (replace your-api-key-here with your actual Gemini API key):
export GOOGLE_API_KEY=your-api-key-here

Windows

  • Set your API key by running this command (replace your-api-key-here with your actual Gemini API key):
set GOOGLE_API_KEY=your-api-key-here

Why an environment variable?

Storing your key in an environment variable keeps it out of your source code. The langchain-google-genai package automatically reads GOOGLE_API_KEY from your environment, so you never need to hardcode it.

This variable only lasts for the current terminal session. If you close and reopen your terminal, you'll need to set it again.

Not sure where to find your API key?

You can generate a Gemini API key at Google AI Studio (aistudio.google.com). Sign in with your Google account, then click Get API key to create one.

Help me get a Gemini API key.

Verify the installation

Now you'll create a quick test script that calls Gemini to confirm your packages and API key are working together.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the blog-writing-team folder, then click Open.
  • Create a new file by clicking File then New File.
  • Save it immediately as test_setup.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Copy the following code into test_setup.py:
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
response = llm.invoke("Say hello in one sentence.")
print(response.content)

What does this code do?

  • Line 1 imports ChatGoogleGenerativeAI, the LangChain class that connects to Google's Gemini models.
  • Line 3 creates an LLM instance pointed at the gemini-3.5-flash model. This is Google's latest model optimized for agentic tasks.
  • Line 4 sends a simple prompt to the model and stores the response.
  • Line 5 prints the model's text response to your terminal.
  • Save the file with Cmd+S (macOS) or Ctrl+S (Windows).
  • Open the terminal in VS Code by pressing Ctrl+` (backtick key, same on both macOS and Windows).
  • Run the test script:
python test_setup.py

You should see a short greeting from Gemini printed in your terminal, something like "Hello! How can I help you today?" The exact wording will vary.

Not seeing a response?

  • If you see a ModuleNotFoundError, your packages may not be installed in the same Python environment VS Code is using. Try running pip3 install langgraph langchain-google-genai langchain-community duckduckgo-search again in the VS Code terminal.
  • If you see an authentication or API key error, double-check that you set the GOOGLE_API_KEY environment variable in the same terminal session where you're running the script. Try setting it again in the VS Code terminal.
  • If python is not found, try python3 test_setup.py instead.

Help me fix my test_setup.py error.

✔️ Awesome, I've got everything!

Great. Double check you've saved test_setup.py and you're ready to build your agents.

ⓧ I'd like to double check the full code

Here's the complete test_setup.py file:

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
response = llm.invoke("Say hello in one sentence.")
print(response.content)

Your environment is ready and Gemini is responding. Next up, you'll define the shared state and build your first agent: the Researcher.

Define State and Build the Researcher Agent

Your environment is working and you have a confirmed connection to the Gemini API. Now it's time to build the first real piece of your multi-agent system.

Every multi-agent pipeline needs two things: a shared state that all agents can read from and write to, and the agents themselves. In this step, you'll define the state shape and build your first specialist agent, the Researcher.

In this step, get ready to:
  • Define the shared state that flows between agents.
  • Build a Researcher agent that searches the web.
  • Test the researcher in isolation to confirm it produces structured research notes.
Define the graph state

In LangGraph, every node in the graph receives a shared state object and returns updates to it. You define the shape of this state with a TypedDict so every agent knows exactly which fields are available.

  • Create a new file called blog_team.py in your project folder by running:
touch blog_team.py
  • Open blog_team.py in VS Code.
  • Add the following imports and state definition at the top of the file:
import operator
from typing import TypedDict

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


# Define the shared state that all agents read from and write to
class BlogState(TypedDict):
    topic: str
    research: str
    draft: str
    next_step: str

What does this code do?

The imports bring in the tools you'll use throughout this project. You don't need all of them right now, but importing them at the top keeps the file organized.

  • BlogState is a TypedDict with four fields: topic holds the blog topic, research stores gathered research notes, draft holds the final blog post, and next_step tells the graph which agent should run next.
  • Every agent function receives this state and returns a dictionary with the fields it wants to update.
  • Save blog_team.py.

You won't see a visible change yet. The next chunk uses this state definition to build the researcher.

Seeing an import error?

  • Make sure you installed all four packages in Step 1: langgraph, langchain-google-genai, langchain-community, and duckduckgo-search.
  • Confirm you're running the script in the same terminal session where you installed the packages.

Help me debug an import error in my LangGraph project.

Create the web search tool and researcher node

The Researcher agent has one job: search the web for information on the given topic, then use the LLM to synthesize raw search results into structured notes.

  • Add the following code below your BlogState class in blog_team.py:
llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
search_tool = DuckDuckGoSearchRun()


def researcher(state: BlogState) -> dict:
    # Get the topic from shared state
    topic = state["topic"]
    # Search the web for information on the topic
    search_results = search_tool.invoke(topic)

    # Use the LLM to synthesize raw search results into structured notes
    response = llm.invoke([
        SystemMessage(content=(
            "You are a research specialist. Analyze the search results and "
            "produce structured research notes with key facts, statistics, "
            "and insights. Organize by subtopic with bullet points."
        )),
        HumanMessage(content=(
            f"Research topic: {topic}\n\n"
            f"Search results:\n{search_results}"
        ))
    ])
    # Return updated state with the research field populated
    return {"research": response.content}

What does this code do?

  • llm initializes the Gemini model you'll use for all agent reasoning throughout this project.
  • search_tool creates a DuckDuckGo search instance. It's free and requires no API key.
  • The researcher function follows the LangGraph node pattern: it takes the full state, does its work, and returns a dictionary with only the fields it wants to update.
  • The LLM receives a SystemMessage (its role instructions) and a HumanMessage (the actual task with data). This two-message pattern gives the model clear context on what to produce.
  • Save blog_team.py.

You won't see output yet. The next substep tests that this function works correctly.

Unsure about the node pattern?

Each node function receives the entire state dictionary and returns only the fields it changed. LangGraph merges these changes back into the shared state automatically.

Help me understand how LangGraph node functions work with the shared state.

Test the researcher in isolation

Before wiring agents into a graph, it's good practice to test each one independently. This confirms the researcher can actually search and produce useful output.

  • Add the following test code at the very bottom of blog_team.py:
# Temporary test - remove after verifying
test_state = {"topic": "artificial intelligence trends 2026", "research": "", "draft": "", "next_step": ""}
result = researcher(test_state)
print(result["research"])

What does this code do?

This creates a minimal state dictionary with just a topic filled in, then calls the researcher function directly. If everything is working, you'll see structured research notes printed to your terminal.

  • Save blog_team.py.
  • Run the file by executing:
python blog_team.py

You should see structured research notes printed to your terminal, organized by subtopic with bullet points containing real data from the web.

Not seeing any output?

  • Check that your GOOGLE_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you'll need to set it again.
  • If you see a timeout error, DuckDuckGo may be rate-limiting your request. Wait 30 seconds and try again.
  • If you see a NameError, make sure the test code is below the researcher function definition.

Help me debug my researcher agent test.

  • Once you've confirmed the output looks good, remove the test code (the last 4 lines starting with # Temporary test) from blog_team.py.
  • Save blog_team.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved blog_team.py and that the temporary test code is removed.

ⓧ I'd like to double check the full code

import operator
from typing import TypedDict

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class BlogState(TypedDict):
    topic: str
    research: str
    draft: str
    next_step: str


llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
search_tool = DuckDuckGoSearchRun()


def researcher(state: BlogState) -> dict:
    topic = state["topic"]
    search_results = search_tool.invoke(topic)

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research specialist. Analyze the search results and "
            "produce structured research notes with key facts, statistics, "
            "and insights. Organize by subtopic with bullet points."
        )),
        HumanMessage(content=(
            f"Research topic: {topic}\n\n"
            f"Search results:\n{search_results}"
        ))
    ])
    return {"research": response.content}

Your Researcher agent is working and ready to be wired into the graph. Next up, you'll build the Writer agent and the Supervisor that decides which specialist works next.

Build the Writer Agent and Supervisor

Your researcher agent is working. It can search the web and synthesize structured notes on any topic.

Now you need the other two pieces of the pipeline: a writer that transforms research into a polished blog post, and a supervisor that decides which agent should work at each step. The supervisor is the brain of the system. It examines the current state and uses the LLM to make routing decisions at runtime.

In this step, get ready to:
  • Build a writer agent that converts research into a blog post.
  • Build a supervisor agent that uses LLM-based routing to coordinate the team.
  • Create a routing function that translates supervisor decisions into graph navigation.
Build the writer node

The writer agent has one job: take the research notes from the previous stage and produce a polished blog post. It reads from the shared state, just like the researcher writes to it.

  • In blog_team.py, scroll to the bottom of your file (below the researcher function).
  • Add the writer function by copying the following code:
def writer(state: BlogState) -> dict:
    response = llm.invoke([
        SystemMessage(content=(
            "You are a blog writer. Write an engaging, well-structured blog "
            "post based on the research provided. Include an introduction, "
            "clear sections with headings, and a conclusion. "
            "Aim for 500-800 words."
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n\n"
            f"Research notes:\n{state['research']}"
        ))
    ])
    return {"draft": response.content}

What does this code do?

The writer follows the same pattern as the researcher. It receives the full BlogState, reads the topic and research fields, then passes them to the LLM with a writing-focused system prompt.

The SystemMessage tells the LLM to act as a blog writer and aim for 500-800 words with clear structure. The function returns a dictionary with just the draft field. LangGraph merges this into the shared state automatically.

  • Save blog_team.py.

You won't see a change yet. The supervisor in the next chunk coordinates when this writer runs.

Build the supervisor node with LLM-based routing

In most multi-agent systems, you need something to decide which agent works next. The supervisor pattern uses the LLM itself to make this routing decision based on the current state.

Why use the LLM for routing?

You could hard-code the order (researcher then writer), but that removes flexibility. With LLM-based routing, the supervisor can adapt. If research is incomplete, it can route back to the researcher. If the task is already done, it can finish early.

This is the supervisor pattern used in production AI systems. The LLM acts as a manager, reading the current state and dispatching work dynamically.

  • Below the writer function in blog_team.py, add the supervisor function:
def supervisor(state: BlogState) -> dict:
    research = state.get("research", "")
    draft = state.get("draft", "")

    response = llm.invoke([
        SystemMessage(content=(
            "You are a supervisor managing a blog writing team. "
            "Your team has two specialists:\n"
            "- researcher: searches the web and gathers information\n"
            "- writer: writes blog posts from research\n\n"
            "Based on the current state, decide who should work next.\n"
            "If research is not done, respond: researcher\n"
            "If research is done but draft is not, respond: writer\n"
            "If both are done, respond: FINISH\n\n"
            "Respond with ONLY one word: researcher, writer, or FINISH"
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n"
            f"Research done: {'yes' if research else 'no'}\n"
            f"Draft done: {'yes' if draft else 'no'}"
        ))
    ])

    decision = response.content.strip().lower()
    if "researcher" in decision:
        return {"next_step": "researcher"}
    elif "writer" in decision:
        return {"next_step": "writer"}
    else:
        return {"next_step": "FINISH"}

What does this code do?

The supervisor checks two fields: research and draft. It uses state.get() with empty-string defaults so it never crashes on missing keys.

The SystemMessage gives the LLM explicit routing rules: if research is empty, pick the researcher. If research exists but no draft, pick the writer. If both are done, finish.

The parsing logic at the bottom uses in keyword checks instead of exact matching. This makes it resilient to slight LLM response variations (like extra whitespace or punctuation).

  • Save blog_team.py.

You won't see a change yet. The routing function in the next chunk connects the supervisor's decision to the graph.

Create the routing function

The supervisor sets a next_step value in the state, but LangGraph needs a function that translates that value into a node name for graph navigation. This is what add_conditional_edges expects.

  • Below the supervisor function in blog_team.py, add the route_next function:
def route_next(state: BlogState) -> str:
    next_step = state.get("next_step", "FINISH")
    if next_step == "researcher":
        return "researcher"
    elif next_step == "writer":
        return "writer"
    else:
        return END

What does this code do?

This function reads the next_step field from the state and returns the matching node name as a string. If the value is FINISH (or anything unexpected), it returns END to terminate the graph.

LangGraph's add_conditional_edges calls this function after each supervisor run to decide where execution flows next. This is how dynamic routing works at runtime.

  • Save blog_team.py.
  • Test the supervisor routing by adding this temporary test code at the very bottom of blog_team.py:
# Temporary test - delete after verifying
test_state = {"topic": "AI trends", "research": "", "draft": "", "next_step": ""}
result = supervisor(test_state)
print(f"No research -> {result['next_step']}")

test_state_2 = {"topic": "AI trends", "research": "Some notes here", "draft": "", "next_step": ""}
result_2 = supervisor(test_state_2)
print(f"Research done -> {result_2['next_step']}")
  • Run the file to verify routing decisions:
python blog_team.py

You should see two lines printed. The first should say No research -> researcher and the second should say Research done -> writer.

  • Delete the temporary test code (everything below the route_next function) and save the file.

Seeing unexpected routing?

If you see FINISH instead of the expected agent name, check that the SystemMessage in your supervisor function matches the code above exactly. The LLM relies on the precise wording of the routing rules.

If you get an API error, confirm your GOOGLE_API_KEY environment variable is still set in this terminal session.

Help me debug my supervisor routing

✔️ Awesome, I've got everything!

Great. Double check you've saved the file and deleted the temporary test code.

ⓧ I'd like to double check the full code

import operator
from typing import TypedDict

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class BlogState(TypedDict):
    topic: str
    research: str
    draft: str
    next_step: str


llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
search_tool = DuckDuckGoSearchRun()


def researcher(state: BlogState) -> dict:
    topic = state["topic"]
    search_results = search_tool.invoke(topic)

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research specialist. Analyze the search results and "
            "produce structured research notes with key facts, statistics, "
            "and insights. Organize by subtopic with bullet points."
        )),
        HumanMessage(content=(
            f"Research topic: {topic}\n\n"
            f"Search results:\n{search_results}"
        ))
    ])
    return {"research": response.content}


def writer(state: BlogState) -> dict:
    response = llm.invoke([
        SystemMessage(content=(
            "You are a blog writer. Write an engaging, well-structured blog "
            "post based on the research provided. Include an introduction, "
            "clear sections with headings, and a conclusion. "
            "Aim for 500-800 words."
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n\n"
            f"Research notes:\n{state['research']}"
        ))
    ])
    return {"draft": response.content}


def supervisor(state: BlogState) -> dict:
    research = state.get("research", "")
    draft = state.get("draft", "")

    response = llm.invoke([
        SystemMessage(content=(
            "You are a supervisor managing a blog writing team. "
            "Your team has two specialists:\n"
            "- researcher: searches the web and gathers information\n"
            "- writer: writes blog posts from research\n\n"
            "Based on the current state, decide who should work next.\n"
            "If research is not done, respond: researcher\n"
            "If research is done but draft is not, respond: writer\n"
            "If both are done, respond: FINISH\n\n"
            "Respond with ONLY one word: researcher, writer, or FINISH"
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n"
            f"Research done: {'yes' if research else 'no'}\n"
            f"Draft done: {'yes' if draft else 'no'}"
        ))
    ])

    decision = response.content.strip().lower()
    if "researcher" in decision:
        return {"next_step": "researcher"}
    elif "writer" in decision:
        return {"next_step": "writer"}
    else:
        return {"next_step": "FINISH"}


def route_next(state: BlogState) -> str:
    next_step = state.get("next_step", "FINISH")
    if next_step == "researcher":
        return "researcher"
    elif next_step == "writer":
        return "writer"
    else:
        return END

Your three agent functions are ready. Next up, you'll wire them into a LangGraph StateGraph with conditional edges so the supervisor can actually dispatch work to the researcher and writer at runtime.

Assemble and Run the Graph

You now have three agents that each do their job well individually. The missing piece is the orchestration layer that coordinates them into a single pipeline.

In this step, you will wire the supervisor, researcher, and writer into a LangGraph StateGraph with conditional edges. When you run the graph, the supervisor will dynamically decide which agent works next based on the current state.

In this step, get ready to:
  • Build a StateGraph with nodes for each agent and conditional routing edges.
  • Compile and run the full pipeline end-to-end on a test topic.
  • Inspect the final output to see how the supervisor routed between agents.
Build the graph

A StateGraph is the core structure in LangGraph that holds your nodes (agents) and edges (connections between them). Conditional edges let the graph choose a different path at runtime based on state values.

  • In blog_team.py, scroll to the bottom of the file (after the route_next function).
  • Add the graph assembly code by copying the following:
# Create the graph with BlogState as the shared state
graph = StateGraph(BlogState)

# Register each agent as a node
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

# Wire the edges: START -> supervisor, then conditional routing
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next, ["researcher", "writer", END])
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", END)

What does this code do?

This creates the structure of your multi-agent pipeline:

  • StateGraph(BlogState) creates a graph where every node reads from and writes to the same BlogState dictionary.
  • add_node registers each function as a named node the graph can invoke.
  • add_edge(START, "supervisor") makes the supervisor the first node that runs.
  • add_conditional_edges uses the route_next function to pick the next node at runtime based on what the supervisor decided.
  • add_edge("researcher", "supervisor") sends the researcher's output back to the supervisor so it can decide the next step.

Why does the researcher loop back to the supervisor?

This is the supervisor pattern in action. After the researcher finishes, control returns to the supervisor so it can re-evaluate the state and decide what happens next. The supervisor sees that research is now done, so it routes to the writer.

The writer does not loop back because once the draft is written, the pipeline is complete.

  • Save blog_team.py.

You won't see a change yet. The next chunk adds the code that actually runs this graph.

Compile and run the pipeline

A graph definition alone does nothing until you compile it into a runnable app. Compiling freezes the structure and prepares it for execution.

  • In blog_team.py, add the following code directly below the graph edges you just added:
# Set up checkpointer for state persistence
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Config identifies this conversation thread
config = {"configurable": {"thread_id": "blog-session-1"}}

print("=== Multi-Agent Blog Writing Team ===")
topic = input("Enter a blog topic: ")

# Run the full pipeline with an initial empty state
result = app.invoke({"topic": topic, "research": "", "draft": "", "next_step": ""}, config)

# Display the results
print("\n=== Final State ===")
print(f"Topic: {result['topic']}")
print(f"\nResearch:\n{result['research']}")
print(f"\nDraft:\n{result['draft']}")
print("\n=== Done! ===")

What does this code do?

  • MemorySaver() creates an in-memory checkpointer that tracks state between graph steps. You will use this for human-in-the-loop in the next step.
  • graph.compile() turns the graph definition into a runnable application.
  • thread_id identifies this specific conversation session for the checkpointer.
  • app.invoke() kicks off the pipeline with an initial state containing the topic and empty fields.
  • Save blog_team.py.

Seeing a syntax error on save?

  • Make sure the new code starts at the top-level indentation (no indent). It should not be inside any function.
  • Check for matching quotes in the f-strings. Each f"...{result['topic']}" uses double quotes outside and single quotes inside the braces.

Help me fix a syntax error in my blog_team.py graph assembly code.

Run the script and inspect the output

Your multi-agent pipeline is fully assembled. Time to run it and watch the supervisor coordinate the team.

  • Run the pipeline by executing this command in your terminal:
python blog_team.py
  • When prompted, enter a blog topic. Try something like the future of remote work.
  • Wait for the pipeline to complete. This takes 15-30 seconds as it makes several LLM and web search calls.

You should see the full final state printed in your terminal. It includes the topic, the structured research notes, and a complete blog post draft.

What just happened behind the scenes?

The supervisor made routing decisions at each step of the pipeline:

  • First call: the supervisor saw that research was empty, so it routed to the researcher.
  • Second call: the researcher completed and control returned to the supervisor. It saw research was done but draft was empty, so it routed to the writer.
  • The writer completed and its edge goes directly to END, finishing the pipeline.

Pipeline hanging or timing out?

  • Check your internet connection. The researcher needs web access for DuckDuckGo searches.
  • Make sure your GOOGLE_API_KEY environment variable is still set in this terminal session.
  • If you see a rate limit error, wait 60 seconds and try again. The free tier has per-minute limits.

Help me debug why my multi-agent pipeline is not completing.

✔️ Awesome, I've got everything!

Great. Double check you have saved blog_team.py.

ⓧ I'd like to double check the full code

Here is the full blog_team.py file at this point:

import operator
from typing import TypedDict

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class BlogState(TypedDict):
    topic: str
    research: str
    draft: str
    next_step: str


llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
search_tool = DuckDuckGoSearchRun()


def researcher(state: BlogState) -> dict:
    topic = state["topic"]
    search_results = search_tool.invoke(topic)

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research specialist. Analyze the search results and "
            "produce structured research notes with key facts, statistics, "
            "and insights. Organize by subtopic with bullet points."
        )),
        HumanMessage(content=(
            f"Research topic: {topic}\n\n"
            f"Search results:\n{search_results}"
        ))
    ])
    return {"research": response.content}


def writer(state: BlogState) -> dict:
    response = llm.invoke([
        SystemMessage(content=(
            "You are a blog writer. Write an engaging, well-structured blog "
            "post based on the research provided. Include an introduction, "
            "clear sections with headings, and a conclusion. "
            "Aim for 500-800 words."
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n\n"
            f"Research notes:\n{state['research']}"
        ))
    ])
    return {"draft": response.content}


def supervisor(state: BlogState) -> dict:
    research = state.get("research", "")
    draft = state.get("draft", "")

    response = llm.invoke([
        SystemMessage(content=(
            "You are a supervisor managing a blog writing team. "
            "Your team has two specialists:\n"
            "- researcher: searches the web and gathers information\n"
            "- writer: writes blog posts from research\n\n"
            "Based on the current state, decide who should work next.\n"
            "If research is not done, respond: researcher\n"
            "If research is done but draft is not, respond: writer\n"
            "If both are done, respond: FINISH\n\n"
            "Respond with ONLY one word: researcher, writer, or FINISH"
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n"
            f"Research done: {'yes' if research else 'no'}\n"
            f"Draft done: {'yes' if draft else 'no'}"
        ))
    ])

    decision = response.content.strip().lower()
    if "researcher" in decision:
        return {"next_step": "researcher"}
    elif "writer" in decision:
        return {"next_step": "writer"}
    else:
        return {"next_step": "FINISH"}


def route_next(state: BlogState) -> str:
    next_step = state.get("next_step", "FINISH")
    if next_step == "researcher":
        return "researcher"
    elif next_step == "writer":
        return "writer"
    else:
        return END


graph = StateGraph(BlogState)

graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next, ["researcher", "writer", END])
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "blog-session-1"}}

print("=== Multi-Agent Blog Writing Team ===")
topic = input("Enter a blog topic: ")

result = app.invoke({"topic": topic, "research": "", "draft": "", "next_step": ""}, config)

print("\n=== Final State ===")
print(f"Topic: {result['topic']}")
print(f"\nResearch:\n{result['research']}")
print(f"\nDraft:\n{result['draft']}")
print("\n=== Done! ===")

Your multi-agent system is running end-to-end. The supervisor coordinates the team dynamically, and you have a researched blog post as output. Next up, you will add a human-in-the-loop checkpoint so you can review the research before the writer starts.

Add Human-in-the-Loop Review

Your multi-agent pipeline runs end-to-end, but there's a problem. The writer blindly trusts whatever the researcher produces. In a real workflow, you'd want to review the research before committing to a full blog draft.

LangGraph solves this with an interrupt pattern. You can pause the graph at any node, inspect the state, and decide whether to continue or modify it. In this step, you'll add a human-in-the-loop checkpoint so the pipeline pauses after research and waits for your approval before writing.

In this step, get ready to:
  • Enable the interrupt so the graph pauses before the writer node.
  • Run the pipeline with pause-and-resume so you review research before writing begins.
  • Learn how to modify state mid-run for full human-in-the-loop control.
Add interrupt_before to the graph compile

The interrupt_before parameter tells LangGraph to pause execution right before a specific node runs. Your graph already has a MemorySaver checkpointer, which stores the graph's state so it can be resumed later.

  • In blog_team.py, scroll down to the graph.compile line near the bottom of the file.
  • Find this line:
app = graph.compile(checkpointer=checkpointer)
  • Add interrupt_before=["writer"] so the line now looks like this:
app = graph.compile(checkpointer=checkpointer, interrupt_before=["writer"])

What does interrupt_before do?

When the supervisor routes to the writer, LangGraph will stop execution right before the writer node starts. The graph's current state gets saved to the checkpointer so you can inspect it, modify it, or resume whenever you're ready.

The thread_id in your config identifies this specific session so the checkpointer knows which state to resume.

Update the invocation to pause and resume

Now that the graph pauses before writing, you need to split the invocation into two phases: run until the interrupt, then resume after review.

  • In blog_team.py, find the output section that starts with this line:
print("\n=== Final State ===")
print(f"Topic: {result['topic']}")
print(f"\nResearch:\n{result['research']}")
print(f"\nDraft:\n{result['draft']}")
print("\n=== Done! ===")
  • Delete those five lines and replace them with the following code that pauses for your review, then resumes the graph:
print("\n=== Research Complete (Paused for Review) ===")
print(result["research"])

input("\nPress Enter to approve and continue to writing...")

# Resume the graph from the interrupt point
result = app.invoke(None, config)

print("\n=== Blog Post Draft ===")
print(result["draft"])
print("\n=== Done! ===")

What does this code do?

The first app.invoke() runs the supervisor and researcher, then pauses. The research is printed so you can review it.

The input() call blocks execution until you press Enter. Calling app.invoke(None, config) with None as the input tells LangGraph to resume from the last saved checkpoint rather than starting fresh.

  • Save blog_team.py.
  • Run the script in your terminal by running:
python blog_team.py
  • Enter a blog topic when prompted (for example, benefits of remote work).
  • Wait for the pipeline to pause. You should see the research notes printed, followed by the prompt: Press Enter to approve and continue to writing...
  • Review the research output. When you're satisfied, press Enter to resume.
  • The writer agent runs and you should see the full blog post draft printed below.

Graph not pausing?

Make sure you added interrupt_before=["writer"] to the graph.compile() line and saved the file. Without this parameter, the graph runs straight through without stopping.

Also confirm the config dictionary with thread_id is passed to both app.invoke() calls. The checkpointer needs a matching thread to know which session to resume.

Help me debug why my LangGraph interrupt is not pausing before the writer node.

Modify state before resuming

Reviewing research is useful, but the real power of human-in-the-loop is editing the state before the next agent runs. LangGraph lets you update any field in the graph state while it's paused.

How does update_state work?

While the graph is paused at an interrupt, you can call app.update_state(config, {"research": "your modified research..."}) to overwrite any field in the state. When you resume with app.invoke(None, config), the writer agent picks up your modified research instead of the original.

This gives you the full pause, review, modify, resume cycle. In production, you might wire this to a web interface where a human editor refines the research before the writer starts drafting.

Try it out. After the graph pauses, open a Python REPL or modify your script to call update_state with revised research notes before pressing Enter to resume. The writer will use your edited version instead.

Extra for Experts: Why not just re-run the whole pipeline?

Re-running the entire pipeline wastes time and LLM calls on steps that already succeeded. The interrupt pattern lets you fix one stage without repeating earlier work. For complex pipelines with many agents, this saves significant time and cost.

✔️ Awesome, I've got everything!

Great. Double check you've saved blog_team.py and confirmed the graph pauses and resumes correctly.

ⓧ I'd like to double check the full code

Here is the complete blog_team.py file with the human-in-the-loop changes applied:

import operator
from typing import TypedDict

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class BlogState(TypedDict):
    topic: str
    research: str
    draft: str
    next_step: str


llm = ChatGoogleGenerativeAI(model="gemini-3.5-flash")
search_tool = DuckDuckGoSearchRun()


def researcher(state: BlogState) -> dict:
    topic = state["topic"]
    search_results = search_tool.invoke(topic)

    response = llm.invoke([
        SystemMessage(content=(
            "You are a research specialist. Analyze the search results and "
            "produce structured research notes with key facts, statistics, "
            "and insights. Organize by subtopic with bullet points."
        )),
        HumanMessage(content=(
            f"Research topic: {topic}\n\n"
            f"Search results:\n{search_results}"
        ))
    ])
    return {"research": response.content}


def writer(state: BlogState) -> dict:
    response = llm.invoke([
        SystemMessage(content=(
            "You are a blog writer. Write an engaging, well-structured blog "
            "post based on the research provided. Include an introduction, "
            "clear sections with headings, and a conclusion. "
            "Aim for 500-800 words."
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n\n"
            f"Research notes:\n{state['research']}"
        ))
    ])
    return {"draft": response.content}


def supervisor(state: BlogState) -> dict:
    research = state.get("research", "")
    draft = state.get("draft", "")

    response = llm.invoke([
        SystemMessage(content=(
            "You are a supervisor managing a blog writing team. "
            "Your team has two specialists:\n"
            "- researcher: searches the web and gathers information\n"
            "- writer: writes blog posts from research\n\n"
            "Based on the current state, decide who should work next.\n"
            "If research is not done, respond: researcher\n"
            "If research is done but draft is not, respond: writer\n"
            "If both are done, respond: FINISH\n\n"
            "Respond with ONLY one word: researcher, writer, or FINISH"
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n"
            f"Research done: {'yes' if research else 'no'}\n"
            f"Draft done: {'yes' if draft else 'no'}"
        ))
    ])

    decision = response.content.strip().lower()
    if "researcher" in decision:
        return {"next_step": "researcher"}
    elif "writer" in decision:
        return {"next_step": "writer"}
    else:
        return {"next_step": "FINISH"}


def route_next(state: BlogState) -> str:
    next_step = state.get("next_step", "FINISH")
    if next_step == "researcher":
        return "researcher"
    elif next_step == "writer":
        return "writer"
    else:
        return END


graph = StateGraph(BlogState)

graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next, ["researcher", "writer", END])
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer, interrupt_before=["writer"])

config = {"configurable": {"thread_id": "blog-session-1"}}

print("=== Multi-Agent Blog Writing Team ===")
topic = input("Enter a blog topic: ")

result = app.invoke({"topic": topic, "research": "", "draft": "", "next_step": ""}, config)

print("\n=== Research Complete (Paused for Review) ===")
print(result["research"])

input("\nPress Enter to approve and continue to writing...")

result = app.invoke(None, config)

print("\n=== Blog Post Draft ===")
print(result["draft"])
print("\n=== Done! ===")

Secret mission

Add a Fact-Checker Agent

Your blog team produces posts, but how do you know the research is accurate? In this extension, you'll add a third specialist agent that verifies claims before the writer starts. This is the same pattern production AI systems use to prevent hallucinated content from reaching users.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with free Gemini API calls, so there are no ongoing costs. Decide whether to keep your files for future experimentation or remove them entirely.

Resources you used:

  • Local project files: blog_team.py and test_setup.py.
  • Python packages: langgraph, langchain-google-genai, langchain-community, duckduckgo-search.
  • Environment variable: GOOGLE_API_KEY.

Keep everything running

No action needed. Choose this if you're still actively building or want to extend your multi-agent system with more agents.

  • Your blog_team.py is a complete, working multi-agent orchestration system you can use as a template for new projects.
  • The installed packages remain available for other LangGraph experiments.
  • Your Gemini API key stays set in your terminal session (it will clear when you close the terminal unless you added it to your shell profile).

Pause - I'll come back to this later

Since everything runs locally with free API calls, there is nothing to pause. Your files will be waiting for you when you come back.

  • Close VS Code to free up memory.
  • Your project files and installed packages remain on disk.
  • When you return, you may need to set your GOOGLE_API_KEY environment variable again if you set it only for the session.

Delete - I don't want to use this again

Remove all project files and uninstall the packages to start fresh.

  • Delete your project files by running this command from the folder that contains them:
rm blog_team.py test_setup.py
  • Uninstall the Python packages by running:
pip uninstall langgraph langchain-google-genai langchain-community duckduckgo-search -y
  • Clear the environment variable from your current terminal session:

macOS

unset GOOGLE_API_KEY

Windows

set GOOGLE_API_KEY=
  • If you added the key to your shell profile (e.g. .bashrc or .zshrc), open that file and remove the export GOOGLE_API_KEY=... line.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade multi-agent orchestration system where autonomous AI agents collaborate to research, verify, and write blog posts.

You've learned how to:

  • Built a multi-agent system with a Supervisor that uses LLM-based dynamic routing to coordinate specialist agents at runtime.
  • Wired a LangGraph StateGraph with conditional edges so the supervisor can route work to a Researcher or Writer based on the current state.
  • Added human-in-the-loop with MemorySaver checkpointing and interrupt_before to pause execution, review intermediate results, and resume on your terms.
  • Secret Mission: Added a Fact-Checker agent that verifies research claims before the writer starts, extending the supervisor's routing to coordinate three specialist agents.

Ready to quiz yourself?