Build a Multi-Agent Research Pipeline

Build a multi-agent pipeline with tools, handoffs, and guardrails on free infra.

Introduction

30 Second Summary

Modern AI assistants rarely work alone. When you ask a complex question, multiple specialized agents often collaborate behind the scenes to research, verify, and format the answer you receive.

In this project, you will build a multi-agent research pipeline where three AI agents collaborate to answer questions. You'll give them tools, teach them to hand off tasks to each other, and add a guardrail that blocks off-topic requests before they waste resources.

What You'll Build

A Python-based multi-agent research assistant powered by the OpenAI Agents SDK and Groq's free LLM API.

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

  • A Triage Agent that receives user questions and decides which specialist agent to delegate to.
  • Custom function tools (web search and calculator) that agents call autonomously to gather information.
  • Agent-to-agent handoffs that route work between a Research Agent and a Writer Agent.
  • An input guardrail that validates requests and blocks off-topic queries before the pipeline runs.
  • Secret Mission: Add a fact-checking agent that verifies research findings before passing them to the writer.

Are there any prerequisites?

You should be comfortable with basic Python (functions, variables, strings). No async or AI experience is needed. The entire project runs on free infrastructure with no credit card required.

Before We Start

Before building your multi-agent pipeline, you need to install the agent framework and connect to a free AI provider.

You'll use the OpenAI Agents SDK as the agent framework and Groq for free LLM inference. By the end of this step, you'll have a verified connection to a powerful language model at zero cost.

In this step, get ready to:
  • Install the OpenAI Agents SDK and project dependencies.
  • Create a Groq account and save your API key.
  • Verify the LLM connection works with a test script.
Create a project folder and install dependencies

Every Python project needs its own isolated environment so package installations don't conflict with other projects on your machine.

  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
  • 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 a virtual environment by running this command:
python3 -m venv .venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, preventing version conflicts between projects.

  • Activate the virtual environment:

macOS

  • Activate by running:
source .venv/bin/activate

Your terminal prompt now shows (.venv) at the beginning, confirming you're inside the virtual environment.

Windows

  • Activate by running:
.venv\Scripts\activate

Your terminal prompt now shows (.venv) at the beginning, confirming you're inside the virtual environment.

  • Install the OpenAI Agents SDK with LiteLLM support and the dotenv package by running:
pip install "openai-agents[litellm]" python-dotenv

What are these packages?

The openai-agents[litellm] package is the OpenAI Agents SDK with LiteLLM integration. LiteLLM lets you connect to non-OpenAI models (like Groq) using the same SDK.

The python-dotenv package loads environment variables from a .env file so you can keep secrets out of your code.

You should see packages downloading and installing. The install may take a minute since openai-agents has several dependencies.

Install failing?

Make sure you see (.venv) in your terminal prompt. If not, the virtual environment is not activated. Re-run the activation command above.

If you see a permissions error, confirm you're using pip from inside the virtual environment (not system pip).

Help me troubleshoot this install issue

Create a Groq account and save your API key

Groq provides free LLM inference with no credit card required. You'll use their API to power your agents.

  • Open your browser and go to console.groq.com.
  • Click Sign Up and create an account. Google OAuth works for a quick sign-up.
  • Once logged in, click API Keys in the left sidebar.
  • Click Create API Key.
  • Give your key a name (e.g. research-pipeline) and click Create.
  • Copy the key immediately. Groq only shows it once.

Why Groq?

Groq runs models on custom LPU hardware that delivers extremely fast inference. Their free tier gives you 30 requests per minute and 14,400 requests per day with no credit card required.

  • Switch back to your terminal. Make sure you're inside the research-pipeline folder.
  • Create a .env file by running:

macOS

touch .env

Windows

New-Item .env
  • Confirm the file exists by running:
ls

You should see .env and .venv in the output.

  • Open VS Code in the current folder by running:
code .
  • In the VS Code Explorer sidebar, click on the .env file to open it.
  • Paste the following into .env, replacing your-groq-api-key-here with the key you just copied:
GROQ_API_KEY=your-groq-api-key-here
  • Save the file with Cmd+S (macOS) or Ctrl+S (Windows).

Lost your API key?

If you closed the dialog without copying, go back to the API Keys section in the Groq console. Revoke the old key and create a new one.

Help me find where to create a Groq API key

✔️ Awesome, I've got everything!

Great. Double check you've saved your .env file.

ⓧ I'd like to double check the full code

Your .env file should look exactly like this (with your own key):

GROQ_API_KEY=your-groq-api-key-here
Verify the connection with a test script

Before building the full pipeline, confirm that your API key works and the SDK can talk to Groq.

  • In VS Code, create a new file called test_connection.py by pressing Cmd+N (macOS) or Ctrl+N (Windows), then saving the file as test_connection.py.
  • Paste the following code into test_connection.py:
import os

from dotenv import load_dotenv

from agents import Agent, Runner, set_tracing_disabled
from agents.extensions.models.litellm_model import LitellmModel

