从零构建RAG系统：Python实现检索增强生成的完整指南

什么是RAG？为什么它如此重要？

RAG（Retrieval-Augmented Generation，检索增强生成）是当前大语言模型应用中最热门的架构之一。它的核心思想很简单：在生成回答之前，先从知识库中检索相关信息，然后将检索到的内容作为上下文提供给大模型，从而生成更准确、更有依据的回答。

传统的纯LLM方案存在几个致命问题：

• 知识截止日期：模型的训练数据有固定的时间边界，无法获取最新信息
• 幻觉问题：模型可能"一本正经地胡说八道"，编造不存在的事实
• 领域知识匮乏：对于企业内部文档、专业领域知识，通用模型往往力不从心
• 可追溯性差：无法验证回答的来源和依据

RAG通过引入外部知识检索环节，有效解决了这些问题。它让AI不再仅仅依赖参数化的记忆，而是能够实时访问和引用外部知识源。

RAG系统架构全景

一个完整的RAG系统通常包含以下核心组件：

用户提问 → 查询预处理 → 向量检索 → 上下文组装 → LLM生成 → 回答返回
                ↑
文档加载 → 文本分块 → 向量化 → 存入向量数据库

让我们从零开始，逐步构建这个系统。

第一步：环境准备与依赖安装

首先创建项目并安装必要的依赖：

mkdir rag-system && cd rag-system
python -m venv venv
source venv/bin/activate

pip install langchain langchain-openai chromadb sentence-transformers
pip install pypdf unstructured tiktoken

创建一个基础的配置文件：

# config.py
import os

# OpenAI API配置（也可替换为其他LLM提供商）
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key-here")
OPENAI_MODEL = "gpt-4o"

## 第二步：文档加载与分块策略

文档分块是RAG系统中极其关键的一步。分块质量直接影响检索的准确性和最终回答的质量。

python

document_loader.py

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import List
from langchain_core.documents import Document

class DocumentProcessor:
"""文档加载与分块处理器"""

def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
    self.chunk_size = chunk_size
    self.chunk_overlap = chunk_overlap
    self.text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""],
    )

def load_document(self, file_path: str) -> List[Document]:
    """根据文件类型自动选择加载器"""
    if file_path.endswith(".pdf"):
        loader = PyPDFLoader(file_path)
    elif file_path.endswith(".md"):
        loader = UnstructuredMarkdownLoader(file_path)
    elif file_path.endswith(".txt"):
        loader = TextLoader(file_path, encoding="utf-8")
    else:
        raise ValueError(f"不支持的文件格式: {file_path}")

    return loader.load()

def process(self, file_path: str) -> List[Document]:

第三步：向量数据库与Embedding

我们使用ChromaDB作为向量数据库，它轻量、易用，非常适合入门和中小规模应用。

# vector_store.py
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from typing import List, Optional
from config import EMBEDDING_MODEL, CHROMA_PERSIST_DIR


class VectorStoreManager:
    """向量数据库管理器"""

    def __init__(self, persist_directory: str = CHROMA_PERSIST_DIR):
        self.embeddings = HuggingFaceEmbeddings(
            model_name=EMBEDDING_MODEL,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        self.persist_directory = persist_directory
        self.vectorstore: Optional[Chroma] = None

    def create_vectorstore(
        self, documents: List[Document], collection_name: str = "default"
    ) -> Chroma:
        """从文档创建向量数据库"""
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory,
            collection_name=collection_name,
        )
        self.vectorstore.persist()
        print(f"向量数据库已创建，共 {len(documents)} 条记录")
        return self.vectorstore

    def load_vectorstore(self, collection_name: str = "default") -> Chroma:
        """加载已有的向量数据库"""
        self.vectorstore = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings,
            collection_name=collection_name,
        )
        return self.vectorstore

## 第四步：检索算法优化

基础的相似度检索往往不够，我们需要更智能的检索策略：

python

retriever.py

from langchain_community.vectorstores import Chroma
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from typing import List

class AdvancedRetriever:
"""高级检索器：混合检索 + 重排序"""

def __init__(self, vectorstore: Chroma, documents: List):
    self.vectorstore = vectorstore
    self.documents = documents

