Building Deep Research Agents with LangGraph: From Search to Deep Research

class State(MessagesState):
    next: str
    web_researcher_reports: Annotated[list[dict], operator.add]
    user_query: str = ""
    next_node: str = ""
    payload: any = None

class FanoutState(TypedDict):
    next_node: str
    payload: any

# Create a separate function for dynamic routing
def route_to_next_node(state: FanoutState)->list[Send]:
    """This function returns Send objects for parallel execution"""
    return [
        Send(state['next_node'], {'payload': p})
        for p in state['payload']
    ]

class SearchQuery(BaseModel):
    query: str = Field(description="search query")
    search_reason: str = Field(description="The explaination for why the search is required")


class SearchOutput(TypedDict):
    query: str
    search_reason: str
    search_output: str

class WebSearcherState(TypedDict):
    user_query: str
    follow_up_query: str
    is_sufficient: bool
    search_outputs: Annotated[list[SearchOutput], operator.add]
    retry_count: int = 0

class WebSearchTeam(TeamsBaseClass):
    """
    A class to build a web search team workflow using LangGraph.
    It initializes the search planner, web searcher, and reflections agents,
    and connects them in a workflow.
    """

    def __init__(self, llm:BaseChatModel)->None:
        self.llm = llm
        self.supervisor = None
        self.web_searcher = WebSeacher(self.llm)
        self.search_planner = SearchPlanner(self.llm, self.web_searcher, ["blog posts"])
        self.report_writer = ReportWriter(self.llm)
        self.reflections = Reflections(self.llm)
        self.reflection_router = ReflectionRouter(false_dest=self.search_planner, true_dest=self.report_writer)
        self.graph_builder = StateGraph(WebSearcherState, input_schema=State, output_schema=State)
        self.graph = None
        self.name = "Web_Search_Team"
        self.description = "A team of web researchers who search the web for information and write reports based on the findings. "
        self._build_graph()

    def _build_graph(self)->None:
        self.graph_builder.add_node(self.search_planner.name, self.search_planner)
        self.graph_builder.add_node(self.web_searcher.name, self.web_searcher)
        self.graph_builder.add_node(self.reflections.name, self.reflections)
        self.graph_builder.add_node(self.report_writer.name, self.report_writer)

        self.graph_builder.add_edge(START, self.search_planner.name)

        self.graph_builder.add_conditional_edges(
            self.search_planner.name,
            route_to_next_node,  # This function returns Send objects
            # No need to specify paths since Send handles routing
        )

        self.graph_builder.add_edge(self.web_searcher.name, self.reflections.name)
        self.graph_builder.add_conditional_edges(self.reflections.name, self.reflection_router)
        self.graph_builder.add_edge(self.report_writer.name, END)
        self.graph = self.graph_builder.compile()

class SearchPlanner:
    def __call__(self, state: WebSearcherState)-> FanoutState:

        logger.info("Starting search planning...")
        llm_promopt = self._get_full_prompt(state)
        response = self.llm.with_structured_output(ListOfSearchQueries).invoke(llm_promopt)
        output = { 'payload'  : response.queries,
                   'next_node': self.searcher.name}
        return output

    def _get_full_prompt(self, state: State):
        system_prompt = self._get_system_prompt()
        if state.get("retry_count", 0) == 0:
            query = state["user_query"]
        else:
            query = state["follow_up_query"]

        messages = [{"role": "system", "content": system_prompt}, 
                    {"role": "user", "content": query}]

        return messages

def route_to_next_node(state: FanoutState)->list[Send]:
    """This function returns Send objects for parallel execution"""
    return [
        Send(state['next_node'], {'payload': p})
        for p in state['payload']
    ]

class WebSeacher:
    def __call__(self, state: FanoutState)-> SearchOutput:

        logger.info("Starting web search...")

        q = state['payload']
        query_outputs= []
        user_prompt = {"messages": [{"role": "user", "content": 
                                              f"search query- {q.query}, search_reason- {q.search_reason}"}]}
        response = self.search_agent.invoke(user_prompt)
        output = {"search_outputs": [{
            "query": q.query,
            "search_reason": q.search_reason,
            "search_output": response['messages'][-1].text()
        }]}
        return output

web_serach_tool = TavilySearch(max_results=10)

