【LLM】Elasticsearch作为向量库入门指南_es向量库

整理不易，请不要令色你的赞和收藏。

1. 前言

这篇文章将介绍如何使用 docker-compose 安装 ES 和 Kibana，如何使用 ES 存储和查询向量数据。这里有全网最详细的 ES 作为向量库参数配置介绍，并且在文章最后会介绍如何使用 Langgraph 搭建一个文本嵌入，并实现相似性查询的工作流。

Elasticsearch 使用倒排索引来加速向量检索，快速定位包含特定向量的文档。在向量检索过程中，Elasticsearch 会根据查询向量的特征，通过倒排索引匹配相似的向量。一旦匹配到倒排索引中的文档，Elasticsearch会计算查询向量与匹配文档向量之间的相似度。常用的相似度计算方法包括余弦相似度、欧几里得距离等。 

Elasticsearch 本身也是一个成熟的全文搜索引擎，支持 BM25、TF-IDF 等文本检索方法，结合向量搜索（k-NN、HNSW、ANN） 可作为提升 RAG 的查询精确度的一种方式。

2. 前提条件

• 已安装 Docker、docker compose。

3. 安装Elasticsearch

为了方便管理和功能，我们使用docker compose安装 Elasticsearch 、 Kibana 以及 IK分词器，如果你需要安装其他版本，请确保 Kibana 的版本和 ES 的版本一致。

3.1 创建docker-compose文件

首先我们先创建一个 docker-compose.yml 文件，文件内容如下：

PS：Docker Compose v2 默认使用最新的 Compose 文件格式，因此不再需要显式指定 version。

# version \"3.9\"services: elasticsearch: image: elasticsearch:8.16.0 container_name: elasticsearch environment: - discovery.type=single-node - ES_JAVA_OPTS=-Xms512m -Xmx512m - xpack.security.enabled=false volumes: - es_data:/usr/share/elasticsearch/data - ./plugins:/usr/share/elasticsearch/plugins # 挂载插件目录 ports: - target: 9200 published: 9200 networks: - elastic command: > bash -c \" if [ -d /usr/share/elasticsearch/plugins/analysis-ik ]; then echo \'Removing existing analysis-ik plugin...\'; rm -rf /usr/share/elasticsearch/plugins/analysis-ik; fi; ./bin/elasticsearch-plugin install https://release.infinilabs.com/analysis-ik/stable/elasticsearch-analysis-ik-8.16.0.zip -b && ./bin/elasticsearch \" kibana: image: kibana:8.16.0 container_name: kibana ports: - target: 5601 published: 5601 depends_on: - elasticsearch networks: - elasticvolumes: es_data: driver: localnetworks: elastic: name: elastic driver: bridge

docker-compose 中的参数解析请参考我之前的文章，这里不在赘述。

IK分词器通过 command 命令安装，二次安装的时候需要先手动删除 plungs 挂载目录，不然可能会报错。

3.2 启动服务

我们先验证配置是否正确：

docker-compose -f docker-compose.yml config

启动服务：

docker-compose up -d

查看是否启动成功：

docker-compose ps

访问 Kibana：http://127.0.0.1:5601/

4. 向量数据嵌入到ES

4.1 创建索引

首先我们创建一个索引 es-embedding-test 。为了方便演示，维度设置为 8 。

