索引（Indexing）API 支持将任何来源的文档加载到向量存储中并保持同步，具体来说有助于：

避免重复数据写入向量数据库
避免重写未更改的数据
避免在未更改的内容上重新计算 embeddings

这些目标可以帮助节省时间和金钱、改进矢量搜索结果

重要的是，即使原始文本经过了一些转换步骤（例如文本切分 chunking）索引依然可以生效

如何工作

LangChain Indexing 使用记录管理器 RecordManager 来跟踪文档写入向量存储的情况

当进行内容索引时，会计算每一个文档的哈希值，并且记录以下内容：

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

删除模式

将文档添加到向量数据库后，可能需要删除一些已经存在的文档

有时可能希望删除与正在索引的新文档来源相同的任何现有文档（删除同源文档）

或者希望批量删除所有现有文档，API 提供的删除模式提供了所需的行为：

模式	重复数据	并行	清理源文件	同源删除	删除时机
None	✅	✅	❌	❌	/
Incremental	✅	✅	❌	✅	立即
Full	✅	❌	✅	✅	索引结束时

None 模式不会做任何自动化的清理，允许用户手动进行删除
Incremental 和 Full 将会进行自动清理
- 如果源文档或派生文档的内容发生了更改，则 Incremental 或 Full 模式都将清除（删除）旧版本的内容
- 如果源文档已被删除（意味着它不包括在当前正在索引的文档中），则 Full 模式将正确地将其从向量存储中删除，但 Incremental 模式不会
当内容发生变化时（例如，源 PDF 文件被修改），在索引期间在一定时间窗口内，新版本和旧版本都可以返回给用户，这种情况发生在写入新内容之后，但在删除旧版本之前（先加后删）
- Incremental 最大限度地处理了这种情况，能够在写入时连续进行清理
- Full 只能在批量写入后进行清理

要求

从实现的功能也可以看出来，想要实现 Indexing 的功能对向量存储的能力有要求

不要对预先写入过数据的向量存储使用，因为这些旧数据没有被管理
仅适用于 LangChain 集成的 vectorstore
- 支持通过 ID 添加（add_documents 方法带有 ids 参数）
- 支持通过 ID 删除（delete 方法带有 ids 参数）

注意

记录管理器基于时间来确定可以清理哪些内容（当使用 Incremental 或 Full 模式时）

如果两个任务接连运行，并且第一个任务在时间更新之前完成，那么第二个任务可能无法清理内容

不过这种情况不太可能发生，因为：

RecordManager 使用更高精度（higher resolution）的时间戳
数据变更在第一个任务和第二个任务这个范围内，不太发生在很小的时间间隔内
索引任务通常需要几毫秒以上的时间

快速开始

需要用到的 API

SQLRecordManager
index
Document
ElasticsearchStore
OpenAIEmbeddings

初始化向量数据库并设置 embeddings

collection_name = "test_index"

embedding = OpenAIEmbeddings()

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

初始化一个记录管理器（record manage）选择合适的命名空间

建议使用一个既能表达向量存储，又能表达向量存储内集合的名字，例如 redis/my_docs、chromadb/my_docs 等

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

使用记录管理器前创建一个 schema

1	record_manager.create_schema()

下面开始索引文档

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

Source

元数据属性中有一个名叫 source 的变量，资源应该指向最终相关的文档

举一个例子，如果有一些文档都是由父文档进行拆分的，那么它们的 source 属性应该相同并且指向相关的父文档

通常 source 应该是存在明确值的，在以下场景可能为 None

不打算使用 Incremental 模式
因为一些原因不能明确文档的来源

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])

[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'})]

索引文档

index(
    new_docs,
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)

1	{'num_added': 5, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

模拟相同 source 的文档，必须设置 source = doggy.txt 才能让文档替换为新版本

changed_doggy_docs = [
    Document(page_content="woof woof", metadata={"source": "doggy.txt"}),
    Document(page_content="woof woof woof", metadata={"source": "doggy.txt"}),
]

index(
    changed_doggy_docs,
    record_manager,
    vectorstore,
    cleanup="incremental",
    source_id_key="source",
)

可以看到新增了两条并且删除了两条

1	{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 2}

做一次查询

1	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'})]

使用加载器（Loader）

索引可以接受文档的 iterable，也可以接受任意的 loader

需要注意：使用 loader 需要明确设置 source

实现一个 BaseLoader，mock 加载了几个文档

from langchain_community.document_loaders.base 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())

从 loader 中读取数据

1 2	loader = MyCustomLoader() loader.load()

1 2	[Document(page_content='woof woof', metadata={'source': 'doggy.txt'}), Document(page_content='woof woof woof', metadata={'source': 'doggy.txt'})]

索引

1	index(loader, record_manager, vectorstore, cleanup="full", source_id_key="source")

1	{'num_added': 2, 'num_updated': 0, 'num_skipped': 0, 'num_deleted': 0}

查询

1	vectorstore.similarity_search("dog", k=30)

1 2	[Document(page_content='woof woof', metadata={'source': 'doggy.txt'}), Document(page_content='woof woof woof', metadata={'source': 'doggy.txt'})]

参考

Indexing | 🦜️🔗 LangChain

贫瘠之地

LangChain 文档学习 No.11 - 索引

如何工作

删除模式

要求

注意

快速开始

Source

使用加载器（Loader）

参考