How to Build an AI Research Agent That Searches the Web and Writes Reports

ai-research-agent/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── schemas.py
│   ├── search_tool.py
│   ├── scraper.py
│   ├── agent_graph.py
│   └── main.py
├── tests/
│   └── test_research_agent.py
├── .env.example
├── docker-compose.yml
└── requirements.txt

python3 -m venv .venv
source .venv/bin/activate
pip install fastapi==0.115.0 uvicorn==0.30.6 langgraph==1.2.18 openai==1.51.0 pydantic==2.9.2 httpx==0.27.2 beautifulsoup4==4.12.3 playwright==1.48.0 tenacity==9.0.0 pytest==8.3.3 pytest-asyncio==0.24.0 python-dotenv==1.0.1
playwright install chromium

OPENAI_API_KEY=sk-proj-your-api-key-here
OPENAI_MODEL=gpt-4o-mini
SEARXNG_BASE_URL=http://localhost:8080
MAX_RESEARCH_ITERATIONS=3
REQUEST_TIMEOUT_SECONDS=15

import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    OPENAI_MODEL: str = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
    SEARXNG_BASE_URL: str = os.getenv("SEARXNG_BASE_URL", "http://localhost:8080")
    MAX_ITERATIONS: int = int(os.getenv("MAX_RESEARCH_ITERATIONS", "3"))
    TIMEOUT_SECONDS: int = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "15"))

settings = Settings()

from typing import List, Dict, TypedDict, Annotated
import operator
from pydantic import BaseModel, Field

class SearchQueryPlan(BaseModel):
    queries: List[str] = Field(
        description="3 to 4 specific, non-overlapping technical web search queries.",
        min_length=1,
        max_length=4,
    )
    rationale: str = Field(description="Why these queries cover the topic.")

class ReflectionEvaluation(BaseModel):
    is_sufficient: bool = Field(
        description="True if gathered sources contain sufficient factual detail to write the full report."
    )
    missing_aspects: List[str] = Field(
        default_factory=list,
        description="Specific technical gaps requiring follow-up search queries.",
    )
    follow_up_queries: List[str] = Field(
        default_factory=list,
        description="Up to 2 targeted follow-up queries if is_sufficient is False.",
    )

class SourceDocument(BaseModel):
    url: str
    title: str
    snippet: str
    extracted_content: str

class ResearchState(TypedDict):
    topic: str
    iteration: int
    queries: Annotated[List[str], operator.add]
    sources: Dict[str, SourceDocument]
    is_sufficient: bool
    draft_report: str
    verified_report: str
    validation_warnings: List[str]

import ipaddress
import socket
from urllib.parse import urlparse
from bs4 import BeautifulSoup
from playwright.async_api import async_playwright
from app.config import settings

def is_safe_public_url(url: str) -> bool:
    """Prevents Server-Side Request Forgery (SSRF) by blocking private/loopback IPs."""
    try:
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https") or not parsed.hostname:
            return False
        addr_info = socket.getaddrinfo(parsed.hostname, None)
        for item in addr_info:
            ip_str = item[4][0]
            ip_obj = ipaddress.ip_address(ip_str)
            if (
                ip_obj.is_private
                or ip_obj.is_loopback
                or ip_obj.is_link_local
                or ip_obj.is_multicast
                or ip_obj.is_reserved
            ):
                return False
        return True
    except Exception:
        return False

def clean_html_to_text(html: str, max_chars: int = 6000) -> str:
    """Strips scripts, styles, navbars, and compresses whitespace."""
    soup = BeautifulSoup(html, "html.parser")
    for tag in soup(["script", "style", "nav", "footer", "header", "noscript", "svg"]):
        tag.decompose()
    lines = [line.strip() for line in soup.get_text(separator="\n").splitlines() if line.strip()]
    return "\n".join(lines)[:max_chars]

async def extract_page_with_playwright(url: str) -> str:
    """Fetches and renders a public webpage using headless Chromium."""
    if not is_safe_public_url(url):
        raise ValueError(f"Blocked unsafe or private URL: {url}")

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True, args=["--disable-dev-shm-usage"])
        context = await browser.new_context(
            user_agent="AniketResearchBot/1.0 (+https://aniket.app)"
        )
        page = await context.new_page()
        try:
            await page.goto(url, timeout=settings.TIMEOUT_SECONDS * 1000, wait_until="domcontentloaded")
            html_content = await page.content()
            return clean_html_to_text(html_content)
        finally:
            await context.close()
            await browser.close()

