How to Build Multi-Stage Research Agents with Specialized Sub-Agents

Compose complex research workflows by chaining focused LLM agents through a SequentialAgent orchestrator that passes data via output keys, as demonstrated in the Google ADK-based trend analyzer from the Arindam200/awesome-ai-apps repository.

The Arindam200/awesome-ai-apps repository provides a production-ready blueprint for building multi-stage research agents with specialized sub-agents. The trend analyzer implementation demonstrates how to decompose a complex AI analysis task into discrete, reusable components that execute sequentially while maintaining state across the pipeline.

Architecture Overview

The system follows the Google ADK (Agent Development Kit) pattern, separating concerns between session management, orchestration, and specialized execution units.

Session and Runner Setup

Execution begins with an InMemorySessionService that maintains conversation state and a Runner that handles the event loop. In advance_ai_agents/trend_analyzer_agent/agent.py, the runner wraps the top-level orchestrator:

# From agent.py lines 95-103

session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME, 
    user_id=USER_ID, 
    session_id=SESSION_ID
)
runner = Runner(agent=pipeline, app_name=APP_NAME, session_service=session_service)

Sequential Orchestration

The AIPipelineAgent acts as a SequentialAgent, executing sub-agents in strict order. This orchestrator feeds the output of each step into the next, creating a linear processing chain ideal for research workflows that require progressive refinement.

The Five Specialized Sub-Agents

Each LlmAgent encapsulates a single responsibility, configured with a shared LiteLlm model instance (Nebius) for consistency unless specific steps require different capabilities.

• ExaAgent: Retrieves latest AI-related tweets and posts using the exa_search_ai tool.
• TavilyAgent: Fetches AI benchmarks and statistical reports via the tavily_search_ai_analysis tool.
• SummaryAgent: Merges raw results from previous agents, applying markdown formatting, headings, and emojis without external API calls.
• FirecrawlAgent: Scrapes the Nebius Token Factory homepage for model specifications using the firecrawl_scrape_nebius tool.
• AnalysisAgent: Performs business-oriented analysis, matching summaries against the Nebius model catalog to generate recommendations.

Data Flow and State Management

The pipeline utilizes output_key parameters to persist intermediate results in session state. As implemented in agent.py, the data flows through these keys:

1. exa_results → ExaAgent output
2. tavily_results → TavilyAgent output
3. final_summary → SummaryAgent output
4. firecrawl_content → FirecrawlAgent output
5. analysis_results → AnalysisAgent final output

Each subsequent agent can reference previous outputs through the session context, enabling progressive document construction without manual data passing.

Implementing the Research Pipeline

The complete execution requires initializing the session, configuring the runner, and processing the event stream. The repository's entry point demonstrates this pattern:

# run_ai_analysis.py – launch the multi-stage research pipeline

from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types
from advance_ai_agents.trend_analyzer_agent.agent import pipeline, APP_NAME, USER_ID, SESSION_ID

# Initialise session

session_service = InMemorySessionService()
session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)

# Build runner around the orchestrator

runner = Runner(agent=pipeline, app_name=APP_NAME, session_service=session_service)

# Start the workflow

def run():
    start_msg = types.Content(role="user", parts=[types.Part(text="Start the AI analysis")])
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=start_msg):
        if event.is_final_response():
            print("\n📢 AI News Analysis and Insights:\n")
            print(event.content.parts[0].text)

if __name__ == "__main__":
    run()

The runner.run() method yields events until event.is_final_response() indicates completion, at which point the final analysis is available in the event content.

Extending the Pipeline with Custom Sub-Agents

Adding new research capabilities requires three steps:

1. Define the tool function that interfaces with your external API:

def pubmed_search(_: str) -> dict:
    """Fetch recent AI research from PubMed API"""
    papers = call_pubmed_api(query="artificial intelligence")
    return {"type": "pubmed", "results": papers}

2. Configure the LlmAgent with instructions and output_key:

pubmed_agent = LlmAgent(
    name="PubMedAgent",
    model=nebius_model,  # Shared LiteLlm instance

    description="Searches PubMed for recent AI-related papers.",
    instruction="""
        Use the pubmed_search tool to fetch the latest AI research articles.
        Prefix your output with "**🧬PubMedAgent:**".
        Format results as a markdown list.
        """,
    tools=[pubmed_search],
    output_key="pubmed_results"
)

3. Insert into the SequentialAgent sub_agents list:

# Append to existing pipeline

pipeline.sub_agents.append(pubmed_agent)

The agent will now execute at the position inserted, with access to all previous output_keys in the session state.

Summary

• Multi-stage research agents require an orchestrator (SequentialAgent) to manage execution order and a session service to maintain state.
• Specialized sub-agents should encapsulate single responsibilities, using dedicated tools for external data retrieval and structured instructions for output formatting.
• State persistence occurs through output_key declarations that deposit results into the session for downstream consumption.
• Extensibility follows a three-step pattern: define tool, configure LlmAgent, append to orchestrator's sub_agents list.

Frequently Asked Questions

What is the difference between LlmAgent and SequentialAgent in the Google ADK?

SequentialAgent is an orchestrator that manages execution flow, running its sub_agents list in order and managing handoffs between them. LlmAgent is a leaf node that performs actual LLM inference, optionally calling tools. The AIPipelineAgent contains multiple LlmAgent instances (ExaAgent, TavilyAgent, etc.) as its children.

How does data flow between sub-agents in the multi-stage pipeline?

Data flows via session state using the output_key parameter. When an LlmAgent completes, it stores its output under its assigned key (e.g., exa_results). Subsequent agents access this data through the session context automatically populated by the Runner, eliminating the need for explicit parameter passing in code.

Can I use different LLM models for different sub-agents?

Yes. While the repository shares a single LiteLlm instance (Nebius) across agents for consistency, each LlmAgent accepts its own model parameter. This allows compute-intensive agents like AnalysisAgent to use larger models while lightweight agents use faster, cheaper alternatives, optimizing cost and latency per stage.

Where should I store API keys for the external tools?

API credentials belong in environment variables following the pattern shown in advance_ai_agents/trend_analyzer_agent/.env.example. The repository expects keys for EXA, TAVILY, FIRECRAWL, and NEBIUS services, loaded via python-dotenv before agent initialization to keep secrets out of source control.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}