跳至主要内容

Elasticsearch

Elasticsearch 是一款分布式、RESTful 搜索和分析引擎,能够执行向量和词法搜索。它构建在 Apache Lucene 库之上。

此笔记本展示了如何使用与Elasticsearch 向量存储相关的功能。

设置

为了使用Elasticsearch 向量搜索,您必须安装langchain-elasticsearch 包。

%pip install -qU langchain-elasticsearch

凭据

主要有两种方法可以设置用于

  1. Elastic Cloud:Elastic Cloud 是一种托管的 Elasticsearch 服务。注册 免费试用

要连接到不需要登录凭据的 Elasticsearch 实例(以启用安全性的方式启动 docker 实例),请将 Elasticsearch URL 和索引名称以及嵌入对象一起传递给构造函数。

  1. 本地安装 Elasticsearch:通过在本地运行 Elasticsearch 来开始使用它。最简单的方法是使用官方的 Elasticsearch Docker 镜像。有关更多信息,请参阅Elasticsearch Docker 文档

通过 Docker 运行 Elasticsearch

示例:运行一个禁用安全性的单节点 Elasticsearch 实例。不建议在生产环境中使用此方法。

%docker run -p 9200:9200 -e "discovery.type=single-node" -e "xpack.security.enabled=false" -e "xpack.security.http.ssl.enabled=false" docker.elastic.co/elasticsearch/elasticsearch:8.12.1

使用身份验证运行

对于生产环境,我们建议您启用安全性后运行。要使用登录凭据连接,您可以使用参数es_api_keyes_useres_password

pip install -qU langchain-openai
import getpass

os.environ["OPENAI_API_KEY"] = getpass.getpass()

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
from langchain_elasticsearch import ElasticsearchStore

elastic_vector_search = ElasticsearchStore(
es_url="http://localhost:9200",
index_name="langchain_index",
embedding=embeddings,
es_user="elastic",
es_password="changeme",
)
API 参考:ElasticsearchStore

如何获取默认“elastic”用户的密码?

要获取默认“elastic”用户的 Elastic Cloud 密码

  1. 登录到 Elastic Cloud 控制台 https://cloud.elastic.co
  2. 转到“安全”>“用户”
  3. 找到“elastic”用户并点击“编辑”
  4. 点击“重置密码”
  5. 按照提示重置密码

如何获取 API 密钥?

要获取 API 密钥

  1. 登录到 Elastic Cloud 控制台 https://cloud.elastic.co
  2. 打开 Kibana 并转到堆栈管理> API 密钥
  3. 点击“创建 API 密钥”
  4. 输入 API 密钥的名称,然后点击“创建”
  5. 复制 API 密钥并将其粘贴到api_key 参数中

Elastic Cloud

要连接到 Elastic Cloud 上的 Elasticsearch 实例,您可以使用es_cloud_id 参数或es_url 参数。

elastic_vector_search = ElasticsearchStore(
es_cloud_id="<cloud_id>",
index_name="test_index",
embedding=embeddings,
es_user="elastic",
es_password="changeme",
)

如果您希望获得一流的模型调用自动跟踪,还可以通过取消以下注释来设置您的 LangSmith API 密钥

# os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
# os.environ["LANGSMITH_TRACING"] = "true"

初始化

Elasticsearch 在 localhost:9200 上使用 docker 在本地运行。有关如何从 Elastic Cloud 连接到 Elasticsearch 的更多详细信息,请参阅上面有关 连接身份验证 的内容。

from langchain_elasticsearch import ElasticsearchStore

vector_store = ElasticsearchStore(
"langchain-demo", embedding=embeddings, es_url="http://localhost:9201"
)
API 参考:ElasticsearchStore

管理向量存储

将项目添加到向量存储

from uuid import uuid4

from langchain_core.documents import Document

document_1 = Document(
page_content="I had chocalate chip pancakes and scrambled eggs for breakfast this morning.",
metadata={"source": "tweet"},
)

document_2 = Document(
page_content="The weather forecast for tomorrow is cloudy and overcast, with a high of 62 degrees.",
metadata={"source": "news"},
)

document_3 = Document(
page_content="Building an exciting new project with LangChain - come check it out!",
metadata={"source": "tweet"},
)

document_4 = Document(
page_content="Robbers broke into the city bank and stole $1 million in cash.",
metadata={"source": "news"},
)

document_5 = Document(
page_content="Wow! That was an amazing movie. I can't wait to see it again.",
metadata={"source": "tweet"},
)

document_6 = Document(
page_content="Is the new iPhone worth the price? Read this review to find out.",
metadata={"source": "website"},
)

document_7 = Document(
page_content="The top 10 soccer players in the world right now.",
metadata={"source": "website"},
)

document_8 = Document(
page_content="LangGraph is the best framework for building stateful, agentic applications!",
metadata={"source": "tweet"},
)

document_9 = Document(
page_content="The stock market is down 500 points today due to fears of a recession.",
metadata={"source": "news"},
)

document_10 = Document(
page_content="I have a bad feeling I am going to get deleted :(",
metadata={"source": "tweet"},
)

documents = [
document_1,
document_2,
document_3,
document_4,
document_5,
document_6,
document_7,
document_8,
document_9,
document_10,
]
uuids = [str(uuid4()) for _ in range(len(documents))]

