Skip to content

RAG 系统搭建入门指南

检索增强生成(RAG)是将 LLM 与外部知识库结合的最常用架构。本指南带你从零搭建一个完整的 RAG 系统,包含文档处理、向量化、检索、重排和生成的每个环节。

架构概览

RAG 系统架构用户查询User Query查询理解Embedding查询向量化知识检索Vector Search向量数据库结果重排Rerank相关性过滤上下文构建Context BuildLLM 生成Generation答案输出Answer核心组件 / CORE COMPONENTS文档处理Document ProcessingEmbedding 模型Embedding Model向量数据库Vector Database重排模型Reranker ModelLLMLarge Language Model检索增强生成 · RETRIEVAL-AUGMENTED GENERATION

核心组件

组件作用常用选择
文档处理将原始文档转换为可检索块LangChain、LlamaIndex、自研
Embedding 模型将文本转换为向量OpenAI text-embedding-3、BGE、E5
向量数据库存储和检索向量Chroma、Milvus、pgvector、Pinecone
LLM基于检索结果生成答案GPT-4o、Claude、Qwen
重排模型对检索结果进行精细排序同 Embedding 模型或专用重排模型

环境准备

bash
# 创建项目目录
mkdir my-rag-system && cd my-rag-system

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install langchain langchain-openai chromadb unstructured
pip install sentence-transformers  # 本地 embedding

第一步:文档处理

加载文档

python
from langchain.document_loaders import DirectoryLoader, TextLoader

# 加载单个文件
loader = TextLoader("docs/knowledge.txt")
documents = loader.load()

# 加载整个目录
loader = DirectoryLoader("docs/", glob="**/*.md")
documents = loader.load()

文本分块

分块策略对比

策略方法适用场景
固定长度每 n 个字符一块简单但可能切断语义
递归分割按标点、段落分割保持语义完整性
语义分块基于相似度合并最优但计算成本高
python
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 推荐:递归分割器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大字符数
    chunk_overlap=50,      # 重叠字符数(保持上下文连续性)
    separators=["\n\n", "\n", ".", " ", ""]  # 优先分割符
)

chunks = splitter.split_documents(documents)
print(f"分割为 {len(chunks)} 个块")

关键参数

  • chunk_size:常用 300-1000,太大会降低检索精度,太小会丢失上下文
  • chunk_overlap:建议为 chunk_size 的 10-20%,确保边界信息不丢失

第二步:向量化

选择 Embedding 模型

python
from langchain.embeddings import OpenAIEmbeddings
from langchain.embeddings import HuggingFaceEmbeddings

# 方案 1:OpenAI(需要 API Key)
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536  # 可缩放维度
)

# 方案 2:本地模型(免费、私有)
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5"  # 中文场景推荐
)

模型选择建议

场景推荐模型维度
英文通用text-embedding-3-large3072
中文场景BAAI/bge-large-zh-v1.51024
跨语言multilingual-e5-large1024
资源有限text-embedding-3-small1536

向量化实例

python
# 单个文本向量化
text = "RAG 是检索增强生成的缩写"
vector = embeddings.embed_query(text)
print(f"向量维度: {len(vector)}")  # 如 1536

# 批量向量化
doc_vectors = embeddings.embed_documents([chunk.page_content for chunk in chunks])

第三步:向量数据库

初始化 Chroma

python
import chromadb
from langchain.vectorstores import Chroma

# 本地持久化
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 数据存储路径
)

# 保存到磁盘
vectorstore.persist()

检索示例

python
# 相似度检索
results = vectorstore.similarity_search(
    query="什么是 RAG 架构?",
    k=5  # 返回 top-5 结果
)

for doc in results:
    print(f"内容: {doc.page_content[:100]}...")
    print(f"来源: {doc.metadata.get('source', 'unknown')}")
    print("---")

高级检索策略

python
# MMR 检索(最大边际相关性)
# 平衡相关性与多样性
results = vectorstore.max_marginal_relevance_search(
    query="RAG 架构",
    k=5,
    fetch_k=20,  # 先检索 20 个,再用 MMR 筛选
    lambda_mult=0.5  # 多样性权重
)

第四步:生成答案

基础 RAG 链

python
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.3  # 低温度保持准确性
)

# 构建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 将所有文档“塞入”提示
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True  # 返回引用来源
)

