向量存储集成测试#

class langchain_tests.integration_tests.vectorstores.VectorStoreIntegrationTests[source]#

向量存储集成测试的基类。

实现者应该继承这个测试套件,并提供一个夹具,为每个测试返回一个空的向量存储。

夹具应使用get_embeddings方法来获取预定义的嵌入模型,该模型应在此测试套件中使用。

这是一个模板:

from typing import Generator

import pytest
from langchain_core.vectorstores import VectorStore
from langchain_parrot_link.vectorstores import ParrotVectorStore
from langchain_tests.integration_tests.vectorstores import VectorStoreIntegrationTests


class TestParrotVectorStore(VectorStoreIntegrationTests):
    @pytest.fixture()
    def vectorstore(self) -> Generator[VectorStore, None, None]:  # type: ignore
        """Get an empty vectorstore."""
        store = ParrotVectorStore(self.get_embeddings())
        # note: store should be EMPTY at this point
        # if you need to delete data, you may do so here
        try:
            yield store
        finally:
            # cleanup operations, or deleting data
            pass

在夹具中,在yield之前,我们实例化了一个空的向量存储。在finally块中,我们调用任何必要的逻辑以使向量存储恢复到干净状态。

示例:

from typing import Generator

import pytest
from langchain_core.vectorstores import VectorStore
from langchain_tests.integration_tests.vectorstores import VectorStoreIntegrationTests

from langchain_chroma import Chroma


class TestChromaStandard(VectorStoreIntegrationTests):
    @pytest.fixture()
    def vectorstore(self) -> Generator[VectorStore, None, None]:  # type: ignore
        """Get an empty vectorstore for unit tests."""
        store = Chroma(embedding_function=self.get_embeddings())
        try:
            yield store
        finally:
            store.delete_collection()
            pass

请注意,默认情况下我们同时启用同步和异步测试。要禁用其中任何一个,请在子类中将has_synchas_async属性覆盖为False。例如:

class TestParrotVectorStore(VectorStoreIntegrationTests):
     @pytest.fixture()
     def vectorstore(self) -> Generator[VectorStore, None, None]:  # type: ignore
         ...

     @property
     def has_async(self) -> bool:
         return False

注意

个别测试方法的API参考包括故障排除提示。

属性

has_async

可配置属性,用于启用或禁用异步测试。

has_sync

可配置属性,用于启用或禁用同步测试。

方法

get_embeddings()

一个预定义的嵌入模型,应该用于此测试。

test_add_documents(vectorstore)

测试将文档添加到向量存储中。

test_add_documents_async(vectorstore)

测试将文档添加到向量存储中。

test_add_documents_by_id_with_mutation(...)

测试我们可以使用add_documents通过ID进行覆盖。

test_add_documents_by_id_with_mutation_async(...)

测试我们可以使用add_documents通过ID进行覆盖。

test_add_documents_documents(vectorstore)

运行 add_documents 测试。

test_add_documents_documents_async(vectorstore)

运行 add_documents 测试。

test_add_documents_with_existing_ids(vectorstore)

测试带有现有ID的add_documents是幂等的。

test_add_documents_with_existing_ids_async(...)

测试带有现有ID的add_documents是幂等的。

test_add_documents_with_ids_is_idempotent(...)

通过ID添加应该是幂等的。

test_add_documents_with_ids_is_idempotent_async(...)

通过ID添加应该是幂等的。

test_delete_missing_content(vectorstore)

删除缺失的内容不应引发异常。

test_delete_missing_content_async(vectorstore)

删除缺失的内容不应引发异常。

test_deleting_bulk_documents(vectorstore)

测试我们可以一次删除多个文档。

test_deleting_bulk_documents_async(vectorstore)

测试我们可以一次删除多个文档。

test_deleting_documents(vectorstore)

测试从向量存储中删除文档。

test_deleting_documents_async(vectorstore)

测试从向量存储中删除文档。

test_get_by_ids(vectorstore)

测试通过ID获取。

test_get_by_ids_async(vectorstore)

测试通过ID获取。

test_get_by_ids_missing(vectorstore)

测试通过缺失的ID获取数据。

test_get_by_ids_missing_async(vectorstore)

测试通过ID获取时缺少ID的情况。

test_vectorstore_is_empty(vectorstore)

测试向量存储是否为空。

test_vectorstore_is_empty_async(vectorstore)

测试向量存储是否为空。

test_vectorstore_still_empty(vectorstore)

此测试应跟随一个添加文档的测试。

test_vectorstore_still_empty_async(vectorstore)

此测试应跟随一个添加文档的测试。

vectorstore()

获取要测试的向量存储类。

static get_embeddings() Embeddings[source]#

