Elasticsearch：崭新的打分机制 - Learning To Rank (LTR)

警告：“学习排名 (Learning To Rank)” 功能处于技术预览版，可能会在未来版本中更改或删除。 Elastic 将努力解决任何问题，但此功能不受官方 GA 功能的支持 SLA 的约束。

注意：此功能是在版本 8.12.0 中引入的，并且仅适用于某些订阅级别。有关更多信息，请参阅 https://www.elastic.co/subscriptions。

Learning To Rank (LTR) 使用经过训练的机器学习（ML）模型来构建搜索引擎的排名函数。通常，该模型用作第二阶段的重新排序器，以提高简单的第一阶段检索算法返回的搜索结果的相关性。LTR 函数接收文档列表和搜索上下文作为输入，并输出排名后的文档：

搜索上下文

除了需要排序的文档列表之外，LTR 函数还需要一个搜索上下文。通常，这个搜索上下文至少包括用户提供的搜索词（上述示例中的 text_query）。搜索上下文还可以提供排名模式中使用的其他信息。这可能是关于执行搜索的用户的信息（例如人口统计数据、地理位置或年龄）；关于查询的信息（例如查询长度）；或者在查询上下文中的文档信息（例如标题字段的得分）。

评判列表 - judgment list

LTR 模型通常是基于一个评判列表进行训练的，这是一组带有相关性等级的查询和文档的集合。评判列表可以由人工或机器生成：通常通过行为分析数据来填充，并经常有人工监督。评判列表确定了给定搜索查询的结果的理想排序。LTR 的目标是使模型尽可能地符合新查询和文档的评判列表排名。

评判列表是训练模型的主要输入。它包括一个数据集，该数据集包含查询和文档的配对，以及它们对应的相关性标签。相关性判断通常是二元的（相关/不相关）或更细致的标签，例如从0（完全不相关）到4（高度相关）的等级。下面的示例使用了分级的相关性判断。

注意事项：评判列表

虽然评判列表可以由人工手动创建，但也有技术可利用用户参与数据（如点击或转化）自动构建评判列表。

评判列表的数量和质量将极大地影响 LTR 模型的整体性能。在构建评判列表时，应非常注意以下几个方面：

大多数搜索引擎可以使用不同类型的查询进行搜索。例如，在电影搜索引擎中，用户可以按照电影标题、演员或导演进行搜索。在评判列表中，保持每种查询类型的示例数量平衡非常重要。这可以防止过度拟合，并使模型能够在所有查询类型上有效地泛化。
用户通常会提供更多的正面示例而不是负面示例。通过平衡正面和负面示例的数量，可以帮助模型更准确地区分相关和不相关的内容。

特征提取 - feature extraction

仅有查询和文档对并不足以提供足够的信息来训练用于 LTR 的机器学习模型。评判列表中的相关性评分依赖于多种属性或特征。必须提取这些特征，以确定各种组件如何组合以确定文档的相关性。评判列表加上提取的特征构成了 LTR 模型的训练数据集。

这些特征分为三个主要类别：

文档特征：这些特征直接从文档属性中派生。例如：电子商务商店中的产品价格。
查询特征：这些特征直接从用户提交的查询中计算得出。例如：查询中的单词数量。
查询-文档特征：用于提供有关查询上下文中的文档信息的特征。例如：标题字段的 BM25 分数。

为了准备训练数据集，将这些特征添加到评判列表中：

在 Elasticsearch 中，可以使用模板化查询来提取特征，这样做可以在构建训练数据集以及在查询时进行推断。以下是一个模板化查询的示例：

[
  {
    "query_extractor": {
      "feature_name": "title_bm25",
      "query": { "match": { "title": "{{query}}" } }
    }
  }
]

模型

LTR 的核心当然是一个机器学习（ML）模型。该模型使用上面描述的训练数据结合一个目标进行训练。在 LTR 的情况下，目标是根据一个排序度量标准（如 nDCG 或 MAP）在最佳方式下对结果文档进行排序，考虑到一个评判列表（judgment list）。该模型仅依赖于来自训练数据的特征和相关性标签。