# 执行查询
result = qa_chain({"query": "请解释 RAG 的工作原理"})
print(result["result"])
print("\n引用来源:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata.get('source', 'unknown')}")

自定义提示模板

python
from langchain.prompts import PromptTemplate

custom_prompt = PromptTemplate(
    template="""你是一个专业的 AI 助手。请基于以下参考文档回答用户问题。
如果文档中没有相关信息,请直接说明。

参考文档:
{context}

用户问题:{question}

请用中文回答,并且在答案末尾列出引用的文档来源。
""",
    input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    chain_type_kwargs={"prompt": custom_prompt}
)

第五步:进阶优化

查询改写

问题:用户查询可能与文档用词不一致

解决:使用 LLM 改写查询

python
from langchain.chains import LLMChain

query_rewrite_prompt = PromptTemplate(
    template="将以下用户查询改写为适合检索的关键词形式:{query}",
    input_variables=["query"]
)

rewrite_chain = LLMChain(llm=llm, prompt=query_rewrite_prompt)
rewritten_query = rewrite_chain.run(query="RAG 是啥东西?")
# 输出: "检索增强生成 RAG 架构 原理"

重排策略

问题:相似度检索可能返回不相关的结果

解决:二次重排

python
# 方案 1:交叉编码器重排
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
compressor = CrossEncoderReranker(model=model, top_n=3)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# 方案 2:多路检索合并
from langchain.retrievers import EnsembleRetriever

# 结合多种检索方法
ensemble_retriever = EnsembleRetriever(
    retrievers=[
        vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 5}),
        vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5})
    ],
    weights=[0.5, 0.5]
)

响应生成优化

python
# 添加引用标记
def format_response(result):
    response = result["result"]
    sources = result.get("source_documents", [])
    
    if sources:
        response += "\n\n---\n**参考来源**\n"
        for i, doc in enumerate(sources, 1):
            source = doc.metadata.get('source', 'unknown')
            response += f"[{i}] {source}\n"
    
    return response

完整示例

python
# complete_rag.py
import os
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class SimpleRAG:
    def __init__(self, docs_path: str, db_path: str = "./chroma_db"):
        self.docs_path = docs_path
        self.db_path = db_path
        self.vectorstore = None
        self.qa_chain = None
        
        # 初始化组件
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
        
    def build(self):
        """构建知识库"""
        # 1. 加载
        loader = TextLoader(self.docs_path)
        documents = loader.load()
        
        # 2. 分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500, chunk_overlap=50
        )
        chunks = splitter.split_documents(documents)
        
        # 3. 存储
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.db_path
        )
        self.vectorstore.persist()
        
        # 4. 构建 QA 链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
            return_source_documents=True
        )
        
        print(f"知识库构建完成,共 {len(chunks)} 个文档块")
    
    def load(self):
        """加载已有知识库"""
        self.vectorstore = Chroma(
            persist_directory=self.db_path,
            embedding_function=self.embeddings
        )
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
            return_source_documents=True
        )
        print("知识库加载完成")
    
    def query(self, question: str) -> dict:
        """提问"""
        if not self.qa_chain:
            raise ValueError("请先调用 build() 或 load()")
        
        result = self.qa_chain({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.metadata.get('source', 'unknown') 
                       for doc in result["source_documents"]]
        }

# 使用示例
if __name__ == "__main__":
    rag = SimpleRAG(docs_path="knowledge.txt")
    
    # 首次运行构建
    rag.build()
    
    # 后续直接加载
    # rag.load()
    
    # 提问
    result = rag.query("什么是 RAG 架构?")
    print(f"答案: {result['answer']}")
    print(f"来源: {result['sources']}")

常见问题与解决

检索结果不准确

可能原因解决方案
分块过大减小 chunk_size
查询表述不清晰添加查询改写
Embedding 模型不匹配换用领域相关模型
缺少重排添加二次重排

生成答案幻觉

可能原因解决方案
检索结果不相关优化检索策略
提示词不足增加系统提示词约束
温度过高降低 temperature
上下文不足增加检索结果数量

进阶方向

  • 多模态 RAG:支持图片、表格、音频等非文本内容
  • Agentic RAG:让 Agent 决定如何检索和使用知识
  • GraphRAG:结合知识图谱增强关系推理
  • 持续学习:自动更新知识库以跟踪最新信息

相关页面

参考资源

AI Knowledge Base — 持续积累