实操｜如何优雅的实现RAG与GraphRAG应用中的知识文档增量更新？

在RAG应用（包括GraphRAG）中，领域知识的导入与索引是后续增强生成的基础。一个常见的问题是，当领域知识发生更新与变化时，如何用最简洁、快速、低成本的方式更新对应的向量或知识图谱索引？让我们来探讨这个问题。

01

需求

一般来说企业的信息系统中都可能有较完善的知识库维护与管理应用，但是如何让变化的知识能够同步更新到RAG应用中则不一样，知识进入到RAG应用通常需要经过拆分（split）、嵌入（embedding）、向量索引（vectorindex）等步骤：

因此当更新发生时，就需要识别出输入的知识文档变化，进而将合适的策略应用到不同的知识块上，比如忽略、新增、删除或者更新。

在实际应用中有两种不同级别的增量更新策略：

一种是文档（Document）级别的简单更新策略。即在导入知识文档时识别出新增或更新的文档，然后对其进行全量解析与向量化，并做索引合并更新。

另外一种是块（Chunk）级别的更新策略。这种更加复杂但也更精细化：在一个文档发生变化的过程中，有新增的块也有发生更新的块，需要识别哪些块需要更新删除、哪些是新增的块，以及哪些块没有发生变化，应该跳过更新。

借助以上的两种策略，你可以在文档发生更新时，降低不必要的计算工作量，消除可能产生的重复块与索引，节约模型使用成本，并提高RAG应用后续检索阶段的有效性与准确性，即保持最新、有效且不重复的上下文。

02

方案

实现增量更新的解决方案通常需要借助于文档或者块的“指纹”来实现，结合必要的持久与缓存方案，在每次进行知识索引时通过“指纹”来识别出本次需要处理的文档或知识块，并执行相应的动作（如插入或者删除），跳过重复的内容，从而达到增量更新的目的。

不管是文档（Document）还是块（Chunk）级别的增量更新策略，都可以基于类似的原理来实现，我们以更细粒度的Chunk级别的增量更新为例，其原理表示如下：

• 在每次处理开始时，计算每个块的hash指纹，这通常是是基于块的内容与元数据，并借助hash函数生成的唯一值

  
  

• 为了实现增量加载更新，需要一个跟踪与保存每次处理的块信息的机制（源文档、块信息、hash指纹、时间戳等），比如LangChain中的RecordManager组件，LlamaIndex中的DocumentStore组件

  
  

• 每次增量更新时，通过与上一次保存的处理信息对比hash指纹，确定数据块的处理动作：

• 如果某数据块的hash指纹在上一次处理中存在，则跳过处理

• 如果某数据块的hash指纹在上一次处理中不存在，则做新增处理

• 对于上一次处理中存在但是本次不存在的hash指纹，则做块删除

  
  

• 根据确定的处理动作对数据块做相应的嵌入与索引更新即可。注意这里可能对向量数据库有一定的能力要求，以实现增量索引更新。

  
  

  
  

03

实现

在现有的两个主流底层LLM应用开发框架：LangChain与LlamaIndex中都提供了文档增量更新的实现方法。两者实现方法各有区别，但核心思想基本类似，这里做一个简单演示与研究。

【LangChain的索引API】

如果你使用了LangChain框架并需要让向量索引与输入知识文档保持同步，那么需要使用LangChain的索引API来创建知识的向量索引，而不是简单的使用from_documents方法来完成。

索引API的主要区别就在于提供了文档增量更新的能力：跳过没有变化的知识块以避免向量库中写入重复知识块、并对新增或者变化的知识块计算嵌入与写入向量库。

为了实现对文档块的跟踪，索引API的使用需要借助一个记录管理器的组件（Record Manager），以跟踪每个知识块的源文档ID、hash指纹以及时间戳等。这里直接给出参考代码：

from langchain.indexes import SQLRecordManager, index
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader

#嵌入模型、向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(
    collection_name="example_collection",
    embedding_function=embeddings,
    persist_directory="./db_chroma"
)

#记录管理器，用来跟踪向量库中已经存储的Document（hash、时间戳、source_id）
namespace = f"chroma/mydocs"
record_manager = SQLRecordManager(
    namespace, db_url="sqlite:///record_manager_cache.sql"
)
record_manager.create_schema()

#文档加载与分割
loader = DirectoryLoader("../data",glob='*.txt')
docs = loader.load()
docs = CharacterTextSplitter(separator='\n',chunk_size=30,chunk_overlap=2).split_documents(docs)

#向量话并索引
result = index(
    docs,
    record_manager,
    vector_store,
    cleanup='incremental',
    source_id_key="source",
)

#打印处理情况
print(result)