class Reflections:
    def __call__(self, state: WebSearcherState)-> WebSearcherState:
        logger.info("Starting reflections...")
        llm_promopt = self._get_full_prompt(state)
        response = self.llm.with_structured_output(ReflectionOutcome).invoke(llm_promopt)

        if response.is_sufficient:
            logger.info("The information is sufficient to answer the user's request. Routing to report writer.")
            return {"is_sufficient": True}

        else:
            logger.info(f"Knowledge Gap Identified: {response.knowledge_gap}")
            logger.info(f"Follow-up Queries Generated: {response.follow_up_query}, and routing back to search planner.")
            return {
                "is_sufficient": False,
                "follow_up_query": response.follow_up_query,
                "retry_count": state.get("retry_count", 0) + 1,
            }

class ReflectionRouter:
    def __call__(self, state: WebSearcherState) -> str:
        if state.get("is_sufficient", False):
            return self.true_dest.name
        else:
            return self.false_dest.name

class ReportWriter:
    def __call__(self, state: WebSearcherState)-> State:
        logger.info("Starting report writing...")
        llm_promopt = self._get_full_prompt(state)
        response = self.llm.invoke(llm_promopt)
        return {
            'payload': response.content
        }

class TeamsBaseClass:
    def __call__(self, state: State) -> State:

        logger.info(f"Starting {self.name} workflow...")
        state =  self.graph.invoke(state)
        updated_mesage = state['payload']
        state['payload'] = None
        if self.supervisor:
                return Command(
                    update={
                        "messages": [
                            HumanMessage(
                                content=updated_mesage, name=self.name
                            )
                        ]
                    },
                    goto=self.supervisor,
                    )
        else:
            state['messages'] += [HumanMessage(content=updated_mesage, name=self.name)]
                ## update messages wit the model output and set payload to None
        return state

{
  "user_query": "Compare cloud accounting adoption trends in Australia and list reliable sources.",
  "retry_count": 0
}

{
  "next_node": "Web_Searcher",
  "payload": [
    {"query": "...", "search_reason": "..."},
    {"query": "...", "search_reason": "..."}
  ]
}

{
  "search_outputs": [
    {"query": "...", "search_reason": "...", "search_output": "..."},
    {"query": "...", "search_reason": "...", "search_output": "..."}
  ]
}

{
  "payload": "Final synthesized research report..."
}

You are a web search planner. 
Given a request produce a set of web searches to gather the context needed. Aim for recent 
headlines, news articles and the following areas if there are any:
{search_areas}

Search for only most upto date information as close as possible to the current date {current_date}

Output between 1 and 3 search terms to query for.

You are a research assistant specializing in doing web based research for search term. 
You have access to the following tools:
{tools}

Given a search term, and a reason for that search, carry out the following tasks  
- use given tools to perform  a web based research to retrieve up‑to‑date context, as close as possible to the current date {current_date}.
- Using only the information acquired via the search, produce a short summary of at most 300 words. Don't use any other information
- use only the infomration closer to the current time. 
- Focus on key information that will support the search reason.

You are an expert reviewer tasked with analyzing summaries from web searches to determine if the information gathered sufficiently addresses the user's query. 
You will be provided with:
- The original user request
- A list of web searches already performed and their outputs

Your responsibilities:
- Assess whether the collected information fully answers the user's request
- Make sure the information is not outdated considering the current date {current_time}
- Identify any knowledge gaps or areas requiring further exploration
- Generate clear, actionable follow-up query if needed

Instructions:
- If the provided summaries are sufficient to answer the user's question, do not generate any follow-up queries.
- If there are knowledge gaps, clearly describe what is missing or unclear.
- For each gap, generate a self-contained, specific follow-up query that includes all necessary context for a web search and targets the missing information.
- Ensure follow-up queries do not repeat information already covered by previous searches.

You are a specialist analyst reporter. You will be provided with the original query and 
a set of raw search summaries. Your task is to synthesize these into a short concise summary that is tailored to original
user query. 

# Instructions:
- Use only the information from the search summaries provided, and do not use any other information
- Report should address the question. If there are no sufficent information, you can include that in the reporter
- Use only the most uptodate information as close as possible to the current date, {current_date}, dont use out dated information
- The report should be less than 500 words