DatabricksVectorSearch

Databricks Vector Search 是一个无服务器相似性搜索引擎，可用于将数据的向量表示形式（包括元数据）存储在向量数据库中。使用矢量搜索，可以从 Unity Catalog 管理的 Delta 表创建自动更新的矢量搜索索引，并使用简单的 API 查询它们以返回最相似的矢量。

此笔记本演示如何将 LangChain 与 Databricks 向量搜索结合使用。

设置

要访问 Databricks 模型，您需要创建一个 Databricks 帐户，设置凭据（仅当您位于 Databricks 工作区之外时）并安装所需的包。

凭据（仅当你在 Databricks 外部时）

如果您在 Databricks 中运行 LangChain 应用程序，则可以跳过此步骤。

否则，需要手动将 Databricks 工作区主机名和个人访问令牌设置为DATABRICKS_HOST和DATABRICKS_TOKEN环境变量。有关如何获取访问令牌的信息，请参阅身份验证文档。

import getpass
import os

os.environ["DATABRICKS_HOST"] = "https://your-databricks-workspace"
if "DATABRICKS_TOKEN" not in os.environ:
    os.environ["DATABRICKS_TOKEN"] = getpass.getpass(
        "Enter your Databricks access token: "
    )

安装

LangChain Databricks 集成位于databricks-langchain包。

%pip install -qU databricks-langchain

创建 Vector Search 终端节点和索引（如果尚未创建）

在本部分中，我们将使用客户端 SDK 创建 Databricks Vector Search 终结点和索引。

如果您已经有一个终端节点和一个索引，则可以跳过该部分并直接进入 “Instantiation” 部分。

首先，实例化 Databricks VectorSearch 客户端：

from databricks.vector_search.client import VectorSearchClient

client = VectorSearchClient()

接下来，我们将创建一个新的 VectorSearch 终端节点。

endpoint_name = "<your-endpoint-name>"

client.create_endpoint(name=endpoint_name, endpoint_type="STANDARD")

最后，我们将创建一个可在终端节点上查询的索引。Databricks Vector Search 中有两种类型的索引，而DatabricksVectorSearch类支持这两种用例。

• Delta Sync Index 自动与源 Delta 表同步，随着 Delta 表中的基础数据的变化，自动以增量方式更新索引。

• Direct Vector Access Index 支持直接读取和写入 vector 和元数据。用户负责使用 REST API 或 Python SDK 更新此表。

此外，对于 delta-sync index，可以选择使用 Databricks 托管的嵌入或自管理的嵌入（通过 LangChain 嵌入类）。

以下代码创建直接访问索引。请参阅 Databricks 文档，了解创建其他类型索引的说明。

index_name = "<your-index-name>"  # Format: "<catalog>.<schema>.<index-name>"

index = client.create_direct_access_index(
    endpoint_name=endpoint_name,
    index_name=index_name,
    primary_key="id",
    # Dimension of the embeddings. Please change according to the embedding model you are using.
    embedding_dimension=3072,
    # A column to store the embedding vectors for the text data
    embedding_vector_column="text_vector",
    schema={
        "id": "string",
        "text": "string",
        "text_vector": "array<float>",
        # Optional metadata columns
        "source": "string",
    },
)

index.describe()

实例

的实例化DatabricksVectorSearch略有不同，具体取决于索引是使用 Databricks 托管的嵌入还是自托管的嵌入，即您选择的 LangChain Embeddings 对象。

如果将增量同步索引与 Databricks 托管的嵌入一起使用：

from databricks_langchain import DatabricksVectorSearch

vector_store = DatabricksVectorSearch(
    endpoint=endpoint_name,
    index_name=index_name,
)

如果您使用的是具有自管理嵌入的直接访问索引或增量同步索引，则 您还需要在源表中提供 embedding model 和 text 列，以便 用于嵌入：

选择嵌入模型：

OpenAI

天蓝色

Google

AWS

Hugging Face

奥拉马

Cohere

MistralAI

诺米克

英伟达

航海 AI

IBM 沃森

假

pip install -qU langchain-openai

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

vector_store = DatabricksVectorSearch(
    endpoint=endpoint_name,
    index_name=index_name,
    embedding=embeddings,
    # The column name in the index that contains the text data to be embedded
    text_column="document_content",
)

管理矢量存储

将项目添加到向量存储

注意：通过以下方式将项目添加到 vector storeadd_documentsmethod 仅支持直接访问索引。

from langchain_core.documents import Document

document_1 = Document(page_content="foo", metadata={"source": "https://example.com"})

document_2 = Document(page_content="bar", metadata={"source": "https://example.com"})

document_3 = Document(page_content="baz", metadata={"source": "https://example.com"})

documents = [document_1, document_2, document_3]

vector_store.add_documents(documents=documents, ids=["1", "2", "3"])

API 参考：文档

['1', '2', '3']

从 vector store 中删除项目

注意：通过以下方式将项目删除到 vector storedeletemethod 仅支持直接访问索引。

vector_store.delete(ids=["3"])

True

查询向量存储

创建矢量存储并添加相关文档后，您很可能希望在链或代理运行期间对其进行查询。

直接查询

可以按如下方式执行简单的相似性搜索：

results = vector_store.similarity_search(
    query="thud", k=1, filter={"source": "https://example.com"}
)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

* foo [{'id': '1'}]

注意：默认情况下，相似性搜索仅返回主键和文本列。如果要检索与文档关联的自定义元数据，请在columns参数。

vector_store = DatabricksVectorSearch(
    endpoint=endpoint_name,
    index_name=index_name,
    embedding=embeddings,
    text_column="text",
    columns=["source"],
)

results = vector_store.similarity_search(query="thud", k=1)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

* foo [{'source': 'https://example.com', 'id': '1'}]

如果要执行相似性搜索并接收相应的分数，可以运行：

results = vector_store.similarity_search_with_score(
    query="thud", k=1, filter={"source": "https://example.com"}
)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

* [SIM=0.414035] foo [{'source': 'https://example.com', 'id': '1'}]

通过转换为 retriever 进行查询

您还可以将 vector store 转换为检索器，以便在您的链中更轻松地使用。

retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 1})
retriever.invoke("thud")

[Document(metadata={'source': 'https://example.com', 'id': '1'}, page_content='foo')]

用于检索增强生成

有关如何使用此向量存储进行检索增强生成 （RAG） 的指南，请参阅以下部分：

• 教程
• 作方法：RAG 的问答
• 检索概念文档

API 参考

有关所有 DatabricksVectorSearch 功能和配置的详细文档，请转到 API 参考：https://api-docs.databricks.com/python/databricks-ai-bridge/latest/databricks_langchain.html#databricks_langchain.DatabricksVectorSearch

相关

• 矢量存储概念指南
• Vector store 操作指南