LTR 领域正在快速发展，许多方法和模型类型正在进行实验。在实践中，Elasticsearch 专门依赖于梯度提升决策树（gradient boosted decision tree - GBDT）模型进行 LTR 推断。

请注意，Elasticsearch 支持模型推断，但训练过程必须在 Elasticsearch 之外进行，使用 GBDT 模型。在今天使用的最受欢迎的 LTR 模型中，LambdaMART 在低推断延迟的情况下提供了强大的排名性能。它依赖于 GBDT 模型，因此非常适合用于 Elasticsearch 中的 LTR。

XGBoost 是一个众所周知的库，提供了 LambdaMART 的实现，使其成为 LTR 的热门选择。我们在 eland 中提供了一些辅助工具，以便将训练的 XBGRanker 模型作为你在 Elasticsearch 中的 LTR 模型进行集成。

部署和管理学习排名模型

使用 Eland 训练和部署模型

通常，XGBoost 模型训练过程使用标准的 Python 数据科学工具，如 Pandas 和 scikit-learn。

我们在 elasticsearch-labs 仓库中开发了一个示例 notebook。这个交互式 Python 笔记本详细介绍了端到端的模型训练和部署工作流程。

我们强烈建议在你的工作流程中使用 eland，因为它提供了在 Elasticsearch 中使用 LTR 所需的重要功能。使用 eland 可以：

配置特征提取
提取用于训练的特征
在 Elasticsearch 中部署模型

在 Eland 中配置特征提取

特征提取器是使用模板化查询定义的。Eland 提供了 eland.ml.ltr.QueryFeatureExtractor 来直接在 Python 中定义这些特征提取器：

from eland.ml.ltr import QueryFeatureExtractor

feature_extractors=[
    # We want to use the score of the match query for the title field as a feature:
    QueryFeatureExtractor(
        feature_name="title_bm25",
        query={"match": {"title": "{{query}}"}}
    ),
    # We can use a script_score query to get the value
    # of the field rating directly as a feature:
    QueryFeatureExtractor(
        feature_name="popularity",
        query={
            "script_score": {
                "query": {"exists": {"field": "popularity"}},
                "script": {"source": "return doc['popularity'].value;"},
            }
        },
    ),
    # We can execute a script on the value of the query
    # and use the return value as a feature:
    QueryFeatureExtractor(
        feature_name="query_length",
        query={
            "script_score": {
                "query": {"match_all": {}},
                "script": {
                    "source": "return params['query'].splitOnToken(' ').length;",
                    "params": {
                        "query": "{{query}}",
                    }
                },
            }
        },
    ),
]

一旦定义了特征提取器，它们就会被封装在一个 eland.ml.ltr.LTRModelConfig 对象中，以便在后续的训练步骤中使用：

from eland.ml.ltr import LTRModelConfig

ltr_config = LTRModelConfig(feature_extractors)

提取特征进行训练

构建数据集是训练过程中的一个关键步骤。这包括提取相关特征并将它们添加到你的判定列表中。我们推荐使用 Eland 的 eland.ml.ltr.FeatureLogger 辅助类来完成此过程。

from eland.ml.ltr import FeatureLogger

# Create a feature logger that will be used to query {es} to retrieve the features:
feature_logger = FeatureLogger(es_client, MOVIE_INDEX, ltr_config)

FeatureLogger 提供了一个 extract_features 方法，它使您能够从你的判定列表中为一系列特定文档提取特征。同时，你可以传递先前定义的查询参数给特征提取器：

feature_logger.extract_features(
    query_params={"query": "foo"},
    doc_ids=["doc-1", "doc-2"]
)

我们的示例 notebook 解释了如何使用 FeatureLogger 构建训练数据集，通过向判定列表添加特征。

有关特征提取需要注意的地方：

