章节导语

你问AI”公司去年Q3的财报表现如何”，AI一脸茫然。它不知道你公司的内部数据，也不知道最新的新闻。

RAG（检索增强生成）就是解决这个问题的技术——让AI”查资料”后再回答，而不是凭空编造。

本文系统讲解RAG的核心原理、架构演进、关键技术，每部分都有完整代码。

一、前置说明

1.1 学习路径

1.2 读者需要的基础

• Python基础：会处理数据
• 向量概念：知道什么是向量 Embedding
• LLM调用：知道怎么调用大模型

1.3 学习目标

学完本文，你将能够：

• 理解RAG的核心原理
• 实现Naive RAG到Advanced RAG的演进
• 掌握RAG的关键技术（分块、检索、重排序）
• 构建完整的RAG系统

二、RAG核心原理

2.1 什么是RAG

RAG = Retrieval-Augmented Generation（检索增强生成）。

核心思想：不只让AI凭空回答，而是先从外部知识库检索相关信息，再用这些信息增强提示词，让AI基于事实回答。

2.2 为什么需要RAG

• 知识过时：LLM训练数据有截止日期，不知道最新信息
• 幻觉问题：AI可能编造不存在的内容
• 私有知识：AI不知道你公司的内部数据
• 无法溯源：AI回答无法验证来源

2.3 RAG工作流程

图1：RAG工作流程

1. 索引：文档切分 → 向量化 → 存入向量数据库
2. 检索：用户问题向量化 → 在数据库中找相似内容
3. 生成：将检索结果注入Prompt → 调用LLM生成答案

2.4 RAG架构演进

三、环境配置

3.1 安装依赖

# 核心依赖
pip install langchain langchain-openai chromadb

# 向量化
pip install sentence-transformers

# 文档处理
pip install pypdf unstructured

3.2 API Key配置

# .env
OPENAI_API_KEY=sk-xxxx

# 向量数据库（可选）
CHROMA_DB_PATH=./chroma_db

四、Naive RAG：基础实现

4.1 完整代码

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

class NaiveRAG:
    """Naive RAG：基础检索增强生成
    
    流程：
    1. 加载文档
    2. 切分文本
    3. 向量化
    4. 存储到向量数据库
    5. 检索相关文档
    6. 生成答案
    """
    
    def __init__(self, persist_dir="./chroma_db"):
        self.persist_dir = persist_dir
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-3.5-turbo")
        self.vectorstore = None
        self.chain = None
    
    def load_documents(self, file_path):
        """加载文档
        
        支持 .txt, .pdf 等格式
        """
        print(f"📄 加载文档: {file_path}")
        loader = TextLoader(file_path)
        documents = loader.load()
        print(f"   加载了 {len(documents)} 个文档")
        return documents
    
    def split_documents(self, documents, chunk_size=500, chunk_overlap=50):
        """切分文档
        
        参数:
            chunk_size: 每个块的大小（字符数）
            chunk_overlap: 块之间的重叠（保持上下文连贯）
        """
        print(f"✂️ 切分文档: chunk_size={chunk_size}, overlap={chunk_overlap}")
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap
        )
        chunks = splitter.split_documents(documents)
        print(f"   切分出 {len(chunks)} 个块")
        return chunks
    
    def create_vectorstore(self, chunks):
        """创建向量数据库
        
        将文本块向量化后存储
        """
        print("🔢 向量化并存储...")
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        print("   向量数据库创建完成")
        return self.vectorstore
    
    def create_chain(self):
        """创建检索问答链"""
        print("🔗 创建检索问答链...")
        self.chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",  # 将检索结果拼接到prompt
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": 3}  # 检索返回3个最相关块
            )
        )
        print("   问答链创建完成")
        return self.chain
    
    def index(self, file_path):
        """索引文档（完整流程）"""
        docs = self.load_documents(file_path)
        chunks = self.split_documents(docs)
        self.create_vectorstore(chunks)
        self.create_chain()
        print("✅ 索引完成！可以开始问答了")
    
    def ask(self, question):
        """问答"""
        if not self.chain:
            raise ValueError("请先调用 index() 索引文档")
        
        print(f"\n❓ 问题: {question}")
        result = self.chain.run(question)
        print(f"🤖 回答: {result}")
        return result