import httpx
from typing import List, Dict
from tenacity import retry, stop_after_attempt, wait_exponential
from app.config import settings

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=6))
async def search_searxng(query: str, max_results: int = 4) -> List[Dict[str, str]]:
    """Queries a SearXNG instance with JSON output enabled."""
    params = {"q": query, "format": "json", "language": "en"}
    async with httpx.AsyncClient(timeout=settings.TIMEOUT_SECONDS) as client:
        response = await client.get(f"{settings.SEARXNG_BASE_URL}/search", params=params)
        response.raise_for_status()
        payload = response.json()
        results = []
        for item in payload.get("results", [])[:max_results]:
            url = item.get("url")
            if url:
                results.append({
                    "url": url,
                    "title": item.get("title", "Untitled Source"),
                    "snippet": item.get("content", ""),
                })
        return results

import re
from typing import Dict
from openai import AsyncOpenAI
from langgraph.graph import StateGraph, START, END
from app.config import settings
from app.schemas import (
    ResearchState,
    SearchQueryPlan,
    ReflectionEvaluation,
    SourceDocument,
)
from app.search_tool import search_searxng
from app.scraper import extract_page_with_playwright

client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)

async def plan_queries_node(state: ResearchState) -> dict:
    if state.get("iteration", 0) == 0:
        completion = await client.beta.chat.completions.parse(
            model=settings.OPENAI_MODEL,
            messages=[
                {"role": "system", "content": "You are a senior technical research planner."},
                {"role": "user", "content": f"Generate initial search queries for: {state['topic']}"},
            ],
            response_format=SearchQueryPlan,
        )
        new_queries = completion.choices[0].message.parsed.queries
    else:
        new_queries = state.get("queries", [])[-2:]
    return {"queries": new_queries, "iteration": state.get("iteration", 0) + 1}

async def execute_search_and_extract_node(state: ResearchState) -> dict:
    sources: Dict[str, SourceDocument] = dict(state.get("sources", {}))
    recent_queries = state["queries"][-4:]

    for query in recent_queries:
        hits = await search_searxng(query, max_results=3)
        for hit in hits:
            url = hit["url"]
            if url in sources:
                continue
            try:
                extracted = await extract_page_with_playwright(url)
            except Exception:
                extracted = hit["snippet"]
            if extracted:
                sources[url] = SourceDocument(
                    url=url,
                    title=hit["title"],
                    snippet=hit["snippet"],
                    extracted_content=extracted,
                )
    return {"sources": sources}

async def reflect_on_evidence_node(state: ResearchState) -> dict:
    evidence_summary = "\n\n".join(
        [f"Source: {doc.title} ({doc.url})\n{doc.extracted_content[:1200]}" for doc in state["sources"].values()]
    )
    completion = await client.beta.chat.completions.parse(
        model=settings.OPENAI_MODEL,
        messages=[
            {"role": "system", "content": "Evaluate whether the gathered evidence is sufficient to answer the research topic thoroughly."},
            {"role": "user", "content": f"Topic: {state['topic']}\n\nEvidence:\n{evidence_summary}"},
        ],
        response_format=ReflectionEvaluation,
    )
    parsed = completion.choices[0].message.parsed
    return {
        "is_sufficient": parsed.is_sufficient,
        "queries": parsed.follow_up_queries if not parsed.is_sufficient else [],
    }

def route_after_reflection(state: ResearchState) -> str:
    if state["is_sufficient"] or state["iteration"] >= settings.MAX_ITERATIONS:
        return "synthesize_report"
    return "execute_search"

async def synthesize_report_node(state: ResearchState) -> dict:
    allowed_sources_block = "\n\n".join(
        [f"[{doc.title}]({doc.url}):\n{doc.extracted_content[:2000]}" for doc in state["sources"].values()]
    )
    prompt = (
        f"Write a comprehensive technical report on: {state['topic']}.\n"
        "CRITICAL RULE: Every factual claim MUST include an inline markdown link [Source Title](URL) "
        "chosen ONLY from the provided sources below. Never invent URLs.\n\n"
        f"Allowed Sources:\n{allowed_sources_block}"
    )
    response = await client.chat.completions.create(
        model=settings.OPENAI_MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
    )
    draft = response.choices[0].message.content or ""
    return {"draft_report": draft}

