Build a Multi-Agent Research Assistant with Gemini Flash Free Tier

What We're Building

We are building a multi-agent research assistant that takes a high-level topic—say, "The impact of transformer models on computational biology"—and autonomously produces a structured research brief. This isn't a single monolithic prompt. It's a pipeline of three specialized agents, each with a narrow, testable responsibility, all driven by Google's Gemini Flash model on the free tier.

Feature List:

• Planner Agent: Decomposes a broad topic into 3-5 specific, answerable sub-questions.
• Search Agent: Executes live web searches for each sub-question using SerpAPI's free credits.
• Writer Agent: Synthesizes raw search snippets into a coherent, cited research brief with an executive summary.
• Orchestrator: Manages state transitions between agents, handling partial failures gracefully.
• Free-tier stack: Zero cost to build and run within reasonable rate limits.

This pattern—decomposition, retrieval, synthesis—is the backbone of most production research pipelines. By building it yourself, you understand exactly where latency lives, how prompts interact, and why naive single-prompt approaches fail on complex topics.

Architecture Overview

The flow is linear but stateful. The Planner outputs a JSON array of strings. The Search Agent iterates over that array, calling SerpAPI for each, and collects results into a context dictionary keyed by sub-question. The Writer receives that dictionary and produces the final brief. We use LangChain primarily for its clean LLM abstraction and output parsing—not for its agent framework, which is overkill here. The orchestration logic is plain Python, giving you full control over retries, timeouts, and error boundaries.

Prerequisites

Everything here is free-tier or has generous free credits. No credit card should be required for initial experimentation.

Create a .env file:

GOOGLE_API_KEY=your-gemini-api-key
SERPAPI_API_KEY=your-serpapi-key

Install dependencies:

pip install langchain langchain-google-genai python-dotenv google-search-results pydantic

Step 1: Project Setup and Dependencies

Create a single Python file—research_assistant.py. We'll keep everything in one file for clarity, but the agent boundaries are clean enough to split into modules later.

import os
import json
import time
from typing import List, Dict
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from serpapi import GoogleSearch

load_dotenv()

# Verify keys exist at startup
assert os.getenv("GOOGLE_API_KEY"), "Missing GOOGLE_API_KEY"
assert os.getenv("SERPAPI_API_KEY"), "Missing SERPAPI_API_KEY"

# Initialize Gemini Flash
llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-flash",
    temperature=0.3,  # Lower temperature for structured output
    max_tokens=2048,
)

We use Pydantic models to enforce structure at every agent boundary. This eliminates the silent failures that happen when one agent produces malformed output that cascades downstream.

Step 2: The Planner Agent

The Planner's job is narrow: receive a topic string, return a list of sub-questions. We constrain the output with a Pydantic model and an explicit parser.

class SubQuestions(BaseModel):
    questions: List[str] = Field(
        description="3-5 specific, answerable sub-questions that decompose the research topic"
    )

planner_parser = PydanticOutputParser(pydantic_object=SubQuestions)

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a research planner. Given a topic, decompose it into 3-5 specific, 
answerable sub-questions. Each sub-question should target a distinct aspect of the topic.
Do not answer the questions—only generate them.

{format_instructions}"""),
    ("human", "Research topic: {topic}")
])

planner_chain = planner_prompt | llm | planner_parser

def plan_research(topic: str) -> List[str]:
    """Decompose a topic into sub-questions."""
    result = planner_chain.invoke({
        "topic": topic,
        "format_instructions": planner_parser.get_format_instructions()
    })
    return result.questions

Why this works: The Pydantic parser guarantees we get a list of strings, not a free-text paragraph we'd have to regex-parse. If Gemini outputs something unparseable, LangChain will retry with the format instructions re-emphasized.

Step 3: The Search Agent

The Search Agent takes a single sub-question, calls SerpAPI, and returns cleaned results. We strip out unnecessary fields to keep context windows tight for the Writer.

def search_question(question: str, num_results: int = 5) -> Dict:
    """Search a single question via SerpAPI and return structured results."""
    params = {
        "q": question,
        "api_key": os.getenv("SERPAPI_API_KEY"),
        "num": num_results,
        "engine": "google"
    }
    search = GoogleSearch(params)
    results = search.get_dict()
    
    organic = results.get("organic_results", [])
    cleaned = []
    for r in organic[:num_results]:
        cleaned.append({
            "title": r.get("title", ""),
            "snippet": r.get("snippet", ""),
            "link": r.get("link", ""),
        })
    
    return {
        "question": question,
        "results": cleaned,
        "searched_at": time.time()
    }

def search_all_questions(questions: List[str]) -> List[Dict]:
    """Execute searches for all sub-questions with basic rate limiting."""
    all_results = []
    for i, q in enumerate(questions):
        print(f"Searching ({i+1}/{len(questions)}): {q}")
        try:
            result = search_question(q)
            all_results.append(result)
        except Exception as e:
            print(f"Search failed for '{q}': {e}")
            all_results.append({
                "question": q,
                "results": [],
                "error": str(e)
            })
        # Respect free-tier rate limits: 1 request per 2 seconds
        time.sleep(2)
    return all_results

Rate limiting matters. SerpAPI's free tier is 100 searches/month. At 5 sub-questions per run, that's 20 research briefs per month. The time.sleep(2) keeps us well under any burst limits.

Step 4: The Writer Agent

The Writer receives the aggregated search results and produces a structured brief. We define the output schema first, then build the prompt around it.

class ResearchBrief(BaseModel):
    title: str = Field(description="Concise title for the research brief")
    executive_summary: str = Field(description="3-5 sentence summary of key findings")
    sections: List[Dict[str, str]] = Field(
        description="List of sections, each with a heading and body synthesizing findings for a sub-question"
    )
    sources: List[str] = Field(description="List of URLs cited in the brief")

writer_parser = PydanticOutputParser(pydantic_object=ResearchBrief)

writer_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a research analyst. Synthesize the provided search results into a 
structured research brief. For each sub-question, write a concise section that integrates 
information from the search snippets. Cite sources inline where appropriate.

{format_instructions}"""),
    ("human", """Original topic: {topic}

Search results by sub-question:
{search_results}

Synthesize a comprehensive brief.""")
])

