Skip to main content
Open In ColabOpen on GitHub

Vespa

Vespa is a fully featured search engine and vector database. It supports vector search (ANN), lexical search, and search in structured data, all in the same query.

此笔记本展示了如何将 Vespa.ai 用作 LangChain 向量存储。

您需要使用 pip install -qU langchain-community 安装 langchain-community 才能使用此集成

为了创建向量存储，我们使用 pyvespa 来创建一个 Vespa 服务的连接。

%pip install --upgrade --quiet  pyvespa

使用 pyvespa 包，您可以连接到一个 Vespa Cloud 实例 或本地的 Docker 实例。 在这里，我们将创建一个新的 Vespa 应用程序，并使用 Docker 进行部署。

创建一个 Vespa 应用程序​

首先，我们需要创建一个应用程序包：

from vespa.package import ApplicationPackage, Field, RankProfile

app_package = ApplicationPackage(name="testapp")
app_package.schema.add_fields(
    Field(
        name="text", type="string", indexing=["index", "summary"], index="enable-bm25"
    ),
    Field(
        name="embedding",
        type="tensor<float>(x[384])",
        indexing=["attribute", "summary"],
        attribute=["distance-metric: angular"],
    ),
)
app_package.schema.add_rank_profile(
    RankProfile(
        name="default",
        first_phase="closeness(field, embedding)",
        inputs=[("query(query_embedding)", "tensor<float>(x[384])")],
    )
)

这将设置一个 Vespa 应用程序，其中每个文档的模式包含两个字段：text 用于存储文档文本，embedding 用于存储嵌入向量。字段 text 设置为使用 BM25 索引以实现高效的文本检索，我们稍后会看到如何使用它以及混合搜索。

embedding 字段被设置为一个长度为 384 的向量，用于存储文本的嵌入表示。有关 Vespa 中张量的更多信息，请参阅 Vespa 的张量指南。

最后，我们添加一个 排名配置文件 来 指示 Vespa 如何对文档进行排序。在这里，我们使用一个 最近邻搜索 来设置它。

现在我们可以将此应用程序本地部署。

from vespa.deployment import VespaDocker

vespa_docker = VespaDocker()
vespa_app = vespa_docker.deploy(application_package=app_package)

这将部署并创建与 Vespa 服务的连接。如果您已经有一个 Vespa 应用程序在运行，例如在云中，请参考 PyVespa 应用程序以了解如何连接。

创建Vespa向量存储​

现在，让我们加载一些文档：

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter

