← Back to DevBytes
AI August 16, 2026 11 min read

Building a Multi-Agent Research Pipeline with Pydantic AI: Complete Guide

Building a Multi-Agent Research Pipeline with Pydantic AI: Complete Guide

Modern research workflows rarely fit neatly into a single prompt-response cycle. Whether you're investigating market trends, summarizing scientific literature, or producing competitive analyses, the work typically involves several distinct stages: gathering sources, extracting key facts, cross-referencing claims, and synthesizing a final report. Each stage benefits from a specialized agent with its own instructions, tools, and structured outputs. Pydantic AI, a framework that pairs LLMs with Pydantic's type validation, is particularly well suited for orchestrating these multi-agent pipelines.

In this guide, you'll learn what a multi-agent research pipeline is, why Pydantic AI is a strong foundation for building one, and how to assemble a working pipeline step by step. By the end, you'll have a runnable architecture you can adapt to your own research domain.

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 task. Rather than asking one model to do everything, you decompose the task into focused subtasks and assign each to an agent with a narrow scope. The output of one agent becomes the input to the next, forming a directed flow of structured data.

A typical research pipeline might include the following roles:

The key advantage is separation of concerns. Each agent can be tested, tuned, and swapped independently. You can also parallelize branches of the pipeline where dependencies allow.

Why Pydantic AI?

Pydantic AI brings several properties that make it ideal for this kind of work:

Because each agent's output is a Pydantic model, the contracts between agents are explicit. If the Search Agent changes its schema, the Extractor Agent will fail at validation time rather than producing silent corruption downstream.

Prerequisites and Setup

Install Pydantic AI and a few supporting packages. The examples below use Python 3.10 or newer.

pip install pydantic-ai pydantic httpx python-dotenv

Create a .env file with your API keys:

OPENAI_API_KEY=sk-...

For this tutorial we'll use OpenAI, but you can swap in any supported model by changing the model argument on each agent.

Step 1: Define Shared Data Models

Start by defining the structured data that flows between agents. These models form the contract for the entire pipeline.

from pydantic import BaseModel, Field
from typing import Literal

class SubQuestion(BaseModel):
    question: str
    rationale: str
    search_terms: list[str]

class ResearchPlan(BaseModel):
    topic: str
    sub_questions: list[SubQuestion]
    overall_strategy: str

class SourceSnippet(BaseModel):
    sub_question: str
    url: str
    title: str
    raw_text: str

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

class CritiqueResult(BaseModel):
    fact_index: int
    verdict: Literal["supported", "contradicted", "uncertain"]
    reasoning: str
    suggested_followup: str | None = None

class ResearchReport(BaseModel):
    topic: str
    executive_summary: str
    key_findings: list[str]
    detailed_analysis: str
    sources: list[str]
    open_questions: list[str]

Notice how each model is self-describing. The confidence field on ExtractedFact is bounded between 0 and 1, and verdict on CritiqueResult is constrained to a literal set of values. These constraints prevent malformed data from propagating.

Step 2: Build the Planner Agent

The Planner takes a research topic and produces a structured plan. We define it as a Pydantic AI Agent with a system prompt and an expected output type.

from pydantic_ai import Agent

planner_agent = Agent(
    model="openai:gpt-4o",
    output_type=ResearchPlan,
    system_prompt=(
        "You are a research planner. Given a topic, break it down into 3-5 "
        "specific sub-questions that together cover the topic. For each "
        "sub-question, provide a rationale and a list of search terms. "
        "Also describe an overall search strategy."
    ),
)

async def plan_research(topic: str) -> ResearchPlan:
    result = await planner_agent.run(f"Topic: {topic}")
    return result.output

Because output_type is set to ResearchPlan, Pydantic AI automatically instructs the model to produce JSON conforming to that schema and validates the response. If the model returns invalid data, the framework retries with corrective feedback.

Step 3: Build the Search Agent with Tools

The Search Agent needs access to external sources. We'll define a simple search tool and inject it as a dependency. In production you'd replace this with a real search API.

from dataclasses import dataclass

@dataclass
class SearchDeps:
    http_client: httpx.AsyncClient

