Building a Multi-Agent Research Pipeline with LlamaIndex: Complete Guide

Building a Multi-Agent Research Pipeline with LlamaIndex: Complete Guide

Modern research workflows rarely fit into a single prompt-response cycle. A realistic research task involves gathering sources, extracting relevant facts, synthesizing findings, fact-checking claims, and producing a polished report. Each of these steps benefits from a specialized agent with its own tools, prompts, and reasoning loop. LlamaIndex's agent framework makes it straightforward to compose these specialized agents into a coordinated pipeline that can tackle complex research tasks autonomously.

In this guide, you'll learn what a multi-agent research pipeline is, why it matters, and how to build one end-to-end using LlamaIndex. We'll cover agent design, tool wiring, orchestration patterns, and production best practices.

What Is a Multi-Agent Research Pipeline?

A multi-agent research pipeline is a system in which several LLM-powered agents collaborate to complete a research objective. Rather than asking one model to do everything, you decompose the work into discrete roles:

• Planner Agent — breaks the research question into sub-questions and assigns them.
• Search Agent — queries external sources (web, vector DBs, APIs) and retrieves raw material.
• Extractor Agent — pulls structured facts and citations from retrieved documents.
• Synthesizer Agent — combines extracted facts into a coherent draft.
• Critic / Fact-Checker Agent — verifies claims against sources and flags hallucinations.
• Writer Agent — produces the final formatted report.

Each agent has a focused system prompt, a curated set of tools, and a defined input/output contract. The pipeline orchestrates them in sequence or via a supervisor that routes messages dynamically.

Why It Matters

Single-agent approaches struggle with deep research for several reasons. Context windows fill up quickly when an agent tries to read dozens of sources. Prompts become overloaded when one agent must simultaneously search, extract, synthesize, and verify. Error rates compound because there is no separation of concerns. A multi-agent design addresses these issues by:

• Keeping each agent's context small and focused, improving reasoning quality.
• Allowing parallel execution of independent sub-tasks, reducing latency.
• Enabling specialized tools and prompts per role, increasing accuracy.
• Providing natural checkpoints where outputs can be validated before proceeding.
• Making the system observable — you can inspect each agent's contribution.

Prerequisites and Setup

Install LlamaIndex and the dependencies we'll use throughout this tutorial:

pip install llama-index llama-index-core \
  llama-index-llms-openai llama-index-embeddings-openai \
  llama-index-tools-google llama-index-tools-wikipedia \
  llama-index-agent-workflows python-dotenv

Create a .env file with your API keys:

OPENAI_API_KEY=sk-...
GOOGLE_API_KEY=...
GOOGLE_CSE_ID=...

Then load them in your entry script:

from dotenv import load_dotenv
load_dotenv()

Designing the Agent Roles

Before writing code, sketch the data flow. Each agent consumes the output of the previous one and produces a typed result. This contract is what keeps the pipeline robust.

ResearchQuestion
  -> Planner: List[SubQuestion]
  -> Searcher: List[SourceDocument]
  -> Extractor: List[ExtractedFact]
  -> Synthesizer: DraftReport
  -> Critic: CritiqueReport
  -> Writer: FinalReport

Define lightweight data classes for these intermediate products:

from pydantic import BaseModel, Field
from typing import List, Optional

class SubQuestion(BaseModel):
    question: str
    rationale: str

class SourceDocument(BaseModel):
    title: str
    url: str
    snippet: str

class ExtractedFact(BaseModel):
    fact: str
    source_url: str
    confidence: float = Field(ge=0.0, le=1.0)

class DraftReport(BaseModel):
    title: str
    sections: List[str]
    citations: List[str]

class CritiqueReport(BaseModel):
    issues: List[str]
    verified_facts: List[ExtractedFact]
    needs_revision: bool

class FinalReport(BaseModel):
    title: str
    body: str
    references: List[str]

Building the Search Agent

The search agent needs tools to query the web and Wikipedia. LlamaIndex provides ready-made tool integrations that wrap these services as function-calling tools.

from llama_index.tools.google import GoogleSearchToolSpec
from llama_index.tools.wikipedia import WikipediaToolSpec
from llama_index.core.tools import FunctionTool

google_spec = GoogleSearchToolSpec()
wiki_spec = WikipediaToolSpec()

search_tools = google_spec.to_tool_list() + wiki_spec.to_tool_list()

def summarize_hits(query: str, hits: List[SourceDocument]) -> str:
    lines = [f"- {h.title} ({h.url}): {h.snippet}" for h in hits]
    return f"Query: {query}\n" + "\n".join(lines)

summarize_tool = FunctionTool.from_defaults(fn=summarize_hits)
search_tools.append(summarize_tool)

Now create the search agent using LlamaIndex's FunctionAgent:

from llama_index.core.agent import FunctionAgent
from llama_index.llms.openai import OpenAI