loader = TextLoader("../../how_to/state_of_the_union.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

from langchain_community.embeddings.sentence_transformer import (
    SentenceTransformerEmbeddings,
)

embedding_function = SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2")

在这里，我们还设置了本地句子嵌入器，将文本转换为嵌入向量。 也可以使用 OpenAI 嵌入，但需要将向量长度更新为 1536，以反映该嵌入更大的尺寸。

为了将这些数据输入到 Vespa，我们需要配置向量存储如何映射到 Vespa 应用程序中的字段。然后，我们可以直接从这组文档创建向量存储。

vespa_config = dict(
    page_content_field="text",
    embedding_field="embedding",
    input_field="query_embedding",
)

from langchain_community.vectorstores import VespaStore

db = VespaStore.from_documents(docs, embedding_function, app=vespa_app, **vespa_config)
API 参考：VespaStore

这创建了一个 Vespa 向量存储，并将该组文档输入到 Vespa 中。向量存储负责为每个文档调用嵌入函数，并将其插入数据库中。

我们现在可以查询向量存储了：

query = "What did the president say about Ketanji Brown Jackson"
results = db.similarity_search(query)

print(results[0].page_content)

这将使用上面给出的嵌入函数为查询创建一个表示，并使用该表示搜索 Vespa。请注意，这将使用 default 排名函数，我们在上面的应用程序包中设置了它。您可以使用 ranking 到 similarity_search 之间的参数来指定要使用的排名函数。

有关更多信息，请参阅pyvespa 文档。

这涵盖了 LangChain 中 Vespa 存储的基本用法。 现在你可以返回结果，并继续在 LangChain 中使用它们。

更新文档​

另一种调用 from_documents 的方法是直接创建向量存储并从该存储中调用 add_texts。这种方法也可以用于更新文档：

query = "What did the president say about Ketanji Brown Jackson"
results = db.similarity_search(query)
result = results[0]

result.page_content = "UPDATED: " + result.page_content
db.add_texts([result.page_content], [result.metadata], result.metadata["id"])

results = db.similarity_search(query)
print(results[0].page_content)

但是，pyvespa 库包含用于操作 Vespa 上内容的方法，您可以直接使用它们。

删除文档​

您可以使用 delete 函数删除文档：

result = db.similarity_search(query)
# docs[0].metadata["id"] == "id:testapp:testapp::32"

db.delete(["32"])
result = db.similarity_search(query)
# docs[0].metadata["id"] != "id:testapp:testapp::32"

再次强调，pyvespa 连接包含用于删除文档的方法。

返回带有分数​

similarity_search 方法仅按相关性顺序返回文档。要检索实际分数：

results = db.similarity_search_with_score(query)
result = results[0]
# result[1] ~= 0.463

这是使用嵌入模型 "all-MiniLM-L6-v2" 并采用余弦距离函数（由应用程序函数中的参数 angular 给出）的结果。

不同的嵌入函数需要不同的距离函数，Vespa 需要知道在排序文档时使用哪种距离函数。 请参阅 关于距离函数的文档 以获取更多信息。

作为检索器​

要将此向量存储用作 LangChain 检索器 ，只需调用 as_retriever 函数，这是一个标准的向量存储方法：

db = VespaStore.from_documents(docs, embedding_function, app=vespa_app, **vespa_config)
retriever = db.as_retriever()
query = "What did the president say about Ketanji Brown Jackson"
results = retriever.invoke(query)

# results[0].metadata["id"] == "id:testapp:testapp::32"

这使得可以从向量存储中进行更通用、无结构的检索。

元数据

到目前为止，我们只使用了文本及其嵌入向量。文档通常包含额外的信息，在LangChain中被称为元数据。

Vespa 可以通过将字段添加到应用程序包中包含许多不同类型的字段。

app_package.schema.add_fields(
    # ...
    Field(name="date", type="string", indexing=["attribute", "summary"]),
    Field(name="rating", type="int", indexing=["attribute", "summary"]),
    Field(name="author", type="string", indexing=["attribute", "summary"]),
    # ...
)
vespa_app = vespa_docker.deploy(application_package=app_package)

我们可以在文档中添加一些元数据字段：

# Add metadata
for i, doc in enumerate(docs):
    doc.metadata["date"] = f"2023-{(i % 12)+1}-{(i % 28)+1}"
    doc.metadata["rating"] = range(1, 6)[i % 5]
    doc.metadata["author"] = ["Joe Biden", "Unknown"][min(i, 1)]

并让 Vespa 向量存储了解这些字段：

vespa_config.update(dict(metadata_fields=["date", "rating", "author"]))

现在，当搜索这些文档时，这些字段将被返回。 此外，这些字段可以进行筛选：

db = VespaStore.from_documents(docs, embedding_function, app=vespa_app, **vespa_config)
query = "What did the president say about Ketanji Brown Jackson"
results = db.similarity_search(query, filter="rating > 3")
# results[0].metadata["id"] == "id:testapp:testapp::34"
# results[0].metadata["author"] == "Unknown"

自定义查询​

如果默认的相似性搜索行为不符合你的需求，你可以随时提供自己的查询。因此，你不需要向向量存储提供所有配置，只需自己编写即可。

首先，让我们在应用程序中添加一个 BM25 排名函数：

from vespa.package import FieldSet

app_package.schema.add_field_set(FieldSet(name="default", fields=["text"]))
app_package.schema.add_rank_profile(RankProfile(name="bm25", first_phase="bm25(text)"))
vespa_app = vespa_docker.deploy(application_package=app_package)
db = VespaStore.from_documents(docs, embedding_function, app=vespa_app, **vespa_config)

然后，基于BM25进行常规文本搜索：

query = "What did the president say about Ketanji Brown Jackson"
custom_query = {
    "yql": "select * from sources * where userQuery()",
    "query": query,
    "type": "weakAnd",
    "ranking": "bm25",
    "hits": 4,
}
results = db.similarity_search_with_score(query, custom_query=custom_query)
# results[0][0].metadata["id"] == "id:testapp:testapp::32"
# results[0][1] ~= 14.384

Vespa 的强大搜索和查询功能可以通过使用自定义查询来实现。有关更多详细信息，请参阅 Vespa 文档中的Query API。

混合搜索意味着同时使用基于经典术语的搜索（如 BM25）和向量搜索，并将结果结合起来。我们需要在 Vespa 上为混合搜索创建一个新的排名配置文件：

app_package.schema.add_rank_profile(
    RankProfile(
        name="hybrid",
        first_phase="log(bm25(text)) + 0.5 * closeness(field, embedding)",
        inputs=[("query(query_embedding)", "tensor<float>(x[384])")],
    )
)
vespa_app = vespa_docker.deploy(application_package=app_package)
db = VespaStore.from_documents(docs, embedding_function, app=vespa_app, **vespa_config)

在这里，我们根据文档的BM25得分和距离得分的组合对其进行评分。我们可以使用自定义查询进行查询：

query = "What did the president say about Ketanji Brown Jackson"
query_embedding = embedding_function.embed_query(query)
nearest_neighbor_expression = (
    "{targetHits: 4}nearestNeighbor(embedding, query_embedding)"
)
custom_query = {
    "yql": f"select * from sources * where {nearest_neighbor_expression} and userQuery()",
    "query": query,
    "type": "weakAnd",
    "input.query(query_embedding)": query_embedding,
    "ranking": "hybrid",
    "hits": 4,
}
results = db.similarity_search_with_score(query, custom_query=custom_query)
# results[0][0].metadata["id"], "id:testapp:testapp::32")
# results[0][1] ~= 2.897

Vespa中的原生嵌入器​

到目前为止，我们使用 Python 中的嵌入函数为文本提供嵌入。Vespa 原生支持嵌入函数，因此您可以将此计算推迟到 Vespa 中。一个好处是，如果您有大量文档集合，可以在嵌入文档时使用 GPU。

请参阅 Vespa 嵌入 以获取更多信息。

首先，我们需要修改我们的应用程序包：

from vespa.package import Component, Parameter

app_package.components = [
    Component(
        id="hf-embedder",
        type="hugging-face-embedder",
        parameters=[
            Parameter("transformer-model", {"path": "..."}),
            Parameter("tokenizer-model", {"url": "..."}),
        ],
    )
]
Field(
    name="hfembedding",
    type="tensor<float>(x[384])",
    is_document_field=False,
    indexing=["input text", "embed hf-embedder", "attribute", "summary"],
    attribute=["distance-metric: angular"],
)
app_package.schema.add_rank_profile(
    RankProfile(
        name="hf_similarity",
        first_phase="closeness(field, hfembedding)",
        inputs=[("query(query_embedding)", "tensor<float>(x[384])")],
    )
)

请参考嵌入文档，了解如何将嵌入模型和分词器添加到应用程序中。请注意，hfembedding字段包含使用hf-embedder进行嵌入的说明。

现在我们可以使用自定义查询进行查询：

query = "What did the president say about Ketanji Brown Jackson"
nearest_neighbor_expression = (
    "{targetHits: 4}nearestNeighbor(internalembedding, query_embedding)"
)
custom_query = {
    "yql": f"select * from sources * where {nearest_neighbor_expression}",
    "input.query(query_embedding)": f'embed(hf-embedder, "{query}")',
    "ranking": "internal_similarity",
    "hits": 4,
}
results = db.similarity_search_with_score(query, custom_query=custom_query)
# results[0][0].metadata["id"], "id:testapp:testapp::32")
# results[0][1] ~= 0.630

请注意，这里的查询包含一个 embed 指令，用于使用与文档相同的模型嵌入查询。

近似最近邻​

在上面的所有示例中，我们使用了精确的最近邻来查找结果。然而，对于大型文档集合来说，这是不可行的，因为必须扫描所有文档才能找到最佳匹配项。为了避免这种情况，我们可以使用 近似最近邻。

首先，我们可以更改嵌入字段以创建 HNSW 索引：

from vespa.package import HNSW

app_package.schema.add_fields(
    Field(
        name="embedding",
        type="tensor<float>(x[384])",
        indexing=["attribute", "summary", "index"],
        ann=HNSW(
            distance_metric="angular",
            max_links_per_node=16,
            neighbors_to_explore_at_insert=200,
        ),
    )
)

这会在嵌入数据上创建一个HNSW索引，从而实现高效的搜索。设置此选项后，我们可以通过将 approximate 参数设为 True 来轻松使用 ANN 进行搜索：

query = "What did the president say about Ketanji Brown Jackson"
results = db.similarity_search(query, approximate=True)
# results[0][0].metadata["id"], "id:testapp:testapp::32")

这涵盖了 LangChain 中 Vespa 向量存储的大部分功能。