PUT /es-embedding-test{ \"mappings\": { \"properties\": { \"vector\": { \"type\": \"dense_vector\", \"element_type\": \"float\", \"dims\": 8, \"index\": true, \"similarity\": \"cosine\", \"index_options\": {  \"type\": \"int8_hnsw\" } }, \"content\": { \"type\": \"text\" } } }}

4.1.1 参数介绍

索引中包含两个字段 vector （存储向量值）和 content（存储对应文本内容）。ES 中使用 dense_vector 数据类型定义。

• element_type：用于编码向量的数据类型。详细见拓展1。

• dims：为向量的维度，必须与嵌入模型的向量维度一致，最高支持4096维。常见的向量维度一般为768、1024、1536等。

• index：设置为 true，用于启用 KNN 查询。

• similarity：用于比较查询向量和文档向量的相似性函数，当 element_type 为 bit 时，默认使用 l2_norm 作为相似性度量方式，否则，默认使用 cosine。可选值：l2_norm（L2距离/欧几里得距离）、dot_product（计算两向量点积）、cosine（余弦）、max_inner_product（最大内积）。

• index_options：用于配置 kNN 索引算法，这个参数只有在 index 设置为 true 时生效。HNSW 算法有两个内部参数，影响数据结构的构建方式。这些参数可以调整以提高结果准确性，但会以较慢的索引速度为代价。见拓展2。

4.1.2 拓展1

element_type 支持的数据类型说明：

• float（默认）：每个维度索引一个 4 字节（32 位）浮点数，适用于 高精度向量计算。

• byte：每个维度索引一个 1 字节（8 位）整数，适用于 存储优化，降低内存占用，但精度可能有所损失。

• bit:每个维度索引 1 位（二进制位）,适用于 超高维向量 或 专门支持比特向量的模型，使用 bit时，维度数必须是 8 的倍数，且表示的是 比特数 而非常规的维度数

4.1.3 拓展2

index_options 高级配置参数说明：

• type（必选）：要使用的 kNN 算法类型，不同类型适用于不同的 搜索需求 和 存储优化。默认：int8_hnsw。见拓展3。

• m（可选）：控制 HNSW 图中每个节点的连接数，默认 16。适用于 type 为 ​​​​​hnsw、int8_hnsw、int4_hnsw 和 bbq_hnsw。

• ef_construction（可选）：控制索引构建时，每个节点考虑的最近邻候选数（默认 100），数值越大，索引构建时间增加，但 查询结果更精确。适用于 type 为 ​​​​​hnsw、int8_hnsw、int4_hnsw 和 bbq_hnsw。

• confidence_interval（可选）：量化向量的置信区间。仅适用于 int8_hnsw 、 int4_hnsw 、 int8_flat 和 int4_flat 索引类型。其可以是 0.90 和 1.0 之间（包括）的任何值或正好为 0 。当值为 0 时，表示应计算动态分位数以优化量化。当介于 0.90 和 1.0 之间时，此值限制在计算量化阈值时使用的值。例如，值为 0.95 时，在计算量化阈值时将仅使用中间 95%的值（例如，将忽略最高和最低 2.5%的值）。对于 int8 量化的向量默认为 1/(dims + 1) ，对于动态分位数计算默认为 0 和 int4 。

4.1.3 拓展3

index_options 的 type 索引类型可选值介绍。

ES 默认使用量化索引（Quantized Index）来优化存储和搜索高维向量。通常情况下高维浮点数向量（float32）会占用大量内存，并且在 最近邻搜索（NN Search） 过程中需要进行大量计算。量化索引通过将浮点向量转换为 更低精度的数据格式（如 int8、int4 或二进制位） 来减少存储需求，并加速搜索计算。

ES 支持以下三种量化策略：

• int8 量化（1 字节整数）：每个浮点数被转换为 int8（1 字节整数）。比原始 float32 版本减少 75%（4 倍）的内存占用。精度损失较小。

• int4 量化（半字节整数）：每个浮点数被转换为 int4（4 位整数，半字节）。比原始 float32 版本减少 87%（8 倍）的内存占用。精度损失较大。

• bbq 量化（Better Binary Quantization）：每个数值量化为 1 bit（二进制位）。比原始 float32 版本 **减少 96%（32 倍）**的内存占用。精度损失最大，但可通过 增加查询时的候选数（oversampling） 和 重排（reranking） 进行补偿。

如果要在 ES 中使用量化索引，可以设置您的索引类型为 int8_hnsw 、 int4_hnsw 或 bbq_hnsw 。当索引 float 向量时，默认索引类型为 int8_hnsw 。

以下是可选值对比：

索引类型

描述

支持的 element_type

存储优化

适用场景

hnsw

HNSW（Hierarchical Navigable Small World）近似 kNN 搜索

float、byte、bit

无

适用于高效、可扩展的向量搜索

int8_hnsw

HNSW + int8 量化

float

减少 4x 内存

节省存储，适用于大规模 kNN

int4_hnsw

HNSW + int4 量化

float

减少 8x 内存

进一步减少存储，占用更少

bbq_hnsw

（预览）

HNSW + 二进制量化

float

减少 32x 内存

超大规模向量数据，但损失较大精度

flat

暴力搜索（精确 kNN）

float、byte、bit

无

小数据集，需要高精度匹配

int8_flat

暴力搜索 + int8 量化

float

减少 4x 内存

节省存储，但搜索速度较慢

int4_flat

暴力搜索 + int4 量化

float

减少 8x 内存

适用于存储受限

但搜索精确的场景

bbq_flat

（预览）

暴力搜索 + 二进制量化

float

减少 32x 内存

超大规模向量搜索，但精度较低

4.2 嵌入索引文档

索引单个文档：

POST /es-embedding-test/_doc{ \"content\": \"索引单个文档\", \"vector\": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8] }

索引多个文档：

POST /_bulk{\"index\": {\"_index\": \"es-embedding-test\",\"_id\": \"2\"}} {\"content\": \"这个是文档1\",\"vector\": [0.11, 0.12, 0.13, 0.14, 0.15, 0.16, 0.17, 0.18]} {\"index\": {\"_index\": \"es-embedding-test\",\"_id\": \"3\"}} {\"content\": \"这个是文档2\",\"vector\": [0.21, 0.22, 0.23, 0.24, 0.25, 0.26, 0.27, 0.28]} {\"index\": {\"_index\": \"es-embedding-test\",\"_id\": \"4\"}} {\"content\": \"这个是文档3\",\"vector\": [0.31, 0.32, 0.33, 0.34, 0.35, 0.36, 0.37, 0.38]} {\"index\": {\"_index\": \"es-embedding-test\",\"_id\": \"5\"}} {\"content\": \"这个是文档4\",\"vector\": [0.41, 0.42, 0.43, 0.44, 0.45, 0.46, 0.47, 0.48]}

4.3 相似性查询

使用 KNN 检索文档，Elasticsearch 使用 HNSW 算法来支持高效的 kNN 搜索，像大多数 kNN 算法一样，HNSW 是一种近似方法，它牺牲了结果准确性以换取速度的提升。

POST /es-embedding-test/_search{ \"retriever\": { \"knn\": { \"field\": \"vector\", \"query_vector\": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8], \"k\": 3, \"num_candidates\": 5 } }}

参数解析：

• k：top-k，返回的最相关结果数量。

• num_candidates：可选，从每个分片选取的候选文档数（越大越精确，但性能开销越高）。

返回示例：

{ \"took\": 79, \"timed_out\": false, \"_shards\": { \"total\": 1, \"successful\": 1, \"skipped\": 0, \"failed\": 0 }, \"hits\": { \"total\": { \"value\": 3, \"relation\": \"eq\" }, \"max_score\": 0.9999957, \"hits\": [ { \"_index\": \"es-embedding-test\", \"_id\": \"14R115UBPMUIQU8gkFHt\", \"_score\": 0.9999957, \"_source\": { \"content\": \"索引单个文档\", \"vector\": [ 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8 ] } }, { \"_index\": \"es-embedding-test\", \"_id\": \"2\", \"_score\": 0.9723628, \"_source\": { \"content\": \"这个是文档1\", \"vector\": [ 0.11, 0.12, 0.13, 0.14, 0.15, 0.16, 0.17, 0.18 ] } }, { \"_index\": \"es-embedding-test\", \"_id\": \"3\", \"_score\": 0.9716257, \"_source\": { \"content\": \"这个是文档2\", \"vector\": [ 0.21, 0.22, 0.23, 0.24, 0.25, 0.26, 0.27, 0.28 ] } } ] }}

参数解析：

• _score：为相似度得分（余弦相似度）。

5. 搭建文本嵌入工作流

工作流使用的嵌入模型是阿里百炼的 text-embedding-v3 ，需要申请百炼的 api_key。

5.1 流程图

下面是工作流的流程图：

流程解析：

• split_text：文本切割节点，将文档内容切分成文本块，这篇文章使用 Langchain 自带的RecursiveCharacterTextSplitter 切割方式，此外 Langchain 还提供一下方式：

  

• store_embedding：向量存储节点，负责文本向量化并存储到 ES 中。

• search_similar_text：相似性查询节点，负责进行相似性查询。

5.2 准备

5.2.1 安装python包

安装必要 python 包

# 安装Langchain、Langgraphpip install langchain==0.3.0 pip install -U langgraph# 安装langchain es包pip install -U langchain-elasticsearch 

5.2.2 重新建ES索引

上面测试的索引，向量维度为8，我们使用的向量模型维度为1024，需要重新创建。

# 先删除原来的DELETE /es-embedding-test# 新增PUT /es-embedding-test{ \"mappings\": { \"properties\": { \"vector\": { \"type\": \"dense_vector\", \"dims\": 1024, \"index\": true, \"similarity\": \"cosine\" }, \"content\": { \"type\": \"text\" }, \"chunk_id\":{ \"type\":\"integer\" } } }}

5.3 代码

5.3.1 定义向量模型

model 为使用的嵌入模型名，dashscope_api_key 为百炼平台的 api_key。

embeddings = DashScopeEmbeddings( model=embedding_model, dashscope_api_key=api_key,)

5.3.2 定义VectorStore

index_name 为 ES 索引名称。

vector_store = ElasticsearchStore( embedding=embeddings, index_name=index_name, es_url=\"http://localhost:9200\",)

5.3.3 完整代码

from dataclasses import dataclass, fieldfrom typing import Listfrom langchain_community.embeddings import DashScopeEmbeddingsfrom langchain_core.documents import Documentfrom langchain_elasticsearch import ElasticsearchStorefrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langgraph.graph import StateGraph# 配置参数api_key = \"\"embedding_model = \"text-embedding-v3\"index_name = \"es-embedding-test\"# 初始化嵌入模型embeddings = DashScopeEmbeddings( model=embedding_model, dashscope_api_key=api_key,)# 创建Elasticsearch向量存储vector_store = ElasticsearchStore( embedding=embeddings, index_name=index_name, es_url=\"http://localhost:9200\",)# 定义工作流状态类@dataclassclass WorkflowState: query_text: str # 查询文本 doc_content: str # 待处理的文档内容 chunks: List[Document] = field(default_factory=list) # 存储分割后的文档块 results: List[Document] = field(default_factory=list) # 存储相似性搜索结果# 定义文本分割节点def split_text(state: WorkflowState) -> WorkflowState: \"\"\"将文档内容分割成多个文本块\"\"\" # 创建文本分割器 text_splitter = RecursiveCharacterTextSplitter(chunk_size=60, chunk_overlap=5) # 分割文本并创建文档对象 texts = text_splitter.split_text(state.doc_content) state.chunks = [Document( page_content=text, metadata={\"chunk_id\": i, \"content\": text} ) for i, text in enumerate(texts)] print(f\"文本分割完成，共生成 {len(state.chunks)} 个文本块\") return state# 定义存储嵌入向量节点def store_embedding(state: WorkflowState) -> WorkflowState: \"\"\"将文本块的嵌入向量存储到Elasticsearch\"\"\" try: if state.chunks: # 批量添加文档到向量存储 vector_store.add_documents(state.chunks) print(f\"成功存储 {len(state.chunks)} 个文档到Elasticsearch\") except Exception as e: print(f\"存储嵌入向量时出错: {str(e)}\") return state# 定义相似性搜索节点def search_similar_text(state: WorkflowState) -> WorkflowState: \"\"\"在Elasticsearch中执行相似性搜索\"\"\" try: if state.query_text: # 执行相似性搜索,返回最相似的5个结果 state.results = vector_store.similarity_search(state.query_text, k=5) print(f\"找到 {len(state.results)} 个相似文档\") except Exception as e: print(f\"执行相似性搜索时出错: {str(e)}\") return state# 创建 LangGraph 工作流graph = StateGraph(WorkflowState)graph.add_node(\"split\", split_text)graph.add_node(\"store\", store_embedding)graph.add_node(\"search\", search_similar_text)graph.set_entry_point(\"split\")graph.add_edge(\"split\", \"store\")graph.add_edge(\"store\", \"search\")# 编译工作流workflow = graph.compile()# 运行工作流query_text = \"LangGraph\"doc_content = \"\"\" LangGraph 是一个强大的AI工作流库，用于构建和执行复杂任务。 它支持多种数据处理方式，并且可以与不同的存储系统集成。 LangGraph提供了灵活的工作流定义方式，可以轻松实现各种AI应用场景。\"\"\"initial_state = WorkflowState(query_text=query_text, doc_content=doc_content)final_state = workflow.invoke(initial_state)print(\"相似文本:\", final_state[\"results\"])

5.3.4 测试

代码运行结果：

6. 参考文档

• 自带密集向量嵌入到 Elasticsearch
• DashScope | 🦜️🔗 LangChain
• Elasticsearch Service 十亿级高性能向量检索