爆肝一周！手把手教你搭建本地RAG系统，从环境配置到完整实现，大模型开发入门必看！

文章详细介绍了一个本地运行的生产级RAG系统构建过程，包括文档处理、嵌入生成、向量搜索和答案生成四大核心组件。使用FAISS、sentence-transformers和llama.cpp等开源工具，在消费级硬件上实现了高性能的文档问答系统。文章提供了完整的环境配置、代码实现和部署指南，帮助开发者从零开始构建一个能够处理文档、检索信息并生成答案的实用AI系统，并涵盖了测试优化和生产部署的考量。

小天才学习机打游戏

572人浏览 · 2026-01-08 13:50:43

小天才学习机打游戏 · 2026-01-08 13:50:43 发布

从零开始构建一个可在本地运行的生产级 RAG（Retrieval-Augmented Generation，检索增强型生成）系统，能够将抽象的概念转化为实际可运行的软件，从而提供真正的价值。我们将完整地介绍实现过程，从安装依赖项到构建一个能够回答关于文档问题的功能性系统。我们不会依赖隐藏复杂性的高级抽象，而是会刻意构建每个组件，确切地了解文档分块、嵌入生成、向量搜索和答案生成是如何协同工作的。

我们使用经过验证的开源工具，这些工具在易用性和功能之间取得了平衡：FAISS 用于向量搜索，sentence-transformers 用于嵌入，llama.cpp 用于本地 LLM（Large Language Model，大型语言模型）推理。这一技术栈在消费级硬件上提供了生产级别的性能，同时又足够简单，便于学习。

一、系统架构与需求

理解完整流程

RAG 实现包括四个相互关联的阶段，这些阶段将信息从文档处理到答案生成。摄入阶段加载文档，将其分割成块，生成嵌入，并将所有内容存储在向量数据库中。这个阶段每处理一次文档或文档发生变化时运行一次，构建了后续查询将要搜索的知识库。

检索阶段接收用户查询，使用与嵌入文档相同的模型将查询转换为嵌入，搜索向量数据库以查找相似的块，并返回最相关的段落。这个阶段针对每个查询运行，必须快速执行——用户期望即使在搜索数千份文档时也能在一秒内得到响应。

生成阶段构建包含查询和检索到的上下文的提示，将它们发送到本地 LLM，并返回生成的答案。这个阶段在管道中引入了最多的延迟——即使在性能良好的硬件上，生成一个 token 也需要几秒钟——但它产生了使 RAG 系统超越关键词搜索的自然语言响应。

协调层协调这些阶段，处理错误，管理资源，并提供用户与之交互的界面。设计良好的协调层可以使复杂的管道使用起来感觉简单，隐藏技术复杂性，同时在需要的地方提供控制。

硬件与软件需求

运行一个功能性的 RAG 系统所需的最低硬件配置包括 16GB 内存、具有 4 个以上核心的现代 CPU 和 50GB 的空闲磁盘空间。这一基线配置可以在 CPU 上运行所有组件，尽管性能会比较有限——预计每个查询（包括检索和生成）需要 5 - 10 秒。舒适的性能配置从 32GB 内存和具有 8GB + 显存的 GPU 开始，将查询延迟降低到 1 - 3 秒。

软件栈需要 Python 3.10 +，它包含了对 async/await 的支持和现代类型提示，这些可以提高代码质量。虽然不是严格要求，但使用虚拟环境可以隔离依赖项，防止与其他 Python 项目发生冲突。Linux 提供了最流畅的体验，兼容性问题最少，尽管 Windows 和 macOS 也可以使用，只是偶尔需要进行一些小调整。

存储方面的考虑取决于文档集合的大小和嵌入的维度。对于 10,000 个文档块的嵌入，384 维度大约占用 15MB 的空间——与原始文档相比微不足道。FAISS 索引增加的开销也很小。计划将源文档大小的 2 - 3 倍作为大致的存储预算。LLM 权重在存储中占据主导地位——一个量化后的 7B 模型需要 4 - 6GB 的空间，无论文档集合的大小如何。

