Skip to main content
Open In ColabOpen on GitHub

如何使用 LangChain 索引 API

在此，我们将使用 LangChain 索引 API 查看一个基本的索引工作流程。

索引 API 允许您将来自任何源的文件加载并保持与 向量存储 同步。具体来说，它有助于：

  • 避免将重复内容写入向量存储
  • 避免重写未更改的内容
  • 避免对未更改的内容重新计算嵌入向量

所有这些都应为您节省时间和金钱，同时提高您的向量搜索效果。

重要的是，索引 API 即使对于相对于原始源文档经过多个转换步骤（例如通过文本分块）的文档也能正常工作。

工作原理​

LangChain 索引功能利用一个记录管理器（RecordManager）来跟踪文档写入向量存储库的过程。

在索引内容时，会为每个文档计算哈希值，并将以下信息存储在记录管理器中：

  • document的哈希值（页面内容和元数据的哈希）
  • 编写时间
  • the source id -- each document should include information in its metadata to allow us to determine the ultimate source of this document

删除模式​

在向向量存储索引文档时，可能需要删除向量存储中某些现有文档。在某些情况下，您可能希望删除所有源自与新索引文档相同来源的现有文档。在另一些情况下，您可能希望批量删除所有现有文档。索引 API 的删除模式允许您选择所需的行为：

清理模式去重内容可并行化清理已删除的源文档清理源文档和/或派生文档的变更清理时机
None✅✅❌❌-
Incremental✅✅❌✅Continuously
Full✅❌✅✅At end of indexing
Scoped_Full✅✅❌✅At end of indexing

None 不会执行任何自动清理，允许用户手动清理旧内容。

incremental, full and scoped_full 提供以下自动清理功能：

  • 如果源文档或衍生文档的内容已更改，所有三种模式都将清理（删除）之前的内容版本。
  • 如果源文档已被删除（意味着它不包含在当前正在索引的文档中），则full清理模式将正确地从向量存储中删除它，但incremental和scoped_full模式则不会。

当内容发生更改（例如，源 PDF 文件被修订）时，在索引过程中会有一段时间，新旧两个版本的内容都可能返回给用户。这种情况发生在新内容写入之后，但在旧版本被删除之前。

  • incremental 索引最小化了这段时间，因为它能够持续进行清理，同时写入。
  • full 和 scoped_full 模式会在所有批次写入后执行清理操作。

需求​

  1. 请勿与已独立于索引 API 预先填充内容的存储一起使用，因为记录管理器将不知道之前已插入记录。
  2. 仅适用于支持以下功能的 LangChain vectorstore：
    • 按 ID 添加文档（add_documents 方法，带有 ids 个参数）
    • 按 ID 删除（delete 方法，带有 ids 个参数）

兼容的向量存储：Aerospike, AnalyticDB, AstraDB, AwaDB, AzureCosmosDBNoSqlVectorSearch, AzureCosmosDBVectorSearch, AzureSearch, Bagel, Cassandra, Chroma, CouchbaseVectorStore, DashVector, DatabricksVectorSearch, DeepLake, Dingo, ElasticVectorSearch, ElasticsearchStore, FAISS, HanaDB, Milvus, MongoDBAtlasVectorSearch, MyScale, OpenSearchVectorSearch, PGVector, Pinecone, Qdrant, Redis, Rockset, ScaNN, SingleStoreDB, SupabaseVectorStore, SurrealDBStore, TimescaleVector, Vald, VDMS, Vearch, VespaStore, Weaviate, Yellowbrick, ZepVectorStore, TencentVectorDB, OpenSearchVectorSearch。

注意​

记录管理器依赖于基于时间的机制来确定可以清理的内容（当使用 full、incremental 或 scoped_full 清理模式时）。

如果两个任务连续运行，且第一个任务在时钟时间变更之前完成，则第二个任务可能无法清理内容。

这种情况在实际场景中不太可能成为问题，原因如下：

  1. RecordManager 使用更高分辨率的时间戳。
  2. 数据需要在第一次和第二次任务运行之间发生变化，如果任务之间的时间间隔很短，这种情况变得不太可能。
  3. 索引任务通常需要超过几毫秒。

快速开始​

from langchain.indexes import SQLRecordManager, index
from langchain_core.documents import Document
from langchain_elasticsearch import ElasticsearchStore
from langchain_openai import OpenAIEmbeddings

初始化向量存储并设置嵌入：

collection_name = "test_index"

embedding = OpenAIEmbeddings()

vectorstore = ElasticsearchStore(
    es_url="http://localhost:9200", index_name="test_index", embedding=embedding
)

使用适当的命名空间初始化记录管理器。

建议：使用一个命名空间，该命名空间同时考虑向量存储和向量存储中的集合名称；例如：'redis/my_docs'、'chromadb/my_docs'或'postgres/my_docs'。

namespace = f"elasticsearch/{collection_name}"
record_manager = SQLRecordManager(
    namespace, db_url="sqlite:///record_manager_cache.sql"
)

在使用记录管理器之前，请先创建模式。

record_manager.create_schema()

让我们索引一些测试文档：

doc1 = Document(page_content="kitty", metadata={"source": "kitty.txt"})
doc2 = Document(page_content="doggy", metadata={"source": "doggy.txt"})

索引到空的向量存储：

def _clear():
    """Hacky helper method to clear content. See the `full` mode section to to understand why it works."""
    index([], record_manager, vectorstore, cleanup="full", source_id_key="source")

None 删除模式​