search_llm = OpenAI(model="gpt-4o-mini", temperature=0.0)

search_agent = FunctionAgent(
    tools=search_tools,
    llm=search_llm,
    system_prompt=(
        "You are a research search specialist. Given a sub-question, "
        "use the available search tools to find authoritative sources. "
        "Return a JSON list of SourceDocument objects with title, url, "
        "and a concise snippet. Prioritize primary sources and recent "
        "publications. If you cannot find reliable sources, return an "
        "empty list rather than guessing."
    ),
)

Building the Extractor Agent

The extractor takes raw source documents and pulls out structured facts. We give it a custom tool that wraps a retrieval-augmented extraction step.

def extract_facts(documents: List[SourceDocument], sub_question: str) -> List[ExtractedFact]:
    facts = []
    for doc in documents:
        # In production, use an LLM call here to extract structured facts.
        # Simplified for illustration:
        facts.append(ExtractedFact(
            fact=doc.snippet,
            source_url=doc.url,
            confidence=0.7,
        ))
    return facts

extract_tool = FunctionTool.from_defaults(fn=extract_facts)

extractor_agent = FunctionAgent(
    tools=[extract_tool],
    llm=OpenAI(model="gpt-4o-mini", temperature=0.0),
    system_prompt=(
        "You are a fact extraction specialist. Given source documents "
        "and a sub-question, extract only facts that directly answer "
        "the question. Each fact must include its source URL and a "
        "confidence score. Discard irrelevant or speculative content."
    ),
)

Building the Synthesizer and Critic Agents

synthesizer_agent = FunctionAgent(
    tools=[],
    llm=OpenAI(model="gpt-4o", temperature=0.3),
    system_prompt=(
        "You are a research synthesizer. Given a list of extracted "
        "facts grouped by sub-question, write a structured draft "
        "report with clear sections and inline citations. Do not "
        "introduce facts that are not present in the input. Mark "
        "uncertain claims explicitly."
    ),
)

critic_agent = FunctionAgent(
    tools=[],
    llm=OpenAI(model="gpt-4o", temperature=0.0),
    system_prompt=(
        "You are a rigorous fact-checker. Review the draft report "
        "against the extracted facts. Flag any claim that is not "
        "supported by a cited source, any contradiction, and any "
        "missing citation. Return a CritiqueReport with specific, "
        "actionable issues."
    ),
)

Orchestrating with AgentWorkflow

LlamaIndex's AgentWorkflow lets you wire multiple agents together with a shared context and explicit handoff rules. Each agent can signal which agent should run next by returning a Handoff event.

from llama_index.agent.workflow import AgentWorkflow, WorkflowContext
from llama_index.core.llms import ChatMessage
import json

async def plan(ctx: WorkflowContext, research_question: str) -> str:
    sub_qs = [
        SubQuestion(question="What is X?", rationale="Define the core concept."),
        SubQuestion(question="What are the latest developments in X?", rationale="Establish currency."),
        SubQuestion(question="What are the criticisms of X?", rationale="Ensure balance."),
    ]
    await ctx.set("sub_questions", [sq.model_dump() for sq in sub_qs])
    return "Planner produced 3 sub-questions. Handing off to search."

async def search_all(ctx: WorkflowContext) -> str:
    sub_qs = await ctx.get("sub_questions")
    all_sources = []
    for sq in sub_qs:
        result = await search_agent.run(sq["question"])
        all_sources.append({"sub_question": sq["question"], "sources": result})
    await ctx.set("sources", all_sources)
    return "Search complete. Handing off to extractor."

async def extract_all(ctx: WorkflowContext) -> str:
    sources = await ctx.get("sources")
    all_facts = []
    for entry in sources:
        result = await extractor_agent.run(
            f"Sub-question: {entry['sub_question']}\nSources: {entry['sources']}"
        )
        all_facts.append({"sub_question": entry["sub_question"], "facts": result})
    await ctx.set("facts", all_facts)
    return "Extraction complete. Handing off to synthesizer."

async def synthesize(ctx: WorkflowContext) -> str:
    facts = await ctx.get("facts")
    draft = await synthesizer_agent.run(f"Extracted facts: {json.dumps(facts)}")
    await ctx.set("draft", str(draft))
    return "Draft complete. Handing off to critic."

async def critique(ctx: WorkflowContext) -> str:
    draft = await ctx.get("draft")
    facts = await ctx.get("facts")
    review = await critic_agent.run(f"Draft: {draft}\nFacts: {json.dumps(facts)}")
    await ctx.set("critique", str(review))
    return "Critique complete. Handing off to writer."

async def write_final(ctx: WorkflowContext) -> str:
    draft = await ctx.get("draft")
    critique = await ctx.get("critique")
    final = await synthesizer_agent.run(
        f"Revise this draft based on the critique.\n\nDraft:\n{draft}\n\nCritique:\n{critique}"
    )
    return str(final)