二、RAG 系统架构

离线组件（运行一次）

文档处理器：加载 PDF 文件，将其分割成块。
嵌入生成器：将文本转换为向量。
向量数据库：使用 FAISS 存储嵌入。

在线组件（针对每个查询）

查询编码器：将用户问题嵌入。
检索器：搜索相似的块。
生成器（LLM）：生成最终答案。

三、环境搭建与安装

创建项目结构

从一开始就将代码组织成清晰的目录结构，可以避免随着项目增长而产生混乱。创建一个项目目录，并在其中创建子目录，分别用于存放源代码、数据、模型和输出结果：

mkdir rag-local && cd rag-localmkdir -p src data/documents data/processed models output# 创建虚拟环境python -m venv venvsource venv/bin/activate # Windows：venv\Scripts\activate# 创建占位文件touch src/__init__.pytouch src/document_processor.pytouch src/embedding_manager.pytouch src/retriever.pytouch src/generator.pytouch src/rag_system.pytouch main.py

这种结构将关注点分离——文档处理、嵌入管理、检索和生成各自拥有专门的模块。main.py文件协调一切，为用户提供交互的入口点。

安装核心依赖项

通过一条命令安装完整的依赖项栈，以确保版本之间的兼容性：

pip install sentence-transformers==2.2.2 \faiss-cpu==1.7.4 \numpy==1.24.3 \pypdf==3.17.1 \llama-cpp-python==0.2.20 \tqdm==4.66.1

如果需要 GPU 加速，可以用 faiss-gpu 替换 faiss-cpu，并安装支持 CUDA 的 llama-cpp-python：

pip uninstall faiss-cpupip install faiss-gpu==1.7.4# 安装支持 CUDA 的 llama-cpp-pythonCMAKE_ARGS="-DLLAMA_CUBLAS=on" pip install llama-cpp-python==0.2.20

通过导入每个库并检查版本来验证安装是否成功：

import sentence_transformersimport faissimport numpy as npimport pypdfimport llama_cppprint(f"sentence-transformers: {sentence_transformers.__version__}")print(f"FAISS: {faiss.__version__}")print(f"NumPy: {np.__version__}")print(f"llama-cpp-python: {llama_cpp.__version__}")

四、构建文档处理器

加载和分割文档

文档处理器负责加载各种文件格式，并将它们分割成适合检索的块。我们首先实现对 PDF 文件的支持，然后使其能够扩展到其他格式：