vector_store.add_documents(documents=documents, ids=uuids)
API 参考:Document
['21cca03c-9089-42d2-b41c-3d156be2b519',
'a6ceb967-b552-4802-bb06-c0e95fce386e',
'3a35fac4-e5f0-493b-bee0-9143b41aedae',
'176da099-66b1-4d6a-811b-dfdfe0808d30',
'ecfa1a30-3c97-408b-80c0-5c43d68bf5ff',
'c0f08baa-e70b-4f83-b387-c6e0a0f36f73',
'489b2c9c-1925-43e1-bcf0-0fa94cf1cbc4',
'408c6503-9ba4-49fd-b1cc-95584cd914c5',
'5248c899-16d5-4377-a9e9-736ca443ad4f',
'ca182769-c4fc-4e25-8f0a-8dd0a525955c']

从向量存储中删除项目

vector_store.delete(ids=[uuids[-1]])
True

查询向量存储

创建向量存储并添加相关文档后,您很可能希望在运行链或代理时查询它。这些示例还展示了如何在搜索时使用过滤。

直接查询

可以按如下方式执行对元数据进行过滤的简单相似性搜索

results = vector_store.similarity_search(
query="LangChain provides abstractions to make working with LLMs easy",
k=2,
filter=[{"term": {"metadata.source.keyword": "tweet"}}],
)
for res in results:
print(f"* {res.page_content} [{res.metadata}]")
* Building an exciting new project with LangChain - come check it out! [{'source': 'tweet'}]
* LangGraph is the best framework for building stateful, agentic applications! [{'source': 'tweet'}]

带分数的相似性搜索

如果要执行相似性搜索并接收相应的得分,可以运行

results = vector_store.similarity_search_with_score(
query="Will it be hot tomorrow",
k=1,
filter=[{"term": {"metadata.source.keyword": "news"}}],
)
for doc, score in results:
print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")
* [SIM=0.765887] The weather forecast for tomorrow is cloudy and overcast, with a high of 62 degrees. [{'source': 'news'}]

通过转换为检索器进行查询

您还可以将向量存储转换为检索器,以便在链中更轻松地使用。

retriever = vector_store.as_retriever(
search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.2}
)
retriever.invoke("Stealing from the bank is a crime")
[Document(metadata={'source': 'news'}, page_content='Robbers broke into the city bank and stole $1 million in cash.'),
Document(metadata={'source': 'news'}, page_content='The stock market is down 500 points today due to fears of a recession.'),
Document(metadata={'source': 'website'}, page_content='Is the new iPhone worth the price? Read this review to find out.'),
Document(metadata={'source': 'tweet'}, page_content='Building an exciting new project with LangChain - come check it out!')]

检索增强生成的使用

有关如何将此向量存储用于检索增强生成 (RAG) 的指南,请参阅以下部分

常见问题

问题:在将文档索引到 Elasticsearch 时,我遇到了超时错误。如何解决此问题?

一个可能的问题是您的文档可能需要更长的时间才能索引到 Elasticsearch 中。ElasticsearchStore 使用 Elasticsearch 批量 API,该 API 有一些您可以调整的默认设置,以减少超时错误的可能性。

当您使用 SparseVectorRetrievalStrategy 时,这也是一个好主意。

默认值为

  • 块大小: 500
  • max_chunk_bytes:100MB

要调整这些设置,您可以将chunk_sizemax_chunk_bytes 参数传递给 ElasticsearchStore 的add_texts 方法。

    vector_store.add_texts(
texts,
bulk_kwargs={
"chunk_size": 50,
"max_chunk_bytes": 200000000
}
)

升级到 ElasticsearchStore

如果您已经在基于 langchain 的项目中使用 Elasticsearch,则可能正在使用旧的实现:ElasticVectorSearchElasticKNNSearch,这些实现现已弃用。我们引入了一种名为ElasticsearchStore 的新实现,它更灵活且更易于使用。此笔记本将指导您完成升级到新实现的过程。

新增功能?

新的实现现在是一个名为ElasticsearchStore的类,它可以通过策略用于近似密集向量、精确密集向量、稀疏向量(ELSER)、BM25检索和混合检索。

我正在使用 ElasticKNNSearch

旧实现


from langchain_community.vectorstores.elastic_vector_search import ElasticKNNSearch

db = ElasticKNNSearch(
elasticsearch_url="http://localhost:9200",
index_name="test_index",
embedding=embedding
)

新实现


from langchain_elasticsearch import ElasticsearchStore, DenseVectorStrategy

db = ElasticsearchStore(
es_url="http://localhost:9200",
index_name="test_index",
embedding=embedding,
# if you use the model_id
# strategy=DenseVectorStrategy(model_id="test_model")
# if you use hybrid search
# strategy=DenseVectorStrategy(hybrid=True)
)

我正在使用 ElasticVectorSearch

旧实现


from langchain_community.vectorstores.elastic_vector_search import ElasticVectorSearch

db = ElasticVectorSearch(
elasticsearch_url="http://localhost:9200",
index_name="test_index",
embedding=embedding
)

API 参考:ElasticVectorSearch

新实现


from langchain_elasticsearch import ElasticsearchStore, DenseVectorScriptScoreStrategy

db = ElasticsearchStore(
es_url="http://localhost:9200",
index_name="test_index",
embedding=embedding,
strategy=DenseVectorScriptScoreStrategy()
)

db.client.indices.delete(
index="test-metadata, test-elser, test-basic",
ignore_unavailable=True,
allow_no_indices=True,
)

API 参考

有关所有ElasticSearchStore功能和配置的详细文档,请访问 API 参考:https://python.langchain.ac.cn/v0.2/api_reference/elasticsearch/vectorstores/langchain_elasticsearch.vectorstores.ElasticsearchStore.html


此页面是否有帮助?


您还可以留下详细的反馈 在 GitHub 上.