# Load environment variables from .env file
load_dotenv()

# Disable tracing (requires an OpenAI key which we are not using)
set_tracing_disabled(disabled=True)

# Set up the model to use Groq via LiteLLM
MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))

# Create a simple test agent
agent = Agent(
    name="Test Agent",
    instructions="You are a helpful assistant. Respond in one sentence.",
    model=MODEL,
)

# Run the agent with a test question
result = Runner.run_sync(agent, "What is 2 + 2?")
print(result.final_output)

What does this code do?

This script loads your Groq API key from .env, creates a LitellmModel pointing at Groq's llama-3.3-70b-versatile model, wraps it in a minimal Agent, and runs a single question through it.

The set_tracing_disabled(disabled=True) call is needed because the SDK's tracing feature expects an OpenAI API key by default. Since you're using Groq instead, you disable tracing.

The Runner.run_sync() method handles all the async machinery internally so you never need to write async/await code yourself.

  • Save test_connection.py.
  • Run the test script in your terminal by running:
python3 test_connection.py

You should see a short text response from the model (something like "2 + 2 equals 4."). The exact wording varies each time.

Seeing an error?

If you see a module not found error, your virtual environment is not activated. Run the activation command again and retry.

If you see an authentication error, double check that your .env file has GROQ_API_KEY= followed by your actual key (no quotes, no spaces).

Help me debug my test script

Your environment is ready and connected to Groq. Next up, you'll build your first agent with custom tools that can search for information and do calculations.

Build Your First Agent with a Custom Tool

An AI agent on its own can only generate text. To make it useful, you give it tools that let it take real actions like searching the web or running calculations.

In this step, you'll create two Python functions and register them as tools with the OpenAI Agents SDK. Then you'll build a Research Agent that decides which tool to call based on the question it receives.

In this step, get ready to:
  • Create function tools that the agent can call autonomously.
  • Build a Research Agent wired up with those tools.
  • Test the agent to observe how it chooses which tool to use.
Create function tools for the Research Agent

Tools are regular Python functions that an agent can choose to call. The SDK's @function_tool decorator handles exposing the function's name, description, and parameters to the LLM automatically.

  • Create a new file called main.py in your project folder.
  • Add the project imports and configuration by pasting this code:
import os

from dotenv import load_dotenv
from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    Runner,
    RunContextWrapper,
    TResponseInputItem,
    function_tool,
    input_guardrail,
    set_tracing_disabled,
)
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

load_dotenv()
set_tracing_disabled(disabled=True)

MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))

What does this code do?

The imports pull in everything you'll need across the entire project. You won't use all of them right away, but having them ready avoids editing the import block in every step.

  • load_dotenv() reads your .env file so the Groq API key is available.
  • set_tracing_disabled(disabled=True) turns off the SDK's built-in tracing, which requires an OpenAI API key you don't have.
  • MODEL creates a reusable model reference pointing at Groq's Llama 3.3 70B model via LiteLLM.
  • Save main.py and run it to confirm there are no import errors:
python main.py
  • The command should complete with no output and no errors. If you see an error, check the troubleshooting callout below.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is still activated. You should see (.venv) at the start of your terminal prompt.
  • Confirm the package is installed by running pip list | grep openai-agents. If it's missing, reinstall with pip install "openai-agents[litellm]".

Help me debug a ModuleNotFoundError when running main.py with the openai-agents SDK

Now that the file is set up, you'll add the first tool. The search_web function simulates looking up information. In a production system this would call a real search API, but mock data lets you see the tool-use pattern without external dependencies.

  • In main.py, add the following code below the MODEL line:

# --- Tools ---