def create_hybrid_retriever(self, top_k: int = 10) -> EnsembleRetriever:
    """创建混合检索器（向量检索 + BM25关键词检索）"""
    # 向量检索器
    vector_retriever = vectorstore.as_retriever(
        search_type="mmr",  # 使用MMR算法增加多样性
        search_kwargs={"k": top_k, "fetch_k": top_k * 3},
    )

    # BM25关键词检索器
    bm25_retriever = BM25Retriever.from_documents(
        self.documents, k=top_k
    )

    # 混合检索
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],

第五步：与LLM集成生成回答

# rag_engine.py
from langchain_openai import ChatOpenAI
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import Chroma
from typing import List
from config import OPENAI_API_KEY, OPENAI_MODEL, TOP_K_RESULTS


class RAGEngine:
    """RAG引擎：检索 + 生成"""

    SYSTEM_PROMPT = """你是一个专业的知识助手。请根据以下检索到的上下文信息回答用户问题。

规则：
1. 仅基于提供的上下文信息回答，不要编造信息
2. 如果上下文中没有足够信息，请明确说明
3. 回答时引用信息来源
4. 使用专业但易懂的语言

上下文信息：
{context}

用户问题：{input}
"""

    def __init__(self, vectorstore: Chroma, top_k: int = TOP_K_RESULTS):
        self.llm = ChatOpenAI(
            api_key=OPENAI_API_KEY,
            model=OPENAI_MODEL,
            temperature=0.1,  # 低温度保证回答稳定性
        )
        self.vectorstore = vectorstore
        self.top_k = top_k
        self.chain = self._build_chain()

    def _build_chain(self):
        """构建RAG链"""
        retriever = self.vectorstore.as_retriever(
            search_type="mmr",
            search_kwargs={
                "k": self.top_k,
                "fetch_k": self.top_k * 3,
                "lambda_mult": 0.7,  # MMR多样性参数
            },
        )


## 第六步：完整系统整合

将所有组件整合到一起：

python

main.py

from document_loader import DocumentProcessor
from vector_store import VectorStoreManager
from rag_engine import RAGEngine
import os

def build_knowledge_base(directory: str):
"""从目录构建知识库"""
processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)
vs_manager = VectorStoreManager()

all_chunks = []
supported_extensions = [".pdf", ".md", ".txt"]

for filename in os.listdir(directory):
    ext = os.path.splitext(filename)[1].lower()
    if ext in supported_extensions:
        file_path = os.path.join(directory, filename)
        chunks = processor.process(file_path)
        all_chunks.extend(chunks)

if not all_chunks:
    print("未找到可处理的文档")
    return None

vectorstore = vs_manager.create_vectorstore(all_chunks)
return vectorstore

def main():
import argparse

parser = argparse.ArgumentParser(description="RAG系统")
parser.add_argument("--build", type=str, help="构建知识库的目录路径")
parser.add_argument("--query", type=str, help="查询问题")
args = parser.parse_args()

性能优化最佳实践

1. 分块优化

分块大小是影响RAG效果最关键的参数之一。建议根据你的文档类型进行实验：

# 针对不同文档类型的推荐配置
CHUNK_CONFIGS = {
    "技术文档": {"chunk_size": 800, "chunk_overlap": 100},
    "法律合同": {"chunk_size": 1200, "chunk_overlap": 200},
    "新闻文章": {"chunk_size": 500, "chunk_overlap": 50},
    "学术论文": {"chunk_size": 1000, "chunk_overlap": 150},
}

2. 查询预处理

对用户查询进行预处理可以显著提升检索质量：

python
from langchain_core.prompts import ChatPromptTemplate

# 查询重写：将用户问题改写为更适合检索的形式
query_rewrite_prompt = ChatPromptTemplate.from_template(
    """请将以下用户问题改写为一个更适合在知识库中检索的查询语句。
要求：提取关键概念，去除无关信息，保持语义完整。

原始问题：{question}


## 总结

本文从零开始构建了一个完整的RAG系统，涵盖了从文档处理到最终生成的全部流程。核心要点回顾：

1. **文档分块**是基础，选择合适的分块策略和参数至关重要
2. **向量数据库**（ChromaDB）提供了高效的相似度检索能力

---

> 📢 **本文为精简版，完整版包含独家工具推荐和深度分析，请访问 [WD Tech Blog](https://wdsega.github.io) 查看！**

*关注我的博客获取最新科技资讯、AI教程和效率工具推荐！*