应该用于此测试的预定义嵌入模型。

目前使用的是来自langchain-coreDeterministicFakeEmbedding,它使用numpy根据输入文本的哈希值生成随机数。

生成的嵌入没有意义,但它们是确定性的。

Return type:

Embeddings

test_add_documents(vectorstore: VectorStore) None[source]#

测试将文档添加到向量存储中。

Troubleshooting

如果此测试失败,请检查:

  1. 我们在vectorestore夹具中正确初始化了一个空向量存储。

  2. 调用 .similarity_search 获取前 k 个相似文档时,不会根据分数进行阈值过滤。

  3. 在将原始文档对象添加到向量存储时,我们不会对其进行突变(例如,通过添加ID)。

Parameters:

vectorstore (VectorStore)

Return type:

async test_add_documents_async(vectorstore: VectorStore) None[source]#

测试将文档添加到向量存储中。

Troubleshooting

如果此测试失败,请检查:

  1. 我们正确地初始化了一个空的向量存储在vectorestore夹具中。

  2. 调用 .asimilarity_search 获取前 k 个相似文档时,不会根据分数进行阈值过滤。

  3. 在将原始文档对象添加到向量存储时,我们不会对其进行修改(例如,通过添加ID)。

Parameters:

vectorstore (VectorStore)

Return type:

test_add_documents_by_id_with_mutation(vectorstore: VectorStore) None[source]#

测试我们可以使用add_documents通过ID进行覆盖。

Troubleshooting

如果此测试失败,请检查当使用已存在于向量存储中的ID调用add_documents时,内容是否被更新而不是重复。

Parameters:

vectorstore (VectorStore)

Return type:

async test_add_documents_by_id_with_mutation_async(vectorstore: VectorStore) None[source]#

测试我们可以使用add_documents通过ID进行覆盖。

Troubleshooting

如果此测试失败,请检查当使用向量存储中已存在的ID调用aadd_documents时,内容是否被更新而不是重复。

Parameters:

vectorstore (VectorStore)

Return type:

test_add_documents_documents(vectorstore: VectorStore) None[source]#

运行 add_documents 测试。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并返回与传入ID顺序相同的文档。