@function_tool
def search_web(query: str) -> str:
    """Search the web for information on a given query.

    Args:
        query: The search query to look up.
    """
    mock_results = {
        "python": "Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes readability and supports multiple paradigms.",
        "machine learning": "Machine learning is a subset of AI that enables systems to learn from data. Key approaches include supervised, unsupervised, and reinforcement learning.",
        "climate change": "Global temperatures have risen approximately 1.1C since pre-industrial times. The Paris Agreement aims to limit warming to 1.5C.",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"Search results for '{query}': This is a general topic. Key facts include recent developments and ongoing research in this area."

What does this code do?

  • @function_tool registers the function so the LLM knows it exists, what it does (from the docstring), and what parameters it accepts.
  • The function checks if the query matches any keys in mock_results and returns the matching data. If nothing matches, it returns a generic response.
  • The docstring is critical. The LLM reads it to decide when to call this tool, so it must clearly describe the function's purpose.

You won't see a change yet. The next chunk adds the second tool.

  • Add the calculate tool directly below the search_web function:

@function_tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A mathematical expression to evaluate (e.g., '2 + 2', '100 / 4').
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression. Only numbers and basic operators are allowed."
        result = eval(expression)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error evaluating expression: {e}"

What does this code do?

  • The allowed_chars check prevents the agent from passing dangerous code into eval(). Only digits and basic math operators are permitted.
  • If the expression is valid, it evaluates and returns a formatted result string. If something goes wrong, it returns a descriptive error the agent can read.
  • Save main.py and run it again to confirm there are no syntax errors:
python main.py
  • The command should complete silently with no errors. Both tools are now registered and ready for an agent to use.

Seeing a SyntaxError?

  • Check that the triple-quoted docstrings are properly closed with three double quotes on their own line.
  • Make sure the @function_tool decorator sits directly above the def line with no blank line between them.

Help me fix a SyntaxError in my function tool definitions

Create the Research Agent

An Agent in the SDK is defined by its name, instructions (a system prompt), a model, and the tools it can access. The LLM reads the instructions to understand its role and uses the tool descriptions to decide when to call each one.

  • Add the Research Agent definition below the calculate function:

# --- Agents ---

research_agent = Agent(
    name="Research Agent",
    instructions="You are a research specialist. Use your tools to gather information and provide detailed, factual answers. Always use the search_web tool for factual questions and the calculate tool for math.",
    model=MODEL,
    tools=[search_web, calculate],
)

What does this code do?

  • name gives the agent a human-readable identity for logging and handoff references.
  • instructions is the system prompt. It tells the LLM what role to play and when to use its tools.
  • tools=[search_web, calculate] gives the agent access to both functions. The LLM sees their docstrings and decides which to call based on the user's question.
  • Save main.py.

You won't see a change yet. The next substep runs the agent so you can see tool-use in action.

Test the Research Agent

The Runner.run_sync() method sends input to an agent and waits for the final response. It handles the entire loop of the agent reading the question, calling tools if needed, and returning a final answer.

  • Add these test lines at the very bottom of main.py:
# --- Test (temporary) ---
result = Runner.run_sync(research_agent, "What is machine learning and what is 25 * 4?")
print(result.final_output)

What does this code do?

This sends a question that requires both tools. The agent should call search_web for the machine learning question and calculate for the math. You're testing that the agent autonomously picks the right tool for each part of the question.

  • Save main.py and run it:
python main.py
  • You should see a response that includes information about machine learning from the search results and the answer to 25 * 4 (which is 100). The agent combines both tool outputs into a single coherent answer.

What is the ReAct pattern?

What you just saw is the ReAct pattern in action. The agent Reasons about what tools to use, Acts by calling them, observes the results, then Reasons again to formulate its final answer. This loop is the foundation of how modern AI agents operate.

Seeing an authentication error?

  • Check that your .env file is in the same folder as main.py and contains GROQ_API_KEY= followed by your actual key (no quotes around the value).
  • Verify your Groq key is valid by checking the API Keys page at console.groq.com.

Help me debug a Groq authentication error in my agent script

  • Now that you've confirmed the agent works, remove the three test lines from the bottom of main.py (the comment, the result line, and the print line).
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and that the test lines have been removed.

ⓧ I'd like to double check the full code

Here's what your main.py should look like at the end of this step:

import os

from dotenv import load_dotenv
from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    Runner,
    RunContextWrapper,
    TResponseInputItem,
    function_tool,
    input_guardrail,
    set_tracing_disabled,
)
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

load_dotenv()
set_tracing_disabled(disabled=True)

MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))


# --- Tools ---

