增强常见问题解答搜索引擎：在 Elasticsearch 中利用 KNN 的力量

在快速准确的信息检索至关重要的时代，开发强大的搜索引擎至关重要。随着大型语言模型和信息检索架构（如 RAG）的出现，在现代软件系统中利用文本表示（向量/嵌入）和向量数据库已变得越来越流行。在本文中，我们深入研究了如何使用 Elasticsearch 的 K 最近邻 (KNN) 搜索和来自强大语言模型的文本嵌入，这是一个强大的组合，有望彻底改变我们访问常见问题 (FAQ) 的方式。通过对 Elasticsearch 的 KNN 功能的全面探索，我们将揭示这种集成如何使我们能够创建尖端的常见问题解答搜索引擎，通过以闪电般的延迟理解查询的语义上下文，从而增强用户体验。

在开始设计解决方案之前，让我们了解信息检索系统中的一些基本概念。

文本表示（嵌入）

你可以通过阅读 “Elasticsearch：什么是向量和向量存储数据库，我们为什么关心？” 来了解更多的关于文本嵌入的知识。

嵌入是一条信息的数字表示，例如文本、文档、图像、音频等。该表示捕获了所嵌入内容的语义，使其对于许多行业应用程序来说都是稳健的。

语义搜索

传统的搜索系统使用词法匹配来检索给定查询的文档。语义搜索旨在使用文本表示（嵌入）来理解查询的上下文，以提高搜索准确性。

语义搜索的类型

对称语义搜索：查询和搜索文本长度相似的搜索用例。例如在数据集中找到类似的问题。
非对称语义搜索：查询和搜索文本长度不同的搜索用例。例如查找给定查询的相关段落。

向量搜索引擎（向量数据库）

向量搜索引擎是专用数据库，可用于将图像、文本、音频或视频等非结构化信息存储为嵌入或向量。在本文中，我们将使用 Elasticsearch 的向量搜索功能。

现在我们了解了搜索系统的构建块，让我们深入了解解决方案架构和实现。

搜索解决方案的第一步是将问题-答案对索引到 Elasticsearch 中。我们将创建一个索引并将问题和答案嵌入存储在同一索引中。我们将根据检索的特征使用两个独立的模型来嵌入问题和答案。
我们将使用步骤 1 中使用的相同模型来嵌入查询，并形成搜索查询（3 个部分，即问题、答案、词汇搜索），将查询嵌入映射到相应的问题和答案嵌入。
我们还将为查询的每个部分提供一个提升值，以表示它们在组合中的重要性。返回的最终结果根据分数总和乘以各自的提升值进行排名。

环境设置

要使用 docker 安装 Elasticsearch，请参阅这篇有关如何设置单节点集群的详细文章。如果你已有集群，请跳过此步骤。如果你想详细了解如何安装 Elasticsearch，请参考文章 “如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch”。在本演示中，我们将使用 Elastic Stack 8.10.4 来进行展示。

设置你的索引。你可以使用以下映射作为起点。我们在 Kibana 的 Dev Tools 中打入如下的命令：

PUT faq-index
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "Question": {
        "type": "text"
      },
      "Answer": {
        "type": "text"
      },
      "question_emb": {
        "type": "dense_vector",
        "dims": 768,
        "index": true,
        "similarity": "dot_product"
      },
      "answer_emb": {
        "type": "dense_vector",
        "dims": 1024,
        "index": true,
        "similarity": "dot_product"
      }
    }
  }
}

模型选择

由于我们使用相当通用的语言处理数据，因此为了进行本实验，我从 MTEB 排行榜的检索（用于答案）和 STS（用于问题）部分中选择了表现最好的模型。

选定型号：

答案：BAAI/bge-large-en-v1.5（您可以使用量化版本以加快推理速度）
如有问题：thenlper/gte-base

如果你有特定领域的常见问题解答并想要检查哪种模型表现最好，你可以使用 Beir。查看本节，其中描述了如何加载自定义数据集以进行评估。

实现

出于本实验的目的，我将使用 Kaggle 的心理健康常见问题解答数据集。

安装所需要的模块

pips install sentence_transformers

1. 装载数据

import pandas as pd
data = pd.read_csv('Mental_Health_FAQ.csv')

2. 生成嵌入

Questions：

from sentence_transformers import SentenceTransformer
question_emb_model = SentenceTransformer('thenlper/gte-base')

data['question_emb'] = data['Questions'].apply(lambda x: question_emb_model.encode(x, normalize_embeddings=True))

