Multi-Agent Research Pipeline

Build a LangGraph multi-agent pipeline with conditional routing and human-in-the-loop.

Introduction

30 Second Summary

When a company researches a complex topic, the work doesn't fall on one person. Someone figures out who should handle it, a subject-matter expert digs deep, and then someone else compiles it all into a clear report. Modern AI systems follow the same pattern, except each "person" is a specialized AI agent.

In this project, you will build a multi-agent research pipeline using LangGraph that routes questions to specialized AI agents, produces structured reports, and includes a human approval checkpoint before delivering the final output.

What You'll Build

A Python application that orchestrates multiple AI agents through a graph-based pipeline to classify queries, research topics, and synthesize structured reports with human oversight.

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

  • A typed state schema defining the data that flows between agents in your pipeline.
  • Three specialized agent nodes (Router, Researcher, Synthesizer) connected via conditional edges that dynamically route work based on intermediate results.
  • A human-in-the-loop approval gate that pauses execution and resumes on user input, demonstrating enterprise-grade oversight patterns.
  • Secret Mission: Add an automated quality feedback loop that sends reports back for revision if they don't meet quality criteria.

Are there any prerequisites?

You should have basic Python experience and an OpenAI API key ready to go. You'll also need Python 3.10 or higher installed on your machine.

Before We Start

Building AI agents that work together requires more than just calling an LLM. You need a way to coordinate multiple agents, pass data between them, and let humans intervene at critical decision points.

In this step, you will install the two Python packages that power your multi-agent pipeline, connect your OpenAI API key, and create your project file.

In this step, get ready to:
  • Install LangGraph and langchain-openai packages.
  • Set up your OpenAI API key as an environment variable.
  • Create the project file for your research pipeline.
Install LangGraph and dependencies

Your research pipeline uses two packages. LangGraph provides the graph-based orchestration framework for coordinating multiple agents. The langchain-openai package connects your pipeline to OpenAI's models.

  • Install both packages by running this command:
pip install langgraph==1.2.8 langchain-openai==1.3.3

What do these packages do?

LangGraph (1.2.8) gives you StateGraph, nodes, edges, and conditional routing for building multi-agent workflows. It also provides checkpointing and interrupt support for human-in-the-loop patterns.

The langchain-openai package (1.3.3) provides the ChatOpenAI class that wraps OpenAI's API. It reads your OPENAI_API_KEY environment variable automatically.

  • Verify LangGraph installed correctly by running:
python -c "import langgraph; print(langgraph.__version__)"

You should see 1.2.8 printed to the terminal.

Using macOS or Linux?

If python does not work on your system, try python3 instead. Some systems use python3 as the default command.

Seeing an error during installation?

  • If you see a message about Python version compatibility, confirm your Python version is 3.10 or higher by running python --version. LangGraph requires Python 3.10+.
  • If you see a network-related error, check your internet connection and try again.
  • If pip is not recognized, try python -m pip install langgraph==1.2.8 langchain-openai==1.3.3 instead.

Help me troubleshoot my LangGraph installation error.

Set up your OpenAI API key

Your agents need access to OpenAI's gpt-4o-mini model to classify queries, research topics, and write reports. The langchain-openai package reads the OPENAI_API_KEY environment variable automatically.

  • Set the environment variable in your terminal. Choose the tab for your operating system:

macOS

  • Set your API key by running this command (replace the placeholder with your actual key):
export OPENAI_API_KEY="your-api-key-here"

Linux users can use the same command above.

Windows

  • Set your API key by running this command in PowerShell (replace the placeholder with your actual key):
$env:OPENAI_API_KEY = "your-api-key-here"

Where do I find my OpenAI API key?

Go to platform.openai.com, sign in, and navigate to the API keys section. Create a new key if you do not have one.

Copy it and paste it in place of your-api-key-here in the command above. This environment variable only lasts for your current terminal session. You will need to set it again if you close and reopen your terminal.

  • Verify your API key works by running this command:
python -c "from langchain_openai import ChatOpenAI; print(ChatOpenAI(model='gpt-4o-mini').invoke('hello').content)"

You should see a short text response from the model printed to your terminal. The exact wording will vary.

Seeing an authentication error?

  • Double-check that you pasted your full API key with no extra spaces before or after.
  • Make sure you ran the export (macOS/Linux) or $env: (Windows) command in the same terminal session where you run the verification.
  • If you just created your OpenAI account, verify it has API credits available on your billing page.

Help me fix my OpenAI API key authentication error.

Create your project file

Your entire research pipeline will live in a single Python file. Let's create a project folder on your Desktop so it's easy to find.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir research-pipeline
cd research-pipeline
  • Create the empty Python file by running:
touch research_pipeline.py

Windows users and the touch command

