RAG 与 GraphRAG 知识文档增量更新的优雅实现方案
在 RAG（Retrieval-Augmented Generation）应用及 GraphRAG 应用中，领域知识的导入与索引是后续增强生成的基础。一个常见且关键的问题是：当领域知识发生更新与变化时，如何用最简洁、快速、低成本的方式更新对应的向量或知识图谱索引？全量重建索引虽然简单，但在数据量大时会导致高昂的算力成本与延迟。本文将深入探讨如何实现高效的增量更新策略。
01 需求分析
一般来说，企业的信息系统中都可能有较完善的知识库维护与管理应用，但是如何让变化的知识能够同步更新到 RAG 应用中则是一个技术难点。知识进入到 RAG 应用通常需要经过拆分（Split）、嵌入（Embedding）、向量索引（Vector Index）等步骤。因此当更新发生时，就需要识别出输入的知识文档变化，进而将合适的策略应用到不同的知识块上，比如忽略、新增、删除或者更新。
在实际应用中有两种不同级别的增量更新策略：
- 文档（Document）级别的简单更新策略：即在导入知识文档时识别出新增或更新的文档，然后对其进行全量解析与向量化，并做索引合并更新。这种方式实现简单，但无法处理文档内部的部分修改。
- 块（Chunk）级别的更新策略：这种更加复杂但也更精细化。在一个文档发生变化的过程中，有新增的块也有发生更新的块，需要识别哪些块需要更新删除、哪些是新增的块，以及哪些块没有发生变化，应该跳过更新。借助以上的两种策略，你可以在文档发生更新时，降低不必要的计算工作量，消除可能产生的重复块与索引，节约模型使用成本，并提高 RAG 应用后续检索阶段的有效性与准确性，即保持最新、有效且不重复的上下文。
02 核心方案原理
实现增量更新的解决方案通常需要借助于文档或者块的'指纹'来实现，结合必要的持久化与缓存方案，在每次进行知识索引时通过'指纹'来识别出本次需要处理的文档或知识块，并执行相应的动作（如插入或者删除），跳过重复的内容，从而达到增量更新的目的。
不管是文档还是块级别的增量更新策略，都可以基于类似的原理来实现。我们以更细粒度的 Chunk 级别的增量更新为例，其核心原理如下：
- 计算哈希指纹：在每次处理开始时，计算每个块的 hash 指纹。这通常是基于块的内容与元数据，并借助 hash 函数（如 MD5 或 SHA-256）生成的唯一值。
- 状态跟踪与持久化：为了实现增量加载更新，需要一个跟踪与保存每次处理的块信息的机制（源文档 ID、块信息、hash 指纹、时间戳等）。例如 LangChain 中的 RecordManager 组件，LlamaIndex 中的 DocumentStore 组件。
- 差异比对：每次增量更新时，通过与上一次保存的处理信息对比 hash 指纹，确定数据块的处理动作：
- 如果某数据块的 hash 指纹在上一次处理中存在，则跳过处理。
- 如果某数据块的 hash 指纹在上一次处理中不存在，则做新增处理。
- 对于上一次处理中存在但是本次不存在的 hash 指纹，则做块删除。
- 索引更新：根据确定的处理动作对数据块做相应的嵌入与索引更新即可。注意这里可能对向量数据库有一定的能力要求，以实现增量索引更新。
03 主流框架实现
在现有的两个主流底层 LLM 应用开发框架：LangChain 与 LlamaIndex 中都提供了文档增量更新的实现方法。两者实现方法各有区别，但核心思想基本类似。
3.1 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 Chroma
langchain_text_splitters CharacterTextSplitter
langchain_community.document_loaders DirectoryLoader
embeddings = OpenAIEmbeddings(model=)
vector_store = Chroma(
collection_name=,
embedding_function=embeddings,
persist_directory=
)
namespace =
record_manager = SQLRecordManager(
namespace, db_url=
)
record_manager.create_schema()
loader = DirectoryLoader(, glob=)
docs = loader.load()
docs = CharacterTextSplitter(separator=, chunk_size=, chunk_overlap=).split_documents(docs)
result = index(
docs,
record_manager,
vector_store,
cleanup=,
source_id_key=,
)
(result)
