智能体之：ReAct、DeepSearch、DeepResearch详解

共计 5789 个字符，预计需要花费 15 分钟才能阅读完成。

自从OpenAI的o1系列、深度求索的Deepseek-r1系列的推理模型推出，随后，推理时计算的概念越发火热，推动智能体的进一步发展，本篇文章主要记录智能体中的核心问题：如何让大模型解决复杂问题。

ReAct的概念早在2023年由Google发表的一篇论文中提出：REACT: SYNERGIZING REASONING AND ACTING IN
LANGUAGE MODELS。论文中对比了普通的CoT方法与ReAct方法，其中ReAct的核心概念为：Reasoning、Acting、Observe的步骤，分别是：推理、执行、观察的循环。
与CoT相比，单纯的在生成过程中让模型推理，模型会受限于生成长度过长会导致模型性能下降，以及在推理时，只能单次交互无法借助大模型外的搜索、计算等工具，导致CoT在非实时性且复杂度不高的问题中表现较好，但需借助外部工具获取实时信息或者与外界交互获取信息时，无法满足需求。
ReAct思想的提出，使智能体能够“自主规划”，自主探索复杂问题解决路径，同时可利用多种外部工具获取资源。

在ReAct思路提出的基础上，结合今年以来的各项新技术，包括更强的推理模型、MCP协议、向量搜索等技术，各大厂商都在竞相推出DeepSearch产品，在宣传上，Deepseek模型的火热促进了DeepXXX系列的产品出圈，但是我们看核心问题仍然逃不开ReAct的核心思想，仍然是推理、执行、观察的循环。
DeepSearch显然是基于Search的思路，笔者认为AI搜索的路线从最开始的RAG路线，到最近的DeepSearch路线，都是尽量在压缩信息，将最核心内容直接呈现给用户，而抛弃以往的给出N个最相关的搜索页面，让用户自己理解，使得人类获取信息、解决问题的效率大大提高，这也是大模型技术发展的重要因素。
常见的产品思路是：用户提出一个复杂的问题，经过大模型进行问题分解、判断是否利用工具、利用工具获取信息、大模型基于获取到的信息判断是否需要额外信息、继续选择合适的工具获取额外信息、如此循环直到大模型认为当前获取到的信息足以解决用户的原始复杂问题，生成最终的答案。

举个例子，利用高德地图的MCP工具，提供了：基于名字查询坐标、基于坐标查询附近地址、基于坐标的路径规划等一系列地图功能，我们把MCP工具开放给大模型，并提出问题：查询成都市银泰城的坐标以及附近的酒店，帮我计算一下步行到酒店的形成，大模型会依次利用MCP工具解决一系列子问题，比如：查询银泰城的坐标、基于银泰城的坐标查询附近1公里的酒店信息、计算最近的酒店与银泰城的步行路线。最终生成答案。

名字同上，都是DeepXXX系列，而且笔者认为Re+Search名字非常妙。“Research”，顾名思义即为研究。DeepResearch是在DeepSearch的基础上，帮用户解决一系列的研究性问题。比如：帮我写一篇关于人类起源的文章。该指令会将原始问题分解为N个大纲问题，再基于大纲问题分解子问题，分解粒度可以自由控制，再基于DeepSearch的方法，依次解决相应的子问题，最终基于子问题生成行文连贯、风格统一的长篇文章，所以DeepResearch的核心在于保持长上下文的关联系、前后段落之间的语义连贯性、整篇文章的结构性以及全文统一的行文风格。

GitHub：deep-researcher