我们强烈建议不要自行实现特征提取。在特征提取方面保持一致性非常重要，即在训练环境和 Elasticsearch 中进行推断时。通过使用与 Elasticsearch 开发和测试一起的 eland 工具，你可以确保它们之间保持一致的功能。
特征提取是通过在 Elasticsearch 服务器上执行查询来执行的。这可能会给你的集群带来很大压力，特别是当你的判定列表包含很多示例或具有许多特征时。我们的特征记录器实现旨在最小化发送到服务器的搜索请求数量并减少负载。但是，最好使用与任何面向用户的生产流量隔离的 Elasticsearch 集群构建您的训练数据集。

将你的模型部署到 Elasticsearch 中

一旦你的模型训练完成，你就可以将其部署到你的 Elasticsearch 集群中。你可以使用 Eland 的 MLModel.import_ltr_model 方法：

from eland.ml import MLModel

LEARNING_TO_RANK_MODEL_ID="ltr-model-xgboost"

MLModel.import_ltr_model(
    es_client=es_client,
    model=ranker,
    model_id=LEARNING_TO_RANK_MODEL_ID,
    ltr_model_config=ltr_config,
    es_if_exists="replace",
)

此方法会将训练后的模型以及学习排名配置（包括特征提取）序列化成 Elasticsearch 能够理解的格式。然后，该模型通过 Create Trained Models API 部署到 Elasticsearch。

目前支持以下类型的模型用于 Elasticsearch 的学习排名（LTR）：

DecisionTreeRegressor
RandomForestRegressor
LGBMRegressor
XGBRanker
XGBRegressor

未来将支持更多类型的模型。

学习排名模型管理

一旦你的模型在 Elasticsearch 中部署，你可以使用 trained model APIs 来管理它。现在，你已经准备好在搜索时使用你的 LTR 模型作为重新评分器。

Demo

安装 Elasticsearch 及 Kibana

接下来，我们将使用一个 Python notebook 来进行展示。

如果你还没有安装好自己的 Elasticsearch 及 Kibana，请参考如下的链接来进行安装：

如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch
Kibana：如何在 Linux，MacOS 及 Windows上安装 Elastic 栈中的 Kibana

在安装的时候，我们选择 Elastic Stack 8.x 来进行安装。

在首次启动 Elasticsearch 的时候，我们可以看到如下的输出：

我们需要记住超级用户 elastic 的密码。

为了能够使得 LTR 正常工作（请参考订阅 | Elastic Stack 产品和支持 | Elastic），我们需要启动白金试用功能：

下载代码并启动 jupyter notebook

我们在地址 GitHub - liu-xiao-guo/elasticsearch-labs: Notebooks & Example Apps for Search & AI Applications with Elasticsearch 下载代码：

git clone https://github.com/liu-xiao-guo/elasticsearch-labs

然后我们进入到如下的目录并启动 jupyter：

$ pwd
/Users/liuxg/python/elasticsearch-labs/notebooks/search
$ ls
00-quick-start.ipynb                README.md
01-keyword-querying-filtering.ipynb _nbtest.setup.ipynb
02-hybrid-search.ipynb              _nbtest.teardown.03-ELSER.ipynb
03-ELSER.ipynb                      _nbtest.teardown.ipynb
04-multilingual.ipynb               articles.json
05-query-rules.ipynb                data.json
06-synonyms-api.ipynb               movies.json
08-learning-to-rank.ipynb           query-rules-data.json
Makefile                            sample_data
$ jupyter notebook

我们选择 08-learning-to-rank.ipynb 文件来进行。

我们在当前的目录下创建一个叫做 .env 的文件：

$ pwd
/Users/liuxg/python/elasticsearch-labs2/notebooks/search
$ ls
00-quick-start.ipynb                _nbtest.setup.ipynb
01-keyword-querying-filtering.ipynb _nbtest.teardown.03-ELSER.ipynb
02-hybrid-search.ipynb              _nbtest.teardown.ipynb
03-ELSER.ipynb                      articles.json
04-multilingual.ipynb               data.json
05-query-rules.ipynb                http_ca.crt
06-synonyms-api.ipynb               movies.json
07-inference.ipynb                  query-rules-data.json
08-learning-to-rank.ipynb           sample_data
README.md
$ ls .env
.env
$ cat .env
ES_USER="elastic"
ES_PASSWORD="wKWgjujKB4gOOKw4Zk=E"
ES_ENDPOINT="localhost"

