How to build a deep research agent with search and citations

Most research agents collapse because they fire one search query and ask the model to write an essay. To build deep research agent that survives real questions, you need to decompose the task, retrieve from multiple angles, and force the synthesizer to anchor every claim to a source. This guide walks through a concrete implementation you can run today.

Step 1: Define the agent loop and pick your models

A deep research agent is a loop: plan → search → read → synthesize → verify. The LLM needs to emit structured output (JSON or function calls) reliably. Use an OpenAI-compatible client so you can swap models without rewriting code.

from openai import OpenAI
import os

# Using n4n.ai as an OpenAI-compatible gateway gives automatic fallback across 240+ models
# when a provider is rate-limited, so the agent loop doesn't die mid-research.
client = OpenAI(
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
)

For search, Tavily’s API returns clean snippets and URLs suited for LLM consumption. A minimal wrapper:

import requests

def search(query: str, max_results: int = 5) -> list[dict]:
    resp = requests.post(
        "https://api.tavily.com/search",
        headers={"Authorization": f"Bearer {os.environ['TAVILY_KEY']}"},
        json={"query": query, "max_results": max_results, "search_depth": "advanced"},
    )
    resp.raise_for_status()
    return resp.json()["results"]

When you build deep research agent loops, treat the model and search as separate failure domains. Isolate their retries.

Step 2: Decompose the question into subqueries

A broad question like “What are the environmental impacts of lithium mining?” needs multiple searches: regulation, water usage, emissions, recycling. Ask the LLM to plan.

import json

def decompose_query(question: str) -> list[str]:
    sys = "You are a research planner. Break the user question into 3-5 specific web search queries that collectively cover it."
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": sys},
                  {"role": "user", "content": question}],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    return data["queries"]

Keep the planner cheap. A smaller model is fine here; the value is in the list of strings, not prose.

Step 3: Run parallel searches and collect sources

Sequential requests waste latency. Use asyncio to hit the search API concurrently.

import asyncio, aiohttp

async def async_search(sess: aiohttp.ClientSession, q: str) -> list[dict]:
    async with sess.post(
        "https://api.tavily.com/search",
        headers={"Authorization": f"Bearer {os.environ['TAVILY_KEY']}"},
        json={"query": q, "max_results": 5, "search_depth": "advanced"},
    ) as r:
        data = await r.json()
        return data["results"]

async def gather_sources(queries: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as sess:
        tasks = [async_search(sess, q) for q in queries]
        results = await asyncio.gather(*tasks)
    sources = []
    for r in results:
        sources.extend(r)
    return sources

Run it from sync code with asyncio.run(gather_sources(queries)). You now have a flat list of candidate sources.

Step 4: Rank and deduplicate

Search APIs return overlapping links. Dedupe by URL and prefer higher-score, diverse domains.

def rank_sources(sources: list[dict], top_k: int = 12) -> list[dict]:
    seen = set()
    unique = []
    for s in sorted(sources, key=lambda x: x.get("score", 0), reverse=True):
        if s["url"] in seen:
            continue
        seen.add(s["url"])
        unique.append(s)
    return unique[:top_k]

Domain diversity matters more than raw score. If three top results are from the same blog, drop two. Build a simple domain counter and cap per-domain entries at two.

Step 5: Synthesize a cited answer

To build deep research agent with citations, the synthesis prompt must be strict. Demand JSON with explicit claim-to-URL mapping.

def synthesize(question: str, sources: list[dict]) -> dict:
    context = "\n\n".join(
        f"URL: {s['url']}\nTITLE: {s.get('title','')}\nSNIPPET: {s['content']}"
        for s in sources
    )
    sys = (
        "You are a research writer. Use ONLY the provided sources. "
        "Return JSON: {answer: string, citations: [{claim: string, url: string}]}. "
        "Every factual sentence in 'answer' must have a matching citation."
    )
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": sys},
            {"role": "user", "content": f"QUESTION: {question}\n\nSOURCES:\n{context}"},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

The citations array lets you render footnotes or inline links. Validate that each url appears in the source list—never trust the model to invent URLs.

Step 6: Verify and iterate

A second pass catches hallucinated claims. Ask the model to check whether the citations support the answer.

def verify(answer: str, citations: list[dict], sources: list[dict]) -> bool:
    url_set = {s["url"] for s in sources}
    for c in citations:
        if c["url"] not in url_set:
            return False
    # Optional LLM check
    sys = "Check if each citation claim is directly supported by the source snippet. Return JSON {ok: bool}."
    ctx = "\n".join(f"{c['claim']} -> {c['url']}" for c in citations)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": sys},
                  {"role": "user", "content": ctx}],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content).get("ok", False)

If verification fails, loop back to Step 2 with a refined planner prompt or pull more sources. Before you build deep research agent into production, wire this loop to a max-iteration guard.

Verify success

Run the full pipeline on a known question:

question = "What are the main criticisms of carbon offset markets?"
queries = decompose_query(question)
sources = asyncio.run(gather_sources(queries))
ranked = rank_sources(sources)
out = synthesize(question, ranked)
assert out["answer"] and out["citations"]
assert verify(out["answer"], out["citations"], ranked)
print(f"Answer length: {len(out['answer'])} chars, sources used: {len(out['citations'])}")

Success criteria: the script completes without assertion errors, the answer contains no URL not present in ranked, and citations is non-empty. For a manual check, open two cited URLs and confirm the claim text matches the page content. If you mock the search layer with fixed fixtures, this doubles as a unit test in CI.

The agent is now capable of multi-step research with traceable output. Swap the search backend or the model name without touching the loop logic, and add persistence for the source set if you need audit trails.