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

Building a Multi-Agent Research Pipeline with OpenAI Agents SDK: Complete Guide

Building a Multi-Agent Research Pipeline with OpenAI Agents SDK: Complete Guide

Modern research workflows rarely fit into a single prompt-response cycle. Whether you're investigating market trends, summarizing scientific literature, or producing competitive analysis reports, the work demands multiple specialized steps: gathering sources, extracting key insights, cross-referencing claims, and synthesizing a final deliverable. The OpenAI Agents SDK is purpose-built for this kind of orchestrated, multi-step reasoning. In this guide, we'll build a complete multi-agent research pipeline from scratch, covering architecture, implementation, and production best practices.

What Is the OpenAI Agents SDK?

The OpenAI Agents SDK is a lightweight Python framework for building agentic applications where one or more LLM-powered agents collaborate to accomplish complex tasks. Unlike a single chat completion call, agents can use tools, hand off work to other agents, maintain guardrails, and execute multi-step plans autonomously. The SDK abstracts away much of the orchestration plumbing — tool calling loops, context propagation, tracing, and handoffs — so you can focus on defining what each agent should do.

At its core, the SDK revolves around four primitives:

Why Multi-Agent Pipelines Matter

A single monolithic agent asked to "research a topic and write a report" tends to produce shallow, unfocused output. It tries to do everything at once and ends up doing nothing particularly well. Multi-agent pipelines solve this by applying the principle of separation of concerns to LLM workflows. Each agent has a narrow, well-defined role, a tailored system prompt, and a curated set of tools. This leads to several concrete benefits:

Prerequisites and Installation

Before we start building, make sure you have Python 3.9 or higher and an OpenAI API key. Install the SDK and a few supporting libraries:

pip install openai-agents python-dotenv httpx

Create a .env file in your project root:

OPENAI_API_KEY=sk-your-key-here

Designing the Research Pipeline Architecture

Our pipeline will consist of four agents working in sequence with one parallel branch:

The Planner hands off to the Searcher, the Searcher hands off to the Analyst, and the Analyst hands off to the Writer. Context — the research question, sub-questions, and accumulated findings — flows through a shared data structure.

Defining the Shared Context

First, let's define a Pydantic model that will carry state between agents. The SDK supports passing a typed context object through the entire run.

from pydantic import BaseModel, Field
from typing import Optional


class ResearchContext(BaseModel):
    """Shared state passed between agents in the research pipeline."""
    original_question: str
    sub_questions: list[str] = Field(default_factory=list)
    raw_sources: list[dict] = Field(default_factory=list)
    findings: list[dict] = Field(default_factory=list)
    final_report: Optional[str] = None

Each agent reads from and writes to this context. Keeping the schema explicit prevents the kind of silent data drift that plagues ad-hoc dictionary-based approaches.

Building the Tools

Agents need tools to interact with the outside world. Our Searcher needs a web search tool, and our Analyst needs a tool to fetch and parse full page content. For this tutorial we'll use a mock search function that you can later replace with a real API like Tavily, SerpAPI, or Brave Search.

import httpx
import json


async def web_search(query: str, max_results: int = 5) -> list[dict]:
    """Search the web and return a list of source dicts.
    
    Replace this stub with a real search API call in production.
    """
    # Stub: return mock results. In production, call Tavily/SerpAPI/etc.
    return [
        {
            "title": f"Source result for: {query}",
            "url": f"https://example.com/source-{i}",
            "snippet": f"This is a mock snippet relevant to '{query}'."
        }
        for i in range(max_results)
    ]


async def fetch_page(url: str) -> str:
    """Fetch and return the text content of a web page."""
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url, follow_redirects=True)
            resp.raise_for_status()
            # In production, use BeautifulSoup or trafilatura to extract text
            return resp.text[:8000]
    except Exception as e:
        return f"Error fetching {url}: {e}"

Now let's wrap these functions as SDK tools using the @function_tool decorator:

from agents import function_tool


@function_tool
async def search_web(query: str) -> str:
    """Search the web for information on a given query.
    
    Args:
        query: The search query string.
    
    Returns:
        JSON string of search results with title, url, and snippet.
    """
    results = await web_search(query)
    return json.dumps(results, indent=2)