writer_chain = writer_prompt | llm | writer_parser

def write_brief(topic: str, search_results: List[Dict]) -> ResearchBrief:
    """Synthesize search results into a structured brief."""
    # Format search results as readable text for the prompt
    formatted = ""
    for item in search_results:
        formatted += f"\n### Sub-question: {item['question']}\n"
        for r in item.get("results", []):
            formatted += f"- {r['title']}: {r['snippet']} ({r['link']})\n"
    
    result = writer_chain.invoke({
        "topic": topic,
        "search_results": formatted,
        "format_instructions": writer_parser.get_format_instructions()
    })
    return result

Step 5: Orchestrating the Multi-Agent Pipeline

The orchestrator is plain Python. It ties the three agents together, handles errors, and prints progress. This is where you'd add logging, retries, or parallel search execution in a production version.

def run_research_pipeline(topic: str) -> ResearchBrief:
    """Full pipeline: plan -> search -> write."""
    print(f"\n{'='*60}")
    print(f"RESEARCH TOPIC: {topic}")
    print(f"{'='*60}\n")
    
    # Phase 1: Plan
    print("[PLANNER] Decomposing topic...")
    questions = plan_research(topic)
    print(f"Generated {len(questions)} sub-questions:")
    for i, q in enumerate(questions, 1):
        print(f"  {i}. {q}")
    
    # Phase 2: Search
    print("\n[SEARCHER] Executing web searches...")
    search_results = search_all_questions(questions)
    successful = sum(1 for s in search_results if s.get("results"))
    print(f"Completed: {successful}/{len(questions)} searches returned results")
    
    # Phase 3: Write
    print("\n[WRITER] Synthesizing research brief...")
    brief = write_brief(topic, search_results)
    
    return brief

def format_brief(brief: ResearchBrief) -> str:
    """Pretty-print the brief for terminal output."""
    output = f"\n{'='*60}\n"
    output += f"{brief.title.upper()}\n"
    output += f"{'='*60}\n\n"
    output += f"EXECUTIVE SUMMARY:\n{brief.executive_summary}\n\n"
    
    for section in brief.sections:
        output += f"## {section.get('heading', 'Section')}\n"
        output += f"{section.get('body', '')}\n\n"
    
    output += f"SOURCES:\n"
    for url in brief.sources:
        output += f"  - {url}\n"
    
    return output

if __name__ == "__main__":
    topic = input("Enter research topic: ")
    brief = run_research_pipeline(topic)
    print(format_brief(brief))
    
    # Save to file
    with open("research_brief.json", "w") as f:
        f.write(brief.model_dump_json(indent=2))
    print("\nBrief saved to research_brief.json")

How to Run It

python research_assistant.py

You'll be prompted for a topic. The pipeline runs sequentially, printing progress at each phase. Output appears in the terminal and is saved as research_brief.json.

Sample run for "Edge computing implications for IoT security":

• Planner generates: "What are the primary attack vectors...", "How does edge architecture reduce latency...", "What standards govern..."
• Searcher fetches 5 results per question (15 total snippets)
• Writer produces a brief with executive summary, 3 sections, and 8-12 cited sources

