# How to Build a Deep Research Agent

- Tool: Any LLM or framework
- Difficulty: Intermediate
- Time required: 45-60 min
- Compatibility: Python 3.10+, langgraph, a search API key (this guide uses Tavily), Claude API key
- Last updated: November 2026

## TL;DR

A deep research agent doesn't just search once and summarize; it plans sub-questions, searches each one, checks its own coverage, and loops back for gaps before writing a final report. This guide builds one with LangGraph's StateGraph that researches a one-line prompt like a competitive landscape question and produces a cited report from real search results, not the model's memory. Budget 45-60 minutes and a search API key.

## What you'll build

You'll build a research agent that takes one line, something like 'competitive landscape of vertical SaaS AI agents in field services,' and comes back with a short, sourced report instead of a single search result pasted into a paragraph. The difference between this and a one-shot 'search and summarize' call is the loop: the agent plans sub-questions first, searches each one, checks whether its own findings actually answer the original question, and goes back for more if they don't.

LangGraph models that loop as an explicit state graph instead of hiding it inside a single agent's internal reasoning. A plan node writes sub-questions into shared state, a search node fills in findings for each one, a synthesize node writes the report from those findings only, and a reflect node decides whether to loop back to planning or stop. Because the state and the routing are explicit, you can inspect exactly what the agent found at each step and cap the loop at a fixed number of rounds instead of trusting the model to know when to stop on its own.

This pattern works with any search tool. This guide uses a direct call to a search API inside the search node rather than Anthropic's hosted web search tool, so you can swap in whichever search provider your team already pays for without changing the graph's shape.

## Before you start

- Python 3.10+ and a terminal
- pip install langgraph langchain langchain-anthropic httpx
- An Anthropic API key (ANTHROPIC_API_KEY) with available credits
- A search API key (this guide uses Tavily; swap the search_web function for any provider you already use)

## Step-by-step guide

### 1. Install dependencies and set your keys

Install LangGraph alongside LangChain's Anthropic integration, which gives you init_chat_model so you can reference a model by a short provider:model string instead of importing a provider-specific client directly. Export both your Anthropic key and your search provider's key; the research loop needs both on every run.

```
pip install langgraph langchain langchain-anthropic httpx
export ANTHROPIC_API_KEY="sk-ant-..."
export TAVILY_API_KEY="tvly-..."
```

**Expected result:** pip install finishes cleanly and python -c "import langgraph" runs with no errors.

### 2. Define the graph's shared state

A TypedDict describes everything that flows between nodes: the original query, the list of sub-questions still to research, the findings collected so far, the current iteration count, and the final report once it's written. Every node reads from and writes back into this same shared state, which is what lets you inspect progress at any point in the loop.

```
from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    sub_questions: list[str]
    findings: list[dict]
    report: str
    iteration: int
```

**Expected result:** A state schema with no logic yet, just the shape every node in the graph will read and write.

### 3. Write the plan node

Ask the model to break the original query into 3-5 focused sub-questions, each narrow enough to answer with a single search. Splitting here is what separates this from a single broad search call: a sub-question like 'who are the top 3 funded competitors' returns far more useful search results than searching the whole original query verbatim.

```
from langchain.chat_models import init_chat_model

llm = init_chat_model("anthropic:claude-sonnet-5-5")

def plan_node(state: ResearchState) -> dict:
    prompt = (
        f"Break this research question into 3-5 focused sub-questions, "
        f"one per line, no numbering: {state['query']}"
    )
    lines = llm.invoke(prompt).content.strip().split("\n")
    return {"sub_questions": [l.strip("- ").strip() for l in lines if l.strip()]}
```

**Expected result:** state['sub_questions'] populated with a handful of narrow, searchable questions derived from the original prompt.

### 4. Write the search node

For each sub-question that doesn't already have findings, call a search function and store the top results with their titles, URLs, and snippets. Keeping findings keyed by sub-question is what makes the reflect step able to tell which sub-questions still have nothing, rather than just a pile of undifferentiated search results.

```
import httpx, os

def search_web(query: str, max_results: int = 4) -> list[dict]:
    resp = httpx.post("https://api.tavily.com/search", json={
        "api_key": os.environ["TAVILY_API_KEY"], "query": query,
        "max_results": max_results,
    }, timeout=20)
    return [{"title": r["title"], "url": r["url"], "snippet": r["content"]}
            for r in resp.json().get("results", [])]

def search_node(state: ResearchState) -> dict:
    researched = {f["sub_question"] for f in state["findings"]}
    new_findings = list(state["findings"])
    for q in state["sub_questions"]:
        if q in researched:
            continue
        for r in search_web(q):
            new_findings.append({"sub_question": q, **r})
    return {"findings": new_findings}
```

**Expected result:** state['findings'] grows by several entries per sub-question, each carrying a real title, URL, and snippet.

### 5. Write the synthesize node