# 示例使用
if __name__ == "__main__":
    # 创建RAG系统
    rag = NaiveRAG()
    
    # 模拟文档内容（实际使用时从文件加载）
    with open("knowledge.txt", "w") as f:
        f.write("""
        人工智能（AI）是当前最热门的技术领域之一。
        机器学习是AI的核心技术，通过数据训练模型。
        深度学习是机器学习的一个分支，使用神经网络。
        GPT是一种基于Transformer的大语言模型。
        RAG是检索增强生成，可以解决LLM知识过时的问题。
        """)
    
    # 索引文档
    rag.index("knowledge.txt")
    
    # 问答
    rag.ask("什么是RAG？")
    rag.ask("GPT是什么？")

4.2 运行结果

$ python naive_rag.py

📄 加载文档: knowledge.txt
   加载了 1 个文档
✂️ 切分文档: chunk_size=500, overlap=50
   切分出 3 个块
🔢 向量化并存储...
   向量数据库创建完成
🔗 创建检索问答链...
   问答链创建完成
✅ 索引完成！可以开始问答了

❓ 问题: 什么是RAG？
🤖 回答: RAG是检索增强生成（Retrieval-Augmented Generation），
它是一种结合了信息检索和文本生成的技术...

❓ 问题: GPT是什么？
🤖 回答: GPT是一种基于Transformer架构的大语言模型...

4.3 Naive RAG的问题

• 检索质量差：简单的向量相似度不总是准确
• 上下文碎片化：块太小可能丢失重要上下文
• 无关信息干扰：检索到的内容可能不完全相关

4.4 小结

Naive RAG是入门基础，但生产环境需要优化。

五、Advanced RAG：检索优化

5.1 优化策略

图2：Advanced RAG优化点

• 预处理优化：更好的切分策略、摘要索引
• 检索优化：混合检索、重排序、Query改写
• 后处理优化：结果去重、压缩、多答案融合

5.2 Query改写

用户问题可能表达不清，需要改写使其更利于检索。

from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate

class QueryRewriter:
    """Query改写：将用户问题改写得更适合检索
    
    解决的问题：
    - 用户问题可能口语化、不清晰
    - 缺少关键实体
    - 表达方式不利于向量检索
    """
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo")
    
    def rewrite(self, query):
        """改写Query
        
        生成3个不同的改写版本，选取最好的一个
        """
        prompt = f"""请将以下用户问题改写，使其更清晰、更适合检索。

原问题：{query}

要求：
1. 明确表达核心意图
2. 添加可能遗漏的实体
3. 使用更专业的术语

直接输出改写后的结果，不需要解释："""
        
        response = self.llm.predict(prompt)
        return response.strip()
    
    def rewrite_multiple(self, query):
        """生成多个改写版本，用于混合检索"""
        prompt = f"""针对以下问题，生成3个不同的改写版本，用于检索。

原问题：{query}

要求：
1. 版本1：更简洁
2. 版本2：添加更多上下文
3. 版本3：使用不同的同义词

格式：
版本1：[改写1]
版本2：[改写2]
版本3：[改写3]"""
        
        response = self.llm.predict(prompt)
        
        versions = []
        for line in response.split('\n'):
            if '：' in line or ':' in line:
                version = line.split('：')[-1] or line.split(':')[-1]
                versions.append(version.strip())
        
        return versions if len(versions) == 3 else [query]

