How to Build a Deep Research Agent

1"""Deep research agent: plans sub-questions, searches, synthesizes a cited
2report, and loops back on real gaps. Run: python deep_research_agent.py
3pip install langgraph langchain langchain-anthropic httpx
4"""
5import os
6from typing import TypedDict
7
8import httpx
9from langchain.chat_models import init_chat_model
10from langgraph.checkpoint.memory import InMemorySaver
11from langgraph.graph import END, START, StateGraph
12
13MAX_ITERATIONS = 2
14llm = init_chat_model("anthropic:claude-sonnet-5-5")
15
16
17class ResearchState(TypedDict):
18    query: str
19    sub_questions: list[str]
20    findings: list[dict]
21    report: str
22    iteration: int
23
24
25def search_web(query: str, max_results: int = 4) -> list[dict]:
26    """Swap this for any search API your team already pays for."""
27    resp = httpx.post(
28        "https://api.tavily.com/search",
29        json={"api_key": os.environ["TAVILY_API_KEY"], "query": query, "max_results": max_results},
30        timeout=20,
31    )
32    return [
33        {"title": r["title"], "url": r["url"], "snippet": r["content"]}
34        for r in resp.json().get("results", [])
35    ]
36
37
38def plan_node(state: ResearchState) -> dict:
39    prompt = (
40        f"Break this research question into 3-5 focused sub-questions, "
41        f"one per line, no numbering: {state['query']}"
42    )
43    lines = llm.invoke(prompt).content.strip().split("\n")
44    return {"sub_questions": [l.strip("- ").strip() for l in lines if l.strip()]}
45
46
47def search_node(state: ResearchState) -> dict:
48    researched = {f["sub_question"] for f in state["findings"]}
49    new_findings = list(state["findings"])
50    for q in state["sub_questions"]:
51        if q in researched:
52            continue
53        for r in search_web(q):
54            new_findings.append({"sub_question": q, **r})
55    return {"findings": new_findings}
56
57
58def synthesize_node(state: ResearchState) -> dict:
59    sources = "\n\n".join(
60        f"[{i + 1}] {f['title']} ({f['url']})\n{f['snippet']}"
61        for i, f in enumerate(state["findings"])
62    )
63    prompt = (
64        f"Write a short report answering: {state['query']}\n\n"
65        f"Use ONLY the sources below, cite them inline as [n], and list them "
66        f"at the end. If the sources don't cover something, say so explicitly.\n\n{sources}"
67    )
68    return {"report": llm.invoke(prompt).content}
69
70
71def reflect_node(state: ResearchState) -> dict:
72    return {"iteration": state["iteration"] + 1}
73
74
75def route_after_reflect(state: ResearchState) -> str:
76    if state["iteration"] >= MAX_ITERATIONS:
77        return "done"
78    check = llm.invoke(
79        f"Does this report fully answer '{state['query']}'? Reply only YES or NO.\n\n{state['report']}"
80    ).content
81    return "done" if "YES" in check.upper() else "replan"
82
83
84def build_graph():
85    graph = StateGraph(ResearchState)
86    graph.add_node("plan", plan_node)
87    graph.add_node("search", search_node)
88    graph.add_node("synthesize", synthesize_node)
89    graph.add_node("reflect", reflect_node)
90    graph.add_edge(START, "plan")
91    graph.add_edge("plan", "search")
92    graph.add_edge("search", "synthesize")
93    graph.add_edge("synthesize", "reflect")
94    graph.add_conditional_edges("reflect", route_after_reflect, {"replan": "plan", "done": END})
95    return graph.compile(checkpointer=InMemorySaver())
96
97
98def main():
99    app = build_graph()
100    result = app.invoke(
101        {
102            "query": "Competitive landscape of vertical SaaS AI agents in field services",
103            "sub_questions": [],
104            "findings": [],
105            "report": "",
106            "iteration": 0,
107        },
108        config={"configurable": {"thread_id": "research-1"}},
109    )
110    print(result["report"])
111
112
113if __name__ == "__main__":
114    main()
115