Feed every finding, grouped by sub-question, into the model and instruct it to write the report using only what's in those findings, with a numbered source list at the end. The explicit instruction matters: without it, the model will quietly blend in things it already knew from training, which is exactly what makes a research report untrustworthy.

```
def synthesize_node(state: ResearchState) -> dict:
    sources = "\n\n".join(
        f"[{i+1}] {f['title']} ({f['url']})\n{f['snippet']}"
        for i, f in enumerate(state["findings"])
    )
    prompt = (
        f"Write a short report answering: {state['query']}\n\n"
        f"Use ONLY the sources below, cite them inline as [n], and list them "
        f"at the end. If the sources don't cover something, say so explicitly.\n\n{sources}"
    )
    return {"report": llm.invoke(prompt).content}
```

**Expected result:** A multi-paragraph report with inline [n] citations that trace back to a real URL in the findings list.

### 6. Write the reflect node and wire the loop

Ask the model whether the report actually answers the original query or is missing something a reasonable reader would expect, and cap how many extra rounds it can request. Without a hard cap, a model that's never fully satisfied with its own coverage will loop until you run out of budget; two or three extra rounds is enough for nearly any question.

```
MAX_ITERATIONS = 2

def reflect_node(state: ResearchState) -> dict:
    return {"iteration": state["iteration"] + 1}

def route_after_reflect(state: ResearchState) -> str:
    if state["iteration"] >= MAX_ITERATIONS:
        return "done"
    check = llm.invoke(
        f"Does this report fully answer '{state['query']}'? Reply only YES or NO.\n\n{state['report']}"
    ).content
    return "done" if "YES" in check.upper() else "replan"
```

**Expected result:** A routing function that sends the graph back to planning on a real gap, and stops once the cap is reached or coverage looks complete.

### 7. Assemble and run the graph

Wire the four nodes together with add_node and add_edge, route the conditional edge from reflect back to plan or on to END, and compile with a checkpointer so a thread_id can resume a research session instead of starting over if it's interrupted. Invoke it with the original query and a thread_id, then print the final report.

```
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

graph = StateGraph(ResearchState)
graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("reflect", reflect_node)
graph.add_edge(START, "plan")
graph.add_edge("plan", "search")
graph.add_edge("search", "synthesize")
graph.add_edge("synthesize", "reflect")
graph.add_conditional_edges("reflect", route_after_reflect, {"replan": "plan", "done": END})
app = graph.compile(checkpointer=InMemorySaver())

result = app.invoke(
    {"query": "...", "sub_questions": [], "findings": [], "report": "", "iteration": 0},
    config={"configurable": {"thread_id": "research-1"}},
)
print(result["report"])
```

**Expected result:** A printed report with inline citations, after the graph loops through plan, search, and synthesize at least once, possibly twice if reflect found a gap.

## Complete code example

File: `deep_research_agent.py`

```python
"""Deep research agent: plans sub-questions, searches, synthesizes a cited
report, and loops back on real gaps. Run: python deep_research_agent.py
pip install langgraph langchain langchain-anthropic httpx
"""
import os
from typing import TypedDict

import httpx
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph

MAX_ITERATIONS = 2
llm = init_chat_model("anthropic:claude-sonnet-5-5")


class ResearchState(TypedDict):
    query: str
    sub_questions: list[str]
    findings: list[dict]
    report: str
    iteration: int


def search_web(query: str, max_results: int = 4) -> list[dict]:
    """Swap this for any search API your team already pays for."""
    resp = httpx.post(
        "https://api.tavily.com/search",
        json={"api_key": os.environ["TAVILY_API_KEY"], "query": query, "max_results": max_results},
        timeout=20,
    )
    return [
        {"title": r["title"], "url": r["url"], "snippet": r["content"]}
        for r in resp.json().get("results", [])
    ]


def plan_node(state: ResearchState) -> dict:
    prompt = (
        f"Break this research question into 3-5 focused sub-questions, "
        f"one per line, no numbering: {state['query']}"
    )
    lines = llm.invoke(prompt).content.strip().split("\n")
    return {"sub_questions": [l.strip("- ").strip() for l in lines if l.strip()]}


def search_node(state: ResearchState) -> dict:
    researched = {f["sub_question"] for f in state["findings"]}
    new_findings = list(state["findings"])
    for q in state["sub_questions"]:
        if q in researched:
            continue
        for r in search_web(q):
            new_findings.append({"sub_question": q, **r})
    return {"findings": new_findings}


def synthesize_node(state: ResearchState) -> dict:
    sources = "\n\n".join(
        f"[{i + 1}] {f['title']} ({f['url']})\n{f['snippet']}"
        for i, f in enumerate(state["findings"])
    )
    prompt = (
        f"Write a short report answering: {state['query']}\n\n"
        f"Use ONLY the sources below, cite them inline as [n], and list them "
        f"at the end. If the sources don't cover something, say so explicitly.\n\n{sources}"
    )
    return {"report": llm.invoke(prompt).content}


def reflect_node(state: ResearchState) -> dict:
    return {"iteration": state["iteration"] + 1}


def route_after_reflect(state: ResearchState) -> str:
    if state["iteration"] >= MAX_ITERATIONS:
        return "done"
    check = llm.invoke(
        f"Does this report fully answer '{state['query']}'? Reply only YES or NO.\n\n{state['report']}"
    ).content
    return "done" if "YES" in check.upper() else "replan"


def build_graph():
    graph = StateGraph(ResearchState)
    graph.add_node("plan", plan_node)
    graph.add_node("search", search_node)
    graph.add_node("synthesize", synthesize_node)
    graph.add_node("reflect", reflect_node)
    graph.add_edge(START, "plan")
    graph.add_edge("plan", "search")
    graph.add_edge("search", "synthesize")
    graph.add_edge("synthesize", "reflect")
    graph.add_conditional_edges("reflect", route_after_reflect, {"replan": "plan", "done": END})
    return graph.compile(checkpointer=InMemorySaver())


def main():
    app = build_graph()
    result = app.invoke(
        {
            "query": "Competitive landscape of vertical SaaS AI agents in field services",
            "sub_questions": [],
            "findings": [],
            "report": "",
            "iteration": 0,
        },
        config={"configurable": {"thread_id": "research-1"}},
    )
    print(result["report"])


if __name__ == "__main__":
    main()
```