@function_tool
def search_web(query: str) -> str:
    """Search the web for information on a given query.

    Args:
        query: The search query to look up.
    """
    mock_results = {
        "python": "Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes readability and supports multiple paradigms.",
        "machine learning": "Machine learning is a subset of AI that enables systems to learn from data. Key approaches include supervised, unsupervised, and reinforcement learning.",
        "climate change": "Global temperatures have risen approximately 1.1C since pre-industrial times. The Paris Agreement aims to limit warming to 1.5C.",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"Search results for '{query}': This is a general topic. Key facts include recent developments and ongoing research in this area."


@function_tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A mathematical expression to evaluate (e.g., '2 + 2', '100 / 4').
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression. Only numbers and basic operators are allowed."
        result = eval(expression)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error evaluating expression: {e}"


# --- Agents ---

research_agent = Agent(
    name="Research Agent",
    instructions="You are a research specialist. Use your tools to gather information and provide detailed, factual answers. Always use the search_web tool for factual questions and the calculate tool for math.",
    model=MODEL,
    tools=[search_web, calculate],
)

Your Research Agent is working and choosing tools on its own. Next up, you'll create a second agent and wire them together with handoffs so a Triage Agent can delegate questions to the right specialist.

Add a Second Agent and Wire Up Handoffs

Your Research Agent can look things up and do math. But what about tasks that need polished writing instead of raw facts?

In this step, you'll create a Writer Agent and a Triage Agent that routes incoming requests to the right specialist. The Triage Agent uses handoffs to delegate control to another agent entirely, letting the specialist handle the rest of the conversation.

In this step, get ready to:
  • Create a Writer Agent with summarization instructions.
  • Create a Triage Agent that delegates to specialists using handoffs.
  • Test the full pipeline with different query types.
Create the Writer Agent

The Writer Agent handles tasks like summarization and restructuring. It does not need tools because it works purely with language.

  • In main.py, find the research_agent definition (the last block in your file).
  • Add the following code below it to create the Writer Agent:
writer_agent = Agent(
    name="Writer Agent",
    instructions="You are a writing specialist. Take information and produce clear, well-structured summaries. Use bullet points for key facts and write in a professional but accessible tone.",
    model=MODEL,
)

What does this code do?

This creates an Agent with no tools. The Writer Agent relies entirely on the LLM's language abilities to produce polished output.

It shares the same MODEL constant as the Research Agent. Since Groq's free tier covers all requests, every agent in your pipeline runs at zero cost.

  • Save main.py.

You won't see a change yet. The next chunk wires this agent into the handoff pipeline.

Create the Triage Agent with handoffs

A handoff transfers full control of the conversation from one agent to another. The receiving agent takes over completely and produces the final response.

  • In main.py, add the following code below writer_agent:
triage_agent = Agent(
    name="Triage Agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are a triage agent. Your job is to route user requests to the right specialist:
    - For questions requiring research, facts, or calculations, hand off to the Research Agent.
    - For requests to write, summarize, or restructure existing information, hand off to the Writer Agent.
    Choose the most appropriate specialist based on the user's needs.""",
    model=MODEL,
    handoffs=[research_agent, writer_agent],
)

What does this code do?

  • handoffs=[research_agent, writer_agent] tells the SDK that this agent can transfer control to either specialist. The LLM reads the agent names and instructions to decide which one fits the user's request.
  • RECOMMENDED_PROMPT_PREFIX is a prompt snippet from the SDK that helps the LLM understand how the handoff mechanism works. It is prepended to your custom instructions.
  • The f-string format lets you embed the prefix variable directly into the instruction text.

Why handoffs instead of calling both agents yourself?

You could write Python logic to route requests manually. Handoffs let the LLM make routing decisions instead. The Triage Agent reads the user's intent and picks the best specialist, adapting to ambiguous or multi-part questions without hardcoded rules.

  • Save main.py.

You won't see a change yet. The next substep runs the pipeline to confirm routing works.

Seeing a syntax error on save?

  • Check that the triple-quoted f-string uses three double quotes at the start and end. Mixing single and double quotes breaks the string.
  • Make sure triage_agent is defined after both research_agent and writer_agent. Python reads top to bottom, so the agents must exist before they are referenced in handoffs.

Help me fix a syntax error in my triage agent definition

✔️ 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

import os

from dotenv import load_dotenv
from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    Runner,
    RunContextWrapper,
    TResponseInputItem,
    function_tool,
    input_guardrail,
    set_tracing_disabled,
)
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

load_dotenv()
set_tracing_disabled(disabled=True)

MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))


# --- Tools ---