# 测试
if __name__ == "__main__":
    rewriter = QueryRewriter()
    
    query = "那个什么GPT怎么来着"
    rewritten = rewriter.rewrite(query)
    print(f"原问题：{query}")
    print(f"改写后：{rewritten}")
    
    print("\n--- 多个版本 ---")
    versions = rewriter.rewrite_multiple(query)
    for i, v in enumerate(versions, 1):
        print(f"版本{i}：{v}")

5.3 混合检索

结合向量检索和关键词检索，取长补短。

from langchain.retrievers import EnsembleRetriever
from langchain.vectorstores import Chroma
from langchain.retrievers import BM25Retriever
from langchain.document_loaders import TextLoader

class HybridRetriever:
    """混合检索器：向量 + 关键词
    
    向量检索：语义相似度高，适合理解意图
    关键词检索：精确匹配，适合专有名词
    """
    
    def __init__(self, documents, embeddings):
        self.documents = documents
        self.embeddings = embeddings
    
    def create_vector_retriever(self, k=3):
        """创建向量检索器"""
        vectorstore = Chroma.from_documents(
            documents=self.documents,
            embedding=self.embeddings
        )
        return vectorstore.as_retriever(search_kwargs={"k": k})
    
    def create_bm25_retriever(self, k=3):
        """创建BM25关键词检索器
        
        BM25是一种成熟的关键词检索算法
        """
        # 简单的文本分割
        texts = [doc.page_content for doc in self.documents]
        metadatas = [doc.metadata for doc in self.documents]
        
        retriever = BM25Retriever.from_texts(
            texts=texts,
            metadatas=metadatas
        )
        retriever.k = k
        return retriever
    
    def create_ensemble(self, k=3):
        """创建混合检索器
        
        权重：向量60%，关键词40%
        """
        vector_retriever = self.create_vector_retriever(k)
        bm25_retriever = self.create_bm25_retriever(k)
        
        ensemble = EnsembleRetriever(
            retrievers=[vector_retriever, bm25_retriever],
            weights=[0.6, 0.4]  # 权重可调
        )
        return ensemble

# 使用示例
if __name__ == "__main__":
    # 假设已有文档
    from langchain.document_loaders import TextLoader
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain.embeddings import OpenAIEmbeddings
    
    loader = TextLoader("knowledge.txt")
    docs = loader.load()
    splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
    chunks = splitter.split_documents(docs)
    
    embeddings = OpenAIEmbeddings()
    
    hybrid = HybridRetriever(chunks, embeddings)
    retriever = hybrid.create_ensemble(k=3)
    
    # 测试检索
    results = retriever.get_relevant_documents("什么是RAG？")
    print(f"检索到 {len(results)} 个相关文档：")
    for i, doc in enumerate(results, 1):
        print(f"{i}. {doc.page_content[:100]}...")

5.4 重排序（Re-Ranking）

检索结果可能不够准确，用LLM重新排序。

class Reranker:
    """重排序：对检索结果重新排序
    
    检索模型和生成模型不同：
    - 检索模型：快速、语义匹配
    - 生成模型：更准确、但慢
    
    重排序：用LLM评估每个文档与问题的相关性
    """
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo")
    
    def rerank(self, query, documents, top_k=3):
        """重排序
        
        评估每个文档与问题的相关程度
        """
        print(f"🔄 对 {len(documents)} 个文档进行重排序...")
        
        scored_docs = []
        
        for doc in documents:
            prompt = f"""请评估以下文档与问题的相关性。

问题：{query}

文档内容：
{doc.page_content}

评分标准：
- 9-10分：文档直接且完整回答了问题
- 7-8分：文档与问题高度相关
- 4-6分：文档与问题部分相关
- 1-3分：文档与问题不相关

请只输出一个数字分数（1-10）："""
            
            response = self.llm.predict(prompt)
            
            try:
                score = float(response.strip())
            except:
                score = 5.0  # 默认分数
            
            scored_docs.append((doc, score))
            print(f"   文档：{doc.page_content[:50]}... → 分数：{score}")
        
        # 按分数排序
        scored_docs.sort(key=lambda x: x[1], reverse=True)
        
        # 返回top_k
        return [doc for doc, score in scored_docs[:top_k]]

