摆脱传统的字面匹配，RocketQA语义检索效果YYDS

 技术原理 
文本搜索
传统的文本搜索系统通过统计学的方法对文本进行特征度量，但这仅考虑了文本间的字面特征，无法充分利用文本中蕴含的丰富的语义知识。因此，在信息搜索领域，如果能够通过编码器将文本映射到具体的语义空间，每段文本对应在该空间下的向量，那么信息搜索问题便可转化为在给定空间下向量间的最近邻搜索问题。例如，DPR（dense passage retriever）通过将文本从高维空间下的稀疏向量，转化为富含语言知识和文本信息的低维空间下的稠密向量，来实现文本的近似搜索，在问答系统、搜索系统等多个领域得到了应用。
预训练语言模型
BERT 系列模型在预训练阶段从大量未标注文本中学习到了丰富的语义知识。较之于自然语言处理领域发展阶段的第二范式，即通过设计模型的拓扑结构来提升模型在某个领域的性能，预训练语言模型（PLM）保证了在较少数据、较低算力的情况下，仅通过对特定任务的微调，便可实现在该任务下 state-of-the-art 的性能。模型蕴含的语义知识使得下游任务不再需要去从头设计一套模型结构并使用大量的任务相关数据重新训练。
Dual encoder 架构使用两个 BERT 模型，query model 和 context model 分别将问题和对应的待查找文本映射到768维的语义空间。为保证在实际检索过程中的准确度，该模型的训练策略为：尽可能的使问题与对应的文本间的距离接近，而与无关的文本间的距离拉远。Cross encoder 则使用一个 BERT 模型同时对问题和文本进行编码。

RocketQA 为百度对 DPR（Dense Passage Retrieval）的训练策略进行优化和改进得到的模型。一般的训练策略为：对于给定问题文本，使其在空间中的表示尽可能靠近正样本同时远离负样本。In batch negatives 训练策略则将同一批次内除当前问题的正样本之外的其他样本均视为负样本（包括当前问题的负样本，和其它问题的正、负样本）。相比于在同一批次内进行采样，RocketQA 基于飞桨的分布式训练能力，使用了跨批次的负采样策略。实验证明，适当增大 batch size 可以较好的提升模型的性能。除此之外，考虑到在实际应用中，训练数据存在漏标、错标等情况，准确率不高。为尽可能降低假负例对模型效果的影响，RocketQA 使用 cross model 对文本进行打分，借此来筛去部分不符合要求的标注数据。同时，RocketQA 也通过使用交互模型来得到更多相关的弱监督数据帮助其训练。

• 戳官方 Repo 了解 RocketQA 详情：

https://github.com/PaddlePaddle/RocketQA

系统架构工具

JINA 提供了一整套搭建搜索系统的开源工具，其主要产品为 Document，Executor 和 Flow。Document 为基础数据结构，Executor 负责对 Document 进行处理，Flow 则负责搭建整个工作流程。

 项目实现 

本文使用百度飞桨框架和 RocketQA 模型，基于 JINA 全家桶构建了含有 Retrieve、Rerank 两个阶段的文本召回系统。接下来将详述项目实现过程。

文本问答的主要流程为：召回（retriever），重排（reranker）和阅读理解（reader）。在召回阶段，本项目使用 RocketQA（Dual model）将问题文本映射为向量的形式，之后，近似最近邻搜索系统会在已有的索引库中搜索与之距离最近的 top-k 向量，并返回符合阈值条件的候选向量；在重排阶段，则会使用 RocketQA 的 Cross model 对所有召回向量进行打分，最后根据排序返回最终得分前三的文本。

首先使用 RocketqaDe 模型来实现第一阶段。在“/index“阶段建立索引库时，Executor将输入的“.content”文本编码为语义向量，并将其嵌入 Document 中；在“/search”阶段搜索时，Executor 将输入的待查询问题编码为语义向量。