如上所示，我们可以根据自己的 Elasticsearch 的配置来定义上面的 ES_USER, ES_PASSWORD 及 ES_EDNPINT。

我们同时需要拷贝 Elasticsearch 的证书到当前的目录下：

$ pwd
/Users/liuxg/python/elasticsearch-labs/notebooks/search
$ cp ~/elastic/elasticsearch-8.13.2/config/certs/http_ca.crt .
$ ls http_ca.crt 
http_ca.crt

这样我们的配置就完成了。

运行 notebook

安装所需的包

!pip3 install -qU elasticsearch eland "eland[scikit-learn]" xgboost tqdm

from tqdm import tqdm

# Setup the progress bar so we can use progress_apply in the notebook.
tqdm.pandas()

配置你的 Elasticsearch 部署

import os
from dotenv import load_dotenv
from elasticsearch import Elasticsearch

load_dotenv()

ES_USER = os.getenv("ES_USER")
ES_PASSWORD = os.getenv("ES_PASSWORD")
ES_ENDPOINT = os.getenv("ES_ENDPOINT")
 
url = f"https://{ES_USER}:{ES_PASSWORD}@{ES_ENDPOINT}:9200"

print(url)

es_client = Elasticsearch(
    hosts=[url], 
    ca_certs = "./http_ca.crt", 
    verify_certs = True
    )

print(es_client.info())

配置数据集

我们将使用源自 MSRD（电影搜索排名数据集）的数据集。

该数据集可在此处获取并包含以下文件：

movie_corpus.jsonl.gz：要索引的电影数据集。
movie_judgements.tsv.gz：一组查询的相关性判断的判断列表。
movie_index_settings.json：要应用于文档和索引的设置。

from urllib.parse import urljoin

#DATASET_BASE_URL = "https://raw.githubusercontent.com/elastic/elasticsearch-labs/main/notebooks/search/sample_data/learning-to-rank/"
DATASET_BASE_URL = "file:/Users/liuxg/python/elasticsearch-labs/notebooks/search/sample_data/learning-to-rank/"

CORPUS_URL = urljoin(DATASET_BASE_URL, "movies-corpus.jsonl.gz")
JUDGEMENTS_FILE_URL = urljoin(DATASET_BASE_URL, "movies-judgments.tsv.gz")
INDEX_SETTINGS_URL = urljoin(DATASET_BASE_URL, "movies-index-settings.json")

我们需要根据自己的文档位置对上面的文件路径进行修改。

导入文档语料库

这一步会将语料库的文档导入到 movies 索引中。

文档包含以下字段：

Field name	Description
`id`	Id of the document
`title`	Movie title
`overview`	A short description of the movie
`actors`	List of actors in the movies
`director`	Director of the movie
`characters`	List of characters that appear in the movie
`genres`	Genres of the movie
`year`	Year the movie was released
`budget`	Budget of the movies in USD
`votes`	Number of votes received by the movie
`rating`	Average rating of the movie
`popularity`	Number use to measure the movie popularity
`tags`	A list of tags for the movies

import json
import elasticsearch.helpers as es_helpers
import pandas as pd
from urllib.request import urlopen

MOVIE_INDEX = "movies"

# Delete index
print("Deleting index if it already exists:", MOVIE_INDEX)
es_client.options(ignore_status=[400, 404]).indices.delete(index=MOVIE_INDEX)

print("Creating index:", MOVIE_INDEX)
index_settings = json.load(urlopen(INDEX_SETTINGS_URL))
es_client.indices.create(index=MOVIE_INDEX, **index_settings)

print(f"Loading the corpus from {CORPUS_URL}")
corpus_df = pd.read_json(CORPUS_URL, lines=True)