# 测试
if __name__ == "__main__":
    reranker = Reranker()
    
    # 模拟文档
    docs = [
        type('obj', (object,), {'page_content': 'RAG是检索增强生成技术'})(),
        type('obj', (object,), {'page_content': '今天天气很好'})(),
        type('obj', (object,), {'page_content': '人工智能正在改变世界'})(),
    ]
    
    reranked = reranker.rerank("什么是RAG？", docs, top_k=3)
    print(f"\n✅ 重排后取top 3")

5.5 小结

Advanced RAG通过Query改写、混合检索、重排序等技术提高检索质量。

六、Modular RAG：模块化架构

6.1 模块化思想

将RAG拆分为独立模块，可根据场景灵活组合。

图3：Modular RAG模块

6.2 核心模块

6.3 完整代码

from abc import ABC, abstractmethod
from typing import List, Optional
from langchain.schema import Document

# ============== 模块接口 ==============

class Loader(ABC):
    """文档加载器基类"""
    @abstractmethod
    def load(self, path: str) -> List[Document]:
        pass

class Splitter(ABC):
    """文档切分器基类"""
    @abstractmethod
    def split(self, documents: List[Document]) -> List[Document]:
        pass

class Embedder(ABC):
    """向量化器基类"""
    @abstractmethod
    def embed(self, texts: List[str]) -> List[List[float]]:
        pass

class Retriever(ABC):
    """检索器基类"""
    @abstractmethod
    def retrieve(self, query: str, k: int) -> List[Document]:
        pass

class Generator(ABC):
    """生成器基类"""
    @abstractmethod
    def generate(self, query: str, context: List[Document]) -> str:
        pass

# ============== 具体实现 ==============

class TextFileLoader(Loader):
    """文本文件加载器"""
    def load(self, path: str) -> List[Document]:
        from langchain.document_loaders import TextLoader
        return TextLoader(path).load()

class RecursiveSplitter(Splitter):
    """递归字符切分器"""
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split(self, documents: List[Document]) -> List[Document]:
        from langchain.text_splitter import RecursiveCharacterTextSplitter
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap
        )
        return splitter.split_documents(documents)

class OpenAIEmbedder(Embedder):
    """OpenAI向量化器"""
    def __init__(self):
        from langchain.embeddings import OpenAIEmbeddings
        self.embedder = OpenAIEmbeddings()
    
    def embed(self, texts: List[str]) -> List[List[float]]:
        return self.embedder.embed_documents(texts)

class VectorRetriever(Retriever):
    """向量检索器"""
    def __init__(self, documents: List[Document], embedder: Embedder):
        from langchain.vectorstores import Chroma
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=embedder.embedder
        )
    
    def retrieve(self, query: str, k: int) -> List[Document]:
        return self.vectorstore.similarity_search(query, k=k)

class LLMGenerator(Generator):
    """LLM生成器"""
    def __init__(self, model="gpt-3.5-turbo"):
        from langchain.chat_models import ChatOpenAI
        self.llm = ChatOpenAI(model=model)
    
    def generate(self, query: str, context: List[Document]) -> str:
        context_text = "\n\n".join([doc.page_content for doc in context])
        
        prompt = f"""基于以下参考资料回答问题。如果参考资料中没有相关信息，请如实说明。

参考资料：
{context_text}

问题：{query}

回答："""
        
        return self.llm.predict(prompt)

# ============== Modular RAG 系统 ==============

