LightRAG实现原理：文档解析与插入详解

概述

本文深度剖析LightRAG框架在文档插入环节的具体实现方式。如果你对其内部运作流程感兴趣，读完本篇应该能梳理出一条清晰的主线，并清楚该从哪些细节入手进一步深入探究。

我们先从一个最基础的例子入手，感受一下它的工作逻辑。

文档查询的基本使用

下面这段代码可以看作是LightRAG的“hello world”——打开一个文件，将内容插入系统，随后即可直接向它提问并获取回答。

WORKING_DIR = "./dickens"

if not os.path.exists(WORKING_DIR):
    os.mkdir(WORKING_DIR)
# 构建LightRAG对象    
rag = LightRAG(
    working_dir=WORKING_DIR,
    llm_model_func=gpt_4o_mini_complete,  # Use gpt_4o_mini_complete LLM model
    # llm_model_func=gpt_4o_complete  # Optionally, use a stronger model
)
# 打开文档，并插入到LightRAG中
with open("./dickens/book.txt", "r", encoding="utf-8") as f:
    rag.insert(f.read())

# 执行原始的本地查询
print(
    rag.query("What are the top themes in this story?", param=QueryParam(mode="naive"))
)

文档插入过程实现分析

那么，具体的实现函数是什么样的呢？其实入口非常简洁，insert方法本质上就是调用了一个异步的ainsert方法。

class LightRAG:
    def insert(self, string_or_strings):
        loop = always_get_an_event_loop()
        return loop.run_until_complete(self.ainsert(string_or_strings))

ainsert()函数实现分析

从整体逻辑来看，ainsert主要完成了以下几步操作：

（1）为每个传入的文档生成唯一ID，同时创建一个字典，并自动清除内容前后的空白字符。

（2）检查这些文档中哪些是全新的——如果全部是已有内容，则记录日志，直接结束流程。

（3）仅处理新文档，按固定大小将其切分为块，再执行一轮去重检查，新的保留，旧的跳过。

（4）将新生成的分块存储起来，同时将其向量表示写入向量数据库。

（5）从这些分块中抽取实体和关系，如果没有发现新的内容，同样直接返回。

（6）将抽取出的实体和关系全部写入图数据库中。

实现代码如下

async def ainsert(self, string_or_strings):
    # 定义异步插入方法，接收字符串或字符串列表参数
    
    update_storage = False
    # 初始化存储更新标志为False
    
    try:
        if isinstance(string_or_strings, str):
            string_or_strings = [string_or_strings]
        # 如果输入是单个字符串，转换为列表形式

        new_docs = {
            compute_mdhash_id(c.strip(), prefix="doc-"): {"content": c.strip()}
            for c in string_or_strings
        }
        # 为每个文档生成唯一ID并创建文档字典，移除首尾空白

        _add_doc_keys = await self.full_docs.filter_keys(list(new_docs.keys()))
        # 检查哪些文档ID是新的(未存储过的)

        new_docs = {k: v for k, v in new_docs.items() if k in _add_doc_keys}
        # 只保留新文档

        if not len(new_docs):
            logger.warning("All docs are already in the storage")
            return
        # 如果没有新文档，记录警告并返回

        update_storage = True
        logger.info(f"[New Docs] inserting {len(new_docs)} docs")
        # 设置更新标志，记录新文档数量

        inserting_chunks = {}
        # 初始化分块字典

        for doc_key, doc in tqdm_async(new_docs.items(), desc="Chunking documents", unit="doc"):
            # 遍历每个文档，显示进度条
            
            chunks = {
                compute_mdhash_id(dp["content"], prefix="chunk-"): {
                    **dp,
                    "full_doc_id": doc_key,
                }
                for dp in chunking_by_token_size(
                    doc["content"],
                    overlap_token_size=self.chunk_overlap_token_size,
                    max_token_size=self.chunk_token_size,
                    tiktoken_model=self.tiktoken_model_name,
                )
            }
            # 对文档内容进行分块，为每个分块生成ID，并关联原始文档ID

            inserting_chunks.update(chunks)
            # 将当前文档的分块添加到总分块字典中

        _add_chunk_keys = await self.text_chunks.filter_keys(list(inserting_chunks.keys()))
        # 检查哪些分块ID是新的

     inserting_chunks = {k: v for k, v in inserting_chunks.items() if k in _add_chunk_keys}
        # 只保留新分块

        if not len(inserting_chunks):
            logger.warning("All chunks are already in the storage")
            return
        # 如果没有新分块，记录警告并返回

        logger.info(f"[New Chunks] inserting {len(inserting_chunks)} chunks")
        # 记录新分块数量

        await self.chunks_vdb.upsert(inserting_chunks)
        # 将分块存入向量数据库

        logger.info("[Entity Extraction]...")
        maybe_new_kg = await extract_entities(
            inserting_chunks,
            knowledge_graph_inst=self.chunk_entity_relation_graph,
            entity_vdb=self.entities_vdb,
            relationships_vdb=self.relationships_vdb,
            global_config=asdict(self),
        )
        # 从分块中提取实体和关系

        if maybe_new_kg is None:
            logger.warning("No new entities and relationships found")
            return
        # 如果没有找到新实体和关系，记录警告并返回

        self.chunk_entity_relation_graph = maybe_new_kg
        # 更新知识图谱

        await self.full_docs.upsert(new_docs)
        await self.text_chunks.upsert(inserting_chunks)
        # 将完整文档和分块存入各自的存储

    finally:
        if update_storage:
            await self._insert_done()
        # 如果进行了存储更新，执行清理工作

小结

总体来看，LightRAG处理新文档的策略可概括为三个核心步骤：先对文档进行分块并计算向量表示，再深入文本挖掘，从中提取出实体及其关系。这张流程图清晰地展示了完整的处理链路，掌握这个主干之后，再深入具体细节便会事半功倍。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：LightRAG实现原理：文档解析与插入详解要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题