@function_tool
def search_web(query: str) -> str:
    """Search the web for information on a given query.

    Args:
        query: The search query to look up.
    """
    mock_results = {
        "python": "Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes readability and supports multiple paradigms.",
        "machine learning": "Machine learning is a subset of AI that enables systems to learn from data. Key approaches include supervised, unsupervised, and reinforcement learning.",
        "climate change": "Global temperatures have risen approximately 1.1C since pre-industrial times. The Paris Agreement aims to limit warming to 1.5C.",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"Search results for '{query}': This is a general topic. Key facts include recent developments and ongoing research in this area."


@function_tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A mathematical expression to evaluate (e.g., '2 + 2', '100 / 4').
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression. Only numbers and basic operators are allowed."
        result = eval(expression)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error evaluating expression: {e}"


# --- Agents ---

research_agent = Agent(
    name="Research Agent",
    instructions="You are a research specialist. Use your tools to gather information and provide detailed, factual answers. Always use the search_web tool for factual questions and the calculate tool for math.",
    model=MODEL,
    tools=[search_web, calculate],
)

writer_agent = Agent(
    name="Writer Agent",
    instructions="You are a writing specialist. Take information and produce clear, well-structured summaries. Use bullet points for key facts and write in a professional but accessible tone.",
    model=MODEL,
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are a triage agent. Your job is to route user requests to the right specialist:
    - For questions requiring research, facts, or calculations, hand off to the Research Agent.
    - For requests to write, summarize, or restructure existing information, hand off to the Writer Agent.
    Choose the most appropriate specialist based on the user's needs.""",
    model=MODEL,
    handoffs=[research_agent, writer_agent],
)
Test the full handoff pipeline

Time to see multi-agent orchestration in action. You will send two different queries to the Triage Agent and watch it route each one to the appropriate specialist.

  • Add the following test code at the bottom of main.py:
# --- Test the pipeline ---

# Test 1: Research question (should route to Research Agent)
result = Runner.run_sync(triage_agent, "What is machine learning and how does it relate to AI?")
print("Research question result:")
print(result.final_output)

print("\n" + "="*60 + "\n")

# Test 2: Writing request (should route to Writer Agent)
result = Runner.run_sync(triage_agent, "Summarize the key benefits of Python for beginners in bullet points.")
print("Writing request result:")
print(result.final_output)

What does this code do?

  • The first query asks a factual question. The Triage Agent should hand off to the Research Agent, which will call search_web to look up information.
  • The second query asks for a written summary. The Triage Agent should hand off to the Writer Agent, which will produce polished output directly.
  • Both calls use Runner.run_sync() so you don't need to write any async code yourself.
  • Save main.py.
  • Run the script by entering this command in your terminal:
python main.py

You should see two separate responses printed. The first response contains factual information about machine learning (pulled from your mock search data). The second response is a polished summary with bullet points.

Not seeing two separate responses?

  • If you see a NameError mentioning triage_agent, check that your test code is below the triage_agent definition.
  • If the output looks the same for both queries (no routing difference), double-check that your handoffs list includes both research_agent and writer_agent.
  • If you see a rate limit error from Groq, wait 60 seconds and try again. The free tier allows 30 requests per minute.

Help me debug why my triage agent isn't routing to the correct specialist

  • Once you have confirmed both queries produce different response styles, delete the test code (everything below the triage_agent definition) to clean up your file.
  • Save main.py.

Your multi-agent pipeline now has intelligent routing. The Triage Agent reads each request and delegates to the right specialist without any hardcoded if-statements. Next up, you will add a guardrail that blocks off-topic requests before the pipeline even starts.

Add a Guardrail to Validate Input

Your multi-agent pipeline works, but right now it processes any request, no matter how irrelevant. If someone asks your research pipeline to write a love poem, it will happily try. In production systems, wasted agent calls cost money and compute.

A guardrail is a lightweight check that runs before the main pipeline. It validates whether the user's request is appropriate, and blocks it if not. In this step, you'll create an input guardrail that uses a separate agent to classify requests before the expensive pipeline runs.

In this step, get ready to:
  • Define a guardrail output model and a guardrail agent that classifies requests.
  • Implement the guardrail function with the @input_guardrail decorator.
  • Attach the guardrail to the Triage Agent and test with on-topic and off-topic inputs.
Define the guardrail output model and agent

The guardrail pattern in the OpenAI Agents SDK works by running a fast, cheap agent call to classify the input. This agent returns structured output (a Pydantic model) so the SDK can programmatically check whether the request should be blocked.

  • In main.py, find the triage_agent definition (the last block of code in your file).
  • Add the following code above the triage_agent definition, below writer_agent:
# --- Guardrail ---

class TopicCheckOutput(BaseModel):
    is_off_topic: bool
    reasoning: str

guardrail_agent = Agent(
    name="Topic Guardrail",
    instructions="Determine if the user request is a legitimate research or writing question. Flag as off-topic if it is: a personal request (poems, stories, jokes), harmful content, or completely unrelated to information gathering. Legitimate requests include questions about facts, science, technology, history, math, or requests to summarize/explain topics.",
    model=MODEL,
    output_type=TopicCheckOutput,
)

What does this code do?

You are defining two things here:

  • TopicCheckOutput is a Pydantic model that forces the guardrail agent to return structured data. The is_off_topic boolean is what the SDK uses to decide whether to block the request.
  • guardrail_agent is a separate, lightweight agent whose only job is classification. The output_type=TopicCheckOutput parameter tells the SDK to parse the agent's response into the Pydantic model automatically.

Why use a separate agent for the guardrail?

The guardrail agent is fast and focused. It only needs to answer one question: is this on-topic? Because it uses the same free Groq model, there is no extra cost. This check runs before your main agents do any real work.

  • Save main.py.
Implement the guardrail function

The guardrail agent can classify requests, but the SDK needs a decorated function that runs the agent and returns a standardized result. This function is what the SDK calls automatically before your Triage Agent processes input.

  • In main.py, add the following code directly below the guardrail_agent definition you just wrote (still above triage_agent):
@input_guardrail
async def topic_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_off_topic,
    )

What does this code do?

  • The @input_guardrail decorator registers this function as a guardrail the SDK can call automatically.
  • Inside the function, Runner.run(guardrail_agent, input, context=ctx.context) sends the user's input to the guardrail agent for classification.
  • The function returns a GuardrailFunctionOutput where tripwire_triggered is set to the value of is_off_topic. When this is True, the SDK raises an exception and the main pipeline never runs.

Why is this function async?

The SDK requires guardrail functions to be async because it runs them concurrently with other processing. You do not need to write any other async code. Runner.run_sync() handles the event loop for you when you run the pipeline.

  • Save main.py.
Attach the guardrail to the Triage Agent and test

The guardrail function exists, but no agent is using it yet. You need to wire it into the Triage Agent so it runs automatically on every incoming request.

  • In main.py, find the triage_agent definition. Locate this line:
    handoffs=[research_agent, writer_agent],
)
  • Replace those two lines with the following, which adds the input_guardrails parameter:
    handoffs=[research_agent, writer_agent],
    input_guardrails=[topic_guardrail],
)

What does this change do?