async def web_search(ctx, query: str) -> str:
    """Search the web and return raw text snippets."""
    deps: SearchDeps = ctx.deps
    # Placeholder: in production, call a real search API.
    # Here we simulate a result.
    return (
        f"Simulated search result for '{query}': "
        f"According to recent reports, {query} shows significant growth "
        f"with multiple contributing factors. Source: https://example.com/{query.replace(' ', '-')}"
    )

search_agent = Agent(
    model="openai:gpt-4o",
    output_type=list[SourceSnippet],
    deps_type=SearchDeps,
    tools=[web_search],
    system_prompt=(
        "You are a research search agent. For each sub-question provided, "
        "use the web_search tool with appropriate search terms. Collect "
        "raw text snippets with their source URLs. Return one SourceSnippet "
        "per sub-question."
    ),
)

async def search_sources(plan: ResearchPlan) -> list[SourceSnippet]:
    async with httpx.AsyncClient() as client:
        deps = SearchDeps(http_client=client)
        prompt = "\n".join(
            f"Sub-question: {sq.question}\nSearch terms: {', '.join(sq.search_terms)}"
            for sq in plan.sub_questions
        )
        result = await search_agent.run(prompt, deps=deps)
        return result.output

The deps_type parameter tells Pydantic AI what dependency object to expect, and the tool function receives a context object that exposes those deps. This pattern keeps the agent decoupled from concrete implementations, which is essential for testing.

Step 4: Build the Extractor Agent

The Extractor reads raw snippets and pulls out structured facts with citations and confidence scores.

extractor_agent = Agent(
    model="openai:gpt-4o",
    output_type=list[ExtractedFact],
    system_prompt=(
        "You are a fact extraction agent. Given raw source snippets, "
        "extract discrete factual claims. For each claim, include the "
        "exact supporting quote, the source URL, and a confidence score "
        "between 0 and 1 reflecting how clearly the source supports the "
        "claim. Map each fact to the sub-question it addresses."
    ),
)

async def extract_facts(snippets: list[SourceSnippet]) -> list[ExtractedFact]:
    formatted = "\n\n".join(
        f"[Sub-question: {s.sub_question}]\nURL: {s.url}\nTitle: {s.title}\nText: {s.raw_text}"
        for s in snippets
    )
    result = await extractor_agent.run(formatted)
    return result.output

Step 5: Build the Critic Agent

The Critic evaluates each extracted fact for support, contradictions, and gaps. This is where the pipeline gains reliability — instead of trusting the model's first pass, we add a verification layer.

critic_agent = Agent(
    model="openai:gpt-4o",
    output_type=list[CritiqueResult],
    system_prompt=(
        "You are a research critic. You receive a list of extracted facts. "
        "For each fact, assess whether it is supported by its cited quote, "
        "whether it contradicts other facts, and whether it is uncertain. "
        "Return a CritiqueResult for each fact. If a fact is contradicted "
        "or uncertain, suggest a follow-up search."
    ),
)

async def critique_facts(facts: list[ExtractedFact]) -> list[CritiqueResult]:
    formatted = "\n\n".join(
        f"Fact {i}: {f.claim}\nQuote: {f.supporting_quote}\n"
        f"URL: {f.source_url}\nConfidence: {f.confidence}\nSub-question: {f.sub_question}"
        for i, f in enumerate(facts)
    )
    result = await critic_agent.run(formatted)
    return result.output

You can then filter facts based on the critic's verdicts before passing them to the synthesizer:

def filter_supported_facts(
    facts: list[ExtractedFact],
    critiques: list[CritiqueResult],
) -> list[ExtractedFact]:
    supported = []
    for fact, critique in zip(facts, critiques):
        if critique.verdict == "supported":
            supported.append(fact)
    return supported

Step 6: Build the Synthesizer Agent

The Synthesizer takes the validated facts and produces the final structured report.

synthesizer_agent = Agent(
    model="openai:gpt-4o",
    output_type=ResearchReport,
    system_prompt=(
        "You are a research synthesizer. Given a topic and a list of "
        "validated facts with citations, produce a structured research "
        "report. Include an executive summary, key findings, a detailed "
        "analysis, a list of source URLs, and any open questions that "
        "remain unanswered. Be precise and cite sources."
    ),
)