该项目采用了ReAct的方法，搭建了一套QA问答系统。
核心代码在deepsearcher/agent/deep_search.py#L206：

    async def async_retrieve(
        self, original_query: str, **kwargs
    ) -> Tuple[List[RetrievalResult], int, dict]:
        max_iter = kwargs.pop("max_iter", self.max_iter)
        ### SUB QUERIES ###
        log.color_print(f"<query> {original_query} </query>\n")
        all_search_res = []
        all_sub_queries = []
        total_tokens = 0

        sub_queries, used_token = self._generate_sub_queries(original_query)
        total_tokens += used_token
        if not sub_queries:
            log.color_print("No sub queries were generated by the LLM. Exiting.")
            return [], total_tokens, {}
        else:
            log.color_print(
                f"<think> Break down the original query into new sub queries: {sub_queries}</think>\n"
            )
        all_sub_queries.extend(sub_queries)
        sub_gap_queries = sub_queries

        for iter in range(max_iter):
            log.color_print(f">> Iteration: {iter + 1}\n")
            search_res_from_vectordb = []
            search_res_from_internet = []  # TODO

            # Create all search tasks
            search_tasks = [
                self._search_chunks_from_vectordb(query, sub_gap_queries)
                for query in sub_gap_queries
            ]
            # Execute all tasks in parallel and wait for results
            search_results = await asyncio.gather(*search_tasks)
            # Merge all results
            for result in search_results:
                search_res, consumed_token = result
                total_tokens += consumed_token
                search_res_from_vectordb.extend(search_res)

            search_res_from_vectordb = deduplicate_results(search_res_from_vectordb)
            # search_res_from_internet = deduplicate_results(search_res_from_internet)
            all_search_res.extend(search_res_from_vectordb + search_res_from_internet)
            if iter == max_iter - 1:
                log.color_print("<think> Exceeded maximum iterations. Exiting. </think>\n")
                break
            ### REFLECTION & GET GAP QUERIES ###
            log.color_print("<think> Reflecting on the search results... </think>\n")
            sub_gap_queries, consumed_token = self._generate_gap_queries(
                original_query, all_sub_queries, all_search_res
            )
            total_tokens += consumed_token
            if not sub_gap_queries or len(sub_gap_queries) == 0:
                log.color_print("<think> No new search queries were generated. Exiting. </think>\n")
                break
            else:
                log.color_print(
                    f"<think> New search queries for next iteration: {sub_gap_queries} </think>\n"
                )
                all_sub_queries.extend(sub_gap_queries)

        all_search_res = deduplicate_results(all_search_res)
        additional_info = {"all_sub_queries": all_sub_queries}
        return all_search_res, total_tokens, additional_info

可以对照着提示词看：
首先基于原始问题，生成最多4个子问题，提示词为：SUB_QUERY_PROMPT。

SUB_QUERY_PROMPT = """To answer this question more comprehensively, please break down the original question into up to four sub-questions. Return as list of str.
If this is a very simple question and no decomposition is necessary, then keep the only one original question in the python code list.

Original Question: {original_query}

<EXAMPLE>
Example input:
"Explain deep learning"

Example output:
[
    "What is deep learning?",
    "What is the difference between deep learning and machine learning?",
    "What is the history of deep learning?"
]
</EXAMPLE>

Provide your response in a python code list of str format:
"""

然后基于4个子问题，在向量库里做召回。召回出来内容，依次经过大模型判断是否跟【当前子问题+所有子问题】相关，使用RERANK_PROMPT，相关则保留，不相关则丢弃。

RERANK_PROMPT = """Based on the query questions and the retrieved chunk, to determine whether the chunk is helpful in answering any of the query question, you can only return "YES" or "NO", without any other information.

Query Questions: {query}
Retrieved Chunk: {retrieved_chunk}

Is the chunk helpful in answering the any of the questions?
"""

将所有相关的召回内容拼在一起形成子问题的全部召回内容。
然后基于REFLECT_PROMPT提示词，将原始问题、所有子问题、所有召回的信息让大模型判断是否还需要额外的子问题，如果还需要额外信息，则输出最多3个子查询，如果没有则进入答案总结阶段。

REFLECT_PROMPT = """Determine whether additional search queries are needed based on the original query, previous sub queries, and all retrieved document chunks. If further research is required, provide a Python list of up to 3 search queries. If no further research is required, return an empty list.

If the original query is to write a report, then you prefer to generate some further queries, instead return an empty list.

Original Query: {question}

Previous Sub Queries: {mini_questions}

Related Chunks: 
{mini_chunk_str}

Respond exclusively in valid List of str format without any other text."""

同样的，也是设置最大迭代次数或者让大模型判断不需要额外信息之后停止迭代。
最后将原始问题、子问题、召回的信息，利用SUMMARY_PROMPT提示词模板让大模型总结答案。

SUMMARY_PROMPT = """You are a AI content analysis expert, good at summarizing content. Please summarize a specific and detailed answer or report based on the previous queries and the retrieved document chunks.

Original Query: {question}

Previous Sub Queries: {mini_questions}

Related Chunks: 
{mini_chunk_str}
"""

GitHub：graph-rag-agent

这个项目非常有意思，使用了GraphRAG作为Agent的知识库，融合了GraphRAG、LightRAG、Neo4j-llm-graph-builder进行知识图谱构建以及搜索，在信息获取方面给智能体提供了更全面的信息，同时结合DeepSearch技术实现知识推理功能。
作为熟悉GraphRAG、DeepSearch技术的具体实现来讲，可以仔细看看其核心代码。