The input_guardrails=[topic_guardrail] parameter tells the SDK to run your guardrail function before the Triage Agent processes any input. If the guardrail trips, the SDK raises InputGuardrailTripwireTriggered and the agent never sees the request.

Now add a test block at the bottom of your file so you can see the guardrail in action.

  • Add the following test code at the very bottom of main.py:
if __name__ == "__main__":
    # Test on-topic request
    try:
        result = Runner.run_sync(triage_agent, "What is machine learning and how does it relate to AI?")
        print(f"\nFinal Output:\n{result.final_output}")
    except InputGuardrailTripwireTriggered:
        print("\n[BLOCKED] Your request was flagged as off-topic. Please ask a research or writing question.")

    # Test off-topic request
    try:
        result = Runner.run_sync(triage_agent, "Write me a love poem about pizza.")
        print(f"\nFinal Output:\n{result.final_output}")
    except InputGuardrailTripwireTriggered:
        print("\n[BLOCKED] Your request was flagged as off-topic. Please ask a research or writing question.")

What does this test code do?

  • The first test sends a legitimate research question. It should pass through the guardrail and produce a response from the Research Agent.
  • The second test sends an off-topic request (a poem). The guardrail should trip, raising InputGuardrailTripwireTriggered, and the except block prints the blocked message.
  • Save main.py.
  • Run the script in your terminal:
python main.py
  • For the first query, you should see the Research Agent produce a response about machine learning.
  • For the second query, you should see [BLOCKED] Your request was flagged as off-topic. printed instead of any agent output.

Not seeing the blocked message?

  • Make sure input_guardrails=[topic_guardrail] is inside the triage_agent definition, not after it.
  • Verify that the topic_guardrail function is defined above triage_agent in the file. Python reads top to bottom, so the function must exist before it is referenced.
  • Check that your TopicCheckOutput class has is_off_topic: bool spelled exactly as shown. A typo means tripwire_triggered will never be set to True.

My guardrail isn't blocking off-topic requests. What should I check?

  • Once both tests pass, delete the entire if __name__ == "__main__": block from the bottom of main.py. You will build a cleaner version in the next step.
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file with the test code removed.

ⓧ I'd like to double check the full code

Here is the full main.py at this point (without the test block):

import os

from dotenv import load_dotenv
from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    Runner,
    RunContextWrapper,
    TResponseInputItem,
    function_tool,
    input_guardrail,
    set_tracing_disabled,
)
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

load_dotenv()
set_tracing_disabled(disabled=True)

MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))


# --- Tools ---

@function_tool
def search_web(query: str) -> str:
    """Search the web for information on a given query.

    Args:
        query: The search query to look up.
    """
    mock_results = {
        "python": "Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes readability and supports multiple paradigms.",
        "machine learning": "Machine learning is a subset of AI that enables systems to learn from data. Key approaches include supervised, unsupervised, and reinforcement learning.",
        "climate change": "Global temperatures have risen approximately 1.1C since pre-industrial times. The Paris Agreement aims to limit warming to 1.5C.",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"Search results for '{query}': This is a general topic. Key facts include recent developments and ongoing research in this area."


@function_tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A mathematical expression to evaluate (e.g., '2 + 2', '100 / 4').
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression. Only numbers and basic operators are allowed."
        result = eval(expression)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error evaluating expression: {e}"


# --- Agents ---

research_agent = Agent(
    name="Research Agent",
    instructions="You are a research specialist. Use your tools to gather information and provide detailed, factual answers. Always use the search_web tool for factual questions and the calculate tool for math.",
    model=MODEL,
    tools=[search_web, calculate],
)

writer_agent = Agent(
    name="Writer Agent",
    instructions="You are a writing specialist. Take information and produce clear, well-structured summaries. Use bullet points for key facts and write in a professional but accessible tone.",
    model=MODEL,
)

# --- Guardrail ---

class TopicCheckOutput(BaseModel):
    is_off_topic: bool
    reasoning: str

guardrail_agent = Agent(
    name="Topic Guardrail",
    instructions="Determine if the user request is a legitimate research or writing question. Flag as off-topic if it is: a personal request (poems, stories, jokes), harmful content, or completely unrelated to information gathering. Legitimate requests include questions about facts, science, technology, history, math, or requests to summarize/explain topics.",
    model=MODEL,
    output_type=TopicCheckOutput,
)


@input_guardrail
async def topic_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_off_topic,
    )


triage_agent = Agent(
    name="Triage Agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are a triage agent. Your job is to route user requests to the right specialist:
    - For questions requiring research, facts, or calculations, hand off to the Research Agent.
    - For requests to write, summarize, or restructure existing information, hand off to the Writer Agent.
    Choose the most appropriate specialist based on the user's needs.""",
    model=MODEL,
    handoffs=[research_agent, writer_agent],
    input_guardrails=[topic_guardrail],
)

Your pipeline now has a protective layer that blocks irrelevant requests before they waste agent resources. Next up, you will consolidate everything into a clean, runnable script with proper error handling and formatted output.

Run the Full Pipeline End-to-End

