LightRAG删除操作：按文档ID和实体名删除数据

【免费下载链接】LightRAG "LightRAG: Simple and Fast Retrieval-Augmented Generation" 项目地址: https://gitcode.com/GitHub_Trending/li/LightRAG

引言：为什么需要精确的数据删除功能？

在RAG（Retrieval-Augmented Generation）系统中，数据管理是核心需求之一。随着知识库的不断增长，用户经常需要删除特定文档或实体信息来：

• 清理过期或错误的数据
• 保护隐私和敏感信息
• 优化存储空间和查询性能
• 维护知识图谱的准确性

LightRAG提供了两种精确的删除方式：按文档ID删除和按实体名删除，让您能够精细控制知识库内容。

删除操作的核心架构

LightRAG的删除功能建立在多层存储架构之上，确保数据一致性：

按文档ID删除：完整清理

功能概述

按文档ID删除会移除指定文档的所有相关数据，包括：

• 文档元数据和状态信息
• 所有文本分块（Chunks）
• 从该文档提取的实体节点
• 相关的实体关系边

核心代码实现

async def adelete_by_doc_id(self, doc_id: str) -> DeletionResult:
    """异步删除指定文档ID的所有相关数据"""
    try:
        # 获取文档状态信息
        doc_status = await self.doc_status.get_by_id(doc_id)
        if not doc_status:
            return DeletionResult(
                status="not_found", 
                doc_id=doc_id, 
                message="Document not found"
            )

        # 获取文档关联的所有分块ID
        chunk_ids = doc_status.get("chunks_list", [])
        
        # 删除文档元数据
        await self.full_docs.delete([doc_id])
        await self.doc_status.delete([doc_id])
        
        # 删除所有文本分块
        if chunk_ids:
            await self.text_chunks.delete(chunk_ids)
            await self.chunks_vdb.delete(chunk_ids)
        
        # 查找并删除关联的实体和关系
        entities_to_delete = set()
        relations_to_delete = set()
        
        # 通过分块ID查找关联的实体和关系
        for chunk_id in chunk_ids:
            # 查找关联实体
            entities = await self.chunk_entity_relation_graph.get_nodes_by_chunk_ids([chunk_id])
            for entity in entities:
                entities_to_delete.add(entity["id"])
            
            # 查找关联关系
            relations = await self.chunk_entity_relation_graph.get_edges_by_chunk_ids([chunk_id])
            for relation in relations:
                relations_to_delete.add((relation["source"], relation["target"]))
        
        # 删除实体和关系
        if entities_to_delete:
            await self.chunk_entity_relation_graph.remove_nodes(list(entities_to_delete))
            await self.entities_vdb.delete(list(entities_to_delete))
        
        if relations_to_delete:
            await self.chunk_entity_relation_graph.remove_edges(list(relations_to_delete))
            await self.relationships_vdb.delete([f"{src}_{tgt}" for src, tgt in relations_to_delete])
        
        return DeletionResult(
            status="success",
            doc_id=doc_id,
            message=f"Document and {len(chunk_ids)} chunks deleted successfully",
            file_path=doc_status.get("file_path")
        )
        
    except Exception as e:
        return DeletionResult(
            status="fail",
            doc_id=doc_id,
            message=f"Deletion failed: {str(e)}",
            status_code=500
        )

使用示例

import asyncio
from lightrag import LightRAG
from lightrag.llm.openai import gpt_4o_mini_complete, openai_embed
from lightrag.kg.shared_storage import initialize_pipeline_status

async def main():
    # 初始化LightRAG实例
    rag = LightRAG(
        working_dir="./my_rag_storage",
        embedding_func=openai_embed,
        llm_model_func=gpt_4o_mini_complete,
    )
    
    await rag.initialize_storages()
    await initialize_pipeline_status()
    
    try:
        # 插入示例文档
        doc_id = await rag.ainsert("这是一篇关于人工智能的测试文档。")
        print(f"插入文档成功，ID: {doc_id}")
        
        # 按文档ID删除
        result = await rag.adelete_by_doc_id(doc_id)
        print(f"删除结果: {result.message}")
        
    finally:
        await rag.finalize_storages()

if __name__ == "__main__":
    asyncio.run(main())

按实体名删除：精准清理

功能概述

按实体名删除允许您移除特定的实体及其相关数据，而不会影响其他文档：

• 删除指定的实体节点
• 移除与该实体相关的所有关系边
• 清理向量存储中的实体嵌入
• 保持其他实体的完整性

核心代码实现

async def adelete_by_entity(self, entity_name: str) -> DeletionResult:
    """异步删除指定实体名的所有相关数据"""
    try:
        # 检查实体是否存在
        entity_node = await self.chunk_entity_relation_graph.get_node(entity_name)
        if not entity_node:
            return DeletionResult(
                status="not_found",
                doc_id=entity_name,
                message="Entity not found"
            )
        
        # 获取实体的所有关系
        entity_edges = await self.chunk_entity_relation_graph.get_node_edges(entity_name)
        relations_to_delete = []
        
        if entity_edges:
            for src, tgt in entity_edges:
                relations_to_delete.append((src, tgt))
        
        # 删除实体节点
        await self.chunk_entity_relation_graph.delete_node(entity_name)
        await self.entities_vdb.delete([entity_name])
        
        # 删除相关关系
        if relations_to_delete:
            await self.chunk_entity_relation_graph.remove_edges(relations_to_delete)
            # 同时从关系向量存储中删除
            relation_ids = [f"{src}_{tgt}" for src, tgt in relations_to_delete]
            await self.relationships_vdb.delete(relation_ids)
        
        return DeletionResult(
            status="success",
            doc_id=entity_name,
            message=f"Entity '{entity_name}' and {len(relations_to_delete)} relations deleted"
        )
        
    except Exception as e:
        return DeletionResult(
            status="fail",
            doc_id=entity_name,
            message=f"Entity deletion failed: {str(e)}",
            status_code=500
        )