注意：我们对嵌入进行归一化，以使用点积作为相似性度量而不是余弦相似性。该计算速度更快，并且在 Elasticsearch 密集向量场文档中得到推荐。

Answers:

answer_emb_model = SentenceTransformer('BAAI/bge-large-en-v1.5')
data['answer_emb'] = data['Answers'].apply(lambda x: answer_emb_model.encode(x, normalize_embeddings=True))

3. 索引文档

我们将使用 Elasticsearch helper 函数。具体来说，我们将使用 streaming_bulk API 来索引我们的文档。

首先，让我们实例化 elasticsearch python 客户端。

我们首先需要把安装好的 Elasticsearch 的证书拷贝到当前目录中：

$ pwd
/Users/liuxg/python/faq
$ cp ~/elastic/elasticsearch-8.10.4/config/certs/http_ca.crt .
$ ls
Mental Health FAQ.ipynb archive (13).zip
Mental_Health_FAQ.csv   http_ca.crt

然后我们打入如下的代码：

from elasticsearch import Elasticsearch

from ssl import create_default_context

context = create_default_context(cafile=r"./http_ca.crt")
es = Elasticsearch('https://localhost:9200',
    basic_auth=('elastic', 'YlGXk9PCN7AUlc*VMtQj'),
    ssl_context=context,
)

接下来，我们需要创建一个可以输入到流式 bulk API 中的文档生成器。

index_name="faq-index"
def generate_docs():
    for index, row in data.iterrows():
        doc = {
                "_index": index_name,
                "_source": {
                    "faq_id":row['Question_ID'],
                    "question":row['Questions'],
                    "answer":row['Answers'],
                    "question_emb": row['question_emb'],
                    "answer_emb": row['answer_emb']
                },
            }

        yield doc

最后，我们可以索引文档。

import tqdm
from elasticsearch.helpers import streaming_bulk
number_of_docs=len(data)
progress = tqdm.tqdm(unit="docs", total=number_of_docs)
successes = 0
for ok, action in streaming_bulk(client=es, index=index_name, actions=generate_docs()):
    progress.update(1)
    successes += ok

print("Indexed %d/%d documents" % (successes, number_of_docs))

4. 查询文档

def faq_search(query="", k=10, num_candidates=10):
    
    if query is not None and len(query) == 0:
        print('Query cannot be empty')
        return None
    else:
        query_question_emb = question_emb_model.encode(query, normalize_embeddings=True)

        instruction="Represent this sentence for searching relevant passages: "

        query_answer_emb = answer_emb_model.encode(instruction + query, normalize_embeddings=True)

        payload = {
          "query": {
            "match": {
              "title": {
                "query": query,
                "boost": 0.2
              }
            }
          },
          "knn": [ {
            "field": "question_emb",
            "query_vector": query_question_emb,
            "k": k,
            "num_candidates": num_candidates,
            "boost": 0.3
          },
          {
            "field": "answer_emb",
            "query_vector": query_answer_emb,
            "k": k,
            "num_candidates": num_candidates,
            "boost": 0.5
          }],
          "size": 10,
          "_source":["faq_id","question", "answer"]
        }

        response = es.search(index=index_name, body=payload)['hits']['hits']

        return response

按照模型页面上的说明，我们需要在将查询转换为嵌入之前将指令附加到查询中。此外，我们使用模型的 v1.5，因为它具有更好的相似度分布。查看型号页面上的常见问题解答以了解更多详细信息。

评估

为了了解所提出的方法是否有效，根据传统的 KNN 搜索系统对其进行评估非常重要。让我们尝试定义这两个系统并评估所提出的系统。

系统 1：非对称 KNN 搜索（查询和答案向量）。
系统2：查询（BM25）、非对称KNN搜索（查询和答案向量）和对称KNN搜索（查询和问题向量）的组合。

为了评估系统，我们必须模仿用户如何使用搜索。简而言之，我们需要从源问题生成与问题复杂性相似的释义问题。我们将使用 t5-small-finetuned-quora-for-paraphrasing 微调模型来解释问题。

让我们定义一个可以生成释义问题的函数。

from transformers import AutoModelWithLMHead, AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("mrm8488/t5-small-finetuned-quora-for-paraphrasing")
model = AutoModelWithLMHead.from_pretrained("mrm8488/t5-small-finetuned-quora-for-paraphrasing")