class RocketqaDeExecutor(Executor):
    def __init__(self,model_name="zh_dureader_de",use_Cuda=True,device_Id=0,batch_Size=32,*args,**kwargs):
        super().__init__(*args, **kwargs)
        self.model = rocketqa.load_model(model=model_name,use_cuda=True,device_id=device_Id,batch_size=batch_Size)

    @requests(on="/index")
    def encode_passage(self,docs:DocumentArray,**kwargs):

        embeddings = self.model.encode_para(para=docs.texts)
        docs.embeddings = [embedding for embedding in embeddings]

    @requests(on="/search")
    def encode_query(self,docs,**kwargs):
        print("retriever is working......")
        start = time.time()
        for doc in  docs:
            generator_temp = self.model.encode_query(query=[doc.text])
            for temp in generator_temp:
                doc.embedding = temp
        end = time.time()
        print("retrieve time: ",end-start,"s")

使用 RocketqaCe 模型来实现第二阶段的打分重排。在”/search”阶段，该 Executor 将前一步处理后得到嵌入文本向量与问题向量一起处理，得到每条召回结果的分数，并根据分数具体排名。在”/index”阶段，该 Executor 不参与工作。

class RocketqaCeExecutor(Executor):
    def __init__(self,model_Name="zh_dureader_ce",use_Cuda=True,device_Id=0,batch_Size=32,*args,**kwargs):
        super().__init__(*args,**kwargs)
        self.model = rocketqa.load_model(model=model_Name,use_cuda=True,device_id=device_Id,batch_size=batch_Size)

    @requests(on="/search")
    def rerank(self,docs,**kwargs):
        print("reranker is working......")
        print("召回结果排序中......")
        start = time.time()
        for doc in docs:
            str_list = []
            for m in doc.matches:
                str_list.append(m.text)
            doc.matches = []
            scores = []
            score_generator = self.model.matching(query=[doc.text]*len(str_list),para=str_list)
            for g in score_generator:
                scores.append(g)

            scores = np.array(scores).argsort()
            doc.matches.append(Document(text=str_list[scores[-1]]))
            doc.matches.append(Document(text=str_list[scores[-2]]))
            doc.matches.append(Document(text=str_list[scores[-3]]))
        end = time.time()
        print("rerank time:",end-start,"s")

即使在使用 GPU 的情况下，每次建立索引也会消耗大量时间。本项目使用以下函数来实现在运行时单独执行建立索引库和搜索阶段。

def main(order):
    if order == 'index':
        if Path('./workspace').exists():
            print('./workspace exists, please deleted it if you want to reindexi')
            return 0
        data_path = sys.argv[2]
        if data_path is None:
            print("No data_path!")
        index(data_path)
    elif order == 'query':
        query()

def index(path):
    with test_flow:
        test_flow.index(inputs=read_file(path), show_progress=True)

def query():
    with test_flow:

        while(True):
            query = input("请输入查询选项：")
            if query == "exit":
                break
            query = Document(text=query)
            docs = test_flow.search(inputs=query)

            matches = docs[0].matches
            print("搜索答案为：")
            ids = 1
            for match in matches:
                print("推荐答案排行，NO.",ids)
                print(match.text)
                ids = ids + 1

 效果展示 
运行 python wow.py index data_path 来建立索引库：

运行 python wow.py query 来执行搜索：

 总结 

本文关注稠密文本召回 DPR（Dense Passage Retrieval）技术在文本搜索和人机问答领域的应用，使用百度飞桨框架和 RocketQA 模型，基于 JINA 全家桶构建了含有 Retrieve、Rerank 两个阶段的文本召回系统。本项目在建库和检索阶段，通过使用预训练语言模型蕴含的丰富的语义知识来对输入文本和输入问题进行编码，能够免去传统的手动标记或词频统计等过程。在后续，本项目使用近似最近邻搜索算法来完成召回，使用交互式模型来完成最终的打分。该文本召回系统将搜索过程简洁化，同时提供了人工智能技术在落地应用方面的典型范例。

此外，飞桨自然语言处理模型库 PaddleNLP 也基于 RocketQA 等前沿模型搭建了完整的检索系统、问答系统，亲测好用。传送门：

• https://github.com/PaddlePaddle/PaddleNLP/tree/develop/applications/experimental/pipelines
• https://github.com/PaddlePaddle/PaddleNLP/tree/develop/applications/neural_search

戳 GitHub 主页与我交流：

https://github.com/Elvisambition