@function_tool
async def read_page(url: str) -> str:
    """Fetch and read the full text content of a web page.
    
    Args:
        url: The URL of the page to read.
    
    Returns:
        The text content of the page (truncated).
    """
    content = await fetch_page(url)
    return content

Creating the Agents

Now we define each agent with tailored instructions. The key to good agent design is writing system prompts that are specific, constrained, and action-oriented.

from agents import Agent, Runner

PLANNER_INSTRUCTIONS = """
You are a research planner. Given a research question, your job is to:
1. Break the question into 3-5 specific sub-questions that together 
   cover the full scope of the original question.
2. Identify what types of sources would be most valuable.
3. Output the sub-questions as a clear numbered list.

Do not attempt to answer the question yourself. Your only job is planning.
Be precise — vague sub-questions lead to vague research.
"""

SEARCHER_INSTRUCTIONS = """
You are a research searcher. You receive a set of sub-questions.
For each sub-question:
1. Use the search_web tool to find relevant sources.
2. Use the read_page tool to read the most promising 2-3 sources per 
   sub-question.
3. Collect the raw source content and URLs.

Prioritize authoritative sources. Skip paywalled or low-quality content.
Return a structured summary of what you found for each sub-question, 
including source URLs.
"""

ANALYST_INSTRUCTIONS = """
You are a research analyst. You receive raw search results and page 
content. Your job is to:
1. Extract the key findings relevant to each sub-question.
2. Note any contradictions between sources.
3. Assess source credibility.
4. Identify gaps where more research might be needed.

Output a structured analysis with findings grouped by sub-question.
Each finding should include a citation to its source URL.
Do not fabricate information. If sources are insufficient, say so.
"""

WRITER_INSTRUCTIONS = """
You are a research report writer. You receive a structured analysis 
with findings and citations. Your job is to synthesize this into a 
polished, well-structured report.

The report should include:
- An executive summary (2-3 paragraphs)
- Key findings organized by theme
- Contradictions and areas of uncertainty
- A conclusion with actionable takeaways
- A sources section listing all URLs cited

Write in a clear, professional tone. Use markdown formatting.
Every factual claim must be traceable to a cited source.
"""

Now let's instantiate the agents and wire up the handoffs:

planner_agent = Agent(
    name="Research Planner",
    instructions=PLANNER_INSTRUCTIONS,
    model="gpt-4o",
)

searcher_agent = Agent(
    name="Research Searcher",
    instructions=SEARCHER_INSTRUCTIONS,
    model="gpt-4o",
    tools=[search_web, read_page],
)

analyst_agent = Agent(
    name="Research Analyst",
    instructions=ANALYST_INSTRUCTIONS,
    model="gpt-4o",
)

writer_agent = Agent(
    name="Report Writer",
    instructions=WRITER_INSTRUCTIONS,
    model="gpt-4o",
)

# Wire up handoffs: each agent can delegate to the next in the pipeline
planner_agent.handoffs = [searcher_agent]
searcher_agent.handoffs = [analyst_agent]
analyst_agent.handoffs = [writer_agent]

Running the Pipeline

With the agents defined and handoffs configured, running the pipeline is straightforward. We start the run with the Planner agent and let handoffs propagate control automatically:

import asyncio
from agents import Runner


async def run_research_pipeline(question: str) -> str:
    """Run the full multi-agent research pipeline.
    
    Args:
        question: The research question to investigate.
    
    Returns:
        The final research report as a markdown string.
    """
    print(f"Starting research pipeline for: {question}\n")
    
    result = await Runner.run(
        starting_agent=planner_agent,
        input=f"Research question: {question}",
    )
    
    return result.final_output


# Example usage
if __name__ == "__main__":
    question = (
        "What are the key differences between RAG and fine-tuning "
        "for adapting LLMs to domain-specific tasks, and when should "
        "each approach be used?"
    )
    
    report = asyncio.run(run_research_pipeline(question))
    print("=" * 60)
    print("FINAL REPORT")
    print("=" * 60)
    print(report)

When you run this, the SDK handles the entire flow: the Planner decomposes the question, hands off to the Searcher, which uses tools to gather sources, hands off to the Analyst for extraction, and finally the Writer produces the report. The built-in tracing lets you inspect every step.