# src/document_processor.pyimport osfrom typing import List, Dictfrom pathlib import Pathimport pypdffrom tqdm import tqdmclass DocumentProcessor:    """处理 RAG 中的文档加载和分块"""    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 100):        """        初始化文档处理器        参数：            chunk_size：每个块的目标单词数            chunk_overlap：块之间重叠的单词数        """        self.chunk_size = chunk_size        self.chunk_overlap = chunk_overlap    def load_pdf(self, filepath: str) -> str:        """        从 PDF 文件中提取文本        参数：            filepath：PDF 文件的路径        返回：            提取的文本        """        text = ""        with open(filepath, 'rb') as file:            pdf_reader = pypdf.PdfReader(file)            for page in pdf_reader.pages:                text += page.extract_text() + "\n"        return text    def load_text(self, filepath: str) -> str:        """加载纯文本文件"""        with open(filepath, 'r', encoding='utf-8') as f:            return f.read()    def load_document(self, filepath: str) -> str:        """        根据文件扩展名加载文档        参数：            filepath：文档的路径        返回：            文档文本        """        ext = Path(filepath).suffix.lower()        if ext == '.pdf':            return self.load_pdf(filepath)        elif ext in ['.txt', '.md']:            return self.load_text(filepath)        else:            raise ValueError(f"不支持的文件类型：{ext}")    def chunk_text(self, text: str, metadata: Dict = None) -> List[Dict]:        """        将文本分割成重叠的块        参数：            text：要分割的输入文本            metadata：可选的元数据，附加到每个块上        返回：            包含文本和元数据的块字典列表        """        # 将文本分割成单词        words = text.split()        chunks = []        # 创建重叠的块        for i in range(0, len(words), self.chunk_size - self.chunk_overlap):            chunk_words = words[i:i + self.chunk_size]            # 跳过过小的块            if len(chunk_words) < 50:                continue            chunk_text = ' '.join(chunk_words)            chunk_data = {                'text': chunk_text,                'word_count': len(chunk_words),                'char_count': len(chunk_text),                'chunk_index': len(chunks)            }            # 添加用户提供的元数据            if metadata:                chunk_data.update(metadata)            chunks.append(chunk_data)        return chunks    def process_directory(self, directory: str) -> List[Dict]:        """        处理目录中的所有支持的文档        参数：            directory：包含文档的目录路径        返回：            所有文档的所有块的列表        """        all_chunks = []        supported_extensions = ['.pdf', '.txt', '.md']        # 查找所有支持的文件        files = []        for ext in supported_extensions:            files.extend(Path(directory).glob(f'**/*{ext}'))        print(f"找到 {len(files)} 个文档需要处理")        # 处理每个文件        for filepath in tqdm(files, desc="处理文档"):            try:                # 加载文档                text = self.load_document(str(filepath))                # 为这个文档创建元数据                metadata = {                    'source': filepath.name,                    'filepath': str(filepath),                    'file_type': filepath.suffix                }                # 将文档分割成块                chunks = self.chunk_text(text, metadata)                all_chunks.extend(chunks)            except Exception as e:                print(f"处理 {filepath} 时出错：{e}")                continue        print(f"从 {len(files)} 个文档中创建了 {len(all_chunks)} 个块")        return all_chunks

这个处理器能够处理最常见的文档类型，并生成具有元数据的结构良好的块，这使得引用和过滤成为可能。重叠的块确保了边界附近的信息会出现在多个块中，从而提高了检索的鲁棒性。

五、实现嵌入和检索

构建嵌入管理器

嵌入管理器负责将文本转换为向量，并管理向量数据库：

# src/embedding_manager.pyimport numpy as npimport faissfrom sentence_transformers import SentenceTransformerfrom typing import List, Dict, Tupleimport picklefrom pathlib import Pathclass EmbeddingManager:    """管理嵌入和向量数据库"""    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):        """        初始化嵌入管理器        参数：            model_name：sentence-transformers 模型的名称        """        print(f"正在加载嵌入模型：{model_name}")        self.model = SentenceTransformer(model_name)        self.dimension = self.model.get_sentence_embedding_dimension()        # 初始化 FAISS 索引        self.index = faiss.IndexFlatIP(self.dimension)  # 使用内积计算余弦相似度        self.chunks = []  # 存储块元数据        print(f"嵌入维度：{self.dimension}")    def embed_texts(self, texts: List[str], show_progress: bool = True) -> np.ndarray:        """        为文本列表生成嵌入        参数：            texts：要嵌入的文本字符串列表            show_progress：是否显示进度条        返回：            嵌入的 NumPy 数组        """        embeddings = self.model.encode(            texts,            convert_to_numpy=True,            show_progress_bar=show_progress,            normalize_embeddings=True# L2 归一化用于计算余弦相似度        )        return embeddings.astype('float32')    def add_chunks(self, chunks: List[Dict]):        """        将块添加到向量数据库        参数：            chunks：包含 'text' 字段的块字典列表        """        ifnot chunks:            print("没有块要添加")            return        print(f"正在为 {len(chunks)} 个块生成嵌入...")        # 从块中提取文本        texts = [chunk['text'] for chunk in chunks]        # 生成嵌入        embeddings = self.embed_texts(texts)        # 添加到 FAISS 索引        self.index.add(embeddings)        # 存储块元数据        self.chunks.extend(chunks)        print(f"已将 {len(chunks)} 个块添加到索引（总数：{len(self.chunks)})")    def search(self, query: str, k: int = 5) -> List[Tuple[Dict, float]]:        """        搜索相似的块        参数：            query：搜索查询            k：要返回的结果数量        返回：            包含（块，分数）元组的列表        """        if len(self.chunks) == 0:            print("警告：数据库中没有块")            return []        # 嵌入查询        query_embedding = self.embed_texts([query], show_progress=False)        # 搜索 FAISS 索引        scores, indices = self.index.search(query_embedding, k)        # 准备结果        results = []        for idx, score in zip(indices[0], scores[0]):            if idx < len(self.chunks):                results.append((self.chunks[idx], float(score)))        return results    def save(self, directory: str):        """        将索引和块保存到磁盘        参数：            directory：保存文件的目录        """        Path(directory).mkdir(parents=True, exist_ok=True)        # 保存 FAISS 索引        index_path = Path(directory) / "faiss_index.bin"        faiss.write_index(self.index, str(index_path))        # 保存块        chunks_path = Path(directory) / "chunks.pkl"        with open(chunks_path, 'wb') as f:            pickle.dump(self.chunks, f)        print(f"已将索引和块保存到 {directory}")    def load(self, directory: str):        """        从磁盘加载索引和块        参数：            directory：包含保存文件的目录        """        # 加载 FAISS 索引        index_path = Path(directory) / "faiss_index.bin"        self.index = faiss.read_index(str(index_path))        # 加载块        chunks_path = Path(directory) / "chunks.pkl"        with open(chunks_path, 'rb') as f:            self.chunks = pickle.load(f)        print(f"已从 {directory} 加载 {len(self.chunks)} 个块")

