Introduction to Multi-Agent Research Pipelines
Modern research tasks rarely fit neatly into a single prompt-response cycle. Whether you're building a market intelligence tool, an academic literature reviewer, or a competitive analysis engine, the workflow typically involves multiple specialized steps: planning, searching, reading, synthesizing, fact-checking, and reporting. Each step benefits from a focused agent with a clear role, its own prompt, and access to specific tools. This is where LangGraph shines.
LangGraph, built by the team behind LangChain, is a library for orchestrating stateful, multi-actor applications using large language models. It treats your agent workflow as a directed graph where nodes represent computation steps (often individual agents or functions) and edges represent the flow of state between them. This graph-based approach gives you fine-grained control over routing, looping, parallelism, and human-in-the-loop checkpoints — all things that become painful quickly when you try to build them with plain LangChain chains.
Why Multi-Agent Systems Matter
Single-agent systems hit a ceiling fast. A single LLM trying to plan, search, read, synthesize, and write a report in one pass tends to produce shallow, unfocused output. Context windows fill up with irrelevant retrieved content, the model loses track of its original goal, and debugging becomes a guessing game because everything happens inside one opaque call.
Multi-agent architectures solve this by decomposing the problem. Each agent gets a narrow responsibility, a tailored system prompt, and only the context it actually needs. The result is better quality, easier debugging, and a system you can extend by adding new nodes rather than rewriting a monolithic prompt.
Core LangGraph Concepts
Before we build, let's lock down the vocabulary LangGraph uses, because the mental model is different from typical LangChain usage.
State
State is the shared data structure that flows through your graph. In LangGraph, state is typically a TypedDict or Pydantic model. Every node receives the current state, does its work, and returns a partial state update. LangGraph merges these updates automatically. This means each node only needs to know about the slice of state it cares about.
Nodes
Nodes are Python functions. Each node takes the state as input and returns a dictionary containing the fields it wants to update. A node might wrap an LLM call, run a tool, execute a search, or even call out to an external API. Nodes are the workhorses of your graph.
Edges
Edges define how control moves between nodes. A standard edge always goes from one node to another. A conditional edge uses a function to decide which node to visit next based on the current state. Conditional edges are what enable looping, branching, and dynamic routing — the behaviors that make agentic workflows feel intelligent.
The Checkpointer
LangGraph can persist state at every node transition using a checkpointer. This enables human-in-the-loop pauses, fault recovery, and time-travel debugging. For a research pipeline, this is invaluable: you can inspect intermediate findings, correct a bad search query, or resume a long-running job after a crash.
Designing the Research Pipeline
For this tutorial, we'll build a research pipeline with four cooperating agents:
- Planner Agent — Decomposes the research question into a list of focused sub-questions.
- Searcher Agent — Runs web searches for each sub-question and collects raw results.
- Analyst Agent — Reads the raw results and extracts key findings relevant to the original question.
- Writer Agent — Synthesizes the findings into a structured final report.
The flow is linear with one loop: after the Analyst produces findings, we check whether we have enough coverage. If not, we route back to the Planner to generate additional sub-questions. This loop is what separates a real research pipeline from a glorified search-and-summarize script.
Setting Up the Project
Install the required packages. We'll use LangGraph, LangChain, and Tavily for web search.
pip install langgraph langchain langchain-openai langchain-community tavily-python
Set your API keys as environment variables:
import os
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
os.environ["TAVILY_API_KEY"] = "tvly-your-key-here"
Defining the State
The state is the contract between every node in the graph. Design it carefully — every field should have a clear purpose and a clear owner (the node that writes to it).
from typing import TypedDict, List, Annotated
import operator
class ResearchState(TypedDict):
question: str
sub_questions: List[str]
search_results: Annotated[List[dict], operator.add]
findings: Annotated[List[str], operator.add]
report: str
iteration: int
max_iterations: int
coverage_sufficient: bool
Notice the Annotated fields with operator.add. This tells LangGraph to concatenate lists returned by multiple nodes rather than overwriting them. This is essential when the Searcher runs multiple times across iterations and we want to accumulate results.
Building the Planner Agent
The Planner takes the original question and produces sub-questions. On subsequent iterations, it receives feedback about what's missing and generates targeted follow-up questions.
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
def planner_node(state: ResearchState) -> dict:
iteration = state.get("iteration", 0)
if iteration == 0:
prompt = [
SystemMessage(content=(
"You are a research planner. Break the user's research question "
"into 3 to 5 focused sub-questions that, when answered together, "
"will fully address the original question. Return only a JSON "
"array of strings, no commentary."
)),
HumanMessage(content=f"Research question: {state['question']}")
]
else:
existing_findings = "\n".join(f"- {f}" for f in state.get("findings", []))
prompt = [
SystemMessage(content=(
"You are a research planner. The current findings do not fully "
"answer the research question. Generate 2 to 3 NEW sub-questions "
"that target the gaps. Avoid repeating previous sub-questions. "
"Return only a JSON array of strings."
)),
HumanMessage(content=(
f"Original question: {state['question']}\n"
f"Existing findings:\n{existing_findings}"
))
]
response = llm.invoke(prompt)
import json
sub_questions = json.loads(response.content)
return {
"sub_questions": sub_questions,
"iteration": iteration + 1
}
Building the Searcher Agent
The Searcher uses Tavily to fetch web results for each sub-question. We map over the sub-questions and collect all results into the accumulated list.
from langchain_community.tools.tavily_search import TavilySearchResults
search_tool = TavilySearchResults(max_results=3)
def searcher_node(state: ResearchState) -> dict:
results = []
for sub_q in state["sub_questions"]:
try:
hits = search_tool.invoke(sub_q)
for hit in hits:
results.append({
"sub_question": sub_q,
"title": hit.get("title", ""),
"url": hit.get("url", ""),
"content": hit.get("content", "")
})
except Exception as e:
results.append({
"sub_question": sub_q,
"title": "Search Error",
"url": "",
"content": f"Search failed: {str(e)}"
})
return {"search_results": results}
Building the Analyst Agent
The Analyst reads the raw search results and extracts concise findings. It also evaluates whether the current findings sufficiently answer the original question.
def analyst_node(state: ResearchState) -> dict:
# Format search results for the LLM
context_parts = []
for i, r in enumerate(state["search_results"]):
context_parts.append(
f"[{i+1}] Sub-question: {r['sub_question']}\n"
f"Source: {r['title']} ({r['url']})\n"
f"Content: {r['content'][:1500]}"
)
context = "\n\n".join(context_parts)
prompt = [
SystemMessage(content=(
"You are a research analyst. Extract key findings from the provided "
"search results that are relevant to the original research question. "
"For each finding, write one clear, factual sentence with a source "
"citation in brackets like [1]. Then, on a new line starting with "
"COVERAGE:, answer 'yes' if the findings collectively provide a "
"thorough answer to the research question, or 'no' if significant "
"gaps remain."
)),
HumanMessage(content=(
f"Research question: {state['question']}\n\n"
f"Search results:\n{context}"
))
]
response = llm.invoke(prompt)
text = response.content
# Split findings from coverage verdict
if "COVERAGE:" in text:
findings_text, coverage_text = text.rsplit("COVERAGE:", 1)
coverage_sufficient = "yes" in coverage_text.strip().lower()
else:
findings_text = text
coverage_sufficient = False
# Each line that starts with a dash or number is a finding
findings = [
line.strip().lstrip("- ").lstrip("0123456789. ")
for line in findings_text.strip().split("\n")
if line.strip() and (line.strip().startswith("-") or line.strip()[0].isdigit())
]
return {
"findings": findings,
"coverage_sufficient": coverage_sufficient
}
Building the Writer Agent
The Writer takes all accumulated findings and produces a polished, structured report.
def writer_node(state: ResearchState) -> dict:
findings_text = "\n".join(f"- {f}" for f in state["findings"])
prompt = [
SystemMessage(content=(
"You are a research report writer. Using only the provided findings, "
"write a structured report answering the research question. Use "
"markdown headings. Include an executive summary, key findings "
"organized by theme, and a conclusion. Preserve source citations "
"from the findings. Do not invent information not present in the "
"findings."
)),
HumanMessage(content=(
f"Research question: {state['question']}\n\n"
f"Findings:\n{findings_text}"
))
]
response = llm.invoke(prompt)
return {"report": response.content}
Adding the Routing Logic
After the Analyst runs, we need a conditional edge that decides whether to loop back to the Planner or proceed to the Writer. We also enforce a maximum iteration count to prevent infinite loops.
def should_continue_research(state: ResearchState) -> str:
if state.get("coverage_sufficient", False):
return "write"
if state.get("iteration", 0) >= state.get("max_iterations", 3):
return "write"
return "plan"
Assembling the Graph
Now we wire everything together using LangGraph's StateGraph API.
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
# Register nodes
graph_builder.add_node("plan", planner_node)
graph_builder.add_node("search", searcher_node)
graph_builder.add_node("analyze", analyst_node)
graph_builder.add_node("write", writer_node)
# Define edges
graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "search")
graph_builder.add_edge("search", "analyze")
# Conditional edge after analysis
graph_builder.add_conditional_edges(
"analyze",
should_continue_research,
{
"plan": "plan",
"write": "write"
}
)
graph_builder.add_edge("write", END)
# Compile the graph
research_graph = graph_builder.compile()
Running the Pipeline
Invoke the compiled graph with an initial state. Only the question and max_iterations fields are required; the rest will be populated as nodes execute.
initial_state = {
"question": "What are the key technical and economic trade-offs between "
"small modular reactors and traditional large nuclear reactors?",
"sub_questions": [],
"search_results": [],
"findings": [],
"report": "",
"iteration": 0,
"max_iterations": 3,
"coverage_sufficient": False
}
final_state = research_graph.invoke(initial_state)
print(final_state["report"])
When you run this, the graph will execute the Planner, Searcher, and Analyst in sequence. If the Analyst determines coverage is insufficient and we haven't hit the iteration cap, control loops back to the Planner with the accumulated findings in state. Once coverage is sufficient or the cap is reached, the Writer produces the final report.
Adding a Checkpointer for Persistence
For production use, you'll want to persist state so you can resume interrupted runs and inspect intermediate results. LangGraph provides an in-memory checkpointer for development and a SQLite checkpointer for durability.
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
# In-memory (development)
memory_checkpointer = MemorySaver()
# SQLite (persistence across runs)
sqlite_checkpointer = SqliteSaver.from_conn_string("research.db")
# Compile with checkpointer
research_graph = graph_builder.compile(checkpointer=memory_checkpointer)
# Run with a thread_id for state tracking
config = {"configurable": {"thread_id": "research-run-001"}}
final_state = research_graph.invoke(initial_state, config=config)
# Inspect intermediate state at any point
state_snapshot = research_graph.get_state(config)
print(state_snapshot.values)
Visualizing the Graph
LangGraph can render your graph as a Mermaid diagram, which is extremely useful for documentation and debugging.
from IPython.display import Image, display
# In a Jupyter notebook
display(Image(research_graph.get_graph().draw_mermaid_png()))
# Or export the Mermaid source
print(research_graph.get_graph().draw_mermaid())
Best Practices
Keep Nodes Focused
Each node should do one thing well. If a node is both searching and analyzing, split it. Focused nodes are easier to test, easier to replace, and easier to reason about when something goes wrong.
Design State Deliberately
Your state schema is the API between agents. Resist the temptation to dump everything into a single context string. Typed fields with clear ownership make the system debuggable and let you use reducers like operator.add for accumulation.
Always Bound Your Loops
Conditional edges that route back to earlier nodes create loops. Without a maximum iteration guard, a stubborn Analyst that never declares coverage sufficient will loop forever. The max_iterations field in our state is a simple, effective safeguard.
Use Structured Outputs for Parsing
In the Planner, we parsed JSON from the LLM response. For production, prefer LangChain's structured output features (llm.with_structured_output()) with Pydantic models. This eliminates fragile string parsing and gives you validation for free.
from pydantic import BaseModel, Field
class SubQuestions(BaseModel):
questions: List[str] = Field(description="Focused sub-questions to research")
structured_llm = llm.with_structured_output(SubQuestions)
result = structured_llm.invoke(prompt)
sub_questions = result.questions
Log at Node Boundaries
Because state transitions are explicit in LangGraph, you can wrap nodes with logging to trace exactly what enters and exits each step. This is far more debuggable than logging inside a monolithic chain.
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("research_pipeline")
def logged_node(func):
def wrapper(state):
logger.info(f"Entering {func.__name__} | iteration={state.get('iteration')}")
result = func(state)
logger.info(f"Exiting {func.__name__} | keys_updated={list(result.keys())}")
return result
return wrapper
# Apply to nodes
graph_builder.add_node("plan", logged_node(planner_node))
Handle Tool Failures Gracefully
Web searches fail. APIs rate-limit. HTML parsing breaks. Wrap external calls in try/except blocks and write error information into state so downstream agents can reason about gaps rather than crashing the entire pipeline.
Test Nodes in Isolation
Because each node is a plain Python function that takes a state dict and returns a partial state dict, you can unit test them without compiling the graph. Construct a mock state, call the node, and assert on the returned fields. This is one of LangGraph's biggest advantages over opaque agent frameworks.
def test_planner_produces_sub_questions():
state = {
"question": "What is quantum computing?",
"iteration": 0
}
result = planner_node(state)
assert "sub_questions" in result
assert len(result["sub_questions"]) >= 3
assert result["iteration"] == 1
Extending the Pipeline
The graph structure makes extension straightforward. Some ideas:
- Add a Fact-Checker node between the Analyst and Writer that verifies each finding against a second search.
- Add parallel Searcher nodes using different sources (web, academic papers, news APIs) and merge their results.
- Add a Human-in-the-Loop interrupt after the Planner so a human can approve or edit sub-questions before searching begins.
- Add a Citation Formatter node after the Writer that converts bracketed citations into properly formatted references.
To add a human approval step, compile with an interrupt before the search node:
research_graph = graph_builder.compile(
checkpointer=memory_checkpointer,
interrupt_before=["search"]
)
# First invocation stops before "search"
config = {"configurable": {"thread_id": "run-002"}}
research_graph.invoke(initial_state, config=config)
# Human reviews and optionally edits state
state = research_graph.get_state(config)
# ... human edits sub_questions ...
# Resume execution
research_graph.invoke(None, config=config)
Conclusion
Building a multi-agent research pipeline with LangGraph gives you a system that is modular, debuggable, and extensible by design. By decomposing the research process into focused agents — a Planner that decomposes, a Searcher that gathers, an Analyst that extracts, and a Writer that synthesizes — you get higher-quality output than any single prompt could produce. The graph abstraction makes the control flow explicit: you can see the loops, trace the state, checkpoint for recovery, and insert human oversight exactly where it adds value. The pipeline we built here is a solid foundation, but the real power emerges when you start adding specialized nodes, parallel execution paths, and human-in-the-loop checkpoints tailored to your specific domain. Start simple, test each node in isolation, and grow the graph incrementally as your research requirements evolve.