print(f"Indexing the corpus into {MOVIE_INDEX} ...")
bulk_result = es_helpers.bulk(
    es_client,
    actions=[
        {"_id": movie["id"], "_index": MOVIE_INDEX, **movie}
        for movie in corpus_df.to_dict("records")
    ],
)
print(f"Indexed {bulk_result[0]} documents into {MOVIE_INDEX}")

我们可以在 Kibana 中进行查看：

上面显示我们写入了 9750 个文档。

加载判断列表

判断列表包含我们将用来训练 “学习排名” 模型的人工评估。

每行代表一个具有关联相关性等级的查询-文档对，并包含以下列：

Column	Description
`query_id`	Pairs for the same query are grouped together and received a unique id.
`query`	Actual text of the query.
`doc_id`	ID of the document.
`grade`	The relevance grade of the document for the query.

注意：在此示例中，相关性等级是二进制值（相关或不相关）。你还可以使用代表相关程度的数字（例如从 0 到 4）。

judgments_df = pd.read_csv(JUDGEMENTS_FILE_URL, delimiter="\t")
judgments_df

配置特征提取

特征是我们模型的输入。它们表示有关单独查询、单独结果文档或查询上下文中的结果文档的信息，例如 BM25 分数。

特征是使用标准模板查询和查询 DSL 定义的。

为了简化训练期间定义和细化特征提取的过程，我们直接在 eland 中合并了许多 primitives。

from eland.ml.ltr import LTRModelConfig, QueryFeatureExtractor

ltr_config = LTRModelConfig(
    feature_extractors=[
        # For the following field we want to use the score of the match query for the field as a features:
        QueryFeatureExtractor(
            feature_name="title_bm25", query={"match": {"title": "{{query}}"}}
        ),
        QueryFeatureExtractor(
            feature_name="actors_bm25", query={"match": {"actors": "{{query}}"}}
        ),
        # We could also use a more strict matching clause as an additional features. Here we want all the terms of our query to match.
        QueryFeatureExtractor(
            feature_name="title_all_terms_bm25",
            query={
                "match": {
                    "title": {"query": "{{query}}", "minimum_should_match": "100%"}
                }
            },
        ),
        QueryFeatureExtractor(
            feature_name="actors_all_terms_bm25",
            query={
                "match": {
                    "actors": {"query": "{{query}}", "minimum_should_match": "100%"}
                }
            },
        ),
        # Also we can use a script_score query to get the document field values directly as a feature.
        QueryFeatureExtractor(
            feature_name="popularity",
            query={
                "script_score": {
                    "query": {"exists": {"field": "popularity"}},
                    "script": {"source": "return doc['popularity'].value;"},
                }
            },
        ),
    ]
)

构建训练数据集

现在我们已经加载了基本数据集并配置了特征提取，我们将使用我们的判断列表来得出最终的训练数据集。数据集将由包含 <query, document> 对的行以及训练模型所需的所有特征组成。为了生成此数据集，我们将从判断列表中运行每个查询，并将提取的特征添加为每个标记结果文档的列。

例如，如果我们有一个包含两个标记文档 d3 和 d9 的查询 q1，训练数据集最终将有两行 — 每对 <q1, d3> 和 <q1, d9> 一行。

请注意，由于这会在您的 Elasticsearch 集群上执行查询，因此运行此操作的时间将根据集群的托管位置以及此笔记本的运行位置而有所不同。例如，如果你在与 Elasticsearch 集群相同的服务器或主机上运行笔记本，则此操作往往会在示例数据集上运行得非常快（< 2 分钟）

import numpy

from eland.ml.ltr import FeatureLogger

# First we create a feature logger that will be used to query Elasticsearch to retrieve the features:
feature_logger = FeatureLogger(es_client, MOVIE_INDEX, ltr_config)


# This method will be applied for each query group in the judgment log:
def _extract_query_features(query_judgements_group):
    # Retrieve document ids in the query group as strings.
    doc_ids = query_judgements_group["doc_id"].astype("str").to_list()

    # Resolve query params for the current query group (e.g.: {"query": "batman"}).
    query_params = {"query": query_judgements_group["query"].iloc[0]}

    # Extract the features for the documents in the query group:
    doc_features = feature_logger.extract_features(query_params, doc_ids)

    # Adding a column to the dataframe for each feature:
    for feature_index, feature_name in enumerate(ltr_config.feature_names):
        query_judgements_group[feature_name] = numpy.array(
            [doc_features[doc_id][feature_index] for doc_id in doc_ids]
        )

    return query_judgements_group


