Skip to main content
Open In ColabOpen on GitHub

如何使用LangChain索引API

在这里，我们将看看使用LangChain索引API的基本索引工作流程。

索引API允许您从任何来源加载文档并将其同步到向量存储中。具体来说，它有助于：

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

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

至关重要的是，索引API即使在文档经历了多个转换步骤（例如，通过文本分块）相对于原始源文档的情况下也能工作。

它是如何工作的​

LangChain 索引使用记录管理器 (RecordManager) 来跟踪文档写入向量存储的过程。

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

  • 文档哈希（页面内容和元数据的哈希）
  • 写入时间
  • 源ID -- 每个文档应在其元数据中包含信息，以便我们能够确定该文档的最终来源

删除模式​

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

清理模式去重内容可并行清理已删除的源文档清理源文档和/或派生文档的变更清理时机
无✅✅❌❌-
增量✅✅❌✅持续
完整✅❌✅✅索引结束时
Scoped_Full✅✅❌✅索引结束时

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

incremental、full 和 scoped_full 提供以下自动清理功能：

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

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

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

需求​

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

兼容的向量存储：Aerospike, AnalyticDB, AstraDB, AwaDB, AzureCosmosDBNoSqlVectorSearch, AzureCosmosDBVectorSearch, 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模式下，用户应将要索引的全部内容传递给索引函数。

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

这种行为对于处理源文档的删除非常有用。

_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}

来源​

元数据属性包含一个名为source的字段。此源应指向与给定文档关联的最终来源。

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

通常，source 应该始终被指定。只有在您从不打算使用incremental模式，并且由于某种原因无法正确指定source字段时，才使用None。

from langchain_text_splitters import CharacterTextSplitter
API Reference: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 Reference:BaseLoader
_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'})]

这个页面有帮助吗？