def paraphrase(question, number_of_questions=3, max_length=128):
    input_ids = tokenizer.encode(question, return_tensors="pt", add_special_tokens=True)

    generated_ids = model.generate(input_ids=input_ids, num_return_sequences=number_of_questions, num_beams=5, max_length=max_length, no_repeat_ngram_size=2, repetition_penalty=3.5, length_penalty=1.0, early_stopping=True)

    preds = [tokenizer.decode(g, skip_special_tokens=True, clean_up_tokenization_spaces=True) for g in generated_ids]

    return preds

现在我们已经准备好了释义函数，让我们创建一个评估数据集，用于测量系统的准确性。

最后，我们将修改 “faq_search” 函数以返回各个系统的 faq_id。

对于系统 1：

def get_faq_id_s1(query="", k=5, num_candidates=10):
    
    if query is not None and len(query) == 0:
        print('Query cannot be empty')
        return None
    else:
        instruction="Represent this sentence for searching relevant passages: "

        query_answer_emb = answer_emb_model.encode(instruction + query, normalize_embeddings=True)

        payload = {
          "knn": [
          {
            "field": "answer_emb",
            "query_vector": query_answer_emb,
            "k": k,
            "num_candidates": num_candidates,
          }],
          "size": 1,
          "_source":["faq_id"]
        }

        response = es.search(index=index_name, body=payload)['hits']['hits']

        return response[0]['_source']['faq_id']

对于系统 2：

def get_faq_id_s2(query="", k=5, num_candidates=10):
    
    if query is not None and len(query) == 0:
        print('Query cannot be empty')
        return None
    else:
        query_question_emb = question_emb_model.encode(query, normalize_embeddings=True)

        instruction="Represent this sentence for searching relevant passages: "

        query_answer_emb = answer_emb_model.encode(instruction + query, normalize_embeddings=True)

        payload = {
          "query": {
            "match": {
              "title": {
                "query": query,
                "boost": 0.2
              }
            }
          },
          "knn": [ {
            "field": "question_emb",
            "query_vector": query_question_emb,
            "k": k,
            "num_candidates": num_candidates,
            "boost": 0.3
          },
          {
            "field": "answer_emb",
            "query_vector": query_answer_emb,
            "k": k,
            "num_candidates": num_candidates,
            "boost": 0.5
          }],
          "size": 1,
          "_source":["faq_id"]
        }

        response = es.search(index=index_name, body=payload)['hits']['hits']

        return response[0]['_source']['faq_id']

注意：boost 值是实验性的。为了这个实验的目的，我根据组合中各个字段的重要性进行了划分。搜索中每个字段的重要性完全是主观的，可能由业务本身定义，但如果不是，系统的一般经验法则是 Answer 向量 > Question 向量 > 查询。

好的！我们一切准备就绪，开始我们的评估。我们将为两个系统生成一个预测列，并将其与原始 faq_id 进行比较。

eval_data['PRED_FAQ_ID_S1'] = eval_data['Question'].apply(get_faq_id_s1)

from sklearn.metrics import accuracy_score

ground_truth = eval_data["FAQ_ID"].values
predictions_s1 = eval_data["PRED_FAQ_ID_S1"].values

s1_accuracy = accuracy_score(ground_truth, predictions_s1)

print('System 1 Accuracy: {}'.format(s1_accuracy))

eval_data['PRED_FAQ_ID_S2'] = eval_data['Question'].apply(get_faq_id_s2)

predictions_s2 = eval_data["PRED_FAQ_ID_S2"].values

s2_accuracy = accuracy_score(ground_truth, predictions_s2)

print('System 2 Accuracy: {}'.format(s2_accuracy))

通过所提出的系统，我们可以看到与非对称 KNN 搜索相比，准确率提高了 7-11%。

我们还可以尝试 ramsrigouthamg/t5_paraphraser，但该模型生成的问题有点复杂和冗长（尽管在上下文中）。

你还可以使用 LLM 生成评估数据集并检查系统的性能。

准确性的提高是主观的，取决于查询的质量，即查询的上下文有多丰富、嵌入的质量和/或使用搜索的用户类型。为了更好地理解这一点，让我们考虑两种最终用户：

想要了解有关您的产品和服务的一些事实的一般用户：在这种情况下，上述系统会做得很好，因为问题简单、直观且上下文充分。
领域/产品特定用户，例如想要了解产品的一些复杂细节以设置系统或解决某些问题的工程师：在这种情况下，查询在词汇组成方面更具特定于领域，因此开箱即用的模型嵌入将无法捕获所有上下文。那么，我们该如何解决这个问题呢？系统的架构将保持不变，但可以通过使用特定领域数据（或预先训练的特定领域模型）微调这些模型来提高搜索系统的整体准确性。