def verify_citations_deterministic(draft: str, allowed_urls: set) -> tuple[str, list[str]]:
    """Deterministically verifies all markdown URLs against the retrieved source registry."""
    pattern = re.compile(r"\[([^\]]+)\]\((https?://[^\)]+)\)")
    warnings = []

    def replace_unverified(match: re.Match) -> str:
        label, url = match.group(1), match.group(2)
        if url not in allowed_urls:
            warnings.append(f"Stripped unverified hallucinated URL: {url}")
            return f"{label} [Unverified Citation Removed]"
        return match.group(0)

    cleaned_report = pattern.sub(replace_unverified, draft)
    return cleaned_report, warnings

async def validate_citations_node(state: ResearchState) -> dict:
    allowed_urls = set(state["sources"].keys())
    verified_text, warnings = verify_citations_deterministic(state["draft_report"], allowed_urls)
    return {"verified_report": verified_text, "validation_warnings": warnings}

def build_research_graph():
    builder = StateGraph(ResearchState)
    builder.add_node("plan_queries", plan_queries_node)
    builder.add_node("execute_search", execute_search_and_extract_node)
    builder.add_node("reflect", reflect_on_evidence_node)
    builder.add_node("synthesize_report", synthesize_report_node)
    builder.add_node("validate_citations", validate_citations_node)

    builder.add_edge(START, "plan_queries")
    builder.add_edge("plan_queries", "execute_search")
    builder.add_edge("execute_search", "reflect")
    builder.add_conditional_edges(
        "reflect",
        route_after_reflection,
        {"execute_search": "execute_search", "synthesize_report": "synthesize_report"},
    )
    builder.add_edge("synthesize_report", "validate_citations")
    builder.add_edge("validate_citations", END)
    return builder.compile()

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from app.agent_graph import build_research_graph

app = FastAPI(title="AI Research Agent API", version="1.0.0")
research_graph = build_research_graph()

class ResearchRequest(BaseModel):
    topic: str = Field(..., min_length=10, max_length=500)

@app.post("/api/v1/research")
async def run_research(req: ResearchRequest):
    initial_state = {
        "topic": req.topic,
        "iteration": 0,
        "queries": [],
        "sources": {},
        "is_sufficient": False,
        "draft_report": "",
        "verified_report": "",
        "validation_warnings": [],
    }
    try:
        final_state = await research_graph.ainvoke(initial_state)
        return {
            "topic": req.topic,
            "iterations_executed": final_state["iteration"],
            "sources_count": len(final_state["sources"]),
            "report": final_state["verified_report"],
            "validation_warnings": final_state["validation_warnings"],
        }
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))

import pytest
from app.scraper import is_safe_public_url, clean_html_to_text
from app.agent_graph import verify_citations_deterministic

def test_ssrf_protection_blocks_private_ips():
    assert is_safe_public_url("http://127.0.0.1:8080/admin") is False
    assert is_safe_public_url("http://169.254.169.254/latest/meta-data/") is False
    assert is_safe_public_url("ftp://example.com/file") is False

def test_clean_html_strips_scripts_and_nav():
    raw_html = "<html><head><script>alert(1)</script></head><body><nav>Menu</nav><main><h1>Benchmark</h1><p>Latency is 42ms.</p></main></body></html>"
    cleaned = clean_html_to_text(raw_html)
    assert "alert(1)" not in cleaned
    assert "Menu" not in cleaned
    assert "Latency is 42ms." in cleaned

def test_citation_validator_strips_hallucinated_links():
    allowed = {"https://docs.langchain.com/oss/python/langgraph/"}
    draft = (
        "LangGraph supports durable execution ([LangGraph Docs](https://docs.langchain.com/oss/python/langgraph/)). "
        "It also has a fake benchmark ([Fake Study](https://fake-domain.example.com/study))."
    )
    verified, warnings = verify_citations_deterministic(draft, allowed)
    assert "https://docs.langchain.com/oss/python/langgraph/" in verified
    assert "https://fake-domain.example.com/study" not in verified
    assert len(warnings) == 1

pytest tests/test_research_agent.py -v

========================= test session starts ==========================
collected 3 items

tests/test_research_agent.py::test_ssrf_protection_blocks_private_ips PASSED [ 33%]
tests/test_research_agent.py::test_clean_html_strips_scripts_and_nav PASSED  [ 66%]
tests/test_research_agent.py::test_citation_validator_strips_hallucinated_links PASSED [100%]
========================== 3 passed in 0.18s ===========================

search:
       formats:
         - html
         - json