Multi-Agent Research System with PydanticAI

Build a production multi-agent system with orchestration, guardrails, and tracing.

Introduction

30 Second Summary

When a research team tackles a complex question, they don't send one person to do everything. They split the work across specialists and have a manager coordinate the results. Production AI systems work the same way.

In this project, you will build a multi-agent research system where an orchestrator agent routes queries to specialist AI agents, runs them concurrently, and iteratively improves its own output through a self-critique loop. You'll wire it all together with guardrails and full observability tracing.

What You'll Build

A production-architecture multi-agent system using PydanticAI and Gemini that demonstrates orchestrator routing, tool calling, async parallel execution, input/output guardrails, self-critique, and real-time observability via Pydantic Logfire.

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

  • Specialist agents with real tool calling that fetch and process web data autonomously, returning type-safe structured outputs.
  • An orchestrator agent that analyzes queries and conditionally routes them to the right specialist, then aggregates results into a final report.
  • Production guardrails that validate inputs before they reach the LLM and check output quality after generation, plus a reflexion loop where a critic agent iteratively improves the report.
  • Secret Mission: Add real-time streaming so you can watch each agent think, call tools, and delegate live in your terminal.

Are there any prerequisites?

This project assumes you have Python experience and familiarity with async/await. You'll also need a Google AI Studio API key for Gemini access (free tier covers this project's usage).

Before We Start

Production AI systems rarely rely on a single agent. They use multiple specialist agents coordinated by an orchestrator that routes work, enforces quality, and runs agents concurrently.

In this step, you will set up your development environment with PydanticAI v2, configure your Gemini API key, and connect Pydantic Logfire for full observability. By the end, you will have a working agent that proves everything is wired up correctly.

In this step, get ready to:
  • Create your project directory and confirm Python 3.10+.
  • Install PydanticAI v2 with Google and Logfire support.
  • Configure your API key and authenticate with Logfire.
Create the project folder and verify Python

PydanticAI v2 requires Python 3.10 or higher. Since you already have Python installed, let's confirm the version first.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

You are good to go. Your Python version meets the requirement.

ⓧ I see an older version

You need Python 3.10+ for this project. On macOS, the system Python may be outdated even after installing a newer version.

  • Download and install Python 3.10+ from python.org/downloads.
  • After installing, try running python3.11 --version or python3.12 --version to verify the new version is accessible.
  • If your system still resolves python3 to an older version, use python3.11 (or your installed version) for all commands in this project.

ⓧ Command not found

Python is not installed or not on your system PATH.

  • Download and install Python 3.10+ from python.org/downloads.
  • During installation on Windows, check the box that says Add Python to PATH.
  • Close and reopen your terminal, then run python3 --version again.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir multi-agent-research
cd multi-agent-research
  • Confirm the folder exists by running:
ls

You should see an empty directory listing since you just created this folder.

Install PydanticAI v2 with dependencies

PydanticAI provides a slim install option that lets you pick exactly which provider and integrations you need. You will install it with the Google provider (for Gemini) and Logfire support in one command.

  • Create a virtual environment and activate it by running these commands:
python3 -m venv venv
source venv/bin/activate

What about Windows?

On Windows, activate the virtual environment with venv\Scripts\activate instead of source venv/bin/activate.

  • Install PydanticAI v2 with Google and Logfire extras by running:
pip install "pydantic-ai-slim[google,logfire]==2.0.0"

What does this install command do?

The pydantic-ai-slim package is the lightweight version of PydanticAI that lets you choose specific extras. The [google,logfire] part adds the Google provider (Gemini access) and Logfire integration.

Pinning to ==2.0.0 ensures you get v2 with the new provider architecture. This also installs httpx as a transitive dependency, which your agents will use for web requests later.

  • Verify the installation succeeded by running:
python3 -c "import pydantic_ai; print('PydanticAI ready')"

You should see PydanticAI ready printed in your terminal.

Seeing an install error?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt. If not, re-run source venv/bin/activate.

If you see a version resolution error, confirm you are running Python 3.10+ inside the venv with python3 --version.

Help me troubleshoot my PydanticAI installation error.

Configure your API key and authenticate with Logfire

PydanticAI reads your Gemini API key from the GOOGLE_API_KEY environment variable. You already have this key from Google AI Studio.

  • Set your Gemini API key as an environment variable by running this command (replace the placeholder with your actual key):
export GOOGLE_API_KEY="your-api-key-here"

Why an environment variable?

Storing API keys in environment variables keeps them out of your source code. PydanticAI's Google provider automatically looks for GOOGLE_API_KEY when you use the google: model prefix.

Now set up Pydantic Logfire for observability. Logfire gives you real-time traces of every agent decision, tool call, and delegation.

  • Open logfire.pydantic.dev in your browser and create a free account.
  • Authenticate the Logfire CLI by running:
logfire auth

What happens when you run this?

The CLI will prompt you to select a data region (US or EU), then open your browser for login. Once you complete the browser login, return to your terminal and it will confirm you are authenticated.

  • Create a new Logfire project for this work by running:
logfire projects new multi-agent-research

The CLI saves your project credentials locally so Logfire knows where to send trace data.

Logfire auth not working?

If your browser does not open automatically, look for a URL printed in the terminal and paste it into your browser manually.

If you see a "command not found" error, the logfire CLI is included in the package you installed. Make sure your virtual environment is still activated.

Help me fix my Logfire authentication issue.

  • Now verify the entire setup works by creating a test file. Open Cursor by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Cursor, and pressing Enter.
  • In Cursor, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop > multi-agent-research and open it.
  • Create a new file called test_setup.py by right-clicking in the Explorer sidebar and selecting New File.
  • Paste the following code into test_setup.py:
"""Quick test to verify PydanticAI + Gemini + Logfire all work together."""

import logfire
from pydantic_ai import Agent

# Initialize Logfire observability
logfire.configure()
logfire.instrument_pydantic_ai()

# Create a minimal agent using Gemini
agent = Agent(
    "google:gemini-2.5-flash",
    instructions="You are a helpful assistant. Keep responses brief.",
)

# Run a test query
result = agent.run_sync("What is PydanticAI in one sentence?")
print(f"Agent response: {result.output}")
print("\nSetup verified! Check your Logfire dashboard for the trace.")
print("View traces at: https://logfire.pydantic.dev")

What does this code do?

This script initializes Logfire with logfire.configure() and enables PydanticAI tracing with logfire.instrument_pydantic_ai(). These two lines are all you need for full observability.