FAISS 提供了快速的相似性搜索，能够扩展到数百万个向量。使用内积与归一化的嵌入可以高效地计算余弦相似度——这种数学运算决定了语义相似性。

集成 LLM 用于生成

设置答案生成

生成器组件负责加载 LLM，并处理提示构建和答案生成：

# src/generator.pyfrom llama_cpp import Llamafrom typing import List, Dictclass Generator:    """使用本地 LLM 处理答案生成"""    def __init__(self, model_path: str, n_ctx: int = 4096, n_threads: int = 8):        """        初始化生成器        参数：            model_path：GGUF 模型文件的路径            n_ctx：上下文窗口大小            n_threads：使用的 CPU 线程数        """        print(f"正在从 {model_path} 加载 LLM")        self.llm = Llama(            model_path=model_path,            n_ctx=n_ctx,            n_threads=n_threads,            verbose=False        )        print("LLM 加载成功")    def build_prompt(self, query: str, context_chunks: List[Dict]) -> str:        """        构建带有查询和检索到的上下文的 RAG 提示        参数：            query：用户的问题            context_chunks：带有元数据的检索到的块        返回：            完整的提示字符串        """        # 构建上下文部分        context_parts = []        for i, chunk in enumerate(context_chunks, 1):            source = chunk.get('source', '未知')            text = chunk['text']            context_parts.append(f"[来源 {i}：{source}]\n{text}")        context = "\n\n".join(context_parts)        # 构建完整的提示        prompt = f"""你是一个基于提供的上下文回答问题的有用助手。上下文：{context}问题：{query}指示：- 只使用提供的上下文回答问题- 如果答案不在上下文中，就说“我没有足够的信息来回答那个问题”- 提及信息时引用来源[来源 X]- 保持简洁直接答案："""        return prompt    def generate(self, query: str, context_chunks: List[Dict],                 max_tokens: int = 512, temperature: float = 0.7) -> Dict:        """        使用检索到的上下文生成答案        参数：            query：用户的问题            context_chunks：检索到的块            max_tokens：最大生成的 token 数量            temperature：采样温度        返回：            包含答案和元数据的字典        """        # 构建提示        prompt = self.build_prompt(query, context_chunks)        # 生成答案        response = self.llm(            prompt,            max_tokens=max_tokens,            temperature=temperature,            stop=["问题：", "\n\n\n"],            echo=False        )        answer = response['choices'][0]['text'].strip()        # 准备结果        result = {            'query': query,            'answer': answer,            'sources': [chunk.get('source', '未知') for chunk in context_chunks],            'num_chunks': len(context_chunks)        }        return result

