参考：Implementation of all RAG techniques in a simpler way

简单的 RAG

从最基础的RAG实现开始，理解其工作原理并评估其效果：

如图所示，简单RAG管道的工作原理如下：

从PDF中提取文本。
将文本分成更小块
将块转换为数字嵌入
根据查询搜索最相关的块
使用检索到的块生成响应
将答案与正确答案进行比较以评估准确性

首先，让我们加载文档，获取文本，并将其分成可管理的块：

class DataPreparationModule:
    """数据准备模块 - 负责数据加载、清洗和预处理"""
    def __init__(self, data_path: str):
        self.data_path = data_path
        self.documents: List[Document] = []  # 父文档(人工智能书籍)
        self.chunks: List[Document] = []  # 子文档(按标题分割的小块)
        self.parent_child_map: Dict[str, List[str]] = {}  # 父文档ID到子文档ID的映射
    
    def load_documents(self) -> List[Document]:
        """批量加载文档数据"""
        documents = []
        data_path = Path(self.data_path)
        print("验证数据路径:", data_path)

        for file_path in tqdm(list(data_path.rglob("*.pdf")), desc="加载文档"):
            # 使用PyMuPDFLoader加载PDF文件
            loader = PyMuPDFLoader(str(file_path))
            loaded_docs = loader.load()
                
            # 为每个父文档分配唯一ID
            parent_id = str(uuid.uuid4())
                
            # 如果加载的文档列表不为空，合并所有页面内容
            if loaded_docs:
                # 合并所有页面内容
                merged_content = ""
                for doc in loaded_docs:
                    merged_content += doc.page_content + "\n\n"
                    
                # 创建Document对象
                doc = Document(
                    page_content=merged_content,
                    metadata={
                        "source": str(file_path),
                        "parent_id": parent_id,
                        "doc_type": "parent",  # 标记为父文档
                    }
                )
                documents.append(doc)
            
        print(f"总共加载了 {len(documents)} 个文档")

        self.documents = documents
        return documents
    
    def chunk_documents(self) -> List[Document]:
        # 切分文档
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
        )
        split_docs = text_splitter.split_documents(self.documents)

        return split_docs

接下来，我们需要将 split_docs 进行模型嵌入到本地数据库当中。

def IndexConstruction(self, persist_path):
        """索引构建，并构建数据库"""
        # 实例化数据库, 为其配置
        # 使用绝对路径确保数据保存在正确位置
        split_docs = self.chunk_documents()
        vectordb = Chroma.from_documents(
            documents=split_docs,
            embedding=TextV4Embeddings(),
            persist_directory=persist_path
        )

        # 显式调用persist()方法确保数据保存到本地
        vectordb.persist()

        # 查询数据库里的数量
        count = vectordb._collection.count()
        print(f"数据库中文档数量: {count}")

        # 验证持久化目录是否存在
        if os.path.exists(persist_path):
            print(f"持久化目录已创建: {persist_path}")
            # 列出目录内容
            files = os.listdir(persist_path)
            print(f"目录内容: {files}")
        else:
            print(f"持久化目录创建失败: {persist_path}")

这里的embedding换成自己所用的嵌入模型，persist_path 是数据库保存的地址。

现在，我们可以执行 semantic_search

哈基鹏的大模型之旅（九）

RAG实战，相关解析：

简单的 RAG