If touch is not recognized in PowerShell, use New-Item research_pipeline.py instead. This creates the same empty file.

  • Confirm the file exists by running:
ls

You should see research_pipeline.py in the output.

  • Open your code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type the name of your editor (e.g. VS Code or Cursor) and press Enter to open it.
  • In your editor, click File in the top menu bar.
  • Click Open Folder.
  • Select the research-pipeline folder on your Desktop, then click Open.
  • In the Explorer sidebar, click on research_pipeline.py to open it.

Your environment is ready. Next up, you will define the typed state schema and build the three agent nodes that power your research pipeline.

Set Up Your Project

Your environment is installed and your API key is working. Now it's time to lay the foundation for the multi-agent pipeline itself.

Every LangGraph application starts with two things: imports that bring in the framework's building blocks, and a typed state schema that defines what data flows between your agent nodes. In this step, you'll set up both.

In this step, get ready to:
  • Add all project imports to research_pipeline.py.
  • Define the typed state schema that carries data between agent nodes.
  • Configure the LLM instance that powers all agent nodes.
Add the project imports

A LangGraph pipeline uses several modules: the graph builder (StateGraph), state markers (START, END), checkpointing (InMemorySaver), human-in-the-loop primitives (interrupt, Command), and the LLM wrapper (ChatOpenAI). You'll import all of these up front so each piece is available as you build out the pipeline.

  • Add the following imports to the top of research_pipeline.py:
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI

What do these imports do?

  • StateGraph, START, END are the core graph-building tools. They let you define nodes and wire them together.
  • InMemorySaver is a checkpointer that saves graph state in memory. This is required for the human-in-the-loop feature you'll add later.
  • interrupt and Command are the primitives for pausing a graph and resuming it with human input.
  • ChatOpenAI is the LangChain wrapper around OpenAI's chat models. It reads your OPENAI_API_KEY environment variable automatically.
  • TypedDict, Annotated, and Literal are Python typing tools used to define the state schema and routing return types.

You won't see a change yet. The next chunk uses these imports to define the pipeline's data structure.

Define the state schema and LLM instance

In LangGraph, a StateGraph is parameterized by a typed state object. This state is the data that flows between every node in your graph. Each node reads from the state, does work, and returns updates to it.

  • Add the state schema class and LLM setup below your imports in research_pipeline.py:
# --- State Schema ---

class ResearchState(TypedDict):
    query: str
    classification: str
    research_findings: str
    report: str
    human_approved: bool


# --- LLM Setup ---

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

What does this code do?

ResearchState defines the shape of data flowing through the pipeline. Each field represents a stage of the research workflow:

  • query holds the user's research question.
  • classification stores the category the router assigns (technical, business, or general).
  • research_findings holds the raw research output.
  • report stores the final structured report.
  • human_approved tracks whether the human reviewer approved the report.

The llm variable creates a shared LLM instance using gpt-4o-mini with temperature=0 for consistent, deterministic outputs. All agent nodes will share this instance.

  • Save research_pipeline.py.
  • Verify the file runs without errors by running:
python research_pipeline.py
  • You should see no output and no errors. The script exits silently because there's no executable logic yet.

Seeing an ImportError?

  • Make sure you installed the packages in the same Python environment you're running now. Run pip show langgraph to confirm it's installed.
  • Check that you're using Python 3.10 or higher. Run python --version to check.
  • If you see an error about typing_extensions, install it explicitly by running pip install typing_extensions.

Help me debug my import error in research_pipeline.py

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete research_pipeline.py file so far:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI


# --- State Schema ---

class ResearchState(TypedDict):
    query: str
    classification: str
    research_findings: str
    report: str
    human_approved: bool


# --- LLM Setup ---

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

Why TypedDict instead of a plain dictionary?

A TypedDict gives your state a fixed structure with named fields and types. LangGraph uses this schema to validate data as it flows between nodes. If you tried passing an unexpected key, you'd get a type-checker warning before runtime.

In production systems, this prevents a common class of agent bugs where one node writes a typo'd key and the downstream node silently reads None.

Your pipeline's foundation is in place. Next up, you'll create the agent node functions that read from this state, call the LLM, and write their results back.

Define Agent Nodes and Routing Logic

Your state schema and LLM connection are ready. Now it's time to build the agent nodes that do the actual work in your pipeline.

Each node in a LangGraph StateGraph is a Python function that receives the current state, performs work (like calling an LLM), and returns a dictionary of state updates. You'll create three specialized nodes and a routing function that controls the flow between them.

In this step, get ready to:
  • Create a router node that classifies research queries into categories.
  • Create researcher and synthesizer nodes that produce findings and structured reports.
  • Create a conditional routing function that controls the pipeline flow.
Create the router node