这里的核心区别就在于index()方法的使用。该方法除了需要输入处理的块（docs）、向量库（vector_store）、记录管理器(record_manager）、表示源文档ID的key名（source_id_key）外，还有一个cleanup参数，该参数决定了LangChain对向量库中现有知识块的清理方式，支持三种方式。三种方式都会根据hash值跳过重复块，并插入新知识块，但对已有块的清理方式则有区别：

• none：不会对已有块做任何清理动作
  

• incremental：如果源文档知识块发生了变更（出现新的块hash指纹），则会清除知识块的旧版本

• full：如果源文档知识块发生了变更（出现新的块hash指纹），或者做了部分块的删除（注意此时未出现新的hash指纹），都会清除知识块的旧版本

  
  

也就是说incremental与full的区别在于：如果源文档中只有部分知识块被删除（即不包含在当前正在被索引的知识块中），incremental模式不会从向量库中清除这些部分知识块，但full模式会清除。

我们用上面的代码样例来对这两种模式做详细测试，假设为如下的知识文档内容创建向量索引：

首次处理后的结果信息如下（无论incremental或full模式），由于采用了按行分割，所以添加了3个知识块：

现在我们把知识内容修改成如下，即删除了最后一行，并修改了第二行：

重新运行上面的代码（无论incremental或者full），处理信息如下：

这里跳过了第一行对应的chunk(num_skipped=1)，新增了第二行对应的chunk(num_added=1)，并且删除了原来的第二行与第三行对应的chunk(num_deleted=2)。可以看到，由于这里出现了知识块的修改（第二行），所以incremental与full模式效果一致。

现在让我们再直接删除第二行，输入文件变成：

此时，两种模式下的处理就会有区别：

incremental清理模式：由于删除了第二个chunk，但是并未出现新的chunk指纹，所以不会做清理动作，只会跳过第一个重复块(num_skipped=1)：

full清理模式：不仅会跳过第一个重复块(num_skipped=1)，还会删除掉第二个chunk（num_deleted=1)：

【LlamaIndex框架的数据摄入管道】

如果你采用LlamaIndex框架，则需要借助LlamaIndex中的数据摄入管道来实现知识增量更新，并指定文档存储（docstore）以及文档存储策略（docstore_strategy），核心代码如下：

......
pipeline = IngestionPipeline(
    transformations=[
        TokenTextSplitter(chunk_size=20, chunk_overlap=0,separator="\n"),
        embedded_model
    ],
    vector_store=vector_store,
    docstore=RedisDocumentStore.from_host_and_port("localhost", 6379, namespace="document_store"),
    docstore_strategy='upserts'
)

docs = SimpleDirectoryReader(input_files=["../data/datafile1.txt"],filename_as_id=True).load_data()
nodes = pipeline.run(documents=docs,show_progress=False)
......

更多的信息可以参考LlamaIndex的官方文档。

04

GraphRAG的增量更新

Graph RAG是最近的一个热点，借助于知识图谱与图数据库对知识中的实体与关系进行组织与表示，同时结合向量检索、社区识别算法等实现复杂知识关系的检索与答案生成。实现Graph RAG的一种方式是借助成熟框架如Microsoft GraphRAG，但目前尚未能够实现增量更新。由于涉及到图与社区等高级数据结构，GraphRAG的知识增量更新要比普通RAG更复杂。

这里推荐一个开源的nano-GraphRAG框架，这是一个保留了Microsoft GraphRAG核心功能，但又更轻量级、更简洁的版本，且提供了一定的知识增量更新的能力。其核心思想也是借助对原始文档与知识块的hash值做分析，识别出需要添加的新知识块，并在上一次生成的Graph图基础上进行图的增量更新，插入新的实体与关系。

nano-GraphRAG也提供了社区识别与生成的功能，所以会在每次图的增量更新基础上，重新进行社区信息的生成，但社区信息的增量更新目前尚未实现，即每次都会对所有社区信息做识别与生成。

有兴趣的朋友可以在Github搜索该项目以了解细节，我们将在后续对该项目进行深入研究与测试。

05

问题

以上探讨了RAG应用中常见的一个知识文档增量更新的问题，这对于企业级的RAG应用、存在大量经常变化的知识文档的场景下的快速同步与降低成本有重要的意义。当然仍然有一些问题可以做进一步优化与思考，比如：

• 基于Chunk指纹来识别知识变化，在简单的基于固定chunk_size分割的RAG应用中，一点小的中间内容变化可能导致大量分割后的chunk的hash指纹发生变化，从而影响增量更新效果。

  
  

• 可能存在少量知识文本发生变化，但实际语义并未发生变化的场景，这也会带来一些无效的更新。但如果借助LLM来识别语义是否变化，又会带来新的性能与成本消耗。

  
  

• 针对复杂知识结构或者知识索引的增量更新。比如多模态的复杂知识文档的增量更新，如何更有效且高效的识别知识变化；以及除了向量索引之外的其他形式索引的增量更新，如上文提到的Graph Index等。

  
  

• 在实际企业应用中，知识文档的动态更新可能需要结合数据特点与业务要求制定更灵活的策略：对于实时性要求较高的数据可以使用更频繁的动态更新策略；而对于实时性要求较低或变化频次很低的数据可以采用更简单的批量更新策略。

  
  

相信未来这些问题都会有更完善的解决方案。