提示设计明确指示模型只使用提供的上下文并引用来源，减少了幻觉现象。停止序列防止模型生成后续问题或无关内容。

六、完整的 RAG 系统集成

协调所有组件

现在，我们将所有内容整合到一个协调的系统中：

# src/rag_system.pyfrom .document_processor import DocumentProcessorfrom .embedding_manager import EmbeddingManagerfrom .generator import Generatorfrom pathlib import Pathfrom typing import List, Dictclass RAGSystem:    """协调所有组件的完整 RAG 系统"""    def __init__(        self,        model_path: str,        index_dir: str = "./data/processed",        embedding_model: str = "all-MiniLM-L6-v2",        chunk_size: int = 500,        chunk_overlap: int = 100    ):        """        初始化完整的 RAG 系统        参数：            model_path：LLM 模型文件的路径            index_dir：保存/加载索引的目录            embedding_model：sentence transformer 模型名称            chunk_size：每个块的单词数            chunk_overlap：块之间重叠的单词数        """        self.index_dir = index_dir        # 初始化组件        self.processor = DocumentProcessor(chunk_size, chunk_overlap)        self.embedding_manager = EmbeddingManager(embedding_model)        self.generator = Generator(model_path)        # 尝试加载现有的索引        if Path(index_dir).exists() and \           (Path(index_dir) / "faiss_index.bin").exists():            print(f"正在从 {index_dir} 加载现有的索引")            self.embedding_manager.load(index_dir)    def index_documents(self, documents_dir: str, save: bool = True):        """        处理并索引目录中的所有文档        参数：            documents_dir：包含文档的目录            save：是否将索引保存到磁盘        """        print(f"\n=== 从 {documents_dir} 索引文档 ===")        # 处理文档        chunks = self.processor.process_directory(documents_dir)        ifnot chunks:            print("没有创建块。请检查文档目录。")            return        # 将块添加到向量数据库        self.embedding_manager.add_chunks(chunks)        # 保存索引        if save:            self.embedding_manager.save(self.index_dir)    def query(        self,        question: str,        k: int = 3,        max_tokens: int = 512,        temperature: float = 0.7,        show_context: bool = False    ) -> Dict:        """        查询 RAG 系统        参数：            question：用户的问题            k：要检索的块的数量            max_tokens：答案的最大 token 数量            temperature：生成温度            show_context：是否显示检索到的上下文        返回：            包含答案和元数据的字典        """        print(f"\n=== 处理查询 ===")        print(f"问题：{question}")        # 检索相关的块        print(f"正在检索前 {k} 个相关的块...")        results = self.embedding_manager.search(question, k)        ifnot results:            return {                'query': question,                'answer': "知识库中未找到相关信息。",                'sources': [],                'num_chunks': 0            }        # 提取块和分数        chunks = [chunk for chunk, score in results]        scores = [score for chunk, score in results]        print(f"检索到 {len(chunks)} 个块（平均相似度：{sum(scores)/len(scores):.3f})")        # 如果请求，显示检索到的上下文        if show_context:            print("\n=== 检索到的上下文 ===")            for i, (chunk, score) in enumerate(zip(chunks, scores), 1):                print(f"\n[块 {i}] (分数：{score:.3f})")                print(f"来源：{chunk.get('source', '未知')}")                print(f"文本预览：{chunk['text'][:200]}...")        # 生成答案        print("\n=== 生成答案 ===")        result = self.generator.generate(            question,            chunks,            max_tokens=max_tokens,            temperature=temperature        )        # 将检索分数添加到结果中        result['retrieval_scores'] = scores        return result# main.py - 示例用法from src.rag_system import RAGSystemimport sysdef main():    # 配置    MODEL_PATH = "./models/llama-2-7b-chat.Q4_K_M.gguf"    DOCUMENTS_DIR = "./data/documents"    INDEX_DIR = "./data/processed"    # 检查模型是否存在    ifnot Path(MODEL_PATH).exists():        print(f"错误：模型未找到于 {MODEL_PATH}")        print("请下载 GGUF 模型并更新 MODEL_PATH")        sys.exit(1)    # 初始化 RAG 系统    print("=== 初始化 RAG 系统 ===")    rag = RAGSystem(        model_path=MODEL_PATH,        index_dir=INDEX_DIR,        embedding_model="all-MiniLM-L6-v2",        chunk_size=500,        chunk_overlap=100    )    # 索引文档（仅当没有现有索引时）    ifnot Path(INDEX_DIR).exists() or \       not (Path(INDEX_DIR) / "faiss_index.bin").exists():        rag.index_documents(DOCUMENTS_DIR, save=True)    # 示例查询    queries = [        "什么是检索增强型生成？",        "向量数据库是如何工作的？",        "本地 AI 部署的好处是什么？"    ]    # 处理查询    for query in queries:        result = rag.query(            query,            k=3,            temperature=0.7,            show_context=False        )        print(f"\n问题：{result['query']}")        print(f"答案：{result['answer']}")        print(f"来源：{', '.join(result['sources'])}")        print("-" * 80)    # 交互模式    print("\n=== 交互模式 ===")    print("请输入问题（或输入 'quit' 退出）：")    whileTrue:        question = input("\n你：").strip()        if question.lower() in ['quit', 'exit', 'q']:            print("再见！")            break        ifnot question:            continue        result = rag.query(question, k=3)        print(f"\n助手：{result['answer']}")        if result['sources']:            print(f"(来源：{', '.join(set(result['sources']))})")if __name__ == "__main__":    main()

