将 Embedding 模型加载到 Elasticsearch 中
本工作簿使用一个由 Elastic 博客标题组成的简单数据集，在 Elasticsearch 中实现 NLP 文本搜索。
你将索引博客文档，并使用 ingest pipeline 生成文本 embedding。随后，通过使用 NLP 模型，你可以使用自然语言对这些博客文档进行查询。
更多阅读：Elasticsearch：如何部署文本嵌入模型并将其用于语义搜索
前提条件
在开始之前，请创建一个 Elastic Cloud deployment，并启用 autoscale，确保至少有一个具有足够（4GB）内存的机器学习（ML）节点。同时确保 Elasticsearch 集群正在运行。
如果你还没有 Elastic deployment，可以注册免费的 Elastic Cloud 试用版。
安装软件包并导入模块
!python3 -m pip install sentence-transformers==2.7.0 eland elasticsearch transformers开始之前，你需要安装所有必需的 Python 依赖项。
!python3 -m pip install sentence-transformers==2.7.0 "eland<9" "elasticsearch<9" transformers # 导入模块 from elasticsearch import Elasticsearch from getpass import getpass from urllib.request import urlopen import json from time import sleep部署 NLP 模型
我们使用eland工具安装一个text_embedding模型。这里使用all-MiniLM-L6-v2模型将搜索文本转换为 dense vector。
该模型会将你的搜索查询转换为向量，用于在存储于 Elasticsearch 中的文档集合上执行搜索。
安装文本 embedding NLP 模型
使用eland_import_hub_model脚本下载并安装all-MiniLM-L6-v2Transformer 模型，并将 NLP 的--task-type设置为text_embedding。
要获取 Cloud ID，请进入 Elastic Cloud，在 deployment 概览页面复制 Cloud ID。
为了验证请求身份，你可以使用 API key。或者，也可以使用 Cloud deployment 的用户名和密码。
# https://www.elastic.co/search-labs/tutorials/install-elasticsearch/elastic-cloud#finding-your-cloud-id ELASTIC_CLOUD_ID = getpass("Elastic Cloud ID: ") # https://www.elastic.co/search-labs/tutorials/install-elasticsearch/elastic-cloud#creating-an-api-key ELASTIC_API_KEY = getpass("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 集群
使用 deployment 的 Cloud ID 和 API Key 创建一个 Elasticsearch client 实例。在本示例中，我们使用上一步中的API_KEY和CLOUD_ID。
你也可以使用 deployment 的用户名和密码进行身份验证。
es = Elasticsearch( cloud_id=ELASTIC_CLOUD_ID, api_key=ELASTIC_API_KEY, request_timeout=600 ) es.info() # 应返回集群信息创建 Ingest Pipeline
我们需要创建一个文本 embedding ingest pipeline，为title字段生成向量（文本）embedding。
下面的 pipeline 定义了一个 processor，用于调用 NLP 模型执行 inference。
# ingest pipeline 定义 PIPELINE_ID = "vectorize_blogs" es.ingest.put_pipeline( id=PIPELINE_ID, processors=[ { "inference": { "model_id": "sentence-transformers__all-minilm-l6-v2", "target_field": "text_embedding", "field_map": {"title": "text_field"}, } } ], )创建带有 mapping 的索引
现在，在索引文档之前，我们先创建一个具有正确 mapping 的 Elasticsearch 索引。我们添加text_embedding字段，用于包含model_id和predicted_value，以存储 embedding。
# 定义索引名称 INDEX_NAME = "blogs" # 标志，用于检查创建索引前是否删除已有索引 SHOULD_DELETE_INDEX = True # 定义索引 mapping INDEX_MAPPING = { "properties": { "title": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } }, }, "text_embedding": { "properties": { "is_truncated": { "type": "boolean" }, "model_id": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } }, }, "predicted_value": { "type": "dense_vector", "dims": 384, "index": True, "similarity": "l2_norm", }, } }, } } INDEX_SETTINGS = { "index": { "number_of_replicas": "1", "number_of_shards": "1", "default_pipeline": PIPELINE_ID, } } # 检查是否需要在创建索引前删除已有索引 if SHOULD_DELETE_INDEX: if es.indices.exists(index=INDEX_NAME): print("Deleting existing %s" % INDEX_NAME) es.indices.delete(index=INDEX_NAME, ignore=[400, 404]) print("Creating index %s" % INDEX_NAME) es.indices.create( index=INDEX_NAME, mappings=INDEX_MAPPING, settings=INDEX_SETTINGS, ignore=[400, 404] )将数据索引到 Elasticsearch
现在，使用 ingest pipeline 索引示例博客数据。
注意：在开始索引之前，请确保你已经启动训练好的模型 deployment。
url = "https://raw.githubusercontent.com/elastic/elasticsearch-labs/main/notebooks/integrations/hugging-face/blogs.json" response = urlopen(url) titles = json.loads(response.read()) actions = [] for title in titles: actions.append({"index": {"_index": "blogs"}}) actions.append(title) es.bulk(index="blogs", operations=actions) sleep(5)查询数据集
下一步是执行查询，搜索相关博客。下面的示例使用我们上传到 Elasticsearch 的sentence-transformers__all-minilm-l6-v2模型，对"model_text": "how to track network connections"进行搜索。
整个过程只需一次查询，尽管内部实际上包含两个步骤。首先，查询会使用 NLP 模型为搜索文本生成一个向量；然后使用该向量在数据集中执行搜索。
最终，输出将显示按与搜索查询接近程度排序的文档列表。
INDEX_NAME = "blogs" source_fields = ["id", "title"] query = { "field": "text_embedding.predicted_value", "k": 5, "num_candidates": 50, "query_vector_builder": { "text_embedding": { "model_id": "sentence-transformers__all-minilm-l6-v2", "model_text": "how to track network connections", } }, } response = es.search( index=INDEX_NAME, fields=source_fields, knn=query, source=False, ) def show_results(results): for result in results: print( f'{result["fields"]["title"]}\n' f'Score: {result["_score"]}\n' ) show_results(response.body["hits"]["hits"])输出：
['Brewing in Beats: Track network connections'] Score: 0.5917864 ['Machine Learning for Nginx Logs - Identifying Operational Issues with Your Website'] Score: 0.40109876 ['Data Visualization For Machine Learning'] Score: 0.39027885 ['Logstash Lines: Introduce integration plugins'] Score: 0.36899462 ['Keeping up with Kibana: This week in Kibana for November 29th, 2019'] Score: 0.35690257原文：https://www.elastic.co/search-labs/tutorials/examples/nlp-model-vector-search-elasticsearch