检索召回率优化探究二：基于 LangChain 0.3集成 Milvus 2.5向量数据库构建的智能问答系统

背景

       基于 LangChain 0.3集成 Milvus 2.5向量数据库构建的 NFRA（National Financial Regulatory Administration，国家金融监督管理总局）政策法规智能问答系统。在此之前，进行了通过文档分块来实现召回率提升的探究，最终结果是未能实现召回率提升到 85%以上的目标。为此，继续。

具体的代码版本（可见）；检索评估召回率详细说明（可见）

目标

        检索召回率 >= 85%

实现方法

       本次探究：在检索前进行查询问题重写或分解来提升检索的质量，从而提升检索召回率，则对应于 RAG系统整体优化思路图（见下图）的“检索前处理-查询优化”。

执行过程

查询重写

       查询重写，是指将用户原始的查询问题重构为合适的形式，以提高系统检索结果的准确性。

查询重写的实现方式可以是：

1. 通过提示指导大模型重写查询；
2. 通过 LangChain 提供的工具类：RePhraseQueryRetriever

      本次选择了第二种方式，通过封装 RePhraseQueryRetriever类来实现。其实，方式二的实现已包含了方式一，往下看便知晓。

       RePhraseQueryRetriever 的核心在于利用 LLM 对用户的自然语言查询进行重新措辞，使其符合向量存储的查询格式。在这个过程中，LLM 会去掉与检索任务无关的信息，同时优化查询结构、扩展关键词或统一查询格式，以生成更加精确的检索请求。

通过封装 RePhraseQueryRetriever 提供的方法：from_llm() 来实现自定义提示词和大语言模型。代码如下：

def query_rewrite_retriever(retriever: BaseRetriever, model: BaseModel) -> BaseRetriever:
    retriever_from_llm = RePhraseQueryRetriever.from_llm(
        retriever=retriever,
        llm=model,
        prompt=RE_QUERY_PROMPT_TEMPLATE
    )
    return retriever_from_llm

RE_QUERY_PROMPT_TEMPLATE：

re_query_prompt_template = """您是 AI 语言模型助手。您的任务是生成给定用户问题的3个不同问法，用来从矢量数据库中检索相关文档。
通过对用户问题生成多个不同的问法，来帮助用户克服基于内积（IP）的相似性检索的一些限制。提供这些用换行符分隔的替代问题，不要给出多余的回答。
问题：{question}"""
RE_QUERY_PROMPT_TEMPLATE = PromptTemplate(
    template=re_query_prompt_template, input_variables=["question"]
)

查询分解

       查询分解，是指将用户查询问题拆分成多个子问题，以从不同角度探索查询的不同方面，使得检索出的内容更加丰富。

       本次实现同样是选择 LangChain 提供的工具类：MultiQueryRetriever。

       MultiQueryRetriever ，核心是通过大语言模型（LLM）生成多个查询变体，提升召回率，解决单一查询匹配度不足的问题，适用于模糊查询、开放域问答、长文档检索等场景。

从这里可以看出，MultiQueryRetriever 对于此项目来说，适配度不高。不过，还是想通过实践检验一下。

       在项目中，也是对它进行了简单的使用封装，让它更符合项目的需求。

代码如下：

class LineListOutputParser(BaseOutputParser[List[str]]):
    def parse(self, text: str) -> List[str]:
        lines = text.strip().split("\n")
        return list(filter(None, lines))  # 过滤空行

def query_multi_retiever(retriever: BaseRetriever, model: BaseModel) -> BaseRetriever:
    # 定义输出格式
    output_parser = LineListOutputParser()
    # 构建执行链
    llm_chain = MULTI_QUERY_PROMPT_TEMPLATE | model | output_parser

    retriever = MultiQueryRetriever(
        retriever=retriever, llm_chain=llm_chain, parser_key="lines"
    )

    return retriever

MULTI_QUERY_PROMPT_TEMPLATE：

multi_query_prompt_template = """您是 AI 语言模型助手。您的任务是生成给定用户问题的3个不同版本，用来从矢量数据库中检索相关文档。
通过对用户问题生成多个视角，来帮助用户克服基于内积（IP）的相似性搜索的一些限制。提供这些用换行符分隔的替代问题，不要给出多余的回答。
问题：{question}"""
MULTI_QUERY_PROMPT_TEMPLATE = PromptTemplate(
    template=multi_query_prompt_template, input_variables=["question"]
)