此模式不会自动清理旧版本的内容；但它仍然会处理内容的去重。

_clear()
index(
    [doc1, doc1, doc1, doc1, doc1],
    record_manager,
    vectorstore,
    cleanup=None,
    source_id_key="source",
)
{'num_added': 1, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}
_clear()
index([doc1, doc2], record_manager, vectorstore, cleanup=None, source_id_key="source")
{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

第二次运行时将跳过所有内容:

index([doc1, doc2], record_manager, vectorstore, cleanup=None, source_id_key="source")
{'num_added': 0, 'num_updated': 0, 'num_skipped': 2, 'num_deleted': 0}

"incremental" 删除模式​

_clear()
index(
    [doc1, doc2],
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)
{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

再次索引应导致两个文档都被跳过——同时也跳过嵌入操作！

index(
    [doc1, doc2],
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)
{'num_added': 0, 'num_updated': 0, 'num_skipped': 2, 'num_deleted': 0}

如果我们不提供任何文档且使用增量索引模式，则不会发生任何更改。

index([], record_manager, vectorstore, cleanup="incremental", source_id_key="source")
{'num_added': 0, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

如果我们修改文档，新版本将被写入，所有共享相同源的老版本都将被删除。

changed_doc_2 = Document(page_content="puppy", metadata={"source": "doggy.txt"})
index(
    [changed_doc_2],
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)
{'num_added': 1, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 1}

"full" 删除模式​

在full模式下，用户应向索引函数传递full内容宇宙（universe of content）以供索引。

任何未传入索引函数且存在于向量存储中的文档都将被删除！

此行为在处理源文档的删除时非常有用。

_clear()
all_docs = [doc1, doc2]
index(all_docs, record_manager, vectorstore, cleanup="full", source_id_key="source")
{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

假设有人删除了第一个文档：

del all_docs[0]
all_docs
[Document(page_content='doggy', metadata={'source': 'doggy.txt'})]

使用完整模式也将清理已删除的内容。

index(all_docs, record_manager, vectorstore, cleanup="full", source_id_key="source")
{'num_added': 0, 'num_updated': 0, 'num_skipped': 1, 'num_deleted': 1}

来源​

metadata 属性包含一个名为 source 的字段。该源应指向与给定文档相关联的 最终 来源。

例如，如果这些文档代表某个父文档的片段，那么这两个文档的 source 应该相同，并引用该父文档。

通常情况下，应始终指定source。仅当您从不打算使用incremental模式，且因某些原因无法正确指定source字段时，才使用None。

from langchain_text_splitters import CharacterTextSplitter
doc1 = Document(
    page_content="kitty kitty kitty kitty kitty", metadata={"source": "kitty.txt"}
)
doc2 = Document(page_content="doggy doggy the doggy", metadata={"source": "doggy.txt"})
new_docs = CharacterTextSplitter(
    separator="t", keep_separator=True, chunk_size=12, chunk_overlap=2
).split_documents([doc1, doc2])
new_docs
[Document(page_content='kitty kit', metadata={'source': 'kitty.txt'}),
 Document(page_content='tty kitty ki', metadata={'source': 'kitty.txt'}),
 Document(page_content='tty kitty', metadata={'source': 'kitty.txt'}),
 Document(page_content='doggy doggy', metadata={'source': 'doggy.txt'}),
 Document(page_content='the doggy', metadata={'source': 'doggy.txt'})]
_clear()
index(
    new_docs,
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)
{'num_added': 5, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}
changed_doggy_docs = [
    Document(page_content="woof woof", metadata={"source": "doggy.txt"}),
    Document(page_content="woof woof woof", metadata={"source": "doggy.txt"}),
]

这应该删除与 doggy.txt 源关联的文档旧版本，并用新版本替换它们。

index(
    changed_doggy_docs,
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)
{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 2}
vectorstore.similarity_search("dog", k=30)
[Document(page_content='woof woof', metadata={'source': 'doggy.txt'}),
 Document(page_content='woof woof woof', metadata={'source': 'doggy.txt'}),
 Document(page_content='tty kitty', metadata={'source': 'kitty.txt'}),
 Document(page_content='tty kitty ki', metadata={'source': 'kitty.txt'}),
 Document(page_content='kitty kit', metadata={'source': 'kitty.txt'})]

使用加载器​

索引功能可以接受文档的可迭代对象，或者任何加载器。

注意：加载器 必须 正确设置源键。

from langchain_core.document_loaders import BaseLoader


class MyCustomLoader(BaseLoader):
    def lazy_load(self):
        text_splitter = CharacterTextSplitter(
            separator="t", keep_separator=True, chunk_size=12, chunk_overlap=2
        )
        docs = [
            Document(page_content="woof woof", metadata={"source": "doggy.txt"}),
            Document(page_content="woof woof woof", metadata={"source": "doggy.txt"}),
        ]
        yield from text_splitter.split_documents(docs)

    def load(self):
        return list(self.lazy_load())
API 参考:基础加载器
_clear()
loader = MyCustomLoader()
loader.load()
[Document(page_content='woof woof', metadata={'source': 'doggy.txt'}),
 Document(page_content='woof woof woof', metadata={'source': 'doggy.txt'})]
index(loader, record_manager, vectorstore, cleanup="full", source_id_key="source")
{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}
vectorstore.similarity_search("dog", k=30)
[Document(page_content='woof woof', metadata={'source': 'doggy.txt'}),
 Document(page_content='woof woof woof', metadata={'source': 'doggy.txt'})]