## Common mistakes

- **Sending the original broad query straight to a search API instead of breaking it into sub-questions** — undefined Fix: Plan 3-5 narrow sub-questions first. A search engine returns far more useful results for 'who are the top funded competitors in X' than for the full original research question verbatim.
- **Letting the synthesis step write from the model's training data instead of only the search findings** — undefined Fix: State explicitly that the report must use only the provided sources and must say so when something isn't covered. Without that instruction, the model quietly blends in outdated or ungrounded claims.
- **No cap on the reflect-and-replan loop** — undefined Fix: A model asked 'is this good enough' will rarely say yes on the first try. Cap iterations at 2-3 rounds regardless of what reflection decides.
- **Re-searching the same sub-question on every loop instead of only the ones still missing findings** — undefined Fix: Key findings by sub-question and skip ones that already have results, the way the search_node in this guide does, so a replan round only fills real gaps.
- **Treating a single search result as sufficient for a sub-question** — undefined Fix: Pull several results per sub-question. One source can be wrong, outdated, or off-topic; multiple results let the synthesis step cross-check.

## Best practices

- Keep sub-questions narrow and specific; a vague sub-question produces vague, low-signal search results.
- Cite inline with a numbered source list so every claim in the report traces back to a real URL.
- Cap replan loops at a fixed number of rounds; don't let the model decide when it's 'done enough.'
- Key findings by the sub-question that produced them, so you can tell which questions still have gaps.
- Use a checkpointer with a stable thread_id so a long research session can resume instead of restarting from scratch.
- Log every sub-question and the search results it returned, not just the final report, so you can audit how a conclusion was reached.

## Use cases

### undefined



### undefined



### undefined



## Frequently asked questions

### How is a deep research agent different from just asking an LLM a question directly?

Asking directly gets you an answer from the model's training data, which can be outdated or simply wrong on specifics. A deep research agent plans sub-questions, runs real searches, and writes its report only from what those searches returned, with citations you can check.

### Why use LangGraph instead of a single agent loop for this?

The plan, search, synthesize, and reflect steps are genuinely different operations with different inputs and outputs. Modeling them as explicit graph nodes lets you inspect state between steps, cap the loop deterministically, and swap one node (like the search function) without touching the others.

### How many rounds of research should I allow before stopping?

Two or three replan rounds covers nearly any question. Past that, additional rounds usually chase diminishing returns rather than closing a real gap, and you're better off narrowing the original query instead.

### Can I use a different search API than Tavily?

Yes. The search_web function is a single, isolated piece; swap its implementation for Exa, Brave Search, Bing, or Anthropic's hosted web search tool without changing anything else in the graph.

### How much does one research run cost?

Budget for one plan call, one search call per sub-question, one synthesis call, and one reflect call per round, each a few thousand tokens on a mid-tier model, plus per-search API costs. A typical 3-5 sub-question run on one round usually lands under a few cents in model cost.

### Can someone build a deep research agent for my specific domain?

Yes. RapidDev builds research agents scoped to a specific industry or data source, including the citation and gap-checking logic in this guide. A free 30-minute consultation is the fastest way to scope your version.

---

Source: https://www.rapidevelopers.com/build-ai-agent/how-to-build-a-deep-research-agent
© RapidDev — https://www.rapidevelopers.com/build-ai-agent/how-to-build-a-deep-research-agent