这里将查询问题分解成三个，再加上返回相似度排名取的是前三的检索结果，这样一共就会有9个检索结果，内容显得有点多，还会存在重复的分块。于是，结合了检索后处理——重排技术来实现。整体实现的流程如下图：

重排 RRF，（Reciprocal Rank Fusion，翻译为倒数排名融合或者互惠排名融合），用来减少结果的冗余，并统一不同检索方法的评分标准。

代码实现如下：

from langchain.load import dumps
from pymilvus import Hits


def reciprocal_rank_fusion(results: list[Hits], k=60) -> list[tuple]:
    """RRF（Reciprocal Rank Fusion）算法实现
    功能：将多个检索结果列表融合成一个统一的排序列表
    算法原理：
        1. 对于每个检索结果列表中的每个文档
        2. 计算该文档的RRF分数：score = 1 / (rank + k)
        3. 如果同一文档出现在多个列表中，累加其分数
        4. 按最终分数对所有文档进行排序
    优势：
        - rank越小（排名越靠前），分数越高
        - k参数防止分母为0，并调节不同排名之间的差距
        - 多次出现的文档会获得更高的累积分数
    :param results: 多个检索结果 Hits 列表，每个列表包含按相关性排序的文档
    :param k: RRF算法的调节参数，默认值60（经验值）
    :return: list[tuple] 融合后的(Hits, 分数)元组列表，按分数降序排序
    """
    used_scores = {}

    # 遍历该列表中的每个文档
    # for rank, doc in enumerate(docs):
    for rank in range(len(results)):
        hits = results[rank]

        # 将 Hits id 作为唯一标识
        hits_id = dumps(hits['id'])

        # 如果该文档首次出现，初始化分数
        if hits_id not in used_scores:
            used_scores[hits_id] = 0

        # 计算RRF分数并累加
        rrf_score = 1 / (rank + k)
        used_scores[hits_id] += rrf_score

    # 按分数降序排序
    reranked_results = [
        (key_id, score)
        for key_id, score in sorted(used_scores.items(), key=lambda x: x[1], reverse=True)
    ]

    # 创建 id 到 hit 对象的映射字典，注意属性类型
    id_to_hit = {str(hit.id): hit for hit in results}

    # 替换元组中的 id 为对应的 hit 对象
    result_list = []
    for id_val, score in reranked_results:
        if id_val in id_to_hit:
            result_list.append((id_to_hit[id_val], score))

    print(f"RRF融合完成，共 {len(reranked_results)} 个唯一文档")

    return result_list


def get_top_n_rrf(results: list[Hits], k=60, top_n=3) -> list[tuple]:
    """将多个检索结果列表融合成一个统一的排序列表，返回 top_n 个（默认为前3个）
    :param results: 多个检索结果 Hits列表，每个列表包含按相关性排序的文档
    :param k: RRF算法的调节参数，默认值60（经验值）
    :param top_n: 融合后按分数降序排序需返回的前 n 个
    :return: 融合后的(Hits, 分数)元组列表，按分数降序排序的前 top_n 个
    """
    reranked_results = reciprocal_rank_fusion(results, k)
    if len(reranked_results) <= top_n:
        return reranked_results
    else:
        return reranked_results[:top_n]

       在这里顺带说一下，不管是网上，还是源码中，都提到了 MultiQueryRetriever 的检索结果，是会不同查询变体检索到的文本分块进行合并且唯一。不过，大家在使用的过程中，要注意检索结果的类型是否符合源码的入参。

       比如，项目中使用了 milvus 的 collection.search()，返回的是 Hits 对象的集合，而不是需要的入参类型 document 集合。因此，得到的结果不会是真的“唯一”。

以下是类 MultiQueryRetriever 实现合并检索结果唯一性的方法源码：

def _unique_documents(documents: Sequence[Document]) -> List[Document]:
    return [doc for i, doc in enumerate(documents) if doc not in documents[:i]]

检索评估（召回率）

查询重写检索评估

RAG 相关处理说明：

切分策略：分块大小： 500； 分块重叠大小： 100； 使用正则表达式，[r"第\S*条 "]
嵌入模型：模型名称： BAAI/bge-base-zh-v1.5 (使用归一化)
向量存储：向量索引类型：IVF_FLAT （倒排文件索引+精确搜索）；

向量度量标准类型：IP（内积）； 聚类数目： 100； 存储数据库： Milvus
向量检索：查询时聚类数目： 10； 检索返回最相似向量数目： N

检索返回最相似向量数目：N = 2

检索结果如下表：

