AI知识库

53AI知识库

学习大模型的前沿技术与行业应用场景


线上RAG应用pdf文档频繁更新,老板下了死命令要节省预算,不能重复做embedding,我这么做.....
发布日期:2024-08-17 06:41:32 浏览次数: 1750 来源:AI 博物院


我们最近在一个项目中遇到了一个问题。项目的场景是这样的:用户将他们的PDF文档存储在磁盘的某个特定目录中,然后有一个定时任务来扫描此目录并从中的PDF文档构建知识库。

一开始,我们采用"增量更新"策略。在扫描目录中的文档时,我们会对每个文档进行哈希运算以生成其指纹,并检查该指纹是否已存在于数据库中。如果指纹不存在,就表示这是一个新文件,我们会对新文件的document做embedding,然后将其加入到知识库中。

然而,这种方法存在一个问题。如果同一文件进行了增量添加,例如我们已经将A.pdf文件加入到了知识库,但后来这个文件添加了新的内容。当我们重新计算其指纹并在数据库中查找时,由于指纹不存在,我们会将这个更新过的文件作为新文件处理,并重新做embedding加入到知识库。这样一来,对于未更新的部分,知识库会有两份相同的数据记录,第二份相同的记录可能会"占据"原本应该被召回的数据记录的位置,从而降低问答效果。

那么应该怎么解决这个问题呢?对于增量更新,做hash指纹这一点毋庸置疑,但是hash的对象不能是文件了,而应该聚焦于真实存到知识库的数据: document.

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

index API允许您将来自任何源的文档加载到矢量存储中并保持同步。具体来说,它有助于:

  • 避免将重复的内容写入vector存储
  • 避免重写未更改的内容
  • 避免在未更改的内容上重新计算embedding

所有这些都可以节省你的时间和金钱,并改善你的矢量搜索结果。

如何工作

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

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

  • 文档hash(页面内容和元数据的散列)
  • 写时间
  • 源id——每个文档应该在其元数据中包含信息,以便我们确定该文档的最终来源

删除模式

将文档索引到矢量存储时,可能会删除矢量存储中的一些现有文档。在某些情况下,您可能希望删除与正在索引的新文档来自相同来源的所有现有文档。在其他情况下,您可能希望批量删除所有现有文档。索引API删除模式可以让你选择你想要的行为:

Cleanup ModeDe-Duplicates ContentParallelizableCleans Up Deleted Source DocsCleans Up Mutations of Source Docs and/or Derived DocsClean Up Timing
None-
IncrementalContinuously
FullAt end of indexing

快速开始

首先,需要明确的是,无论使用何种清理模式,index函数都会自动去重。也就是说,调用index([doc1, doc1, doc2])的效果等同于调用index([doc1, doc2])。然而,在我们的实际应用场景中,情况并不完全如此。

可能在第一次运行时,我们对[doc1, doc2]进行了索引操作,而在下次定时任务执行时,我们又对[doc1, doc3]进行了索引。换言之,我们从源文档中删除了一部分内容,并添加了一些新的内容。这才是我们真正面临的场景:我们希望保持doc1不变,新增doc3,并能够自动删除doc2。这种需求可以通过Incremental增量模式得到满足。

话不多说,我们来看看三种模式的使用效果吧。

None

None模式的功能可以理解为去重和添加,而不包括删除。例如,如果你首次调用index([doc1, doc2]),然后再次调用index([doc1, doc3]),那么在向量库中的数据就会是[doc1, doc2, doc3]。需要注意的是,这种模式下,旧版本的doc2并不会被删除。

from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from langchain.vectorstores.elasticsearch import ElasticsearchStore
from langchain.indexes import SQLRecordManager, index


collection_name = "test_index"