The router is the first decision-maker in your pipeline. It takes the incoming research query and classifies it into a category (technical, business, or general) so downstream nodes can tailor their behavior accordingly.

  • In research_pipeline.py, add the following code below the llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) line:


# --- Agent Nodes ---

def router_node(state: ResearchState) -> dict:
    """Classify the research query into a category."""
    response = llm.invoke(
        f"Classify this research query into exactly one category: "
        f"'technical', 'business', or 'general'. "
        f"Respond with only the category word.\n\n"
        f"Query: {state['query']}"
    )
    classification = response.content.strip().lower()
    print(f"[Router] Classified as: {classification}")
    return {"classification": classification}

What does this code do?

The router_node function follows the LangGraph node pattern: it accepts the current state as input and returns a dictionary of state updates.

  • It sends the research query to gpt-4o-mini with a prompt that forces a single-word classification response.
  • The .strip().lower() ensures consistent formatting regardless of how the model capitalizes its response.
  • The print() statement gives you visibility into the pipeline's decisions as it runs.
  • Save research_pipeline.py.

You won't see a change yet. The next chunk adds the researcher and synthesizer nodes that use this classification.

Create the researcher and synthesizer nodes

The researcher node uses the classification from the router to select a specialized system prompt. This means a technical query gets deep architecture analysis while a business query gets ROI-focused research.

  • Add the researcher node below the router_node function:


def researcher_node(state: ResearchState) -> dict:
    """Conduct research based on the classification."""
    system_prompts = {
        "technical": "You are a senior technical researcher. Provide detailed technical analysis with specific technologies, architectures, and implementation details.",
        "business": "You are a business strategy researcher. Focus on market trends, ROI, competitive advantages, and business impact.",
        "general": "You are a research analyst. Provide a balanced overview covering key facts, context, and practical implications.",
    }
    system_prompt = system_prompts.get(state["classification"], system_prompts["general"])

    response = llm.invoke(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Research this topic thoroughly:\n\n{state['query']}"},
        ]
    )
    print(f"[Researcher] Completed research ({len(response.content)} chars)")
    return {"research_findings": response.content}

What does this code do?

  • The system_prompts dictionary maps each classification to a tailored persona. This is how the same node produces different research styles.
  • The .get() with a fallback to "general" ensures the node works even if the router returns an unexpected classification.
  • The message list format (system + user messages) gives the LLM both a persona and a specific task, producing more focused output than a single string prompt.
  • Save research_pipeline.py.

You won't see a change yet. The synthesizer node below uses these research findings to build the final report.

The synthesizer takes raw research findings and structures them into a readable report with clear sections. This is the final output-producing node in the pipeline.

  • Add the synthesizer node below the researcher_node function:


def synthesizer_node(state: ResearchState) -> dict:
    """Synthesize research findings into a structured report."""
    response = llm.invoke(
        f"Based on the following research findings, create a structured report "
        f"with these sections:\n"
        f"1. Summary (2-3 sentences)\n"
        f"2. Key Points (bullet list)\n"
        f"3. Recommendations (actionable next steps)\n\n"
        f"Research findings:\n{state['research_findings']}"
    )
    print(f"[Synthesizer] Report generated")
    return {"report": response.content}

What does this code do?

  • The prompt enforces a consistent output structure: Summary, Key Points, and Recommendations. Every report follows the same format regardless of the topic.
  • It reads state['research_findings'] (written by the researcher) and writes to state['report']. This is how data flows between nodes through the shared state.
  • Save research_pipeline.py.
  • Verify the file has no errors by running:
python -c "import research_pipeline; print('All nodes defined successfully')"

You should see All nodes defined successfully printed in your terminal. If you see an error, check for missing colons or mismatched quotes in the functions you just added.

Seeing a SyntaxError or IndentationError?

  • Check that each function definition uses consistent 4-space indentation. Mixing tabs and spaces causes silent failures in Python.
  • Verify each f-string has matching opening and closing quotes. The multi-line f-strings in the router and synthesizer are a common place for missing quotes.
  • Make sure your file still has the ResearchState class and llm variable above the node functions.

Help me debug my node functions

Add conditional routing logic

In a real pipeline, you don't always want to proceed to the next step blindly. A conditional edge inspects the current state and decides which node to visit next. This is how enterprise workflows handle dynamic routing based on intermediate results.

  • Add the routing function below the synthesizer_node function:


# --- Routing Logic ---

def route_after_research(state: ResearchState) -> Literal["synthesizer_node", END]:
    """Route based on whether research findings are sufficient."""
    if state.get("research_findings") and len(state["research_findings"]) > 50:
        return "synthesizer_node"
    return END

What does this code do?

