在本教程中。我们将展示如何将 Embedding 模型嵌入 Elasticsearch，实现语义搜索。通过使用 NLP 模型，您可以直接用自然语言查询博客文档。而不必担心关键词匹配的局限性。

前提条件

在开始之前。请先完成以下准备工作：

• 创建一个 Elastic Cloud 部署，并启用 Autoscale。确保至少有一台具有 4 GB 内存的机器学习节点。
• 确保 Elasticsearch 集群正在运行。
• 如果还没有 Elastic Deployment，请注册免费的 Elastic Cloud 试用版。

常见痛点：很多团队在启动时会遇到资源不足导致模型推理缓慢或超时的问题。建议先通过 Autoscale 配置动态伸缩，以避免峰值负载下的性能瓶颈。

安装组件包并导入模块

# 安装所需 Python 包
pip install sentence-transformers eland elasticsearch transformers

请确认已安装所有必要的依赖项，随后导入所需模块：

from elasticsearch import Elasticsearch
from getpass import getpass
from urllib.request import urlopen
import json
from time import sleep

部署 NLP 模型

我们使用 工具安装一个文本嵌入模型。这里选用，它可以将搜索文本转换为 dense vector。

常见痛点：下载大模型时网络速度慢会导致部署时间过长；而小模型虽然速度快，但可能无法满足高精度需求。很关键，

安装文本 embedding NLP 模型

使用 脚本下载并安装 Transformer 模型，并将任务类型设为 text_embedding

# 获取 Cloud ID 和 API Key
ELASTIC_CLOUD_ID=$
ELASTIC_API_KEY=$
# 下载并部署模型
eland_import_hub_model \
--cloud-id $ELASTIC_CLOUD_ID \
--hub-model-id sentence-transformers/all-MiniLM-L6-v2 \
--task-type text_embedding \
--es-api-key $ELASTIC_API_KEY \
--start \
--clear-previous

连接到 Elasticsearch 集群

Create an Elasticsearch client instance using your Cloud ID and API key:

es = Elasticsearch(
cloud_id=ELASTIC_CLOUD_ID。api_key=ELASTIC_API_KEY,request_timeout=300,)
print)

常见痛点：ECS 集群与 ELK 的网络延迟可能导致请求超时； 请检查 VPC 配置和安全组规则，确保客户端能正常访问集群端口。

创建 Ingest Pipeline

A pipeline is required to generate embeddings for incoming documents:

{
"processors":
}

提示： 请先确认模型已成功部署，否则 Ingest Processor 会报错。

创建带有 Mapping 的索引

INDEXNAME = "blogs"
PIPELINEID = "vectorize_blogs"

INDEXMAPPING = { "properties": { "title": { "type": "text","fields": { "keyword": { "type": "keyword","ignoreabove": 256 } } }。"textembedding": { "properties":{ "pseudoistruncated"&colon{"type"&colon:":boolean"},"pseudomodelid"&colon{"type"&colon:"text","fields"&colon:{"pseudokeyword"&colon{"type"&colon:"keyword","ignoreabove"&colon:256}}},"pseudopredictedvalue"&colon{&qott&amp，amp;qot，"densevector"&qot"dims"&qot"384"&qot"index"&qot"true"&qot"similarity"&qot"l2_norm"}} } } }

INDEXSETTINGS = { \"index\":{ \"numberofreplicas\":1,\"numberofshards\":1。\"defaultpipeline\": PIPELINE_ID } }

if es.indices.exists: es.indices.delete

es.indices.create( index=INDEXNAME,mappings=INDEXMAPPING,settings=INDEX_SETTINGS) print

User Pain Point: Mapping 定义错误是最常见的故障原因之一；是在 dense_vector 字段配置上。如果 dims 与实际向量维度不符，将导致索引失败或搜索结果失效。建议在开发阶段先使用小批量数据验证 mapping 正确性。

将数据索引到 Elasticsearch

url = 'https://raw.githubusercontent.com/elastic/elasticsearch-labs/main/notebooks/integrations/hugging-face/blogs.json'
response = urlopen
titles = json.loads)

actions = for title in titles: actions.append actions.append

es.bulk sleep # 等待索引完成 print

User Pain Point: Bulk 索引过程中过多文档一次性发送会消耗大量内存，导致客户端崩溃。建议分批次提交，例如每 batch 1000 条，并监控内存使用情况。

查询数据集

The following query uses same embedding model to search for relevant blogs based on a natural language query:

{
“field”: “textembedding.predictedvalue”,“k”: 5,“numcandidates”: 10,“queryvectorbuilder”: {
“textembedding”: {
“modelid”: “sentence-transformersall-minilm-l6-v2”。“modeltext”: “how to track network connections”
}
}
}

python response = es.search( index=INDEXNAME,fields=,knn=querybody,source=False)

def show_results: for result in results: print

show_results

• The entire process is executed in one query call.

User Pain Point:

• KNN 查询对 CPU/GPU 的占用很高；若集群资源有限，可考虑开启 GPU inference 或减少 num_candidates 参数，以降低负载。

NLP 推理延迟问题：

• NLP 推理往往是搜索流程中的瓶颈，特别是在高并发场景下。可以通过预热缓存、使用更轻量级模型或部署专用推理服务来缓解此问题。

原文来源：

---