Total execution time: ~30-40 seconds, dominated by the time.sleep(2) between searches. Without rate limiting, it's ~10 seconds.

Extensions and Production Hardening

This pipeline works, but it's a prototype. Here's what production-grade looks like:

Parallel search execution. Replace the sequential loop with concurrent.futures.ThreadPoolExecutor. SerpAPI can handle parallel requests; your free-tier monthly cap is the real constraint.

from concurrent.futures import ThreadPoolExecutor, as_completed

def search_all_parallel(questions: List[str], max_workers: int = 3) -> List[Dict]:
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(search_question, q): q for q in questions}
        results = []
        for future in as_completed(futures):
            results.append(future.result())
        return results

Streaming output. For long briefs, stream the Writer's output token-by-token. LangChain's .stream() method on the chain gives you this for free.

Caching. Cache search results by question hash. If a user researches a similar topic, reuse cached snippets instead of burning SerpAPI credits. A simple SQLite cache adds 20 lines of code.

Better source handling. The current Writer hallucinates citations occasionally. For a rigorous brief, pass only URLs that appear in search results and instruct the model to cite only from the provided list. Better yet, fetch full page content with requests and BeautifulSoup for the top 2 results per question, then run a smaller extraction LLM call to pull key claims.

If you're interested in building more autonomous agent systems, the patterns here—decomposition, tool use, structured output—are the same ones that power calendar-negotiating agents. Check out our guide on building a calendar-scheduling agent with Gemini for a deeper dive into stateful agent loops.

Common Pitfalls and Debugging Tips

Pitfall 1: Planner generates vague sub-questions. Fix: Increase the specificity in the system prompt. Add examples: "Instead of 'What is the impact?', ask 'How did transformer models reduce protein folding prediction error rates between 2020 and 2024?'"

Pitfall 2: Writer ignores search results and hallucinates. Fix: This is the most common failure mode. Add an explicit constraint: "Only use information present in the provided search snippets. If a sub-question has insufficient results, state that clearly rather than fabricating content." Also, lower temperature to 0.1 for the Writer.

Pitfall 3: SerpAPI returns empty results. Fix: Check your query format. Overly long or oddly phrased sub-questions confuse Google. Truncate to the core keywords before calling search_question. A simple heuristic: take the first 10 words.

Pitfall 4: Gemini rate limits (429 errors). Fix: The free tier is 15 RPM. Our pipeline makes 2 LLM calls (planner + writer), so you're well under the limit. If you add a refinement loop or retries, add time.sleep(4) between LLM calls.

Pitfall 5: Pydantic parsing fails on malformed LLM output. Fix: LangChain's PydanticOutputParser includes automatic retry logic. If it still fails, wrap the chain invocation in a try/except and fall back to a simpler prompt that asks for JSON without the schema. Parse it manually as a last resort.

For a deeper look at why structured output matters in production LLM applications, see our piece on DSLs as the missing link for reliable LLM applications.

FAQ

Q: Why three agents instead of one big prompt? A: Separation of concerns. A single prompt that plans, searches, and writes would need tool-use capabilities and complex output parsing. Three agents let you debug each phase independently, swap components (e.g., replace SerpAPI with Brave Search), and control exactly what context passes between stages. It also keeps each prompt small, which reduces hallucination and improves Gemini Flash's output quality.

Q: Can I use a different search API? A: Yes. The Search Agent is the only component that touches SerpAPI. Swap it for Brave Search API (also has free credits), Bing Web Search, or even a local DuckDuckGo scraping library. The interface is just search_question(question) -> Dict.

Q: How do I make the brief longer or more detailed? A: Increase the number of search results per question (currently 5) and instruct the Writer to produce longer sections. You can also add a "Refinement Agent" that takes the first draft and expands each section with additional context from the search results.

Q: What if I want to research a topic that changes frequently (e.g., stock prices)? A: Add a timestamp to each search result and instruct the Writer to note when information was retrieved. For real-time data, you'd need a different tool than SerpAPI—Google's search index has latency. Consider adding a dedicated "News Search" agent using SerpAPI's engine: google_news parameter.

Q: Is this pattern useful beyond research briefs? A: Absolutely. The plan-search-write pipeline is a general pattern for any task requiring decomposition, external data, and synthesis. It's the same architecture behind competitive intelligence dashboards, due diligence reports, and even the Slack digest bot we built that summarizes channels every morning. Once you internalize this pattern, you'll see it everywhere.

Q: How do I deploy this as a simple web app? A: Wrap the pipeline in a FastAPI endpoint, add a minimal React frontend (or use Streamlit for even faster prototyping), and deploy on Render or Railway free tiers. The LLM and search API calls happen server-side, so no API key exposure. Add request queuing if you expect concurrent users.