async def synthesize_report(
    topic: str,
    facts: list[ExtractedFact],
) -> ResearchReport:
    formatted = "\n\n".join(
        f"- {f.claim} (Source: {f.source_url}, Confidence: {f.confidence})"
        for f in facts
    )
    prompt = f"Topic: {topic}\n\nValidated facts:\n{formatted}"
    result = await synthesizer_agent.run(prompt)
    return result.output

Step 7: Orchestrate the Full Pipeline

Now wire everything together. The orchestrator runs each stage in sequence, passing structured outputs between agents.

import asyncio
from dotenv import load_dotenv

load_dotenv()

async def run_research_pipeline(topic: str) -> ResearchReport:
    print(f"[1/5] Planning research for: {topic}")
    plan = await plan_research(topic)
    print(f"  Plan has {len(plan.sub_questions)} sub-questions")

    print("[2/5] Searching sources")
    snippets = await search_sources(plan)
    print(f"  Collected {len(snippets)} snippets")

    print("[3/5] Extracting facts")
    facts = await extract_facts(snippets)
    print(f"  Extracted {len(facts)} facts")

    print("[4/5] Critiquing facts")
    critiques = await critique_facts(facts)
    supported = filter_supported_facts(facts, critiques)
    print(f"  {len(supported)} facts supported out of {len(facts)}")

    print("[5/5] Synthesizing report")
    report = await synthesize_report(topic, supported)
    return report

if __name__ == "__main__":
    topic = "The impact of remote work on commercial real estate"
    report = asyncio.run(run_research_pipeline(topic))

    print("\n=== RESEARCH REPORT ===\n")
    print(f"Topic: {report.topic}")
    print(f"\nExecutive Summary:\n{report.executive_summary}")
    print(f"\nKey Findings:")
    for finding in report.key_findings:
        print(f"  - {finding}")
    print(f"\nDetailed Analysis:\n{report.detailed_analysis}")
    print(f"\nSources:")
    for src in report.sources:
        print(f"  - {src}")
    print(f"\nOpen Questions:")
    for q in report.open_questions:
        print(f"  - {q}")

Running this script executes the full pipeline end to end. Each stage prints its progress, and the final report is a validated ResearchReport instance you can serialize to JSON, store in a database, or render into a template.

Adding Parallelism

Some stages can run in parallel. For example, if sub-questions are independent, you can search and extract for each one concurrently:

async def search_and_extract_one(
    sub_question: SubQuestion,
    client: httpx.AsyncClient,
) -> list[ExtractedFact]:
    deps = SearchDeps(http_client=client)
    search_prompt = (
        f"Sub-question: {sub_question.question}\n"
        f"Search terms: {', '.join(sub_question.search_terms)}"
    )
    search_result = await search_agent.run(search_prompt, deps=deps)
    snippets = search_result.output

    formatted = "\n\n".join(
        f"[Sub-question: {s.sub_question}]\nURL: {s.url}\nText: {s.raw_text}"
        for s in snippets
    )
    extract_result = await extractor_agent.run(formatted)
    return extract_result.output

async def search_and_extract_parallel(
    plan: ResearchPlan,
) -> list[ExtractedFact]:
    async with httpx.AsyncClient() as client:
        tasks = [
            search_and_extract_one(sq, client)
            for sq in plan.sub_questions
        ]
        results = await asyncio.gather(*tasks)
    return [fact for batch in results for fact in batch]

This can significantly reduce latency for pipelines with many sub-questions, since the bottleneck shifts from sequential LLM calls to the provider's rate limits.

Best Practices

Extending the Pipeline

Once the core pipeline is working, consider these extensions:

Conclusion

Building a multi-agent research pipeline with Pydantic AI gives you a structured, type-safe, and maintainable way to tackle complex research tasks. By decomposing the work into specialized agents — planner, searcher, extractor, critic, and synthesizer — you gain fine-grained control over each stage, the ability to test components independently, and the flexibility to swap models or tools without rewriting the entire system. The typed contracts between agents, enforced by Pydantic models, turn what would otherwise be fragile prompt-chaining into a robust data pipeline. Start with the architecture described here, adapt the schemas and tools to your domain, and iterate by adding verification layers, parallelism, and persistence as your needs grow. The result is a research system that is not only more capable than a single monolithic prompt, but also far easier to debug, extend, and trust.

— Ad —

Google AdSense will appear here after approval

← Back to all articles