这个完整的实现提供了一个具有适当错误处理、进度跟踪以及批量和交互式查询模式的生产级 RAG 系统。

七、测试和优化

运行

在实现所有组件后，使用示例文档和查询来测试系统，以验证其正常运行。在 data/documents/中放置一些 PDF 或文本文件，然后运行：

python main.py

系统将处理文档，生成嵌入，构建索引，并进入交互模式。尝试提出一些可以从你的文档中找到答案的问题，以及一些不能找到答案的问题，以验证系统能否正确区分可检索和不可检索的信息。

在测试过程中，监控性能指标：文档处理速度（每秒的块数）、嵌入生成时间（通常在 CPU 上为每秒 50 - 200 个块）、检索延迟（应低于 100 毫秒）以及生成速度（根据硬件不同而变化，但每秒 5 - 20 个 token 是典型的）。这些基准有助于在添加功能时识别性能退化。

质量评估需要同时评估检索和生成。检索是否为查询返回了相关的块？生成器是否产生了准确的答案，并且适当地引用了来源？记录失败模式——检索到与检索到的上下文相矛盾的答案、检索到与检索到的上下文相矛盾的答案，或者在来源中不存在的幻觉事实。这些失败指导优化的优先级。

参数调整以获得更好的结果

块大小显著影响检索质量。较小的块（200 - 300 个单词）可以提供精确匹配，但可能缺乏复杂问题所需的上下文。较大的块（600 - 800 个单词）包含更多的上下文，但可能会包含稀释相关性得分的无关信息。使用不同的块大小测试你的特定文档集合，测量正确答案出现在检索到的块中的频率。

检索到的块的数量（k 参数）在上下文的完整性与噪声之间取得了平衡。检索到的块太少可能会错过相关的信息，当它分布在多个段落中时。检索到的块太多会包含混淆生成器或超出上下文窗口限制的无关内容。从 k=3 - 5 开始，只有当答案经常引用存在于你的文档中的缺失信息时，才增加它。

生成温度控制创造力与一致性的平衡。较低的温度（0.3 - 0.5）产生更确定性、更集中的答案，适用于事实性问题。较高的温度（0.7 - 1.0）增加了多样性，但可能会引入幻觉或风格上的不一致性。对于 RAG 系统来说，准确性最为重要，因此优先选择较低的温度，保持接近检索到的上下文。