Your pipeline now has all the building blocks: agents with tools, handoffs for delegation, and a guardrail that blocks off-topic requests. The final step is bringing everything together into a clean, runnable script.

Right now your main.py defines agents and tools but has no entry point to actually run them. You need a function that accepts user input, runs the pipeline, and handles errors gracefully.

In this step, get ready to:
  • Add a run_pipeline function with error handling and formatted output.
  • Add test queries that exercise all three pipeline behaviors.
  • Run the full pipeline end-to-end and verify all three query types.
Add the run_pipeline function

The pipeline needs a single entry point that takes user input, passes it to the Triage Agent, and prints the result. It also needs to catch the InputGuardrailTripwireTriggered exception so off-topic requests fail gracefully instead of crashing the script.

  • In main.py, find the line that defines triage_agent = Agent( and add this section comment directly above it:
# --- Triage Agent ---
  • Scroll to the very bottom of main.py (after the closing parenthesis of triage_agent) and add the run_pipeline function by pasting this code:

# --- Run Pipeline ---

def run_pipeline(user_input: str) -> None:
    print(f"\n{'='*60}")
    print(f"User: {user_input}")
    print(f"{'='*60}")

    try:
        result = Runner.run_sync(triage_agent, user_input)
        print(f"\nFinal Output:\n{result.final_output}")
    except InputGuardrailTripwireTriggered:
        print("\n[BLOCKED] Your request was flagged as off-topic. Please ask a research or writing question.")
    except Exception as e:
        print(f"\n[ERROR] Something went wrong: {e}")

What does this code do?

  • run_pipeline prints a formatted header showing the user's question, then passes it to Runner.run_sync(triage_agent, user_input).
  • If the guardrail fires, InputGuardrailTripwireTriggered is caught and a friendly [BLOCKED] message prints instead of a traceback.
  • The generic except Exception catches any other issue (network errors, model timeouts) and prints a clear [ERROR] message.
  • Save main.py.
Add test queries to the main block

The if __name__ == "__main__": block ensures the test queries only run when you execute main.py directly. You will add three queries that exercise each pipeline path: research, writing, and off-topic.

  • At the very bottom of main.py (below the run_pipeline function), add the main block by pasting this code:

if __name__ == "__main__":
    print("Multi-Agent Research Pipeline")
    print("Powered by Groq (free tier) + OpenAI Agents SDK")
    print("-" * 60)

    run_pipeline("What is machine learning and how does it relate to AI?")
    run_pipeline("Summarize the key benefits of Python for beginners in bullet points.")
    run_pipeline("Write me a love poem about pizza.")

What does this code do?

  • The first query is a research question. The Triage Agent should hand off to the Research Agent, which calls the search_web tool.
  • The second query is a writing/summarization request. The Triage Agent should hand off directly to the Writer Agent.
  • The third query is an off-topic request. The guardrail should block it before any agent processes it.
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before running the pipeline.

ⓧ I'd like to double check the full code

import os

from dotenv import load_dotenv
from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    Runner,
    RunContextWrapper,
    TResponseInputItem,
    function_tool,
    input_guardrail,
    set_tracing_disabled,
)
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

load_dotenv()
set_tracing_disabled(disabled=True)

MODEL = LitellmModel(model="groq/llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))


# --- Tools ---

@function_tool
def search_web(query: str) -> str:
    """Search the web for information on a given query.

    Args:
        query: The search query to look up.
    """
    mock_results = {
        "python": "Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes readability and supports multiple paradigms.",
        "machine learning": "Machine learning is a subset of AI that enables systems to learn from data. Key approaches include supervised, unsupervised, and reinforcement learning.",
        "climate change": "Global temperatures have risen approximately 1.1C since pre-industrial times. The Paris Agreement aims to limit warming to 1.5C.",
    }
    for key, value in mock_results.items():
        if key in query.lower():
            return value
    return f"Search results for '{query}': This is a general topic. Key facts include recent developments and ongoing research in this area."


@function_tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A mathematical expression to evaluate (e.g., '2 + 2', '100 / 4').
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression. Only numbers and basic operators are allowed."
        result = eval(expression)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error evaluating expression: {e}"


# --- Agents ---

research_agent = Agent(
    name="Research Agent",
    instructions="You are a research specialist. Use your tools to gather information and provide detailed, factual answers. Always use the search_web tool for factual questions and the calculate tool for math.",
    model=MODEL,
    tools=[search_web, calculate],
)

writer_agent = Agent(
    name="Writer Agent",
    instructions="You are a writing specialist. Take information and produce clear, well-structured summaries. Use bullet points for key facts and write in a professional but accessible tone.",
    model=MODEL,
)

# --- Guardrail ---

class TopicCheckOutput(BaseModel):
    is_off_topic: bool
    reasoning: str

guardrail_agent = Agent(
    name="Topic Guardrail",
    instructions="Determine if the user request is a legitimate research or writing question. Flag as off-topic if it is: a personal request (poems, stories, jokes), harmful content, or completely unrelated to information gathering. Legitimate requests include questions about facts, science, technology, history, math, or requests to summarize/explain topics.",
    model=MODEL,
    output_type=TopicCheckOutput,
)


@input_guardrail
async def topic_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_off_topic,
    )