class ModularRAG:
    """Modular RAG：模块化检索增强生成
    
    优点：
    - 每个模块可替换
    - 方便调试和优化
    - 可根据场景组合
    """
    
    def __init__(self):
        self.loader: Optional[Loader] = None
        self.splitter: Optional[Splitter] = None
        self.embedder: Optional[Embedder] = None
        self.retriever: Optional[Retriever] = None
        self.generator: Optional[Generator] = None
    
    def build(self, config: dict):
        """根据配置构建系统"""
        print("🔧 构建Modular RAG系统...")
        
        # 初始化各模块
        if config.get("loader") == "text":
            self.loader = TextFileLoader()
        
        if config.get("splitter") == "recursive":
            self.splitter = RecursiveSplitter(
                chunk_size=config.get("chunk_size", 500),
                chunk_overlap=config.get("chunk_overlap", 50)
            )
        
        if config.get("embedder") == "openai":
            self.embedder = OpenAIEmbedder()
        
        if config.get("retriever") == "vector":
            # 需要先有documents
            pass
        
        if config.get("generator") == "llm":
            self.generator = LLMGenerator(
                model=config.get("model", "gpt-3.5-turbo")
            )
        
        print("✅ 构建完成")
    
    def index(self, file_path: str):
        """索引文档"""
        # 加载
        docs = self.loader.load(file_path)
        print(f"   加载了 {len(docs)} 个文档")
        
        # 切分
        chunks = self.splitter.split(docs)
        print(f"   切分了 {len(chunks)} 个块")
        
        # 创建检索器
        self.retriever = VectorRetriever(chunks, self.embedder)
        print("   向量数据库创建完成")
    
    def ask(self, query: str) -> str:
        """问答"""
        # 检索
        docs = self.retriever.retrieve(query, k=3)
        print(f"   检索到 {len(docs)} 个相关文档")
        
        # 生成
        answer = self.generator.generate(query, docs)
        return answer

# 使用示例
if __name__ == "__main__":
    # 配置
    config = {
        "loader": "text",
        "splitter": "recursive",
        "embedder": "openai",
        "retriever": "vector",
        "generator": "llm",
        "chunk_size": 300,
        "model": "gpt-3.5-turbo"
    }
    
    # 创建系统
    rag = ModularRAG()
    rag.build(config)
    
    # 索引
    rag.index("knowledge.txt")
    
    # 问答
    answer = rag.ask("RAG是什么？")
    print(f"\n🤖 回答：{answer}")

6.4 小结

Modular RAG将系统拆分为独立模块，便于优化和扩展。

七、实战：企业知识库问答

7.1 需求分析

构建一个完整的企业知识库问答系统：

• 支持多格式文档（PDF、Word、TXT）
• 混合检索（向量+关键词）
• 重排序优化
• 带来源标注的回答

7.2 完整代码

import os
from typing import List
from langchain.document_loaders import PyPDFLoader, TextLoader, UnstructuredWordDocumentLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate

