章节导语

你问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. 索引：文档切分 → 向量化 → 存入向量数据库
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 优化策略

• 预处理优化：更好的切分策略、摘要索引
• 检索优化：混合检索、重排序、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拆分为独立模块，可根据场景灵活组合。

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，能回答需要综合多份文档的问题。