# --- Triage Agent ---

triage_agent = Agent(
    name="Triage Agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are a triage agent. Your job is to route user requests to the right specialist:
    - For questions requiring research, facts, or calculations, hand off to the Research Agent.
    - For requests to write, summarize, or restructure existing information, hand off to the Writer Agent.
    Choose the most appropriate specialist based on the user's needs.""",
    model=MODEL,
    handoffs=[research_agent, writer_agent],
    input_guardrails=[topic_guardrail],
)


# --- Run Pipeline ---

def run_pipeline(user_input: str) -> None:
    print(f"\n{'='*60}")
    print(f"User: {user_input}")
    print(f"{'='*60}")

    try:
        result = Runner.run_sync(triage_agent, user_input)
        print(f"\nFinal Output:\n{result.final_output}")
    except InputGuardrailTripwireTriggered:
        print("\n[BLOCKED] Your request was flagged as off-topic. Please ask a research or writing question.")
    except Exception as e:
        print(f"\n[ERROR] Something went wrong: {e}")


if __name__ == "__main__":
    print("Multi-Agent Research Pipeline")
    print("Powered by Groq (free tier) + OpenAI Agents SDK")
    print("-" * 60)

    run_pipeline("What is machine learning and how does it relate to AI?")
    run_pipeline("Summarize the key benefits of Python for beginners in bullet points.")
    run_pipeline("Write me a love poem about pizza.")
Run the pipeline with test queries

Time to see all three agent behaviors in action. Each query exercises a different path through the pipeline.

  • Run the pipeline by executing this command in your terminal:
python main.py

What should I see?

The script runs three queries back-to-back. You should see:

  • Query 1 (machine learning): The Research Agent responds with factual information about machine learning, pulling from the search_web tool's results.
  • Query 2 (Python benefits): The Writer Agent produces a polished bullet-point summary.
  • Query 3 (love poem): The guardrail blocks it and prints the [BLOCKED] message.

The exact wording of the AI responses will vary each run, but the routing behavior should be consistent.

Seeing an error instead of output?

  • If you see a rate limit error, wait 60 seconds and try again. Groq's free tier allows 30 requests per minute.
  • If you see an authentication error, check that your .env file contains a valid GROQ_API_KEY and that your virtual environment is active.
  • If the guardrail does not block the poem request, the LLM may have interpreted it differently. Run the script again. LLM classification can occasionally vary.

Still stuck? Help me debug my multi-agent pipeline run

Secret mission

Add a Fact-Checker Agent

Your pipeline trusts whatever the Research Agent returns. In production, that is dangerous. Add a Fact-Checker Agent that verifies claims before the Writer Agent polishes them. This introduces the powerful "agent as tool" pattern where the orchestrator keeps control instead of delegating it entirely.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Groq's free tier. There are no ongoing costs or running services to worry about.

Decide whether to keep your resources for future experimentation or delete them entirely.

Resources you used:

  • Project folder on your Desktop (contains main.py, .env, and virtual environment with installed packages).
  • Groq API key (stored in your Groq console at console.groq.com).

Keep everything running

No action needed. Choose this if you want to keep experimenting with your multi-agent pipeline.

  • Your project folder stays on your Desktop, ready to run anytime.
  • Your Groq API key continues working within the free tier limits (30 requests/minute, 14,400 requests/day).
  • Try swapping in different Groq models or adding more specialized agents to your pipeline.

Pause - I'll come back to this later

Since this project is a local Python script (not a running server), there is nothing to pause. Your files and API key will be exactly where you left them whenever you return.

  • No running processes need to be stopped.
  • Your virtual environment deactivates automatically when you close the terminal.
  • Your Groq free tier has no expiry date.

Delete - I don't want to use this again

Remove your project folder and revoke your API key to start completely fresh.

  • Delete your project folder by running this command from your Desktop:

macOS

cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]

Windows

cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]

This removes main.py, .env, your virtual environment, and all installed packages in one step.

  • To revoke your Groq API key, go to console.groq.com, open the API Keys section, and delete the key you created for this project.

Nice Work!

Nice Work!

You just built a production-style multi-agent research pipeline where specialized AI agents collaborate to answer complex questions. The entire system runs on completely free infrastructure.

You've learned how to:

  • Build a multi-agent pipeline with three collaborating agents (Triage, Research, Writer) that route questions, gather information with custom tools, and produce polished summaries.
  • Create custom function tools and wire up agent handoffs so agents autonomously decide which tools to call and which specialist to delegate to.
  • Add an input guardrail that validates user requests before the main pipeline runs, blocking off-topic queries and saving resources.
  • Secret Mission: Added a Fact-Checker Agent using the agent-as-tool pattern for multi-step verification before handing results to the Writer.

Ready to quiz yourself?