1 为什么要在 Elasticsearch 上做 LTR？

适用版本： Elasticsearch ≥ 8.12.0
前置条件： 需拥有包含 “Serverless LTR” 的订阅等级（详见官方订阅矩阵）
技术栈： Elasticsearch + Python + Eland + XGBoost / LightGBM / scikit-learn

相关性瓶颈：经典 BM25 只能“打基础”，很难兼顾业务特征（点击率、销量、评分……）。
端到端成本：把向量检索或深度模型接到 ES 之外，需要额外的服务网关、时延和 DevOps。
Elasticsearch 8.12.0 新特性：官方 Serverless LTR 能直接在 search → rescore 阶段调用训练好的模型，既继承 ES 的伸缩性，又省去了微服务运维。

2 整体流程概览

在这里插入图片描述

3 环境准备

pip install "elasticsearch>=8.12" eland pandas scikit-learn xgboost lightgbm
export ELASTICSEARCH_URL="http://localhost:9200"
export ELASTIC_PASSWORD="elastic_password"

贴士：离线训练最好用一套隔离的 ES 集群，避免大规模特征抽取冲击线上查询。

4 定义特征抽取模板（QueryFeatureExtractor）

from eland.ml.ltr import QueryFeatureExtractor, LTRModelConfigfeature_extractors = [# 1️⃣  BM25 分数QueryFeatureExtractor(feature_name="title_bm25",query={"match": {"title": "{{query}}"}}),# 2️⃣  Title 命中词数QueryFeatureExtractor(feature_name="title_matched_term_count",query={"script_score": {"query": {"match": {"title": "{{query}}"}},"script": {"source": "return _termStats.matchedTermsCount();"}}}),# 3️⃣  直接把文档字段当特征QueryFeatureExtractor(feature_name="popularity",query={"script_score": {"query": {"exists": {"field": "popularity"}},"script": {"source": "return doc['popularity'].value;"}}}),# 4️⃣  查询词长度QueryFeatureExtractor(feature_name="query_term_count",query={"script_score": {"query": {"match": {"title": "{{query}}"}},"script": {"source": "return _termStats.uniqueTermsCount();"}}}),
]ltr_config = LTRModelConfig(feature_extractors)

为什么不用自己写脚本？
Eland 与 Elasticsearch 同步开发&测试，可确保训练阶段与线上推理的特征完全一致，避免“训练-服务漂移”。

5 采集训练数据（FeatureLogger）

from eland.ml.ltr import FeatureLogger
from elasticsearch import Elasticsearch
import pandas as pdes = Elasticsearch(ELASTICSEARCH_URL, basic_auth=("elastic", ELASTIC_PASSWORD))
MOVIE_INDEX = "movies_demo"logger = FeatureLogger(es, MOVIE_INDEX, ltr_config)# 假设我们已有一份人工评判文件 judgements.csv
judgements = pd.read_csv("judgements.csv")  # query, doc_id, relevancerecords = []
for _, row in judgements.iterrows():feats = logger.extract_features(query_params={"query": row.query},doc_ids=[row.doc_id])record = feats[0].to_dict()record["label"] = row.relevancerecords.append(record)df_train = pd.DataFrame(records)
df_train.to_csv("ltr_train.csv", index=False)

性能提示：FeatureLogger 会自动合并批量请求、最小化查询次数，但大量特征 × 大量样本仍可能对 ES 造成压力。调低 size 或切分批次能进一步减负。

6 训练 Ranker

以下以 XGBRanker 为例（LightGBM、RandomForest、DecisionTree 均可）：

from xgboost import XGBRanker
import numpy as npX = df_train[feature_extractors_names].values
y = df_train["label"].values
group = df_train.groupby("query").size().to_numpy()  # 每个 query 的样本数ranker = XGBRanker(objective="rank:pairwise",n_estimators=300,max_depth=6,learning_rate=0.1)
ranker.fit(X, y, group=group)

7 模型部署一键化（MLModel.import_ltr_model）

from eland.ml import MLModelMODEL_ID = "movies_ltr_xgboost_v1"MLModel.import_ltr_model(es_client=es,model=ranker,model_id=MODEL_ID,ltr_model_config=ltr_config,es_if_exists="replace"       # 本例重复执行时覆盖
)

执行后，Eland 会自动：

将模型序列化为 ES 接受的 [Eland ML Pack] 格式；
调用 Create Trained Model API 上传；
生成 inference_config，包含上文 4 个特征查询模板。

8 线上调用：Search + Rescore

POST movies_demo/_search
{"query": {"match": {"title": "science fiction"}},"rescore": {"window_size": 100,"query": {"rescore_query": {"ltr_model": {"model_id": "movies_ltr_xgboost_v1","params": { "query": "science fiction" }}},"query_weight": 0.2,"rescore_query_weight": 1.0}}
}

window_size 控制重新排序的命中窗口，100-500 较常见；
params.query 必须与特征模板中的 {{query}} 对应；
query_weight 越小，模型重排权重越大。

9 模型管理运维

操作	API	说明
查看模型	`GET _ml/trained_models/<id>`	包含元数据、大小、创建时间
移除旧版	`DELETE _ml/trained_models/<id>`	勿忘同时清理索引模板里旧 rescore 配置
批量迁移	`POST _ml/trained_models/_revert`	结合 CI/CD 做蓝绿发布
性能监控	`_nodes/stats` `inference`	关注推理延迟 & CPU/内存

10 最佳实践与踩坑

场景	建议
特征库膨胀	控制在 20-40 个以内；复杂逻辑交给模型学习权重，而不是写 N 个脚本评分
交叉验证	使用 group k-fold，确保同一 query 不同时出现在 train&test
线上流量回滚	给旧 BM25 排序保留 20% 流量，异常时快速切换
版本化	`movies_ltr_xgboost_v1` → `v2`，避免热更新导致 cache 冲突
安全与订阅	开启 HTTPS + token；无服务器版本仅在满足订阅级别时可用