The agent uses the google:gemini-2.5-flash model string, which tells PydanticAI to connect via the Google provider using your GOOGLE_API_KEY. The run_sync method sends a prompt and waits for the response synchronously.

  • Save test_setup.py.
  • Open the integrated terminal in Cursor with Ctrl+` (backtick key).
  • Make sure your virtual environment is active (you should see (venv) in your prompt), then run:
python3 test_setup.py

You should see a one-sentence response from Gemini printed in your terminal, followed by a confirmation message.

  • Open logfire.pydantic.dev in your browser. You should see a trace showing the agent run, including the model request and response.

Not seeing a response?

If you see an authentication error, double-check that GOOGLE_API_KEY is set in your current terminal session. Run echo $GOOGLE_API_KEY to verify it is not empty.

If you see an import error for pydantic_ai, confirm your virtual environment is activated. You should see (venv) in your prompt.

Help me debug my test_setup.py script error.

Your environment is ready. Next up, you will define the structured output schemas that let your agents communicate with type safety, and build the first specialist agents with real tool calling.

Set Up Your Environment

Your Python environment is ready, your Gemini API key is configured, and Logfire is connected. Now it's time to scaffold the project structure that your multi-agent system will live in.

A well-organized multi-agent system separates concerns into distinct modules. You'll create four Python files, each owning a specific layer of the architecture.

In this step, get ready to:
  • Create the project file structure with four Python modules.
  • Verify all dependencies are importable and ready to use.
Create your project files

Your multi-agent system needs four files, each with a distinct responsibility:

  • models.py defines the Pydantic schemas that agents use to communicate with each other in a type-safe way.
  • tools.py holds the tool functions and guardrails that agents call during execution.
  • agents.py defines the specialist agents (web researcher, data analyst, critic) with their tool bindings.
  • main.py wires everything together with the orchestrator, async execution, and the reflexion loop.
  • Create all four files by running this command in your Cursor terminal:

macOS

touch models.py tools.py agents.py main.py

Windows

New-Item -ItemType "File" -Path "models.py", "tools.py", "agents.py", "main.py"
  • Confirm the files exist by running:
ls

You should see models.py, tools.py, agents.py, and main.py listed in the output.

Why four separate files?

Splitting responsibilities across files mirrors how production agent systems are built. Each file maps to one architectural layer: data contracts, tooling, agent definitions, and orchestration.

This separation makes it easy to test, swap, or scale individual parts without touching the rest of the system.

Files not appearing in Cursor's sidebar?

Make sure you ran the command in the integrated terminal inside your project folder. If the files still don't appear, click the refresh icon in Cursor's Explorer panel.

help me debug why my files aren't showing in Cursor

Verify your dependencies

Before writing any agent code, confirm that all three key packages are importable in your project's Python environment.

  • Run this import check in your terminal:
python -c "import pydantic_ai; import logfire; import httpx; print('All imports successful')"

What does this command do?

This one-liner tries to import the three packages your agents depend on: PydanticAI for agent orchestration, logfire for observability, and httpx for async HTTP requests inside tools.

If any import fails, Python raises an error immediately. Catching this now saves you from confusing tracebacks later when agents try to run.

✔️ I see 'All imports successful'

All three dependencies are available. Your environment is fully ready for agent development.

ⓧ I see a ModuleNotFoundError

The missing module name in the error tells you which package to reinstall.

  • Reinstall the full PydanticAI package with Google and Logfire extras by running:
pip install "pydantic-ai-slim[google,logfire]==2.0.0"

Then re-run the import check above.

help me fix my PydanticAI import error

Your project structure is in place and all dependencies are confirmed. Next up, you'll define the Pydantic models that let your agents communicate with type-safe, validated data.

Define Structured Output Schemas and Specialist Agents

Your environment is wired up and ready to go. Now it's time to give your multi-agent system a shared language.

In a multi-agent system, agents pass data to each other constantly. Without a shared contract defining what that data looks like, agents would misunderstand each other or crash with unexpected formats. Pydantic models solve this by defining type-safe schemas that every agent must conform to.

In this step, get ready to:
  • Define Pydantic models that act as communication contracts between agents.
  • Build tool functions that give your agents real capabilities like fetching web content.
  • Create specialist agents with typed structured outputs and tool calling.
Define the inter-agent communication schemas

Every agent in your system produces structured data that another agent consumes. These Pydantic models define exactly what shape that data takes.

  • Switch back to Cursor and open models.py.
  • Add the following code to define the input query model and the web researcher's output schema:
"""Structured output schemas for type-safe inter-agent communication."""

from pydantic import BaseModel, Field


class ResearchQuery(BaseModel):
    """Validated input query with routing metadata."""

    topic: str = Field(description="The research topic or company name")
    depth: str = Field(
        default="standard",
        description="Research depth: 'quick', 'standard', or 'deep'",
    )
    focus_areas: list[str] = Field(
        default_factory=list,
        description="Specific areas to focus research on",
    )


class WebResearchResult(BaseModel):
    """Structured output from the web researcher agent."""

    sources: list[str] = Field(description="URLs or sources consulted")
    key_findings: list[str] = Field(description="Main findings from research")
    raw_content: str = Field(description="Summarized raw content from sources")
    confidence: float = Field(
        ge=0.0, le=1.0, description="Confidence score of findings"
    )

What do these models do?

Each model defines a strict contract. ResearchQuery validates incoming topics with metadata about desired depth and focus areas.

WebResearchResult forces the web researcher agent to always return sources, findings, raw content, and a confidence score. The Field constraints like ge=0.0, le=1.0 enforce value ranges at runtime.

  • Add the remaining output schemas below your existing code in models.py:

class AnalysisResult(BaseModel):
    """Structured output from the data analyst agent."""

    summary: str = Field(description="Executive summary of analysis")
    insights: list[str] = Field(description="Key analytical insights")
    metrics: dict[str, str] = Field(
        default_factory=dict, description="Quantitative metrics extracted"
    )
    recommendations: list[str] = Field(description="Actionable recommendations")


class FinalReport(BaseModel):
    """Aggregated final report combining all specialist outputs."""

    topic: str = Field(description="Original research topic")
    executive_summary: str = Field(description="High-level summary")
    research_findings: list[str] = Field(description="Key findings from research")
    analysis_insights: list[str] = Field(description="Analytical insights")
    recommendations: list[str] = Field(description="Final recommendations")
    confidence_score: float = Field(
        ge=0.0, le=1.0, description="Overall confidence"
    )
    sources: list[str] = Field(description="All sources used")


class CritiqueResult(BaseModel):
    """Output from the critic agent evaluating report quality."""

    score: float = Field(
        ge=0.0, le=10.0, description="Quality score out of 10"
    )
    passed: bool = Field(description="Whether the report meets quality threshold")
    feedback: str = Field(description="Specific feedback for improvement")
    weak_areas: list[str] = Field(description="Areas that need improvement")

What do these three models represent?

AnalysisResult is what the data analyst agent returns. It includes a summary, insights, extracted metrics as key-value pairs, and recommendations.

FinalReport is the orchestrator's aggregated output, combining findings from all specialists into one unified structure.

CritiqueResult is what the critic agent produces when evaluating report quality. The passed field drives the reflexion loop: if False, the system re-generates.

  • Save models.py.

Seeing import errors?

Make sure you installed the dependencies in Step 1. If pydantic is not found, run pip install "pydantic-ai-slim[google,logfire]==2.0.0" again to ensure Pydantic is installed.

help me debug a Pydantic import error in my models.py file

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

"""Structured output schemas for type-safe inter-agent communication."""

from pydantic import BaseModel, Field


class ResearchQuery(BaseModel):
    """Validated input query with routing metadata."""

    topic: str = Field(description="The research topic or company name")
    depth: str = Field(
        default="standard",
        description="Research depth: 'quick', 'standard', or 'deep'",
    )
    focus_areas: list[str] = Field(
        default_factory=list,
        description="Specific areas to focus research on",
    )


class WebResearchResult(BaseModel):
    """Structured output from the web researcher agent."""

    sources: list[str] = Field(description="URLs or sources consulted")
    key_findings: list[str] = Field(description="Main findings from research")
    raw_content: str = Field(description="Summarized raw content from sources")
    confidence: float = Field(
        ge=0.0, le=1.0, description="Confidence score of findings"
    )


class AnalysisResult(BaseModel):
    """Structured output from the data analyst agent."""

    summary: str = Field(description="Executive summary of analysis")
    insights: list[str] = Field(description="Key analytical insights")
    metrics: dict[str, str] = Field(
        default_factory=dict, description="Quantitative metrics extracted"
    )
    recommendations: list[str] = Field(description="Actionable recommendations")


class FinalReport(BaseModel):
    """Aggregated final report combining all specialist outputs."""

    topic: str = Field(description="Original research topic")
    executive_summary: str = Field(description="High-level summary")
    research_findings: list[str] = Field(description="Key findings from research")
    analysis_insights: list[str] = Field(description="Analytical insights")
    recommendations: list[str] = Field(description="Final recommendations")
    confidence_score: float = Field(
        ge=0.0, le=1.0, description="Overall confidence"
    )
    sources: list[str] = Field(description="All sources used")


class CritiqueResult(BaseModel):
    """Output from the critic agent evaluating report quality."""

    score: float = Field(
        ge=0.0, le=10.0, description="Quality score out of 10"
    )
    passed: bool = Field(description="Whether the report meets quality threshold")
    feedback: str = Field(description="Specific feedback for improvement")
    weak_areas: list[str] = Field(description="Areas that need improvement")
Build the tool functions

Agents on their own can only generate text. Tools give them the ability to interact with the outside world. Your tools will let agents fetch web content and validate inputs and outputs.

  • Open tools.py in Cursor.
  • Add the following code to create a utility function for fetching web content:
"""Tool implementations for specialist agents."""

import httpx


async def fetch_url_content(url: str) -> str:
    """Fetch and return text content from a URL.

    Args:
        url: The URL to fetch content from.

    Returns:
        The text content of the page, truncated to 5000 chars.
    """
    async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response.text[:5000]
        except httpx.HTTPError as e:
            return f"Error fetching {url}: {str(e)}"

What does this function do?

fetch_url_content creates an httpx async HTTP client with a 10-second timeout and fetches a page. It truncates the response to 5000 characters to keep LLM context windows manageable.

The follow_redirects=True setting handles URL redirects automatically. If the request fails, it returns a descriptive error string instead of crashing.

  • Add the guardrail validation functions below your existing code in tools.py:

def validate_query(topic: str) -> tuple[bool, str]:
    """Input guardrail: validate the research query.

    Args:
        topic: The research topic to validate.

    Returns:
        Tuple of (is_valid, error_message).
    """
    if len(topic.strip()) < 3:
        return False, "Query too short. Please provide a meaningful research topic."
    if len(topic) > 500:
        return False, "Query too long. Please keep it under 500 characters."
    blocked_terms = ["ignore previous", "system prompt", "jailbreak"]
    for term in blocked_terms:
        if term.lower() in topic.lower():
            return False, f"Query contains blocked content: '{term}'"
    return True, ""


def validate_report(report: "FinalReport") -> tuple[bool, str]:
    """Output guardrail: validate the final report quality.

    Args:
        report: The FinalReport to validate.

    Returns:
        Tuple of (is_valid, error_message).
    """
    from models import FinalReport

    if len(report.research_findings) < 2:
        return False, "Report has insufficient research findings (minimum 2)."
    if len(report.executive_summary) < 50:
        return False, "Executive summary is too brief (minimum 50 characters)."
    if report.confidence_score < 0.3:
        return False, "Confidence score too low. Research may be unreliable."
    return True, ""

What are guardrails?

Guardrails are validation functions that run before or after an LLM processes data. validate_query is an input guardrail. It rejects queries that are too short, too long, or contain prompt injection attempts.

validate_report is an output guardrail. It checks that the final report has enough findings, a sufficiently detailed summary, and a reasonable confidence score. You'll wire these up in the orchestrator later.

  • Save tools.py.

Seeing a type hint error for FinalReport?

The "FinalReport" string annotation in validate_report is intentional. It avoids a circular import since tools.py does not import from models.py at the top level. The actual import happens inside the function body.

help me understand the string type annotation pattern in Python

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

"""Tool implementations for specialist agents."""

import httpx


async def fetch_url_content(url: str) -> str:
    """Fetch and return text content from a URL.

    Args:
        url: The URL to fetch content from.

    Returns:
        The text content of the page, truncated to 5000 chars.
    """
    async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response.text[:5000]
        except httpx.HTTPError as e:
            return f"Error fetching {url}: {str(e)}"


def validate_query(topic: str) -> tuple[bool, str]:
    """Input guardrail: validate the research query.

    Args:
        topic: The research topic to validate.

    Returns:
        Tuple of (is_valid, error_message).
    """
    if len(topic.strip()) < 3:
        return False, "Query too short. Please provide a meaningful research topic."
    if len(topic) > 500:
        return False, "Query too long. Please keep it under 500 characters."
    blocked_terms = ["ignore previous", "system prompt", "jailbreak"]
    for term in blocked_terms:
        if term.lower() in topic.lower():
            return False, f"Query contains blocked content: '{term}'"
    return True, ""


def validate_report(report: "FinalReport") -> tuple[bool, str]:
    """Output guardrail: validate the final report quality.

    Args:
        report: The FinalReport to validate.

    Returns:
        Tuple of (is_valid, error_message).
    """
    from models import FinalReport

    if len(report.research_findings) < 2:
        return False, "Report has insufficient research findings (minimum 2)."
    if len(report.executive_summary) < 50:
        return False, "Executive summary is too brief (minimum 50 characters)."
    if report.confidence_score < 0.3:
        return False, "Confidence score too low. Research may be unreliable."
    return True, ""
Build specialist agents with tool calling

Now you'll create the agents themselves. Each agent has a specific role, a model it talks to, tools it can call, and a structured output type it must return.

  • Open agents.py in Cursor.
  • Add the following code to define the shared dependencies and the web researcher agent with its tools:
"""Specialist agent definitions with tool calling and structured outputs."""

from dataclasses import dataclass

import httpx
from pydantic_ai import Agent, RunContext

from models import AnalysisResult, CritiqueResult, FinalReport, WebResearchResult


@dataclass
class ResearchDeps:
    """Shared dependencies for all agents."""

    http_client: httpx.AsyncClient
    max_retries: int = 3


# --- Web Researcher Agent ---
web_researcher = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=WebResearchResult,
    instructions=(
        "You are a web research specialist. Your job is to gather information "
        "about the given topic by searching the web and extracting key findings. "
        "Always cite your sources and rate your confidence in the findings."
    ),
)


@web_researcher.tool
async def search_web(ctx: RunContext[ResearchDeps], query: str) -> str:
    """Search the web for information on a topic.

    Args:
        ctx: The run context with dependencies.
        query: The search query to look up.

    Returns:
        Text content from the search results.
    """
    url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={query}&format=json&srlimit=3"
    try:
        response = await ctx.deps.http_client.get(url)
        response.raise_for_status()
        data = response.json()
        results = data.get("query", {}).get("search", [])
        if not results:
            return "No results found."
        summaries = []
        for r in results:
            snippet = r.get("snippet", "").replace("<span class=\"searchmatch\">", "").replace("</span>", "")
            summaries.append(f"- {r['title']}: {snippet}")
        return "\n".join(summaries)
    except httpx.HTTPError as e:
        return f"Search error: {str(e)}"


@web_researcher.tool
async def fetch_page(ctx: RunContext[ResearchDeps], url: str) -> str:
    """Fetch content from a specific URL.

    Args:
        ctx: The run context with dependencies.
        url: The URL to fetch.

    Returns:
        The page content, truncated to 3000 characters.
    """
    try:
        response = await ctx.deps.http_client.get(url)
        response.raise_for_status()
        return response.text[:3000]
    except httpx.HTTPError as e:
        return f"Fetch error: {str(e)}"

What's happening here?

ResearchDeps is a dependency container that all agents share. It holds an httpx.AsyncClient so agents reuse one HTTP connection pool instead of creating new ones per request.

The web_researcher agent is created with Agent() specifying the model (google:gemini-2.5-flash), the dependency type, the expected output schema, and system instructions.

The @web_researcher.tool decorator registers tool functions. PydanticAI passes the tool's docstring to the LLM as the tool description, so the model knows when and how to call each tool. The RunContext parameter gives tools access to shared dependencies.

Why use @agent.tool instead of @agent.tool_plain?

Tools decorated with @agent.tool receive a RunContext as their first argument, giving them access to dependencies like the shared HTTP client. Tools decorated with @agent.tool_plain do not receive context and operate independently. Use @agent.tool when your tool needs shared resources.

  • Add the data analyst agent and critic agent below your existing code in agents.py:

# --- Data Analyst Agent ---
data_analyst = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=AnalysisResult,
    instructions=(
        "You are a data analyst specialist. Your job is to analyze raw research "
        "findings and extract structured insights, metrics, and actionable "
        "recommendations. Be quantitative where possible and always provide "
        "clear, evidence-based conclusions."
    ),
)


@data_analyst.tool_plain
def extract_metrics(text: str) -> str:
    """Extract numerical metrics and statistics from text.

    Args:
        text: The text to analyze for metrics.

    Returns:
        Extracted metrics as a formatted string.
    """
    import re

    patterns = [
        r"\d+\.?\d*%",
        r"\$\d+[\d,.]*[BMK]?",
        r"\d+[\d,]*\s*(users|customers|employees|revenue|growth)",
    ]
    found = []
    for pattern in patterns:
        matches = re.findall(pattern, text, re.IGNORECASE)
        found.extend(matches)
    if found:
        return "Extracted metrics: " + "; ".join(found[:10])
    return "No specific metrics found in the provided text."


# --- Critic Agent ---
critic_agent = Agent(
    "google:gemini-2.5-flash",
    output_type=CritiqueResult,
    instructions=(
        "You are a research quality critic. Evaluate the provided report on: "
        "1) Completeness - does it cover the topic thoroughly? "
        "2) Accuracy - are claims supported by sources? "
        "3) Coherence - is it well-structured and logical? "
        "4) Actionability - are recommendations practical? "
        "Score from 0-10. Pass threshold is 7.0. "
        "Be specific in your feedback about what needs improvement."
    ),
)

What makes these agents different?

The data_analyst uses @data_analyst.tool_plain because extract_metrics is a pure function that uses regex to pull numbers from text. It does not need the shared HTTP client, so it skips RunContext.

The critic_agent has no tools at all. It relies purely on the LLM's reasoning ability to evaluate report quality against the criteria in its instructions. Notice it also has no deps_type since it does not need external resources.

  • Save agents.py.
  • Verify the imports work by running this in your terminal:
python -c "from agents import web_researcher, data_analyst, critic_agent; print('All agents loaded successfully')"

You should see All agents loaded successfully printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure you are running the command from inside your project folder where models.py, tools.py, and agents.py all live in the same directory.

If you see an error about pydantic_ai, confirm your virtual environment is activated.

help me fix a ModuleNotFoundError when importing my agents

✔️ Awesome, I've got everything!

Great. Double check you've saved all three files before moving on.

ⓧ I'd like to double check the full code

"""Specialist agent definitions with tool calling and structured outputs."""

from dataclasses import dataclass

import httpx
from pydantic_ai import Agent, RunContext

from models import AnalysisResult, CritiqueResult, FinalReport, WebResearchResult


@dataclass
class ResearchDeps:
    """Shared dependencies for all agents."""

    http_client: httpx.AsyncClient
    max_retries: int = 3


# --- Web Researcher Agent ---
web_researcher = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=WebResearchResult,
    instructions=(
        "You are a web research specialist. Your job is to gather information "
        "about the given topic by searching the web and extracting key findings. "
        "Always cite your sources and rate your confidence in the findings."
    ),
)


@web_researcher.tool
async def search_web(ctx: RunContext[ResearchDeps], query: str) -> str:
    """Search the web for information on a topic.

    Args:
        ctx: The run context with dependencies.
        query: The search query to look up.

    Returns:
        Text content from the search results.
    """
    url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={query}&format=json&srlimit=3"
    try:
        response = await ctx.deps.http_client.get(url)
        response.raise_for_status()
        data = response.json()
        results = data.get("query", {}).get("search", [])
        if not results:
            return "No results found."
        summaries = []
        for r in results:
            snippet = r.get("snippet", "").replace("<span class=\"searchmatch\">", "").replace("</span>", "")
            summaries.append(f"- {r['title']}: {snippet}")
        return "\n".join(summaries)
    except httpx.HTTPError as e:
        return f"Search error: {str(e)}"


@web_researcher.tool
async def fetch_page(ctx: RunContext[ResearchDeps], url: str) -> str:
    """Fetch content from a specific URL.

    Args:
        ctx: The run context with dependencies.
        url: The URL to fetch.

    Returns:
        The page content, truncated to 3000 characters.
    """
    try:
        response = await ctx.deps.http_client.get(url)
        response.raise_for_status()
        return response.text[:3000]
    except httpx.HTTPError as e:
        return f"Fetch error: {str(e)}"


# --- Data Analyst Agent ---
data_analyst = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=AnalysisResult,
    instructions=(
        "You are a data analyst specialist. Your job is to analyze raw research "
        "findings and extract structured insights, metrics, and actionable "
        "recommendations. Be quantitative where possible and always provide "
        "clear, evidence-based conclusions."
    ),
)


@data_analyst.tool_plain
def extract_metrics(text: str) -> str:
    """Extract numerical metrics and statistics from text.

    Args:
        text: The text to analyze for metrics.

    Returns:
        Extracted metrics as a formatted string.
    """
    import re

    patterns = [
        r"\d+\.?\d*%",
        r"\$\d+[\d,.]*[BMK]?",
        r"\d+[\d,]*\s*(users|customers|employees|revenue|growth)",
    ]
    found = []
    for pattern in patterns:
        matches = re.findall(pattern, text, re.IGNORECASE)
        found.extend(matches)
    if found:
        return "Extracted metrics: " + "; ".join(found[:10])
    return "No specific metrics found in the provided text."


# --- Critic Agent ---
critic_agent = Agent(
    "google:gemini-2.5-flash",
    output_type=CritiqueResult,
    instructions=(
        "You are a research quality critic. Evaluate the provided report on: "
        "1) Completeness - does it cover the topic thoroughly? "
        "2) Accuracy - are claims supported by sources? "
        "3) Coherence - is it well-structured and logical? "
        "4) Actionability - are recommendations practical? "
        "Score from 0-10. Pass threshold is 7.0. "
        "Be specific in your feedback about what needs improvement."
    ),
)

Your schemas, tools, and specialist agents are all defined. Next up, you'll build the orchestrator that decides which specialists to call and enforces guardrails on every request.

Build the Orchestrator with Guardrails

Your specialist agents can research and analyze independently. But in a production system, you need a central coordinator that decides what to do with an incoming query before handing it off.

In this step, you will build the orchestrator agent and wrap it in guardrails. The input guardrail blocks bad queries before they ever reach the LLM. The output guardrail validates the quality of the final report after generation.

In this step, get ready to:
  • Set up observability instrumentation in a new main.py file.
  • Define the orchestrator agent that synthesizes specialist outputs into a final report.
  • Implement input and output guardrails that protect the research pipeline.
Create main.py with observability instrumentation

The orchestrator lives in a new file called main.py. Before defining any agents, you need to wire up Logfire so every agent call, tool execution, and HTTP request gets traced automatically.

  • Create a new file called main.py in your project folder.
  • Add the following imports and observability setup:
"""Multi-agent research system with orchestrator, guardrails, and reflexion loop."""

import asyncio
import time

import httpx
import logfire
from pydantic_ai import Agent, RunContext

from agents import (
    ResearchDeps,
    critic_agent,
    data_analyst,
    web_researcher,
)
from models import CritiqueResult, FinalReport, ResearchQuery
from tools import validate_query, validate_report

# --- Observability Setup ---
logfire.configure()
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

What does this code do?

  • logfire.configure() initializes the Logfire SDK, reading your authentication token from the .logfire directory you set up in Step 1.
  • logfire.instrument_pydantic_ai() instruments all PydanticAI agents globally. Every agent run generates a trace with spans for model calls and tool functions.
  • logfire.instrument_httpx() instruments all HTTP requests made by the Google provider SDK. You will see every API call to Gemini in your Logfire dashboard.
  • Save main.py.

Seeing import errors?

Make sure you are running from the same folder that contains agents.py, models.py, and tools.py. All local imports resolve relative to the working directory.

help me debug import errors in my PydanticAI multi-agent project

Define the orchestrator agent

The orchestrator is the brain of the system. It does not research or analyze directly. Instead, it takes the outputs from the specialist agents and synthesizes them into a single FinalReport.

  • Add the orchestrator agent definition below the observability setup in main.py:
# --- Orchestrator Agent ---
orchestrator = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=FinalReport,
    instructions=(
        "You are a research orchestrator. Your job is to synthesize findings "
        "from the web researcher and data analyst into a comprehensive final "
        "report. Combine their outputs into a coherent narrative with clear "
        "executive summary, findings, insights, and recommendations."
    ),
)

What does this code do?

The orchestrator uses the same google:gemini-2.5-flash model as the specialists. Its output_type=FinalReport forces Gemini to return a structured response matching the FinalReport Pydantic model you defined earlier.

The deps_type=ResearchDeps gives the orchestrator access to the shared HTTP client and configuration. This is how dependencies flow through the multi-agent system.

  • Save main.py.

Why an orchestrator instead of chaining agents directly?

You could call each specialist in sequence and manually stitch their outputs together. But an orchestrator agent can reason about the combined results, fill gaps, resolve contradictions, and produce a coherent narrative. It mirrors how production systems at companies like Lyft use a supervisor pattern to coordinate multiple agents.

Implement the guardrailed research function

Guardrails are validation checks that run before and after the LLM processes a request. The input guardrail rejects bad queries (too short, too long, or containing prompt injection attempts) before they cost you API tokens. The output guardrail checks the generated report for quality issues.

  • Add the research() function below the orchestrator definition in main.py:
async def research(topic: str) -> FinalReport | str:
    """Main research pipeline with guardrails, routing, and reflexion."""
    # --- Input Guardrail ---
    is_valid, error_msg = validate_query(topic)
    if not is_valid:
        logfire.warn("Input guardrail rejected query: {error}", error=error_msg)
        return f"Input rejected: {error_msg}"

    query = ResearchQuery(topic=topic)
    logfire.info("Starting research pipeline for: {topic}", topic=query.topic)

What does this code do?

  • validate_query(topic) is the input guardrail you wrote in tools.py. It checks length constraints and blocks prompt injection attempts like "ignore previous instructions".
  • If validation fails, the function returns an error string immediately. No LLM call happens, saving you tokens and latency.
  • logfire.warn() logs the rejection to your Logfire dashboard so you can monitor how often guardrails trigger in production.
  • ResearchQuery(topic=topic) wraps the raw string in a validated Pydantic model, adding structure and metadata for downstream agents.

What about the output guardrail?

You imported validate_report at the top of the file. This function checks that the final report has enough findings, a sufficiently detailed summary, and a reasonable confidence score. You will wire it into the pipeline in the next step, after the orchestrator actually generates a report to validate.

  • Save main.py.

Seeing a syntax error on the union type hint?

The FinalReport | str syntax requires Python 3.10 or higher. If you see an error, confirm your Python version by running python3 --version in your terminal.

help me fix a Python version error with union type hints in my PydanticAI project

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py with all three sections: imports and observability, orchestrator agent, and the research() function with the input guardrail.

ⓧ I'd like to double check the full code

"""Multi-agent research system with orchestrator, guardrails, and reflexion loop."""

import asyncio
import time

import httpx
import logfire
from pydantic_ai import Agent, RunContext

from agents import (
    ResearchDeps,
    critic_agent,
    data_analyst,
    web_researcher,
)
from models import CritiqueResult, FinalReport, ResearchQuery
from tools import validate_query, validate_report

# --- Observability Setup ---
logfire.configure()
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

# --- Orchestrator Agent ---
orchestrator = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=FinalReport,
    instructions=(
        "You are a research orchestrator. Your job is to synthesize findings "
        "from the web researcher and data analyst into a comprehensive final "
        "report. Combine their outputs into a coherent narrative with clear "
        "executive summary, findings, insights, and recommendations."
    ),
)


async def research(topic: str) -> FinalReport | str:
    """Main research pipeline with guardrails, routing, and reflexion."""
    # --- Input Guardrail ---
    is_valid, error_msg = validate_query(topic)
    if not is_valid:
        logfire.warn("Input guardrail rejected query: {error}", error=error_msg)
        return f"Input rejected: {error_msg}"

    query = ResearchQuery(topic=topic)
    logfire.info("Starting research pipeline for: {topic}", topic=query.topic)

Your orchestrator is defined and your input guardrail is in place. Next up, you will wire the orchestrator to delegate work to the specialist agents and run them concurrently for maximum speed.

Wire Up Parallel Specialist Execution

Your orchestrator knows what to do, and your guardrails keep bad data out. Now it's time to connect the actual delegation pipeline so the orchestrator dispatches work to your specialist agents and synthesizes their results.

In production multi-agent systems, agents rarely run in isolation. The orchestrator needs to hand off tasks, wait for results, and combine outputs into a single coherent response. This step wires up that full lifecycle in main.py.

In this step, get ready to:
  • Build a delegation function that dispatches work to specialist agents.
  • Wire the full research pipeline from input through synthesis to output.
  • Create the entry point and run your multi-agent system end-to-end.
Build the specialist delegation function

The orchestrator needs a way to dispatch research tasks to the right specialist agents and collect their structured outputs. Because the data analyst needs the researcher's findings as context, the pipeline runs them sequentially: researcher first, then analyst with the raw content as input.

  • In main.py, find the closing parenthesis of the orchestrator Agent definition. Add the following function below it, before the existing research() function:
async def run_specialists_parallel(
    topic: str, deps: ResearchDeps
) -> tuple[str, str]:
    """Run specialist agents concurrently using asyncio.gather."""
    logfire.info("Dispatching specialists in parallel for: {topic}", topic=topic)

    async def run_researcher():
        result = await web_researcher.run(
            f"Research the following topic thoroughly: {topic}",
            deps=deps,
        )
        return result.output

    async def run_analyst(research_context: str):
        result = await data_analyst.run(
            f"Analyze these research findings and extract insights:\n{research_context}",
            deps=deps,
        )
        return result.output

    # Run researcher first to get context, then analyst in parallel with synthesis
    research_result = await run_researcher()
    analysis_result = await run_analyst(research_result.raw_content)

    return research_result, analysis_result

What does this code do?

This function coordinates the two specialist agents in a data-dependent chain.

  • run_researcher() invokes the web_researcher agent, passing the shared deps (which includes the HTTP client). It returns the structured WebResearchResult.
  • run_analyst() receives the researcher's raw content as input context, so the analyst can extract metrics and insights from real data rather than generating from scratch.
  • Both agents share the same ResearchDeps instance, propagating the HTTP client and retry config across all delegated runs.

Why not asyncio.gather here?

The analyst depends on the researcher's output. You can't analyze data that hasn't been gathered yet. In production systems, you'd use asyncio.gather when specialists are truly independent (e.g., researching different sub-topics simultaneously).

The architecture still supports parallel execution. If you add a third specialist that doesn't depend on the researcher, you can gather it alongside the analyst.

  • Save main.py.

Seeing a NameError for web_researcher or data_analyst?

  • Check that your imports at the top of main.py include both web_researcher and data_analyst from agents. These were added in the previous step.
  • Make sure the function is placed above the research() function, since research() will call it.

Help me debug a NameError in my multi-agent delegation function.

Complete the research pipeline with orchestrator synthesis

The research() function currently validates input and stops. Now you'll wire in the full lifecycle: create the HTTP client, dispatch specialists, feed their outputs to the orchestrator for synthesis, and validate the final report.

  • In main.py, find the last line of the research() function. It currently ends with:
    logfire.info("Starting research pipeline for: {topic}", topic=query.topic)
  • Add the following code directly below that line to complete the function:
    start_time = time.time()

    async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
        deps = ResearchDeps(http_client=client)

        # --- Parallel Specialist Execution ---
        research_result, analysis_result = await run_specialists_parallel(
            query.topic, deps
        )

        # --- Orchestrator Synthesizes Results ---
        synthesis_prompt = (
            f"Synthesize a comprehensive research report on: {query.topic}\n\n"
            f"WEB RESEARCH FINDINGS:\n"
            f"Sources: {research_result.sources}\n"
            f"Key Findings: {research_result.key_findings}\n"
            f"Content: {research_result.raw_content}\n"
            f"Confidence: {research_result.confidence}\n\n"
            f"DATA ANALYSIS:\n"
            f"Summary: {analysis_result.summary}\n"
            f"Insights: {analysis_result.insights}\n"
            f"Metrics: {analysis_result.metrics}\n"
            f"Recommendations: {analysis_result.recommendations}\n\n"
            f"Create a final report combining all findings."
        )

        result = await orchestrator.run(synthesis_prompt, deps=deps)
        report = result.output

    # --- Output Guardrail ---
    is_valid, error_msg = validate_report(report)
    if not is_valid:
        logfire.warn("Output guardrail flagged report: {error}", error=error_msg)

    elapsed = time.time() - start_time
    logfire.info("Research pipeline complete in {elapsed:.1f}s", elapsed=elapsed)

    return report

What does this code do?

  • httpx.AsyncClient creates a shared HTTP client that all agents reuse through ResearchDeps. This avoids opening a new connection for every tool call.
  • The synthesis_prompt feeds the structured outputs from both specialists into the orchestrator. The orchestrator sees typed fields (sources, findings, metrics) rather than raw text, because your Pydantic models enforce that structure.
  • orchestrator.run() sends the combined context to Gemini and validates the response against your FinalReport schema automatically.
  • The output guardrail (validate_report) checks the final report for minimum quality: at least 2 findings, a substantive summary, and reasonable confidence.
  • time.time() captures the total pipeline duration so you can see how long the full agent coordination takes.
  • Save main.py.

Seeing indentation errors?

  • The entire block starting with start_time = time.time() must be indented one level inside the research() function body (4 spaces).
  • The code inside async with needs a second level of indentation (8 spaces). The lines after the async with block (is_valid, elapsed, return) drop back to 4 spaces.

Help me fix indentation in my research() function.

Write the entry point and run the pipeline

Your pipeline is complete but has no way to start. You need a main() function that accepts user input, calls the research pipeline, and formats the final report for the terminal.

  • At the bottom of main.py, after the research() function, add the entry point by pasting this code:
async def main():
    """Entry point: run the multi-agent research system."""
    print("=" * 60)
    print("MULTI-AGENT RESEARCH SYSTEM")
    print("Production Patterns: Orchestrator | Guardrails | Reflexion")
    print("=" * 60)

    topic = input("\nEnter research topic: ").strip()
    if not topic:
        topic = "PydanticAI framework for production AI agents"

    print(f"\nResearching: {topic}")
    print("-" * 40)

    result = await research(topic)

    if isinstance(result, str):
        print(f"\nError: {result}")
    else:
        print(f"\n{'=' * 60}")
        print(f"FINAL REPORT: {result.topic}")
        print(f"{'=' * 60}")
        print(f"\nExecutive Summary:\n{result.executive_summary}")
        print(f"\nKey Findings:")
        for i, finding in enumerate(result.research_findings, 1):
            print(f"  {i}. {finding}")
        print(f"\nAnalysis Insights:")
        for i, insight in enumerate(result.analysis_insights, 1):
            print(f"  {i}. {insight}")
        print(f"\nRecommendations:")
        for i, rec in enumerate(result.recommendations, 1):
            print(f"  {i}. {rec}")
        print(f"\nConfidence Score: {result.confidence_score:.0%}")
        print(f"Sources: {', '.join(result.sources)}")
        print(f"\nView full traces at: https://logfire.pydantic.dev")


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

What does this code do?

  • main() is an async function because it calls await research(topic). The asyncio.run(main()) at the bottom starts the async event loop.
  • If the user presses Enter without typing a topic, it defaults to researching "PydanticAI framework for production AI agents".
  • The isinstance(result, str) check handles the case where the input guardrail rejected the query and returned an error string instead of a FinalReport.
  • Save main.py.
  • Run your multi-agent system by executing this command in your terminal:
python main.py
  • When prompted, type a research topic (or press Enter for the default). You should see a formatted report with an executive summary, key findings, analysis insights, recommendations, and a confidence score.

What should I see?

The pipeline takes 15-45 seconds depending on Gemini's response time. You'll see the header banner, your topic echoed back, then the full report printed with numbered findings, insights, and recommendations. The final line points you to your Logfire dashboard where you can inspect every agent call.

Getting a timeout or connection error?

  • Make sure your GOOGLE_API_KEY environment variable is still set in your terminal session. If you opened a new terminal, you'll need to export it again.
  • The Wikipedia API calls in the web researcher tool require internet access. Check your connection if you see fetch errors in the output.
  • If the script hangs for more than 60 seconds, press Ctrl+C to cancel and try again. Gemini occasionally has high latency during peak times.

Help me debug a connection or timeout error when running my multi-agent system.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and that your pipeline ran successfully.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

"""Multi-agent research system with orchestrator, guardrails, and reflexion loop."""

import asyncio
import time

import httpx
import logfire
from pydantic_ai import Agent, RunContext

from agents import (
    ResearchDeps,
    critic_agent,
    data_analyst,
    web_researcher,
)
from models import CritiqueResult, FinalReport, ResearchQuery
from tools import validate_query, validate_report

# --- Observability Setup ---
logfire.configure()
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

# --- Orchestrator Agent ---
orchestrator = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=FinalReport,
    instructions=(
        "You are a research orchestrator. Your job is to synthesize findings "
        "from the web researcher and data analyst into a comprehensive final "
        "report. Combine their outputs into a coherent narrative with clear "
        "executive summary, findings, insights, and recommendations."
    ),
)


async def run_specialists_parallel(
    topic: str, deps: ResearchDeps
) -> tuple[str, str]:
    """Run specialist agents concurrently using asyncio.gather."""
    logfire.info("Dispatching specialists in parallel for: {topic}", topic=topic)

    async def run_researcher():
        result = await web_researcher.run(
            f"Research the following topic thoroughly: {topic}",
            deps=deps,
        )
        return result.output

    async def run_analyst(research_context: str):
        result = await data_analyst.run(
            f"Analyze these research findings and extract insights:\n{research_context}",
            deps=deps,
        )
        return result.output

    research_result = await run_researcher()
    analysis_result = await run_analyst(research_result.raw_content)

    return research_result, analysis_result


async def research(topic: str) -> FinalReport | str:
    """Main research pipeline with guardrails, routing, and reflexion."""
    # --- Input Guardrail ---
    is_valid, error_msg = validate_query(topic)
    if not is_valid:
        logfire.warn("Input guardrail rejected query: {error}", error=error_msg)
        return f"Input rejected: {error_msg}"

    query = ResearchQuery(topic=topic)
    logfire.info("Starting research pipeline for: {topic}", topic=query.topic)

    start_time = time.time()

    async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
        deps = ResearchDeps(http_client=client)

        # --- Parallel Specialist Execution ---
        research_result, analysis_result = await run_specialists_parallel(
            query.topic, deps
        )

        # --- Orchestrator Synthesizes Results ---
        synthesis_prompt = (
            f"Synthesize a comprehensive research report on: {query.topic}\n\n"
            f"WEB RESEARCH FINDINGS:\n"
            f"Sources: {research_result.sources}\n"
            f"Key Findings: {research_result.key_findings}\n"
            f"Content: {research_result.raw_content}\n"
            f"Confidence: {research_result.confidence}\n\n"
            f"DATA ANALYSIS:\n"
            f"Summary: {analysis_result.summary}\n"
            f"Insights: {analysis_result.insights}\n"
            f"Metrics: {analysis_result.metrics}\n"
            f"Recommendations: {analysis_result.recommendations}\n\n"
            f"Create a final report combining all findings."
        )

        result = await orchestrator.run(synthesis_prompt, deps=deps)
        report = result.output

    # --- Output Guardrail ---
    is_valid, error_msg = validate_report(report)
    if not is_valid:
        logfire.warn("Output guardrail flagged report: {error}", error=error_msg)

    elapsed = time.time() - start_time
    logfire.info("Research pipeline complete in {elapsed:.1f}s", elapsed=elapsed)

    return report


async def main():
    """Entry point: run the multi-agent research system."""
    print("=" * 60)
    print("MULTI-AGENT RESEARCH SYSTEM")
    print("Production Patterns: Orchestrator | Guardrails | Reflexion")
    print("=" * 60)

    topic = input("\nEnter research topic: ").strip()
    if not topic:
        topic = "PydanticAI framework for production AI agents"

    print(f"\nResearching: {topic}")
    print("-" * 40)

    result = await research(topic)

    if isinstance(result, str):
        print(f"\nError: {result}")
    else:
        print(f"\n{'=' * 60}")
        print(f"FINAL REPORT: {result.topic}")
        print(f"{'=' * 60}")
        print(f"\nExecutive Summary:\n{result.executive_summary}")
        print(f"\nKey Findings:")
        for i, finding in enumerate(result.research_findings, 1):
            print(f"  {i}. {finding}")
        print(f"\nAnalysis Insights:")
        for i, insight in enumerate(result.analysis_insights, 1):
            print(f"  {i}. {insight}")
        print(f"\nRecommendations:")
        for i, rec in enumerate(result.recommendations, 1):
            print(f"  {i}. {rec}")
        print(f"\nConfidence Score: {result.confidence_score:.0%}")
        print(f"Sources: {', '.join(result.sources)}")
        print(f"\nView full traces at: https://logfire.pydantic.dev")


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

Your multi-agent system is running end-to-end. The orchestrator dispatches, the specialists execute with real tool calling, and the output guardrail validates quality. Next up, you'll add a self-critique loop that evaluates the report and iteratively improves it.

Implement the Self-Critique Reflexion Loop

Your multi-agent system can now research a topic, analyze findings, and synthesize a final report. But how do you know the report is actually good?

In this step, you'll build a reflexion loop where a critic agent evaluates the report and, if it falls short, feeds specific improvement instructions back to the orchestrator for re-generation. This is the same self-improvement pattern used in production AI systems to iteratively raise output quality without human intervention.

In this step, get ready to:
  • Build the reflexion loop that uses the critic agent to score and improve reports.
  • Wire the loop into your research pipeline so every report goes through quality checks.
  • Run the complete system and observe improvement iterations in Logfire traces.
Build the reflexion loop function

A reflexion loop is a pattern where an AI agent evaluates its own output, identifies weaknesses, and iteratively improves until it meets a quality threshold. Your critic_agent (already defined in agents.py) scores reports on completeness, accuracy, coherence, and actionability. The loop passes if the score hits 7.0 or above.

What is a reflexion loop?

Think of it like a code review cycle. A developer writes code, a reviewer gives feedback, and the developer revises. The reflexion loop automates this: the critic provides structured feedback, and the orchestrator generates an improved version addressing every critique point.

Capping iterations prevents infinite loops where the critic is never satisfied. Two iterations is a good balance between quality improvement and API cost.

  • In main.py, add the following reflexion_loop() function directly below the run_specialists_parallel() function:
async def reflexion_loop(
    report: FinalReport, max_iterations: int = 2
) -> FinalReport:
    """Self-critique loop that iteratively improves the report."""
    current_report = report

    for iteration in range(max_iterations):
        logfire.info(
            "Reflexion loop iteration {iteration}/{max}",
            iteration=iteration + 1,
            max=max_iterations,
        )

        # Ask the critic agent to evaluate the current report
        critique = await critic_agent.run(
            f"Evaluate this research report:\n\n"
            f"Topic: {current_report.topic}\n"
            f"Summary: {current_report.executive_summary}\n"
            f"Findings: {current_report.research_findings}\n"
            f"Insights: {current_report.analysis_insights}\n"
            f"Recommendations: {current_report.recommendations}\n"
            f"Confidence: {current_report.confidence_score}"
        )

        critique_result: CritiqueResult = critique.output
        logfire.info(
            "Critique score: {score}, passed: {passed}",
            score=critique_result.score,
            passed=critique_result.passed,
        )

        # If the report passes quality threshold, stop iterating
        if critique_result.passed:
            logfire.info("Report passed quality check at iteration {i}", i=iteration + 1)
            break

        # Feed critic feedback back to the orchestrator for improvement
        improvement_prompt = (
            f"Improve this research report based on critic feedback.\n\n"
            f"Current report topic: {current_report.topic}\n"
            f"Current summary: {current_report.executive_summary}\n"
            f"Current findings: {current_report.research_findings}\n"
            f"Current insights: {current_report.analysis_insights}\n\n"
            f"CRITIC FEEDBACK (score {critique_result.score}/10):\n"
            f"{critique_result.feedback}\n"
            f"Weak areas: {critique_result.weak_areas}\n\n"
            f"Generate an improved version addressing ALL feedback points."
        )

        improved = await orchestrator.run(improvement_prompt)
        current_report = improved.output

    return current_report

What does this code do?

  • The function takes a FinalReport and loops up to max_iterations times (default 2), preventing infinite improvement cycles.
  • Each iteration sends the report to critic_agent which returns a CritiqueResult with a score, pass/fail boolean, feedback text, and weak areas.
  • If critique_result.passed is True (score >= 7.0), the loop exits early. Otherwise, it constructs an improvement_prompt containing the critique feedback and asks the orchestrator to generate an improved version.
  • Every iteration is logged to Logfire so you can trace exactly how scores improve across cycles.
  • Save main.py.

Seeing a NameError for critic_agent or CritiqueResult?

Make sure your imports at the top of main.py include critic_agent from agents and CritiqueResult from models. These were added in an earlier step.

Help me debug a NameError in my reflexion loop

Wire the reflexion loop into the research pipeline

The reflexion_loop() function exists but your research() pipeline doesn't call it yet. You need to pass the report through the loop after the output guardrail check.

  • In main.py, scroll down to the research() function.
  • Find these lines near the end of the function:
    # --- Output Guardrail ---
    is_valid, error_msg = validate_report(report)
    if not is_valid:
        logfire.warn("Output guardrail flagged report: {error}", error=error_msg)

    elapsed = time.time() - start_time
    logfire.info("Research pipeline complete in {elapsed:.1f}s", elapsed=elapsed)

    return report
  • Replace everything from the output guardrail comment to the end of the function with this updated version:
    is_valid, error_msg = validate_report(report)
    if not is_valid:
        logfire.warn("Output guardrail flagged report: {error}", error=error_msg)

    # --- Reflexion Loop: self-critique and iterative improvement ---
    final_report = await reflexion_loop(report)

    elapsed = time.time() - start_time
    logfire.info(
        "Research pipeline complete in {elapsed:.1f}s",
        elapsed=elapsed,
    )

    return final_report

What changed?

  • The report now passes through reflexion_loop(report) which either improves it or returns it unchanged if it already passes quality checks.
  • The function now returns final_report (the improved version) instead of the original report.
  • The timing measurement now includes the reflexion loop duration, giving you the true end-to-end pipeline time.
  • Save main.py.

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

Here is the complete main.py at this point:

"""Multi-agent research system with orchestrator, guardrails, and reflexion loop."""

import asyncio
import time

import httpx
import logfire
from pydantic_ai import Agent, RunContext

from agents import (
    ResearchDeps,
    critic_agent,
    data_analyst,
    web_researcher,
)
from models import CritiqueResult, FinalReport, ResearchQuery
from tools import validate_query, validate_report

# --- Observability Setup ---
logfire.configure()
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

# --- Orchestrator Agent ---
orchestrator = Agent(
    "google:gemini-2.5-flash",
    deps_type=ResearchDeps,
    output_type=FinalReport,
    instructions=(
        "You are a research orchestrator. Your job is to synthesize findings "
        "from the web researcher and data analyst into a comprehensive final "
        "report. Combine their outputs into a coherent narrative with clear "
        "executive summary, findings, insights, and recommendations."
    ),
)


async def run_specialists_parallel(
    topic: str, deps: ResearchDeps
) -> tuple[str, str]:
    """Run specialist agents concurrently using asyncio.gather."""
    logfire.info("Dispatching specialists in parallel for: {topic}", topic=topic)

    async def run_researcher():
        result = await web_researcher.run(
            f"Research the following topic thoroughly: {topic}",
            deps=deps,
        )
        return result.output

    async def run_analyst(research_context: str):
        result = await data_analyst.run(
            f"Analyze these research findings and extract insights:\n{research_context}",
            deps=deps,
        )
        return result.output

    research_result = await run_researcher()
    analysis_result = await run_analyst(research_result.raw_content)

    return research_result, analysis_result


async def reflexion_loop(
    report: FinalReport, max_iterations: int = 2
) -> FinalReport:
    """Self-critique loop that iteratively improves the report."""
    current_report = report

    for iteration in range(max_iterations):
        logfire.info(
            "Reflexion loop iteration {iteration}/{max}",
            iteration=iteration + 1,
            max=max_iterations,
        )

        critique = await critic_agent.run(
            f"Evaluate this research report:\n\n"
            f"Topic: {current_report.topic}\n"
            f"Summary: {current_report.executive_summary}\n"
            f"Findings: {current_report.research_findings}\n"
            f"Insights: {current_report.analysis_insights}\n"
            f"Recommendations: {current_report.recommendations}\n"
            f"Confidence: {current_report.confidence_score}"
        )

        critique_result: CritiqueResult = critique.output
        logfire.info(
            "Critique score: {score}, passed: {passed}",
            score=critique_result.score,
            passed=critique_result.passed,
        )

        if critique_result.passed:
            logfire.info("Report passed quality check at iteration {i}", i=iteration + 1)
            break

        improvement_prompt = (
            f"Improve this research report based on critic feedback.\n\n"
            f"Current report topic: {current_report.topic}\n"
            f"Current summary: {current_report.executive_summary}\n"
            f"Current findings: {current_report.research_findings}\n"
            f"Current insights: {current_report.analysis_insights}\n\n"
            f"CRITIC FEEDBACK (score {critique_result.score}/10):\n"
            f"{critique_result.feedback}\n"
            f"Weak areas: {critique_result.weak_areas}\n\n"
            f"Generate an improved version addressing ALL feedback points."
        )

        improved = await orchestrator.run(improvement_prompt)
        current_report = improved.output

    return current_report


async def research(topic: str) -> FinalReport | str:
    """Main research pipeline with guardrails, routing, and reflexion."""
    is_valid, error_msg = validate_query(topic)
    if not is_valid:
        logfire.warn("Input guardrail rejected query: {error}", error=error_msg)
        return f"Input rejected: {error_msg}"

    query = ResearchQuery(topic=topic)
    logfire.info("Starting research pipeline for: {topic}", topic=query.topic)

    start_time = time.time()

    async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
        deps = ResearchDeps(http_client=client)

        research_result, analysis_result = await run_specialists_parallel(
            query.topic, deps
        )

        synthesis_prompt = (
            f"Synthesize a comprehensive research report on: {query.topic}\n\n"
            f"WEB RESEARCH FINDINGS:\n"
            f"Sources: {research_result.sources}\n"
            f"Key Findings: {research_result.key_findings}\n"
            f"Content: {research_result.raw_content}\n"
            f"Confidence: {research_result.confidence}\n\n"
            f"DATA ANALYSIS:\n"
            f"Summary: {analysis_result.summary}\n"
            f"Insights: {analysis_result.insights}\n"
            f"Metrics: {analysis_result.metrics}\n"
            f"Recommendations: {analysis_result.recommendations}\n\n"
            f"Create a final report combining all findings."
        )

        result = await orchestrator.run(synthesis_prompt, deps=deps)
        report = result.output

    is_valid, error_msg = validate_report(report)
    if not is_valid:
        logfire.warn("Output guardrail flagged report: {error}", error=error_msg)

    final_report = await reflexion_loop(report)

    elapsed = time.time() - start_time
    logfire.info(
        "Research pipeline complete in {elapsed:.1f}s",
        elapsed=elapsed,
    )

    return final_report


async def main():
    """Entry point: run the multi-agent research system."""
    print("=" * 60)
    print("MULTI-AGENT RESEARCH SYSTEM")
    print("Production Patterns: Orchestrator | Guardrails | Reflexion")
    print("=" * 60)

    topic = input("\nEnter research topic: ").strip()
    if not topic:
        topic = "PydanticAI framework for production AI agents"

    print(f"\nResearching: {topic}")
    print("-" * 40)

    result = await research(topic)

    if isinstance(result, str):
        print(f"\nError: {result}")
    else:
        print(f"\n{'=' * 60}")
        print(f"FINAL REPORT: {result.topic}")
        print(f"{'=' * 60}")
        print(f"\nExecutive Summary:\n{result.executive_summary}")
        print(f"\nKey Findings:")
        for i, finding in enumerate(result.research_findings, 1):
            print(f"  {i}. {finding}")
        print(f"\nAnalysis Insights:")
        for i, insight in enumerate(result.analysis_insights, 1):
            print(f"  {i}. {insight}")
        print(f"\nRecommendations:")
        for i, rec in enumerate(result.recommendations, 1):
            print(f"  {i}. {rec}")
        print(f"\nConfidence Score: {result.confidence_score:.0%}")
        print(f"Sources: {', '.join(result.sources)}")
        print(f"\nView full traces at: https://logfire.pydantic.dev")


if __name__ == "__main__":
    asyncio.run(main())
Run the complete system and observe improvements in Logfire

Your multi-agent research system is now fully assembled. The pipeline runs through input guardrails, specialist agents, orchestrator synthesis, output guardrails, and the reflexion loop.

  • Run the complete system by executing this command in your terminal:
python main.py
  • When prompted, enter a research topic (or press Enter to use the default).

The system takes 30-60 seconds to complete. You should see the final report printed with an executive summary, findings, insights, recommendations, and a confidence score.

  • Open your Logfire dashboard at https://logfire.pydantic.dev.
  • Look for the trace spans labelled with reflexion loop iterations. You should see the critique scores and whether the report passed or required improvement.

What should I see in Logfire?

Each reflexion iteration creates its own trace span showing the critic's score and pass/fail status. If the report scored below 7.0, you'll see a second iteration where the orchestrator regenerates an improved version.

The full trace tree shows every agent decision: specialist tool calls, orchestrator synthesis, critic evaluation, and any re-generation cycles. This is the same observability pattern production teams use to debug multi-agent failures.

Pipeline timing out or returning errors?

If you see timeout errors, your httpx.AsyncClient may be closing before the reflexion loop finishes. The reflexion loop runs outside the async with httpx.AsyncClient block, which is correct because the critic and orchestrator don't need the HTTP client for re-evaluation.

If you see Gemini API rate limit errors, wait 60 seconds and try again. The reflexion loop makes multiple LLM calls, so free-tier rate limits can be a factor.

Help me debug my reflexion loop pipeline

Secret mission

Add Real-Time Streaming

Your multi-agent system works, but you have to wait for the full pipeline to finish before seeing anything. What if you could watch each agent think, call tools, and generate text token-by-token in real time? In this mission, you'll use PydanticAI's streaming API to build a live feed of agent reasoning.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs mostly locally. The Gemini API free tier and Logfire free tier mean there are no ongoing costs from your usage during this project.

Resources you used:

  • Local project files (models.py, tools.py, agents.py, main.py, main_streaming.py).
  • Local Logfire data directory (created by the SDK during authentication).
  • Google AI Studio API key (GOOGLE_API_KEY environment variable).

Keep everything running

No action needed. Choose this if you plan to keep building on your multi-agent system or experimenting with PydanticAI.

  • Your local project files remain on your machine for future reference.
  • Your Logfire project at logfire.pydantic.dev stays active on the free tier (10M spans/month, no credit card required).
  • Your Gemini API key remains active for future projects.

Pause - I'll come back to this later

There are no persistent running processes to pause. Everything in this project runs on demand when you execute the script.

  • Unset the environment variable from your current terminal session to avoid accidental API usage by running:
unset GOOGLE_API_KEY

Your files and accounts are ready whenever you return.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below.

  • Delete your local project files by running this command from the folder containing your Python files:
rm models.py tools.py agents.py main.py main_streaming.py
  • Remove the local Logfire data and logs by running:
logfire clean --logs
  • Delete your Gemini API key if you no longer need it:
  • Open the Google Cloud API Credentials page at https://console.cloud.google.com/apis/credentials.
  • Find the API key you used for this project and click the Actions icon.
  • Select Delete API key.
  • In the Delete credential modal, select Delete. Deletion takes a few minutes to propagate.
  • Remove the GOOGLE_API_KEY from your shell profile (.bashrc, .zshrc, or .bash_profile) if you added it there permanently.

Nice Work!

Nice Work!

Nice work! You just built a production-architecture multi-agent research system using PydanticAI and Gemini. This is the same orchestrator-delegate-critique pattern used by companies deploying AI agents at scale.

You've learned how to:

  • Built specialist agents with tool calling that autonomously fetch and process data, returning type-safe structured outputs via Pydantic models.
  • Implemented an orchestrator with conditional routing and input/output guardrails that validate data before and after LLM processing.
  • Added a self-critique reflexion loop where a critic agent evaluates output quality and triggers iterative re-generation until the report passes.
  • Secret Mission: Streamed agent reasoning in real-time using agent.iter() and node-level streaming to watch each agent think, call tools, and generate token-by-token.

Ready to quiz yourself?