class EnterpriseKnowledgeBase:
    """企业知识库问答系统
    
    完整功能：
    - 多格式文档支持
    - 混合检索
    - 重排序
    - 带来源的回答
    """
    
    def __init__(self, persist_dir="./enterprise_kb"):
        self.persist_dir = persist_dir
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o")
        self.vectorstore = None
        self.hybrid_retriever = None
    
    def load_document(self, file_path: str):
        """加载文档，根据扩展名选择加载器"""
        ext = os.path.splitext(file_path)[1].lower()
        
        if ext == ".pdf":
            loader = PyPDFLoader(file_path)
        elif ext == ".txt":
            loader = TextLoader(file_path)
        elif ext in [".doc", ".docx"]:
            loader = UnstructuredWordDocumentLoader(file_path)
        else:
            raise ValueError(f"不支持的文件格式: {ext}")
        
        return loader.load()
    
    def index_documents(self, file_paths: List[str]):
        """索引多个文档"""
        print("📚 开始索引文档...")
        
        all_chunks = []
        
        for path in file_paths:
            print(f"\n处理: {path}")
            try:
                docs = self.load_document(path)
                print(f"   加载了 {len(docs)} 页")
                
                splitter = RecursiveCharacterTextSplitter(
                    chunk_size=500,
                    chunk_overlap=50
                )
                chunks = splitter.split_documents(docs)
                print(f"   切分为 {len(chunks)} 个块")
                
                all_chunks.extend(chunks)
            except Exception as e:
                print(f"   ⚠️ 处理失败: {e}")
        
        # 创建向量数据库
        print(f"\n🔢 向量化 {len(all_chunks)} 个块...")
        self.vectorstore = Chroma.from_documents(
            documents=all_chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        print("✅ 索引完成！")
    
    def retrieve_with_rerank(self, query: str, k: int = 5):
        """检索 + 重排序"""
        # 初步检索（多取一些）
        docs = self.vectorstore.similarity_search(query, k=k * 2)
        
        # 重排序
        reranked = self._rerank_documents(query, docs, top_k=k)
        
        return reranked
    
    def _rerank_documents(self, query: str, docs: List, top_k: int = 3):
        """LLM重排序"""
        scored = []
        
        for doc in docs:
            prompt = f"""评估文档与问题的相关性，输出1-10分：

问题：{query}

文档：{doc.page_content[:200]}

只输出数字："""
            
            score = float(self.llm.predict(prompt))
            scored.append((doc, score))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        return [doc for doc, score in scored[:top_k]]
    
    def generate_answer(self, query: str):
        """生成带来源的答案"""
        # 检索
        docs = self.retrieve_with_rerank(query, k=3)
        
        # 构建上下文
        context_with_sources = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source", "未知")
            page = doc.metadata.get("page", "N/A")
            context_with_sources.append(
                f"[来源{i}]（{source} 第{page}页）\n{doc.page_content}"
            )
        
        context_text = "\n\n".join(context_with_sources)
        
        # 生成答案
        prompt = f"""基于以下参考资料回答问题。答案中需要标注参考来源。

参考资料：
{context_text}

问题：{query}

要求：
1. 如果参考资料中有答案，基于资料回答
2. 每引用一段资料，用 [来源X] 标注
3. 如果资料不足，如实说明"""
        
        answer = self.llm.predict(prompt)
        
        return {
            "answer": answer,
            "sources": [
                {"content": doc.page_content[:100], "source": doc.metadata}
                for doc in docs
            ]
        }
    
    def chat(self, query: str):
        """对话入口"""
        print(f"\n👤 用户：{query}")
        
        result = self.generate_answer(query)
        
        print(f"\n🤖 回答：\n{result['answer']}")
        print(f"\n📚 参考来源：")
        for i, src in enumerate(result['sources'], 1):
            print(f"   [{i}] {src['content']}...")
        
        return result

# 测试
if __name__ == "__main__":
    kb = EnterpriseKnowledgeBase()
    
    # 创建测试文档
    os.makedirs("test_docs", exist_ok=True)
    with open("test_docs/公司制度.txt", "w") as f:
        f.write("""
        公司年假制度：
        - 工作满1年：5天年假
        - 工作满3年：10天年假
        - 工作满5年：15天年假
        
        加班调休：
        - 工作日加班按1.5倍计算
        - 周末加班按2倍计算
        - 节假日加班按3倍计算
        """)
    
    # 索引
    kb.index_documents(["test_docs/公司制度.txt"])
    
    # 问答
    kb.chat("工作3年有多少天年假？")

八、总结与练习

8.1 要点回顾

• Naive RAG：基础流程，适合入门
• Advanced RAG：Query改写、混合检索、重排序
• Modular RAG：模块化架构，灵活可扩展

8.2 选型指南

8.3 延伸阅读

• LangChain RAG：官方RAG教程
• RAG论文：Retrieval-Augmented Generation
• 向量数据库对比：Day9 向量数据库深入对比

8.4 课后练习

基础题：实现一个简单的RAG系统，索引自己的文档并问答。

进阶题：为RAG添加PDF文档支持和关键词检索。

挑战题：实现多跳推理RAG，能回答需要综合多份文档的问题。