嵌入模型选择在速度与质量之间进行了权衡。all-MiniLM-L6-v2 模型在 CPU 上提供了良好的速度和适用于大多数用例的质量。升级到 all-mpnet-base-v2 可以将检索准确性提高 5 - 10%，但运行速度会慢 2 - 3 倍。对于生产系统来说，质量至关重要，升级是值得的。对于实验，从速度更快的模型开始。

处理边缘情况和错误

当查询在语义上与任何文档都不匹配时，会出现空的检索结果。这可能是由于问题超出了文档范围、存在拼写错误，或者知识库中确实没有相关的信息。系统应该检测到空的结果，并告知用户，而不是尝试从空白中生成答案。实现最小相似度阈值，以拒绝低置信度的检索。

长文档可能会产生数百个块，可能会使系统不堪重负或减慢索引速度。实现进度跟踪和错误处理，即使个别文档失败，也要继续处理。考虑并行处理极其大的文档，或者将它们分解为逻辑部分（章节、部分），这些部分成为单独的可索引单元。

在资源受限的系统上，内存约束需要谨慎管理。在索引过程中监控内存使用情况——如果同时处理所有文档会耗尽内存，则将文档分批处理，每次处理 10 - 100 个文档。FAISS 索引随着文档数量的增加而线性增长，因此在索引大型集合之前估算内存需求。

八、生产考虑

API 包装

为 RAG 系统包装 API，使其能够被其他应用程序访问。一个简单的 Flask 或 FastAPI 服务器暴露了查询和文档管理的端点：

from fastapi import FastAPI, UploadFile, Filefrom src.rag_system import RAGSystemapp = FastAPI()rag = RAGSystem(model_path="./models/model.gguf")@app.post("/query")asyncdef query(question: str, k: int = 3):    result = rag.query(question, k=k)    return result@app.post("/upload")asyncdef upload_document(file: UploadFile = File(...)):    # 保存上传的文件    # 重新索引文档    return {"status": "success"}

这个 API 使得 Web 应用程序、移动应用程序或其他服务能够使用你的 RAG 系统，而无需直接集成 Python 代码。为生产部署添加身份验证、速率限制和日志记录。

增量更新

允许添加新文档而无需重新构建整个索引。存储文档的哈希值或修改时间戳，在重新启动时检查是否有新文档。只处理已更改的文档，并将它们的块添加到现有的索引中。这种增量方法可以扩展到大型的、不断增长的文档集合，其中完整的重新索引变得过于昂贵。

监控和日志记录

监控和日志记录可以捕获系统行为，以便进行调试和优化。记录检索查询、相似度分数、生成时间和错误。聚合日志以识别常见的失败模式、流行的查询和性能瓶颈。这种遥测技术指导优化工作，并揭示使用模式，从而为系统改进提供信息。

九、高级增强

多模态文档支持

扩展到文本之外，支持图像、表格和图表，需要额外的处理。对于图像，使用 tesseract 进行 OCR，从图表和图表中提取文本。对于表格，使用专门的提取器保留结构——将表格转换为格式化的文本描述，这些描述嵌入了有意义的关系。RAG 核心管道保持不变；只有文档处理需要适应。

代码文档受益于语法感知分块，它尊重函数边界，而不是任意的单词计数。在函数定义、类声明或逻辑代码块上进行分割。将每个块周围的上下文（导入、类定义）包含在内，以便检索到的代码片段能够独立理解。

混合搜索实现

将密集嵌入与稀疏关键词搜索结合起来，可以改善从精确匹配中受益的查询的检索。实现 BM25 与向量搜索并行，然后使用互惠排名融合合并结果。包含技术术语、产品名称或缩写的查询，使用混合方法比纯语义搜索检索得更好。

查询重写和扩展

复杂或模糊的查询在检索之前受益于重写。使用 LLM 生成替代措辞、扩展缩写或将复杂问题拆分为子问题。检索所有变体的文档，并合并结果。这种预处理改善了用户措辞模糊或使用特定领域术语时的检索。