This routing function gets called by LangGraph after the researcher node finishes. It examines the state and returns the name of the next node to visit.

  • If research_findings exists and has more than 50 characters, the pipeline moves to the synthesizer.
  • If the findings are empty or too short, the pipeline ends early. This guards against wasting an LLM call on empty data.
  • The Literal type annotation tells LangGraph exactly which nodes this function might route to. This enables static validation of the graph structure.
  • Save research_pipeline.py.
  • Test the router node with a real API call by running:
python -c "from research_pipeline import router_node; print(router_node({'query': 'What is Kubernetes?', 'classification': '', 'research_findings': '', 'report': '', 'human_approved': False}))"

You should see output like [Router] Classified as: technical followed by the returned state update dictionary. This confirms your node is calling the OpenAI API and returning results in the correct format.

Seeing an AuthenticationError?

  • Confirm your OPENAI_API_KEY environment variable is still set in this terminal session. PowerShell variables reset when you close the window.
  • Check that your API key starts with sk- and has not expired on your OpenAI dashboard.

Help me fix my OpenAI API connection

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the full research_pipeline.py file as it should look at the end of this step:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI


# --- State Schema ---

class ResearchState(TypedDict):
    query: str
    classification: str
    research_findings: str
    report: str
    human_approved: bool


# --- LLM Setup ---

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


# --- Agent Nodes ---

def router_node(state: ResearchState) -> dict:
    """Classify the research query into a category."""
    response = llm.invoke(
        f"Classify this research query into exactly one category: "
        f"'technical', 'business', or 'general'. "
        f"Respond with only the category word.\n\n"
        f"Query: {state['query']}"
    )
    classification = response.content.strip().lower()
    print(f"[Router] Classified as: {classification}")
    return {"classification": classification}


def researcher_node(state: ResearchState) -> dict:
    """Conduct research based on the classification."""
    system_prompts = {
        "technical": "You are a senior technical researcher. Provide detailed technical analysis with specific technologies, architectures, and implementation details.",
        "business": "You are a business strategy researcher. Focus on market trends, ROI, competitive advantages, and business impact.",
        "general": "You are a research analyst. Provide a balanced overview covering key facts, context, and practical implications.",
    }
    system_prompt = system_prompts.get(state["classification"], system_prompts["general"])

    response = llm.invoke(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Research this topic thoroughly:\n\n{state['query']}"},
        ]
    )
    print(f"[Researcher] Completed research ({len(response.content)} chars)")
    return {"research_findings": response.content}


def synthesizer_node(state: ResearchState) -> dict:
    """Synthesize research findings into a structured report."""
    response = llm.invoke(
        f"Based on the following research findings, create a structured report "
        f"with these sections:\n"
        f"1. Summary (2-3 sentences)\n"
        f"2. Key Points (bullet list)\n"
        f"3. Recommendations (actionable next steps)\n\n"
        f"Research findings:\n{state['research_findings']}"
    )
    print(f"[Synthesizer] Report generated")
    return {"report": response.content}


# --- Routing Logic ---

def route_after_research(state: ResearchState) -> Literal["synthesizer_node", END]:
    """Route based on whether research findings are sufficient."""
    if state.get("research_findings") and len(state["research_findings"]) > 50:
        return "synthesizer_node"
    return END

Why three separate nodes instead of one big function?

Splitting logic into separate nodes is the core enterprise pattern for agentic AI. Each node has a single responsibility, which means you can test, swap, or scale them independently.

In a production system, you might replace the researcher node with a retrieval-augmented generation (RAG) pipeline, or add a caching layer before the synthesizer. The graph structure makes these changes surgical instead of requiring a rewrite.

Your agent nodes are defined and tested. Next up, you'll wire them together into a functioning StateGraph with conditional edges and run the full pipeline end-to-end.

Build the StateGraph with Conditional Routing

In the previous step, you created the building blocks of your pipeline: a typed state schema and three specialized agent nodes.

Now you need to orchestrate them. LangGraph's StateGraph connects your nodes into a directed workflow where data flows through edges and routing decisions happen dynamically at runtime.

In this step, get ready to:
  • Wire the three agent nodes into a StateGraph.
  • Add a conditional edge that routes based on research quality.
  • Compile and test the full pipeline end-to-end.
Construct the StateGraph and wire the edges

The StateGraph builder pattern lets you register nodes, connect them with edges, and then compile the whole thing into an executable graph.

  • In research_pipeline.py, scroll to the bottom of the file (below the route_after_research function) and add the graph construction code:
# --- Build the Graph ---

builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("router_node", router_node)
builder.add_node("researcher_node", researcher_node)
builder.add_node("synthesizer_node", synthesizer_node)