embedding = OpenAIEmbeddings()

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

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"})
doc3 = Document(page_content="doggy1", 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")

_clear()

res = index(
[doc1, doc1, doc2],
record_manager,
vectorstore,
cleanup=None,
source_id_key="source",
)
print(res)

得到的结果:

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

我们发现做了去重并且帮我们增加了两条数据。

然后我们再执行index操作:

res = index(
[doc1, doc3],
record_manager,
vectorstore,
cleanup=None,
source_id_key="source",
)
print(res)

执行结果发现添加了doc3, 跳过了doc1, doc2 还在数据库记录里:

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

full

full 含义是用户应该将所有需要进行索引的全部内容传递给index函数,任何没有传递到索引函数并且存在于vectorstore中的文档将被删除! 此行为对于处理源文档的删除非常有用。我们还是使用上面的代码,这次只是把模式换成 full. 首先,我们需要重置并清空数据,这可以通过调用_clear()函数实现。


res = index(
    [doc1, doc1, doc2],
    record_manager,
    vectorstore,
    cleanup="full",
    source_id_key="source",
)
print(res)

我们发现添加了2个文档:

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

接着我们执行:

res = index(
[doc1, doc3],
record_manager,
vectorstore,
cleanup="full",
source_id_key="source",
)
print(res)

我们发现添加了一个文档doc3,跳过了一个文档doc1,删除了一个文档doc2:

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

incremental

"增量模式"是我们最常用的一种。顾名思义,这种模式主要进行增量操作,即添加最新记录并删除旧版记录。在这种模式下,如果我们传入一个空的文档数组,即index([]),将不会发生任何操作。然而,如果我们在"全量模式"下传入同样的空数组,系统则会清除所有数据。

首先,执行以下操作:

_clear()

res = index(
[doc1, doc1, doc2],
record_manager,
vectorstore,
cleanup="incremental",
source_id_key="source",
)
print(res)

res = index(
[doc1, doc3],
record_manager,
vectorstore,
cleanup="incremental",
source_id_key="source",
)
print(res)

得到的结果如下:

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

可以看出,第一次操作添加了两个文档。在第二次操作中,系统跳过了doc1,并删除了之前属于"doggy.txt"的doc2,因为现在我们只传入了doc3。因此,增量模式会将这个旧版本(doc2)删除。

然后执行以下操作:

res = index(
[doc1],
record_manager,
vectorstore,
cleanup="incremental",
source_id_key="source",
)
print(res)

这次对于"doggy.txt"没有任何新的文档被传入,所以数据没有任何改动,结果如下:

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

但是,如果我们只传入doc2,则会发现系统增加了doc2,并删除了同一源文件("doggy.txt")的doc3。结果如下:

{'num_added': 1, 'num_updated': 0, 'num_skipped':

源码

def index(
    docs_source: Union[BaseLoader, Iterable[Document]],
    record_manager: RecordManager,
    vector_store: VectorStore,
    *,
    batch_size: int = 100,
    cleanup: Literal["incremental""full", None] = None,
    source_id_key: Union[str, Callable[[Document], str], None] = None,
    cleanup_batch_size: int = 1_000,
)
 -> IndexingResult:

    ...

    if isinstance(docs_source, BaseLoader):
        try:
            doc_iterator = docs_source.lazy_load()
        except NotImplementedError:
            doc_iterator = iter(docs_source.load())
    else:
        doc_iterator = iter(docs_source)

    source_id_assigner = _get_source_id_assigner(source_id_key)

    # Mark when the update started.
    index_start_dt = record_manager.get_time()
    num_added = 0
    num_skipped = 0
    num_updated = 0
    num_deleted = 0
    
    
    for doc_batch in _batch(batch_size, doc_iterator):
        hashed_docs = list(
            _deduplicate_in_order(
                [_HashedDocument.from_document(doc) for doc in doc_batch]
            )
        )

        source_ids: Sequence[Optional[str]] = [
            source_id_assigner(doc) for doc in hashed_docs
        ]

        ....

        exists_batch = record_manager.exists([doc.uid for doc in hashed_docs])

        # Filter out documents that already exist in the record store.
        uids = []
        docs_to_index = []
        
        # 判断哪些是要更新,哪些是要添加的
        for hashed_doc, doc_exists in zip(hashed_docs, exists_batch):
            if doc_exists:
                # Must be updated to refresh timestamp.
                record_manager.update([hashed_doc.uid], time_at_least=index_start_dt)
                num_skipped += 1
                continue
            uids.append(hashed_doc.uid)
            docs_to_index.append(hashed_doc.to_document())

      
        # 知识入向量库
        if docs_to_index:
            vector_store.add_documents(docs_to_index, ids=uids)
            num_added += len(docs_to_index)

        # 更新数据库记录时间
        record_manager.update(
            [doc.uid for doc in hashed_docs],
            group_ids=source_ids,
            time_at_least=index_start_dt,
        )

        # 根据时间和source_ids 清理旧版本数据
        if cleanup == "incremental":
            ...

            uids_to_delete = record_manager.list_keys(
                group_ids=_source_ids, before=index_start_dt
            )
            if uids_to_delete:
                vector_store.delete(uids_to_delete)
                record_manager.delete_keys(uids_to_delete)
                num_deleted += len(uids_to_delete)

    if cleanup == "full":
        while uids_to_delete := record_manager.list_keys(
            before=index_start_dt, limit=cleanup_batch_size
        ):
            # First delete from record store.
            vector_store.delete(uids_to_delete)
            # Then delete from record manager.
            record_manager.delete_keys(uids_to_delete)
            num_deleted += len(uids_to_delete)

    return {
        "num_added": num_added,
        "num_updated": num_updated,
        "num_skipped": num_skipped,
        "num_deleted": num_deleted,
    }

通过上述代码,我们可以了解到一个常见的优化策略:对于涉及大量数据操作的数据库和向量库,我们通常使用批处理(batch)方式进行操作。上面代码的流程图如下:



53AI,企业落地应用大模型首选服务商

产品:大模型应用平台+智能体定制开发+落地咨询服务

承诺:先做场景POC验证,看到效果再签署服务协议。零风险落地应用大模型,已交付160+中大型企业

联系我们

售前咨询
186 6662 7370
预约演示
185 8882 0121

微信扫码

与创始人交个朋友

回到顶部

 
扫码咨询