Elasticsearch检索机制及自定义预处理管道、评分函数的Python实现咨询

1. Elasticsearch 文档检索流程 #

ES 检索全程分为两个核心阶段，底层基于倒排索引实现快速匹配：

• 查询阶段（Query Phase）：请求先到达协调节点，协调节点将请求转发到目标索引的所有分片（主副分片均可响应），每个分片在本地执行查询逻辑，通过倒排索引匹配得到符合条件的文档 ID 及对应评分，排序后返回 top N 结果到协调节点，协调节点对所有分片返回的结果做全局排序、分页，得到最终要返回的文档 ID 列表。
• 提取阶段（Fetch Phase）：协调节点根据文档 ID 列表，向对应分片拉取完整的文档内容，聚合后返回给调用方。

2. 自定义预处理管道（替换分词器为 BertTokenizer） #

ES 的 Ingest Pipeline 是文档写入索引前执行的预处理规则链，你要替换默认分词逻辑有两种适配 Python 技术栈的方案：

方案1：写入端提前预处理（推荐） #

不需要改动 ES 侧配置，直接在 Python 写入文档前完成 BertTokenizer 分词，再把 token 数组写入 ES 字段，开发调试成本最低，示例代码：

from transformers import BertTokenizer
from elasticsearch import Elasticsearch

# 初始化ES客户端和分词器
es = Elasticsearch(["http://localhost:9200"])
tokenizer = BertTokenizer.from_pretrained("你的本地Bert模型路径")

def bert_tokenize(text: str) -> list:
    tokens = tokenizer.tokenize(text)
    # 可自定义过滤特殊token、截断等逻辑
    return [t for t in tokens if t not in ["[CLS]", "[SEP]", "[PAD]"]]

# 写入文档时预处理
doc = {
    "raw_content": "原始文本内容",
    "bert_tokens": bert_tokenize("原始文本内容")
}
es.index(index="你的索引名", document=doc)

方案2：ES 侧自定义 Ingest Pipeline #

如果必须在 ES 内部执行预处理，7.5.1 版本可以先安装ingest-python插件，再创建包含 Python 分词逻辑的管道，示例管道配置：

{
  "processors": [
    {
      "python": {
        "code": """
from transformers import BertTokenizer
tokenizer = BertTokenizer.from_pretrained("所有ES节点都能访问的模型路径")
ctx['bert_tokens'] = [t for t in tokenizer.tokenize(ctx['raw_content']) if t not in ["[CLS]", "[SEP]", "[PAD]"]]
        """
      }
    }
  ]
}

注意该方案需要所有 ES 节点提前安装 Python 依赖、存放模型文件，性能开销比写入端预处理更高。

3. 自定义评分函数实现 #

对比 TF-IDF 与 BM25 效果 #

ES 7.5.1 默认相似度算法是 BM25，TF-IDF 在 ES 中内置为classic相似度，不需要额外开发，直接在索引 mapping 中为字段指定即可：

{
  "mappings": {
    "properties": {
      "raw_content": {
        "type": "text",
        "similarity": "BM25" // 要对比TF-IDF就填classic
      }
    }
  }
}

你可以给同一个文本设置两个子字段，分别绑定 BM25 和 TF-IDF 相似度，查询时分别指定查询字段即可直接对比两种算法的评分结果。

神经模型自定义打分 #

推荐用预计算向量 + 向量检索的方案，不需要修改 ES 内核，适配 Python 技术栈：

1. 提前在 Python 侧用神经模型计算所有文档的语义向量，存入 ES 的dense_vector类型字段
2. 查询时先计算查询词的语义向量，用script_score查询计算向量相似度作为评分，示例代码：

from elasticsearch import Elasticsearch

es = Elasticsearch(["http://localhost:9200"])
# 调用你自己的神经模型计算查询向量
query_vec = your_neural_model.encode("查询文本")

resp = es.search(
    index="你的索引名",
    query={
        "script_score": {
            "query": {"match": {"raw_content": "查询文本"}}, // 可叠加全文查询缩小范围，提升性能
            "script": {
                "source": "cosineSimilarity(params.query_vec, 'doc_vec_field') + 1.0",
                "params": {"query_vec": query_vec}
            }
        }
    }
)

Python 开发注意事项 #

• 安装elasticsearch-py客户端时选择 7.x 版本，和你部署的 ES 7.5.1 大版本对齐，避免兼容性问题
• 所有模型推理、预处理逻辑都建议放在 Python 侧实现，再把结果写入 ES，比在 ES 侧开发自定义插件更灵活，调试成本更低

内容的提问来源于stack exchange，提问作者namespace-Pt