# Add edges
builder.add_edge(START, "router_node")
builder.add_edge("router_node", "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_research, ["synthesizer_node", END])
builder.add_edge("synthesizer_node", END)

# Compile without checkpointer for initial test
graph = builder.compile()

What does this code do?

builder = StateGraph(ResearchState) creates a new graph parameterized by your typed state schema. Every node in this graph reads from and writes to ResearchState.

add_node() registers each agent function as a named node. add_edge() creates a fixed connection between two nodes, guaranteeing execution order.

add_conditional_edges() is the key enterprise pattern here. Instead of a fixed path, it calls route_after_research at runtime to decide the next node based on the current state.

builder.compile() validates the graph structure (no orphaned nodes, valid edges) and returns an executable object you can call with .invoke().

  • Save research_pipeline.py.

Seeing a NameError?

Make sure the graph construction code is placed after all your function definitions. Python reads top to bottom, so router_node, researcher_node, and synthesizer_node must all be defined before you reference them in add_node().

Also check that StateGraph, START, and END are all imported at the top of your file.

Help me debug my StateGraph construction

Understand conditional routing

The conditional edge after researcher_node is what makes this pipeline intelligent rather than just sequential.

Why use conditional routing?

In a production AI system, you rarely want a fixed pipeline. Different inputs need different paths. Conditional edges let the graph inspect intermediate state and branch dynamically.

Here, route_after_research checks whether the research findings are long enough to be useful. If findings are too short (under 50 characters), the graph skips synthesis and ends early. Otherwise, it proceeds to synthesizer_node.

This is the enterprise pattern of dynamic workflow routing based on intermediate results. In larger systems, you might route to entirely different agent branches depending on query complexity or confidence scores.

Test the pipeline end-to-end

Your graph is compiled and ready to run. Time to send a real query through all three nodes and see a structured report come out the other end.

  • Test the pipeline by running this command in your terminal:
python -c "from research_pipeline import graph; result = graph.invoke({'query': 'What are the best practices for microservices architecture?'}); print(result['report'])"

What should I see?

You should see three print statements from your nodes showing the pipeline flow:

  • [Router] Classified as: technical (or similar classification).
  • [Researcher] Completed research (XXXX chars) showing research was generated.
  • [Synthesizer] Report generated confirming synthesis completed.

After those status lines, you should see a full structured report with Summary, Key Points, and Recommendations sections.

Seeing an API key error?

If you see an authentication error, your OPENAI_API_KEY environment variable may not be set in this terminal session. Re-run the export command from Step 1.

If you see a module import error, make sure you saved research_pipeline.py and that you are running the command from the same directory where the file is located.

Help me debug my LangGraph pipeline test

✔️ Awesome, I've got everything!

Great. Double check you have saved your file.

ⓧ I'd like to double check the full code

Here is the complete research_pipeline.py file at this point:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI


# --- State Schema ---

class ResearchState(TypedDict):
    query: str
    classification: str
    research_findings: str
    report: str
    human_approved: bool


# --- LLM Setup ---

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


# --- Agent Nodes ---

def router_node(state: ResearchState) -> dict:
    """Classify the research query into a category."""
    response = llm.invoke(
        f"Classify this research query into exactly one category: "
        f"'technical', 'business', or 'general'. "
        f"Respond with only the category word.\n\n"
        f"Query: {state['query']}"
    )
    classification = response.content.strip().lower()
    print(f"[Router] Classified as: {classification}")
    return {"classification": classification}


def researcher_node(state: ResearchState) -> dict:
    """Conduct research based on the classification."""
    system_prompts = {
        "technical": "You are a senior technical researcher. Provide detailed technical analysis with specific technologies, architectures, and implementation details.",
        "business": "You are a business strategy researcher. Focus on market trends, ROI, competitive advantages, and business impact.",
        "general": "You are a research analyst. Provide a balanced overview covering key facts, context, and practical implications.",
    }
    system_prompt = system_prompts.get(state["classification"], system_prompts["general"])

    response = llm.invoke(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Research this topic thoroughly:\n\n{state['query']}"},
        ]
    )
    print(f"[Researcher] Completed research ({len(response.content)} chars)")
    return {"research_findings": response.content}


def synthesizer_node(state: ResearchState) -> dict:
    """Synthesize research findings into a structured report."""
    response = llm.invoke(
        f"Based on the following research findings, create a structured report "
        f"with these sections:\n"
        f"1. Summary (2-3 sentences)\n"
        f"2. Key Points (bullet list)\n"
        f"3. Recommendations (actionable next steps)\n\n"
        f"Research findings:\n{state['research_findings']}"
    )
    print(f"[Synthesizer] Report generated")
    return {"report": response.content}


# --- Routing Logic ---

def route_after_research(state: ResearchState) -> Literal["synthesizer_node", END]:
    """Route based on whether research findings are sufficient."""
    if state.get("research_findings") and len(state["research_findings"]) > 50:
        return "synthesizer_node"
    return END