使用示例

import asyncio
from lightrag import LightRAG
from lightrag.llm.openai import gpt_4o_mini_complete, openai_embed
from lightrag.kg.shared_storage import initialize_pipeline_status

async def main():
    # 初始化LightRAG实例
    rag = LightRAG(
        working_dir="./my_rag_storage",
        embedding_func=openai_embed,
        llm_model_func=gpt_4o_mini_complete,
    )
    
    await rag.initialize_storages()
    await initialize_pipeline_status()
    
    try:
        # 插入包含实体的文档
        text = """
        苹果公司由史蒂夫·乔布斯创立，总部位于加利福尼亚州。
        微软是另一家重要的科技公司，由比尔·盖茨创立。
        """
        await rag.ainsert(text)
        
        # 按实体名删除"苹果公司"
        result = await rag.adelete_by_entity("苹果公司")
        print(f"删除结果: {result.message}")
        
        # 验证删除效果
        remaining_entities = await rag.chunk_entity_relation_graph.get_all_labels()
        print(f"剩余实体: {remaining_entities}")
        
    finally:
        await rag.finalize_storages()

if __name__ == "__main__":
    asyncio.run(main())

删除操作的数据一致性保障

LightRAG通过以下机制确保删除操作的数据一致性：

1. 事务性操作

2. 错误处理机制

LightRAG实现了完善的错误处理策略：

class DeletionResult:
    """删除操作结果数据结构"""
    status: Literal["success", "not_found", "fail"]
    doc_id: str
    message: str
    status_code: int = 200
    file_path: str | None = None

3. 回滚机制

在删除过程中，如果任何步骤失败，系统会尝试回滚已执行的操作：

async def safe_delete_operation(self, doc_id: str):
    """安全的删除操作，支持回滚"""
    backup_data = {}
    
    try:
        # 备份关键数据
        backup_data["doc_status"] = await self.doc_status.get_by_id(doc_id)
        backup_data["chunks"] = await self.get_chunks_by_doc_id(doc_id)
        
        # 执行删除操作
        result = await self._perform_deletion(doc_id)
        
        if result.status == "fail":
            # 回滚操作
            await self._rollback_deletion(doc_id, backup_data)
            
        return result
        
    except Exception as e:
        # 异常情况下的回滚
        if backup_data:
            await self._rollback_deletion(doc_id, backup_data)
        raise e

性能优化建议

1. 批量删除操作

对于大量数据的删除，建议使用批量操作：

async def batch_delete_documents(self, doc_ids: list[str]):
    """批量删除文档"""
    results = []
    for doc_id in doc_ids:
        result = await self.adelete_by_doc_id(doc_id)
        results.append(result)
    return results

2. 异步处理优化

利用LightRAG的异步特性提高删除效率：

async def concurrent_delete_entities(self, entity_names: list[str]):
    """并发删除多个实体"""
    tasks = []
    for entity_name in entity_names:
        task = asyncio.create_task(self.adelete_by_entity(entity_name))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

3. 存储后端优化

不同存储后端的删除性能对比：

存储类型	删除性能	适用场景	注意事项
JsonKVStorage	中等	开发测试	文件IO操作较多
PGKVStorage	高	生产环境	需要数据库连接
RedisKVStorage	非常高	高性能需求	内存消耗较大
MongoKVStorage	高	大规模数据	需要MongoDB集群

常见问题解答

Q1: 删除操作是否立即生效？

A: 是的，删除操作会立即从所有存储层中移除数据。但对于某些存储后端（如文件存储），数据持久化可能需要短暂时间。

Q2: 删除后数据能否恢复？

A: LightRAG默认不提供数据恢复功能。建议在执行重要删除操作前进行数据备份。

Q3: 删除操作是否影响查询性能？

A: 删除操作会优化存储空间，通常对查询性能有正面影响。但大量删除后建议重新索引以获得最佳性能。

Q4: 如何监控删除操作的状态？

A: 可以通过检查DeletionResult对象的status字段来监控操作状态：

result = await rag.adelete_by_doc_id("doc_123")
if result.status == "success":
    print("删除成功")
elif result.status == "not_found":
    print("文档不存在")
else:
    print(f"删除失败: {result.message}")

总结

LightRAG的删除功能提供了灵活而强大的数据管理能力：

1. 精确控制：支持按文档ID和实体名两种删除方式
2. 数据一致性：通过多层存储架构确保删除操作的原子性
3. 高性能：利用异步处理和批量操作优化删除效率
4. 安全可靠：完善的错误处理和回滚机制

通过合理使用删除功能，您可以有效管理LightRAG知识库，保持数据的准确性和新鲜度，为高质量的RAG应用提供坚实基础。

提示：在生产环境中执行删除操作前，建议先在测试环境中验证操作效果，并确保有适当的数据备份策略。

【免费下载链接】LightRAG "LightRAG: Simple and Fast Retrieval-Augmented Generation" 项目地址: https://gitcode.com/GitHub_Trending/li/LightRAG