Adding Guardrails for Safety

In production, you want to validate inputs and outputs. The SDK supports guardrails that run alongside your agents. Let's add an input guardrail that rejects harmful or out-of-scope queries, and an output guardrail that checks the final report for uncited claims:

from agents import GuardrailFunctionOutput, input_guardrail, output_guardrail


@input_guardrail
async def scope_guardrail(ctx, agent, input_data):
    """Reject queries that are not research-related."""
    blocked_keywords = ["hack", "exploit", "malware", "phishing"]
    input_lower = str(input_data).lower()
    
    is_blocked = any(kw in input_lower for kw in blocked_keywords)
    return GuardrailFunctionOutput(
        output_info={"blocked_reason": "out of scope" if is_blocked else None},
        tripwire_triggered=is_blocked,
    )


@output_guardrail
async def citation_guardrail(ctx, agent, output):
    """Check that the final report contains citations."""
    output_str = str(output)
    has_urls = "http" in output_str or "www." in output_str
    return GuardrailFunctionOutput(
        output_info={"has_citations": has_urls},
        tripwire_triggered=not has_urls,
    )

Attach the input guardrail to the Planner and the output guardrail to the Writer:

planner_agent = Agent(
    name="Research Planner",
    instructions=PLANNER_INSTRUCTIONS,
    model="gpt-4o",
    input_guardrails=[scope_guardrail],
)

writer_agent = Agent(
    name="Report Writer",
    instructions=WRITER_INSTRUCTIONS,
    model="gpt-4o",
    output_guardrails=[citation_guardrail],
)

If a guardrail's tripwire is triggered, the run stops immediately and raises a GuardrailTripwireTriggered exception that you can catch and handle gracefully.

Tracing and Observability

One of the SDK's most valuable features is automatic tracing. Every agent invocation, tool call, and handoff is recorded. You can view traces in the OpenAI dashboard or export them programmatically:

from agents import trace


async def run_research_pipeline_with_tracing(question: str) -> str:
    with trace("Research Pipeline Run"):
        result = await Runner.run(
            starting_agent=planner_agent,
            input=f"Research question: {question}",
        )
    
    print(f"Trace completed. View in OpenAI dashboard.")
    return result.final_output

Traces show you exactly where time is spent, which tools were called, how many tokens each agent consumed, and where handoffs occurred. This is indispensable for debugging and cost optimization.

Best Practices

After building several production pipelines, the following practices consistently lead to better results:

Extending the Pipeline

Once you have the basic four-agent pipeline working, there are several valuable extensions:

Parallel search agents: Instead of one Searcher handling all sub-questions sequentially, spawn a separate Searcher for each sub-question and run them concurrently with asyncio.gather. This can cut latency dramatically for multi-part questions.

async def parallel_search(sub_questions: list[str]) -> list[dict]:
    """Run searches for all sub-questions in parallel."""
    tasks = [
        Runner.run(
            starting_agent=searcher_agent,
            input=f"Search for: {sq}",
        )
        for sq in sub_questions
    ]
    results = await asyncio.gather(*tasks)
    return [r.final_output for r in results]

Fact-checker agent: Add a fifth agent that runs after the Writer and independently verifies key claims against the source material, flagging any unsupported assertions before the report is delivered to the user.

Human-in-the-loop: Use the SDK's input_fn parameter to pause the pipeline after the Planner produces sub-questions, let a human review and edit them, then resume execution. This is especially valuable for high-stakes research where getting the decomposition right matters.

Conclusion

Building a multi-agent research pipeline with the OpenAI Agents SDK transforms a fragile, hard-to-maintain mega-prompt into a modular, observable, and extensible system. By decomposing research into planning, searching, analysis, and writing — each handled by a specialized agent with its own tools and guardrails — you get higher-quality output, better debuggability, and a architecture that scales as your needs grow. Start with the four-agent pipeline in this guide, add tracing from day one, and iterate on individual agents using the best practices covered here. The result is a research system that produces consistently deeper, better-cited, and more reliable reports than any single-agent approach can match.

— Ad —

Google AdSense will appear here after approval

← Back to all articles