# --- Build the Graph ---

builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("router_node", router_node)
builder.add_node("researcher_node", researcher_node)
builder.add_node("synthesizer_node", synthesizer_node)

# Add edges
builder.add_edge(START, "router_node")
builder.add_edge("router_node", "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_research, ["synthesizer_node", END])
builder.add_edge("synthesizer_node", END)

# Compile without checkpointer for initial test
graph = builder.compile()

Your pipeline produces a structured research report in a single pass. Next up, you will add a human-in-the-loop approval gate that pauses the pipeline and waits for your decision before finalizing the output.

Add Human-in-the-Loop Approval Gate

Your pipeline currently runs from start to finish without any human oversight. The router classifies, the researcher investigates, and the synthesizer produces a report. But in production AI systems, you often need a human to review and approve outputs before they're finalized.

In this step, you'll add a human-in-the-loop approval gate using LangGraph's interrupt() and Command(resume=...) pattern. The graph will pause after synthesis, surface the draft report for your review, and wait for you to approve or reject it before continuing.

In this step, get ready to:
  • Create a human review node that pauses execution for approval.
  • Rewire the graph with a checkpointer to support state persistence.
  • Build an interactive execution loop that handles interrupt and resume.
Create the human review node

The interrupt() function is LangGraph's way of pausing a running graph mid-execution. When a node calls interrupt(), the graph freezes its state, surfaces a payload to the caller, and waits for a human to resume it with a decision.

  • In research_pipeline.py, find the # --- Routing Logic --- comment.
  • Add the following function directly above that comment, after the synthesizer_node function:
def human_review_node(state: ResearchState) -> dict:
    """Pause for human approval of the report."""
    decision = interrupt(
        {
            "report": state["report"],
            "message": "Please review the report above. Type 'approve' to accept or 'revise' to request changes.",
        }
    )
    if decision == "approve":
        print("[Human Review] Report approved!")
        return {"human_approved": True}
    else:
        print("[Human Review] Report rejected.")
        return {"human_approved": False}

What does this code do?

This node calls interrupt() with a payload dictionary containing the draft report and a message prompt. When the graph hits this node, execution freezes.

Later, when a human resumes the graph with Command(resume=...), the resume value becomes the return value of interrupt(). So decision will hold whatever the user typed ("approve" or anything else).

  • Save the file. You won't see a change yet. The next chunk wires this node into the graph.
Rewire the graph with a checkpointer

For interrupt() to work, the graph needs a checkpointer. A checkpointer persists the graph's state between invocations so it can freeze, wait for human input, and resume exactly where it left off.

  • Find the # --- Build the Graph --- section in your file.
  • Replace the entire section (from that comment down to the end of the file) with the following:
# --- Build the Graph ---

builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("router_node", router_node)
builder.add_node("researcher_node", researcher_node)
builder.add_node("synthesizer_node", synthesizer_node)
builder.add_node("human_review_node", human_review_node)

# Add edges
builder.add_edge(START, "router_node")
builder.add_edge("router_node", "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_research, ["synthesizer_node", END])
builder.add_edge("synthesizer_node", "human_review_node")
builder.add_edge("human_review_node", END)

# Compile with checkpointer for human-in-the-loop
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

What changed here?

  • You registered human_review_node as a new node in the graph.
  • The edge from synthesizer_node now points to human_review_node instead of END. The human review node then connects to END.
  • InMemorySaver() provides the checkpointer that persists state across the interrupt. Without it, the graph would lose its progress when paused.

What is a checkpointer?

A checkpointer saves a snapshot of the graph's state after every node executes. InMemorySaver stores these snapshots in memory (fine for development). In production, you'd use a database-backed checkpointer so state survives process restarts.

  • Save the file. You won't see a change yet. The next chunk adds the execution loop that actually runs the graph and handles the interrupt.
Build the interactive execution loop

The graph now pauses at the human review node, but you need code that detects the pause, shows you the report, collects your decision, and resumes the graph. This is the interrupt/resume cycle.

  • Add the following code at the very bottom of research_pipeline.py, after the graph = builder.compile(checkpointer=checkpointer) line:
# --- Main Execution ---

def run_pipeline():
    """Run the research pipeline with human-in-the-loop."""
    query = input("Enter your research question: ")

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

    # First invocation - runs until interrupt
    print("\n--- Starting Research Pipeline ---\n")
    result = graph.invoke({"query": query}, config=config)

    # Check if we hit an interrupt
    state = graph.get_state(config)
    while state.next:
        # We're at an interrupt - get the interrupt payload
        interrupt_value = state.tasks[0].interrupts[0].value
        print(f"\n--- HUMAN REVIEW REQUIRED ---")
        print(f"\nReport:\n{interrupt_value['report']}")
        print(f"\n{interrupt_value['message']}")

        decision = input("\nYour decision: ").strip().lower()

        # Resume the graph with the human's decision
        result = graph.invoke(Command(resume=decision), config=config)
        state = graph.get_state(config)

    print("\n--- Pipeline Complete ---")
    print(f"\nFinal Report:\n{result.get('report', 'No report generated')}")
    print(f"Approved: {result.get('human_approved', False)}")


if __name__ == "__main__":
    run_pipeline()

What does this code do?

  • config with a thread_id tells the checkpointer which session to save state for. This is how LangGraph tracks separate conversations.
  • graph.invoke() runs the graph until it either completes or hits an interrupt.
  • graph.get_state(config) checks the current state. If state.next is non-empty, there are pending nodes (the graph is paused).
  • state.tasks[0].interrupts[0].value retrieves the payload you passed to interrupt() (the report and message).
  • Command(resume=decision) injects the human's decision back into the graph. The graph unfreezes and interrupt() returns that value.
  • Save research_pipeline.py.
  • Run the pipeline by executing this command in your terminal:
python research_pipeline.py
  • When prompted, enter a research question like: What are the best practices for microservices architecture?
  • Watch the pipeline flow through the router and researcher nodes. After synthesis, it should pause and display your draft report with the prompt asking you to approve or revise.
  • Type approve and press Enter. You should see [Human Review] Report approved! followed by the final report and Approved: True.

Test the rejection path too

Run the script again and type anything other than approve (like revise) to see the rejection path. You should see [Human Review] Report rejected. and Approved: False.

Pipeline not pausing for input?

  • Make sure you have checkpointer = InMemorySaver() and are passing it to builder.compile(checkpointer=checkpointer). Without a checkpointer, interrupt() raises an error instead of pausing.
  • Check that the edge goes from synthesizer_node to human_review_node, not directly to END. If the graph skips the human node, your edge wiring is off.
  • Verify that your config dictionary includes "configurable": {"thread_id": "research-session-1"}. The checkpointer needs a thread ID to track state.

Why isn't my LangGraph pipeline pausing at the human review node?

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete research_pipeline.py at this point:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI


# --- State Schema ---

class ResearchState(TypedDict):
    query: str
    classification: str
    research_findings: str
    report: str
    human_approved: bool


# --- LLM Setup ---

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


# --- Agent Nodes ---

def router_node(state: ResearchState) -> dict:
    """Classify the research query into a category."""
    response = llm.invoke(
        f"Classify this research query into exactly one category: "
        f"'technical', 'business', or 'general'. "
        f"Respond with only the category word.\n\n"
        f"Query: {state['query']}"
    )
    classification = response.content.strip().lower()
    print(f"[Router] Classified as: {classification}")
    return {"classification": classification}


def researcher_node(state: ResearchState) -> dict:
    """Conduct research based on the classification."""
    system_prompts = {
        "technical": "You are a senior technical researcher. Provide detailed technical analysis with specific technologies, architectures, and implementation details.",
        "business": "You are a business strategy researcher. Focus on market trends, ROI, competitive advantages, and business impact.",
        "general": "You are a research analyst. Provide a balanced overview covering key facts, context, and practical implications.",
    }
    system_prompt = system_prompts.get(state["classification"], system_prompts["general"])

    response = llm.invoke(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Research this topic thoroughly:\n\n{state['query']}"},
        ]
    )
    print(f"[Researcher] Completed research ({len(response.content)} chars)")
    return {"research_findings": response.content}


def synthesizer_node(state: ResearchState) -> dict:
    """Synthesize research findings into a structured report."""
    response = llm.invoke(
        f"Based on the following research findings, create a structured report "
        f"with these sections:\n"
        f"1. Summary (2-3 sentences)\n"
        f"2. Key Points (bullet list)\n"
        f"3. Recommendations (actionable next steps)\n\n"
        f"Research findings:\n{state['research_findings']}"
    )
    print(f"[Synthesizer] Report generated")
    return {"report": response.content}


def human_review_node(state: ResearchState) -> dict:
    """Pause for human approval of the report."""
    decision = interrupt(
        {
            "report": state["report"],
            "message": "Please review the report above. Type 'approve' to accept or 'revise' to request changes.",
        }
    )
    if decision == "approve":
        print("[Human Review] Report approved!")
        return {"human_approved": True}
    else:
        print("[Human Review] Report rejected.")
        return {"human_approved": False}


# --- Routing Logic ---

def route_after_research(state: ResearchState) -> Literal["synthesizer_node", END]:
    """Route based on whether research findings are sufficient."""
    if state.get("research_findings") and len(state["research_findings"]) > 50:
        return "synthesizer_node"
    return END


# --- Build the Graph ---

builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("router_node", router_node)
builder.add_node("researcher_node", researcher_node)
builder.add_node("synthesizer_node", synthesizer_node)
builder.add_node("human_review_node", human_review_node)

# Add edges
builder.add_edge(START, "router_node")
builder.add_edge("router_node", "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_research, ["synthesizer_node", END])
builder.add_edge("synthesizer_node", "human_review_node")
builder.add_edge("human_review_node", END)

# Compile with checkpointer for human-in-the-loop
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)


# --- Main Execution ---

def run_pipeline():
    """Run the research pipeline with human-in-the-loop."""
    query = input("Enter your research question: ")

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

    # First invocation - runs until interrupt
    print("\n--- Starting Research Pipeline ---\n")
    result = graph.invoke({"query": query}, config=config)

    # Check if we hit an interrupt
    state = graph.get_state(config)
    while state.next:
        # We're at an interrupt - get the interrupt payload
        interrupt_value = state.tasks[0].interrupts[0].value
        print(f"\n--- HUMAN REVIEW REQUIRED ---")
        print(f"\nReport:\n{interrupt_value['report']}")
        print(f"\n{interrupt_value['message']}")

        decision = input("\nYour decision: ").strip().lower()

        # Resume the graph with the human's decision
        result = graph.invoke(Command(resume=decision), config=config)
        state = graph.get_state(config)

    print("\n--- Pipeline Complete ---")
    print(f"\nFinal Report:\n{result.get('report', 'No report generated')}")
    print(f"Approved: {result.get('human_approved', False)}")


if __name__ == "__main__":
    run_pipeline()

Why this pattern matters for enterprise AI

In production AI systems, automated outputs often need human sign-off before triggering downstream actions (sending emails, publishing content, executing trades). The interrupt/resume pattern decouples the AI pipeline from the approval process.

The checkpointer means the graph can wait hours or days for a human response without losing progress. This is the same architecture pattern used at companies like Klarna and Uber for their agentic AI workflows.

Your pipeline now has a genuine human oversight gate. The graph pauses, surfaces the draft for review, and only continues after explicit human approval. Next up, you'll extend this into a quality feedback loop in the Secret Mission.

Secret mission

Add a Quality Feedback Loop

A production agentic system doesn't just get human approval. It also has automated quality gates. Can you add a quality-checking node that evaluates the synthesized report and conditionally routes back to the researcher for more detail if it doesn't meet standards?

Clean Up Your Resources

Clean Up Your Resources

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

This project runs locally and uses the OpenAI API on a pay-per-use basis. Once you stop running the script, there are no ongoing costs.

Resources you used:

  • research_pipeline.py local file.
  • langgraph and langchain-openai Python packages.
  • OPENAI_API_KEY environment variable (session-only, clears automatically when you close your terminal).

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the pipeline further.

  • Your research_pipeline.py file stays on your machine ready to run.
  • The langgraph and langchain-openai packages remain installed in your Python environment.
  • Your OpenAI API key is only active for the current terminal session. You will need to set it again next time you open a new terminal.

Pause - I'll come back to this later

Since this project runs only when you execute the script manually, there is nothing to pause. No background processes or services are running.

  • Your file and packages stay installed and ready for whenever you come back.
  • The OPENAI_API_KEY environment variable already clears itself when you close the terminal.
  • No ongoing API charges occur while the script is not running.

Delete - I don't want to use this again

Remove all project files and packages to start fresh.

  • Delete the project file by running this command in the folder where you created it:

macOS

rm research_pipeline.py

Windows

del research_pipeline.py
  • Uninstall the Python packages by running:
pip uninstall langgraph langchain-openai -y
  • The OPENAI_API_KEY environment variable clears automatically when you close your terminal. No action needed.
  • Your OpenAI account remains active. You can manage or delete API keys from the OpenAI API keys dashboard if you no longer need them.

Nice Work!

Nice Work!

Nice work! You just built a production-grade multi-agent research pipeline using LangGraph. Your system routes queries through specialized agents, synthesizes structured reports, and includes a human approval gate that pauses and resumes execution on demand.

You've learned how to:

  • Build a multi-agent research pipeline using LangGraph's StateGraph with typed state and conditional routing between specialized agent nodes.
  • Implement a human-in-the-loop approval gate using LangGraph's interrupt() and Command(resume=...) pattern with InMemorySaver checkpointing for state persistence.
  • Demonstrate enterprise agentic AI architecture patterns directly referenced in AI Engineer job listings: graph-based orchestration, typed state management, dynamic routing, and human oversight.
  • Secret Mission: Added an automated quality feedback loop with cycle detection and max-retry logic that sends reports back for revision when they don't meet quality criteria.

Ready to quiz yourself?