十、排查常见问题

检索返回不相关的块

当检索到的块始终遗漏相关信息时，调查嵌入质量和块边界。尝试不同的嵌入模型——在类似文本上训练的特定于领域的模型通常比通用模型表现更好。调整块大小和重叠部分，以确保完整的想法保持在一起，而不是被分割到边界之外。

可视化有助于调试检索问题。使用降维（t-SNE 或 UMAP）在 2D 空间中绘制嵌入，按文档来源着色。检索不佳的查询往往在嵌入空间中与相关文档聚集得较远，表明词汇或语义不匹配。这种视觉反馈指导是否尝试不同的嵌入模型或添加查询扩展。

生成的答案忽略了检索到的上下文

当 LLM 生成的答案与检索到的上下文相矛盾或忽略检索到的上下文时，加强提示指令。明确多次说明“只使用提供的上下文”。在提示中包含示例，展示期望的行为——当有可用信息时从上下文中回答，当信息缺失时说“我不知道”。降低生成温度，减少可能会偏离上下文的创造性。

由于训练差异，某些 LLM 更容易忽略指令。如果提示工程没有帮助，考虑尝试不同的基础模型。专门针对指令遵循进行微调的模型（如 Llama 2 Chat、Mistral Instruct）通常比基础模型更能尊重上下文。

性能瓶颈

通过在每个管道阶段进行计时测量来识别瓶颈。如果检索占主导地位，优化向量搜索——使用近似最近邻索引（如 FAISS IVF）而不是平面搜索。如果生成速度慢，尝试使用更小的模型、更积极的量化，或者如果有可用的话，使用 GPU 加速。

在索引期间出现内存瓶颈表明批处理大小过大。以较小的批次处理文档，一次为 100 - 200 个块创建嵌入。这种权衡速度以换取内存效率的做法，使得有限内存的系统能够处理大型文档集合。

十、部署和维护

容器化

Docker 容器简化了在不同系统上的部署。创建一个包含所有依赖项、模型和代码的 Dockerfile：

FROM python:3.10-slimWORKDIR /app# 安装系统依赖项RUN apt-get update && apt-get install -y \    build-essential \    && rm -rf /var/lib/apt/lists/*# 安装 Python 依赖项COPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txt# 复制应用程序代码COPY src/ ./src/COPY main.py .# 创建目录RUN mkdir -p data/documents data/processed models# 运行应用程序CMD ["python", "main.py"]

使用挂载的卷构建并运行容器，用于文档和模型：

docker build -t rag-local .docker run -v $(pwd)/data:/app/data -v $(pwd)/models:/app/models rag-local

这种容器化部署确保了在开发、测试和生产环境中的一致行为。

持续更新

实现一个文档监视系统，它能够检测到新文件或已修改的文件，并自动重新索引：

import timefrom watchdog.observers import Observerfrom watchdog.events import FileSystemEventHandlerclass DocumentWatcher(FileSystemEventHandler):    def __init__(self, rag_system):        self.rag_system = rag_system    def on_created(self, event):        ifnot event.is_directory and event.src_path.endswith(('.pdf', '.txt')):            print(f"检测到新文档：{event.src_path}")            self.rag_system.index_documents(event.src_path, save=True)# 使用方法observer = Observer()observer.schedule(DocumentWatcher(rag), "./data/documents", recursive=True)observer.start()

这种自动化操作使得知识库能够随着文档的变化而保持最新状态，无需手动干预。

从零开始在本地实现一个完整的 RAG 系统，能够深入了解检索增强型生成的实际工作原理，从抽象的概念转变为能够处理文档、检索信息并生成答案的具体软件。我们构建的模块化架构——将文档处理、嵌入生成、检索和生成分别作为独立的组件——使得针对特定用例进行定制成为可能，同时保持了清晰的关注点分离。代码的每一部分都有明确的目的，使得系统易于维护和扩展，以适应不断变化的需求。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述