judgments_with_features = judgments_df.groupby(
    "query_id", group_keys=False
).progress_apply(_extract_query_features)

judgments_with_features

从上面，我们可以看出提取的特征值。

创建并训练模型

LTR rescorer 支持 XGBRanker 训练的模型。

在 XGBoost 文档中了解更多信息。

from xgboost import XGBRanker
from sklearn.model_selection import GroupShuffleSplit


# Create the ranker model:
ranker = XGBRanker(
    objective="rank:ndcg",
    eval_metric=["ndcg@10"],
    early_stopping_rounds=20,
)

# Shaping training and eval data in the expected format.
X = judgments_with_features[ltr_config.feature_names]
y = judgments_with_features["grade"]
groups = judgments_with_features["query_id"]

# Split the dataset in two parts respectively used for training and evaluation of the model.
group_preserving_splitter = GroupShuffleSplit(n_splits=1, train_size=0.7).split(
    X, y, groups
)
train_idx, eval_idx = next(group_preserving_splitter)

train_features, eval_features = X.loc[train_idx], X.loc[eval_idx]
train_target, eval_target = y.loc[train_idx], y.loc[eval_idx]
train_query_groups, eval_query_groups = groups.loc[train_idx], groups.loc[eval_idx]

# Training the model
ranker.fit(
    X=train_features,
    y=train_target,
    group=train_query_groups.value_counts().sort_index().values,
    eval_set=[(eval_features, eval_target)],
    eval_group=[eval_query_groups.value_counts().sort_index().values],
    verbose=True,
)

将模型导入 Elasticsearch

模型训练完成后，我们可以使用 Eland 将其加载到 Elasticsearch 中。

请注意，MLModel.import_ltr_model 方法包含 LTRModelConfig 对象，该对象定义如何为导入的模型提取特征。

from eland.ml import MLModel

LEARNING_TO_RANK_MODEL_ID = "ltr-model-xgboost"

MLModel.import_ltr_model(
    es_client=es_client,
    model=ranker,
    model_id=LEARNING_TO_RANK_MODEL_ID,
    ltr_model_config=ltr_config,
    es_if_exists="replace",
)

一旦部署完毕，我们可以在 Kibana 中进行查看：

使用 rescorer

将模型上传到 Elasticsearch 后，你将能够在 _search API 中将其用作 rescorer，如本示例所示：

GET /movies/_search
{
   "query" : {
      "multi_match" : {
         "query": "star wars",
         "fields": ["title", "overview", "actors", "director", "tags", "characters"]
      }
   },
   "rescore" : {
      "window_size" : 50,
      "learning_to_rank" : {
         "model_id": "ltr-model-xgboost",
         "params": { 
            "query": "star wars"
         }
      }
   }
}

query = "star wars"

# First let's display the result when not using the rescorer:
search_fields = ["title", "overview", "actors", "director", "tags", "characters"]
bm25_query = {"multi_match": {"query": query, "fields": search_fields}}

bm25_search_response = es_client.search(index=MOVIE_INDEX, query=bm25_query)

[
    (movie["_source"]["title"], movie["_score"], movie["_id"])
    for movie in bm25_search_response["hits"]["hits"]
]

# Now let's display result when using the rescorer:

ltr_rescorer = {
    "learning_to_rank": {
        "model_id": LEARNING_TO_RANK_MODEL_ID,
        "params": {"query": query},
    },
    "window_size": 100,
}

rescored_search_response = es_client.search(
    index=MOVIE_INDEX, query=bm25_query, rescore=ltr_rescorer
)

[
    (movie["_source"]["title"], movie["_score"], movie["_id"])
    for movie in rescored_search_response["hits"]["hits"]
]