表格说明：

• TOP1、TOP2 个数：是指检索回来的文本块（被最终用于回复问题的文本块在检索返回时相似度的排名）的数量。越是位于 TOP1，说明检索效率越高；
• TOP N 策略：就是在问题检索时，需要返回最相似向量个数。（本次评估，N=2）

检索返回最相似向量数目：N = 3

检索结果如下表：

表格说明：

• TOP1、TOP2 、TOP3个数：是指检索回来的文本块（被最终用于回复问题的文本块在检索返回时相似度的排名）的数量。越是位于 TOP1，说明检索效率越高；
• TOP N 策略：就是在问题检索时，需要返回最相似向量个数。（本次评估，N=3）

从两次不同的表格数据对比来看，检索返回最相似向量数目 N = 3 比 N = 2，最终的检索召回率，是高的，是有一定提升的。

       但是，TOP 1、2的数目是有波动的，这个应是查询问题重写的不稳定性造成的，这也是使用 LLM 对查询问题进行重写不可避免的。在实际的使用中，我们就要注意进行多次评估，来得到一个较为稳定的波动范围。

看到 TOP 3，能得到一个不错的召回率提升。为此，又进行了一个对比评估。

RAG 相关处理如上，N = 3，但是，不做查询问题的重写，直接使用查询问题进行检索。

最终的检索结果，检索召回个数为：70个，检索召回率是：84.34%

       当看到这样一个结果时，整个人都有点懵了，辛辛苦苦地进行问题的重写，得到的一个提升，还比直接查询得到的召回率差点。而且，对查询问题使用 LLM 进行重写，还会消耗一定的 token，并且还会在一定程度上延长了整个过程的总时间。

       后面的查询分解，差点都不想进行试验了，因为它同样也是会用到 LLM，只不过使用的方式有所不同，也结合了检索后处理技术——RRF。

       不过，作为以学习为目的，是要继续的。

查询分解检索评估

RAG 相关处理说明：

切分策略：分块大小： 500； 分块重叠大小： 100； 使用正则表达式，[r"第\S*条 "]
嵌入模型：模型名称： BAAI/bge-base-zh-v1.5 (使用归一化)
向量存储：向量索引类型：IVF_FLAT （倒排文件索引+精确搜索）；

向量度量标准类型：IP（内积）； 聚类数目： 100； 存储数据库： Milvus
向量检索：查询时聚类数目： 10； 检索返回最相似向量数目： 3

检索结果如下表：

表格说明，同上述检索返回最相似向量数目 N=3一样。

        不做查询问题处理，TOP 3得到的检索召回率是 84.34%，而做了查询分解和检索后处理（RRF重排）得到的是 79.52%。考虑到 LLM 具有不确定性，而这也只是一次的检索结果而已。为此，是不能下定论的。

检索评估小结

       根据上述不同的查询问题处理方法，检索召回率得有一定的提升。

• 使用 LangChain 提供的工具类 RePhraseQueryRetriever，进行查询重写，在检索返回最相似向量数目N 为 3 时，得到的检索召回率还是不错的，83.13%；
• 使用 LangChain 提供的 MultiQueryRetriever，进行查询分解，在检索返回最相似向量数目N ，同样是 3 时，得到的检索召回率就不理想了，仅为：79.52%。

它们两个使用的 LLM 都是 deepseek-chat，主要的不同是 prompt，后者的 prompt 偏向多样性，引导大语言模型从不同视角对查询问题进行重写；而前者是偏向同一个意思下的不同问法，偏向的是不同表述方法。

总结

       本次探究结果仍是未达到预期的目标。而本次的收获是：

• LangChain 提供的 MultiQueryRetriever，不适合对领域准确性要求较高的场景，比如政策法规、法律条文等的检索。（当然，这里不包括查询变体所用模型是经过领域微调的，主要是指通用的 LLM）
• 合适的检索返回最相似向量数目 N，会使召回率得到一定程度上的提升。

       接下来，会继续按 RAG系统整体优化思路图进行优化，提升检索召回率。在大方向上，还是会集中在检索前处理，而重点是在 RePhraseQueryRetriever 的进一步探究上，此次是跑通试验，缺乏太多细节上的考虑，比如重写模型的选择，重写之后的问题分析等等。

文中基于的项目代码地址：https://gitee.com/qiuyf180712/rag_nfra

本文关联项目的文章：RAG项目实战：LangChain 0.3集成 Milvus 2.5向量数据库，构建大模型智能应用-CSDN博客