基于Open Search 2.9的LTR插件与模型服务框架技术问询

可以通过成熟的模型服务框架（搭配MLOps流水线工具）覆盖你列出的全部LTR步骤。现代模型服务框架（如MLflow、TorchServe）通常配套全生命周期管理能力，结合自定义API封装，可串联特征创建、日志采集、判断列表生成、模型训练、模型上传的完整流程。

各步骤的实现方式 #

• 1. 通过API创建特征集：在流水线中集成特征存储组件（如Feast），通过API接口定义特征schema并同步到特征库，实现特征的标准化管理。
• 2. 通过API调用日志功能：在搜索请求API中嵌入日志采集逻辑，自动记录query、返回文档、特征值、用户反馈等数据，或利用模型服务框架自带的日志模块实现统一收集。
• 3. 创建判断列表：基于日志数据，通过API批量导出历史搜索记录，或对接标注工具生成人工/自动标注的相关性判断结果（如点击、停留时长、人工打分）。
• 4. 训练模型（如XGBoost）：通过流水线触发训练任务，读取特征集和判断列表，使用XGBoost的排序专用接口（如rank:pairwise目标函数）训练LTR模型。
• 5. 通过API上传模型：训练完成后，流水线自动打包模型，调用模型服务框架的模型上传API，将模型部署到服务集群。

示例实现（基于FastAPI+MLflow+XGBoost） #

1. 特征集创建API #

from fastapi import FastAPI
import feast

app = FastAPI()
# 初始化特征存储
fs = feast.FeatureStore(repo_path="./search_feature_repo")

@app.post("/api/feature-set/create")
def create_feature_set(feature_set_name: str, feature_names: list):
    # 定义特征集结构
    feature_set = feast.FeatureSet(
        name=feature_set_name,
        features=[feast.Feature(name=f, dtype=feast.ValueType.FLOAT) for f in feature_names],
        entities=[feast.Entity(name="query_id", dtype=feast.ValueType.STRING)]
    )
    # 同步到特征库
    fs.apply([feature_set])
    return {"status": "success", "feature_set": feature_set_name}

2. 搜索请求与日志采集API #

import logging
from datetime import datetime
import json

# 配置日志存储
logging.basicConfig(
    filename='search_requests.log',
    format='%(asctime)s|%(message)s',
    level=logging.INFO
)

@app.post("/api/search")
def search(query_id: str, query: str, doc_ids: list):
    # 在线获取特征
    feature_data = fs.get_online_features(
        features=["search_features:tfidf_score", "search_features:bm25_score"],
        entity_rows=[{"query_id": query_id}]
    ).to_dict()
    # 记录请求日志
    log_content = json.dumps({
        "query_id": query_id,
        "query": query,
        "docs": doc_ids,
        "features": feature_data
    })
    logging.info(log_content)
    # 模拟返回排序结果
    return {"query_id": query_id, "results": doc_ids, "features": feature_data}

3. 生成判断列表API #

import pandas as pd

@app.get("/api/judgments/generate")
def generate_judgments(query_id: str = None):
    # 读取日志数据
    logs = pd.read_csv(
        'search_requests.log',
        sep='|',
        names=["timestamp", "log_data"]
    )
    logs["log_data"] = logs["log_data"].apply(json.loads)
    
    # 筛选指定query或全部数据
    if query_id:
        target_logs = logs[logs["log_data"].apply(lambda x: x["query_id"] == query_id)]
    else:
        target_logs = logs
    
    # 生成判断列表（示例用前3个文档标记为相关）
    judgments = []
    for _, row in target_logs.iterrows():
        log = row["log_data"]
        for idx, doc_id in enumerate(log["docs"]):
            judgments.append({
                "query_id": log["query_id"],
                "doc_id": doc_id,
                "relevance": 1 if idx < 3 else 0  # 模拟相关性标注
            })
    return {"judgments": judgments}

4. XGBoost LTR模型训练 #

import xgboost as xgb
import mlflow
import pandas as pd

def train_ltr_model():
    # 加载判断列表和特征数据
    judgments = pd.DataFrame(generate_judgments()["judgments"])
    # 获取历史特征数据
    feature_df = fs.get_historical_features(
        entity_df=judgments[["query_id", "doc_id"]],
        features=["search_features:tfidf_score", "search_features:bm25_score"]
    ).to_df()
    
    # 合并数据，准备XGBoost训练格式
    train_data = pd.merge(judgments, feature_df, on="query_id")
    # 按query分组，XGBoost排序需要分组信息
    query_groups = train_data.groupby("query_id").size().values
    dtrain = xgb.DMatrix(train_data[["tfidf_score", "bm25_score"]], label=train_data["relevance"])
    dtrain.set_group(query_groups)
    
    # 配置排序训练参数
    params = {
        "objective": "rank:pairwise",
        "eval_metric": "ndcg",
        "max_depth": 3,
        "learning_rate": 0.1,
        "seed": 42
    }
    
    # 用MLflow跟踪训练过程
    with mlflow.start_run(run_name="ltr_xgboost_train"):
        model = xgb.train(params, dtrain, num_boost_round=100)
        # 记录模型和参数
        mlflow.xgboost.log_model(model, "ltr_ranker_model")
        mlflow.log_params(params)
        mlflow.log_metric("train_ndcg", float(model.eval(dtrain).split()[-1]))
    
    return model

5. 模型上传到服务框架 #

import requests

def upload_model():
    # 从MLflow获取模型存储路径
    model_uri = mlflow.get_artifact_uri("ltr_ranker_model")
    # 调用模型服务的上传API（示例为自定义服务接口）
    response = requests.post(
        "http://model-service:8080/api/model/upload",
        json={
            "model_name": "search_ltr_ranker",
            "model_path": model_uri,
            "framework": "xgboost"
        }
    )
    return response.json()

# 触发训练并上传
if __name__ == "__main__":
    trained_model = train_ltr_model()
    upload_result = upload_model()
    print(upload_result)

内容的提问来源于stack exchange，提问作者anshuk_pal