Now assemble the workflow. We use FunctionWorkflow with named events to chain the steps:

from llama_index.core.workflow import Workflow, StopEvent, step

class ResearchWorkflow(Workflow):
    @step
    async def plan_step(self, ctx, ev) -> str:
        return await plan(ctx, ev.get("question", ""))

    @step
    async def search_step(self, ctx, ev) -> str:
        return await search_all(ctx)

    @step
    async def extract_step(self, ctx, ev) -> str:
        return await extract_all(ctx)

    @step
    async def synthesize_step(self, ctx, ev) -> str:
        return await synthesize(ctx)

    @step
    async def critique_step(self, ctx, ev) -> str:
        return await critique(ctx)

    @step
    async def write_step(self, ctx, ev) -> StopEvent:
        result = await write_final(ctx)
        return StopEvent(result=result)

workflow = ResearchWorkflow(timeout=300, verbose=True)

Running the Pipeline

import asyncio

async def main():
    question = (
        "What are the current capabilities and limitations of "
        "retrieval-augmented generation systems in production?"
    )
    result = await workflow.run(question=question)
    print("=== FINAL REPORT ===")
    print(result)

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

Adding Parallelism for Speed

The search and extraction steps are embarrassingly parallel across sub-questions. Use asyncio.gather to run them concurrently:

async def search_all_parallel(ctx: WorkflowContext) -> str:
    sub_qs = await ctx.get("sub_questions")
    tasks = [search_agent.run(sq["question"]) for sq in sub_qs]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    sources = []
    for sq, res in zip(sub_qs, results):
        if isinstance(res, Exception):
            sources.append({"sub_question": sq["question"], "sources": []})
        else:
            sources.append({"sub_question": sq["question"], "sources": str(res)})
    await ctx.set("sources", sources)
    return "Parallel search complete."

This can cut total latency significantly when you have many sub-questions.

Best Practices

• Keep agent prompts narrow. Each agent should have one job. Resist the temptation to add "just one more" responsibility.
• Use structured outputs. Pydantic models or JSON schemas prevent agents from producing free-form text that breaks downstream parsing.
• Validate between stages. Add lightweight checks — non-empty source lists, confidence thresholds, citation presence — before handing off.
• Log every handoff. Persist each agent's input and output so you can debug failures and audit the reasoning trail.
• Cap tool calls. Set a maximum number of iterations per agent to prevent infinite loops when a tool returns unexpected results.
• Choose models per role. Use cheaper, faster models for search and extraction; reserve larger models for synthesis and critique where reasoning depth matters.
• Make the critic non-negotiable. A pipeline without a verification step will silently propagate hallucinations. Always include a fact-checking stage.
• Cache search results. Repeated research questions should hit a cache rather than re-querying external APIs, saving cost and rate-limit budget.
• Handle failures gracefully. Wrap each agent call in try/except and decide whether to retry, skip, or escalate. A single failed search should not abort the entire pipeline.

Extending the Pipeline

Once the core pipeline works, you can extend it in several directions. Add a reflection loop where the writer's output is sent back to the critic for a second pass if needs_revision is true. Add a human-in-the-loop checkpoint after extraction so a domain expert can approve facts before synthesis. Integrate specialized tools like arXiv search, SEC filings, or a private vector store for proprietary knowledge. You can also add a planner-v2 agent that re-plans mid-pipeline if the critic discovers that a sub-question was poorly scoped.

A reflection loop looks like this:

MAX_REVISIONS = 2

async def synthesize_with_revision(ctx: WorkflowContext) -> str:
    facts = await ctx.get("facts")
    draft = await synthesizer_agent.run(f"Extracted facts: {json.dumps(facts)}")
    for i in range(MAX_REVISIONS):
        review = await critic_agent.run(
            f"Draft:\n{draft}\n\nFacts:\n{json.dumps(facts)}"
        )
        review_text = str(review)
        if "needs_revision: false" in review_text.lower():
            break
        draft = await synthesizer_agent.run(
            f"Revise based on critique.\nDraft:\n{draft}\nCritique:\n{review_text}"
        )
    await ctx.set("draft", str(draft))
    return "Synthesis with revision complete."

Conclusion

Building a multi-agent research pipeline with LlamaIndex lets you decompose complex research into focused, reliable, and observable stages. By giving each agent a clear role, structured inputs and outputs, and a curated toolset, you get a system that is easier to debug, cheaper to run, and more accurate than a monolithic single-agent approach. Start with the sequential pipeline shown here, add parallelism where it helps, and iterate on prompts and tools until each stage produces trustworthy results. With a critic in the loop and structured contracts between agents, you have a foundation that scales from simple questions to deep, multi-source research investigations.