还要检查如果没有提供字符串ID,add_documents 是否能正确生成字符串ID。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
def test_add_documents_documents(self, vectorstore: VectorStore) -> None:
    super().test_add_documents_documents(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

async test_add_documents_documents_async(vectorstore: VectorStore) None[source]#

运行 add_documents 测试。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现并返回与传入ID顺序相同的文档。

还要检查如果没有提供字符串ID,aadd_documents 是否能正确生成字符串ID。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
async def test_add_documents_documents(self, vectorstore: VectorStore) -> None:
    await super().test_add_documents_documents(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

test_add_documents_with_existing_ids(vectorstore: VectorStore) None[source]#

测试带有现有ID的add_documents是幂等的。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并返回与传入ID顺序相同的文档。

此测试还验证了:

  1. Document.id字段中指定的ID在添加文档时分配。

  2. 如果某些文档包含ID而其他文档不包含,则为后者生成字符串ID。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
def test_add_documents_with_existing_ids(self, vectorstore: VectorStore) -> None:
    super().test_add_documents_with_existing_ids(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

async test_add_documents_with_existing_ids_async(vectorstore: VectorStore) None[source]#

测试带有现有ID的add_documents是幂等的。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并返回与传入ID顺序相同的文档。

此测试还验证了:

  1. Document.id字段中指定的ID在添加文档时分配。

  2. 如果某些文档包含ID而其他文档不包含,则为后者生成字符串ID。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
async def test_add_documents_with_existing_ids(self, vectorstore: VectorStore) -> None:
    await super().test_add_documents_with_existing_ids(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

test_add_documents_with_ids_is_idempotent(vectorstore: VectorStore) None[source]#

按ID添加应该是幂等的。

Troubleshooting

如果此测试失败,请检查使用相同ID两次添加同一文档是否与添加一次具有相同的效果(即,它不会重复文档)。

Parameters:

vectorstore (VectorStore)

Return type:

async test_add_documents_with_ids_is_idempotent_async(vectorstore: VectorStore) None[source]#

按ID添加应该是幂等的。

Troubleshooting

如果此测试失败,请检查使用相同ID两次添加同一文档是否与添加一次具有相同的效果(即,它不会重复文档)。

Parameters:

vectorstore (VectorStore)

Return type:

test_delete_missing_content(vectorstore: VectorStore) None[source]#

删除缺失的内容不应引发异常。

Troubleshooting

如果此测试失败,请检查delete在删除不存在的ID时不会引发异常。

Parameters:

vectorstore (VectorStore)

Return type:

async test_delete_missing_content_async(vectorstore: VectorStore) None[源代码]#

删除缺失的内容不应引发异常。

Troubleshooting

如果此测试失败,请检查adelete在删除不存在的ID时不会引发异常。

Parameters:

vectorstore (VectorStore)

Return type:

test_deleting_bulk_documents(vectorstore: VectorStore) None[source]#

测试我们可以一次删除多个文档。

Troubleshooting

如果此测试失败,请检查delete在给定ID列表时是否正确删除了多个文档。

Parameters:

vectorstore (VectorStore)

Return type:

async test_deleting_bulk_documents_async(vectorstore: VectorStore) None[source]#

测试我们可以一次删除多个文档。

Troubleshooting

如果此测试失败,请检查adelete在给定ID列表时是否正确删除了多个文档。

Parameters:

vectorstore (VectorStore)

Return type:

test_deleting_documents(vectorstore: VectorStore) None[source]#

测试从向量存储中删除文档。

Troubleshooting

如果此测试失败,请检查add_documents是否保留了通过ids传递的标识符,以及delete是否正确删除了文档。

Parameters:

vectorstore (VectorStore)

Return type:

async test_deleting_documents_async(vectorstore: VectorStore) None[source]#

测试从向量存储中删除文档。

Troubleshooting

如果此测试失败,请检查aadd_documents是否保留了通过ids传递的标识符,以及delete是否正确删除了文档。

Parameters:

vectorstore (VectorStore)

Return type:

test_get_by_ids(vectorstore: VectorStore) None[source]#

测试通过ID获取。

此测试要求向量存储上实现get_by_ids

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并返回与传入ID顺序相同的文档。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
def test_get_by_ids(self, vectorstore: VectorStore) -> None:
    super().test_get_by_ids(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

async test_get_by_ids_async(vectorstore: VectorStore) None[source]#

测试通过ID获取。

此测试要求向量存储上实现get_by_ids

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并返回与传入ID顺序相同的文档。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
async def test_get_by_ids(self, vectorstore: VectorStore) -> None:
    await super().test_get_by_ids(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

test_get_by_ids_missing(vectorstore: VectorStore) None[source]#

测试通过缺失的ID获取数据。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并且在给定不存在的ID时不会引发异常。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
def test_get_by_ids_missing(self, vectorstore: VectorStore) -> None:
    super().test_get_by_ids_missing(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

async test_get_by_ids_missing_async(vectorstore: VectorStore) None[source]#

测试通过缺失的ID获取数据。

Troubleshooting

如果此测试失败,请检查get_by_ids是否已实现,并且在给定不存在的ID时不会引发异常。

注意

get_by_idslangchain-core 版本 0.2.11 中被添加到 VectorStore 接口中。如果难以实现,可以使用 pytest 的 xfail 在测试类上跳过此测试:

@pytest.mark.xfail(reason=("get_by_ids not implemented."))
async def test_get_by_ids_missing(self, vectorstore: VectorStore) -> None:
    await super().test_get_by_ids_missing(vectorstore)
Parameters:

vectorstore (VectorStore)

Return type:

test_vectorstore_is_empty(vectorstore: VectorStore) None[source]#

测试向量存储是否为空。

Troubleshooting

如果此测试失败,请检查测试类(即VectorStoreIntegrationTests的子类)是否在vectorestore夹具中初始化了一个空的向量存储。

Parameters:

vectorstore (VectorStore)

Return type:

async test_vectorstore_is_empty_async(vectorstore: VectorStore) None[source]#

测试向量存储是否为空。

Troubleshooting

如果此测试失败,请检查测试类(即VectorStoreIntegrationTests的子类)是否在vectorestore夹具中初始化了一个空的向量存储。

Parameters:

vectorstore (VectorStore)

Return type:

test_vectorstore_still_empty(vectorstore: VectorStore) None[source]#

此测试应跟随添加文档的测试。

这只是验证了在每个测试之后,夹具被正确设置为空。

Troubleshooting

如果此测试失败,请检查测试类(即VectorStoreIntegrationTests的子类)是否正确地在finally块中清除了向量存储。

Parameters:

vectorstore (VectorStore)

Return type:

async test_vectorstore_still_empty_async(vectorstore: VectorStore) None[source]#

此测试应跟随添加文档的测试。

这只是验证了在每个测试之后,夹具被正确设置为空。

Troubleshooting

如果此测试失败,请检查测试类(即VectorStoreIntegrationTests的子类)是否正确地在finally块中清除了向量存储。

Parameters:

vectorstore (VectorStore)

Return type:

abstract vectorstore() VectorStore[source]#

获取要测试的向量存储类。

返回的向量存储应该是空的。

Return type:

VectorStore