前言：大模型的"知识困境"

大模型很强大，但有几个致命问题：

问题 1：知识有截止日期

用户: 2024 年诺贝尔物理学奖颁给了谁？

GPT-4 (2023年训练): 抱歉，我的知识截止到 2023 年 4 月，
                    无法回答 2024 年的问题。

问题 2：不知道你的私有数据

用户: 我们公司的报销流程是什么？

大模型: 一般公司的报销流程是... [给了一个通用答案]
        （并不知道你们公司的具体规定）

问题 3：会"幻觉"——一本正经地胡说八道

用户: 《三体》作者刘慈欣的代表作有哪些？

大模型: 刘慈欣的代表作包括《三体》《流浪地球》《球状闪电》
        《北京折叠》...  ← 这是郝景芳的作品！

RAG（Retrieval-Augmented Generation，检索增强生成） 就是解决这些问题的方案。

一、RAG 的核心思想

1.1 一句话解释

RAG = 开卷考试

• 闭卷考试：只能靠记忆（纯大模型）
• 开卷考试：可以翻书查资料（RAG）

1.2 RAG 的两个核心步骤

Step 1: Retrieval（检索）

• 把用户问题转换为向量
• 在知识库中找到最相关的文档
• 返回 Top-K 个结果

Step 2: Generation（生成）

• 把检索到的文档作为上下文
• 让大模型基于上下文回答问题

1.3 为什么 RAG 有效？

RAG 的优势：

1. 知识可更新：只需更新知识库，无需重新训练模型
2. 减少幻觉：回答有据可查
3. 可解释：可以展示引用来源
4. 成本低：比微调模型便宜得多

二、向量检索：RAG 的基础

2.1 文本向量化（Embedding）

把文本转换为向量，语义相近的文本向量距离更近。

from sentence_transformers import SentenceTransformer

# 加载 Embedding 模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 文本向量化
texts = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络",
    "今天天气很好",
]

embeddings = model.encode(texts)
print(f"向量维度: {embeddings.shape}")  # (3, 1024)

# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity

sim_matrix = cosine_similarity(embeddings)
print("相似度矩阵:")
print(sim_matrix)
# [[1.0, 0.82, 0.15],
#  [0.82, 1.0, 0.12],
#  [0.15, 0.12, 1.0]]
# 前两个文本语义相近，第三个不相关

2.2 常用 Embedding 模型

2.3 向量数据库

存储和检索向量需要专门的数据库：

2.4 Chroma 快速入门

import chromadb
from chromadb.utils import embedding_functions

# 创建客户端
client = chromadb.Client()

# 使用 BGE 作为 embedding 函数
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-large-zh-v1.5"
)

# 创建集合
collection = client.create_collection(
    name="my_knowledge_base",
    embedding_function=embedding_fn,
)

# 添加文档
documents = [
    "机器学习是人工智能的一个重要分支，它使计算机能够从数据中学习。",
    "深度学习是机器学习的子领域，使用多层神经网络。",
    "自然语言处理让计算机能够理解和生成人类语言。",
    "计算机视觉使机器能够从图像和视频中获取信息。",
]

collection.add(
    documents=documents,
    ids=[f"doc_{i}" for i in range(len(documents))],
    metadatas=[{"source": "ai_textbook"} for _ in documents],
)

# 检索
results = collection.query(
    query_texts=["什么是深度学习？"],
    n_results=2,
)

print("检索结果:")
for doc, distance in zip(results['documents'][0], results['distances'][0]):
    print(f"  [{distance:.4f}] {doc[:50]}...")

三、构建完整的 RAG 系统

3.1 RAG 系统架构

3.2 文档分块（Chunking）

文档太长无法直接处理，需要分块：

from langchain.text_splitter import RecursiveCharacterTextSplitter


def chunk_documents(documents: list[str], chunk_size=500, chunk_overlap=50):
    """文档分块"""
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    )
    
    chunks = []
    for doc in documents:
        chunks.extend(splitter.split_text(doc))
    
    return chunks


# 分块策略对比
strategies = {
    "固定大小": {"chunk_size": 500, "chunk_overlap": 50},
    "小块多重叠": {"chunk_size": 200, "chunk_overlap": 100},
    "大块少重叠": {"chunk_size": 1000, "chunk_overlap": 50},
}

# 选择建议：
# - 精确检索：小块 (200-500)
# - 保持上下文：大块 (500-1000)
# - 重叠：通常 10-20% 的 chunk_size

3.3 分块策略对比

3.4 完整 RAG 实现

import os
from typing import List, Dict
import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI


class SimpleRAG:
    """简单的 RAG 系统"""
    
    def __init__(
        self,
        embedding_model: str = "BAAI/bge-large-zh-v1.5",
        llm_model: str = "gpt-3.5-turbo",
        collection_name: str = "rag_collection",
    ):
        # Embedding 模型
        self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name=embedding_model
        )
        
        # 向量数据库
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn,
        )
        
        # LLM 客户端
        self.llm_client = OpenAI()
        self.llm_model = llm_model
    
    def add_documents(
        self,
        documents: List[str],
        metadatas: List[Dict] = None,
        chunk_size: int = 500,
    ):
        """添加文档到知识库"""
        from langchain.text_splitter import RecursiveCharacterTextSplitter
        
        # 分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=50,
        )
        
        all_chunks = []
        all_metadatas = []
        
        for i, doc in enumerate(documents):
            chunks = splitter.split_text(doc)
            all_chunks.extend(chunks)
            
            # 元数据
            meta = metadatas[i] if metadatas else {}
            all_metadatas.extend([{**meta, "chunk_index": j} for j in range(len(chunks))])
        
        # 添加到向量数据库
        self.collection.add(
            documents=all_chunks,
            ids=[f"chunk_{i}" for i in range(len(all_chunks))],
            metadatas=all_metadatas,
        )
        
        print(f"添加了 {len(all_chunks)} 个文档块")
    
    def retrieve(self, query: str, top_k: int = 3) -> List[str]:
        """检索相关文档"""
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
        )
        
        return results['documents'][0]
    
    def generate(self, query: str, context: List[str]) -> str:
        """基于上下文生成回答"""
        
        # 构造 Prompt
        context_str = "\n\n".join([f"[文档 {i+1}]: {doc}" for i, doc in enumerate(context)])
        
        prompt = f"""请基于以下参考文档回答用户的问题。如果文档中没有相关信息，请说明。

参考文档：
{context_str}

用户问题：{query}

回答："""
        
        # 调用 LLM
        response = self.llm_client.chat.completions.create(
            model=self.llm_model,
            messages=[
                {"role": "system", "content": "你是一个helpful assistant，基于提供的文档回答问题。"},
                {"role": "user", "content": prompt},
            ],
            temperature=0.7,
        )
        
        return response.choices[0].message.content
    
    def query(self, question: str, top_k: int = 3) -> Dict:
        """完整的 RAG 查询"""
        
        # 1. 检索
        context = self.retrieve(question, top_k)
        
        # 2. 生成
        answer = self.generate(question, context)
        
        return {
            "question": question,
            "answer": answer,
            "sources": context,
        }


# 使用示例
if __name__ == "__main__":
    # 创建 RAG 系统
    rag = SimpleRAG()
    
    # 添加文档
    documents = [
        """
        机器学习（Machine Learning）是人工智能的一个分支，它使计算机系统能够从数据中学习和改进，
        而无需进行明确的编程。机器学习算法通过分析大量数据来识别模式，并使用这些模式来做出决策或预测。
        主要的机器学习类型包括：监督学习、无监督学习和强化学习。
        """,
        """
        深度学习（Deep Learning）是机器学习的一个子领域，它使用多层神经网络来学习数据的层次表示。
        深度学习在图像识别、自然语言处理和语音识别等领域取得了突破性的成果。
        常见的深度学习架构包括：卷积神经网络（CNN）、循环神经网络（RNN）和Transformer。
        """,
        """
        大型语言模型（LLM）是一种基于深度学习的自然语言处理模型，通过在海量文本数据上进行预训练，
        学习语言的统计规律和语义知识。代表性的大型语言模型包括 GPT 系列、BERT、LLaMA 等。
        LLM 可以执行多种任务，如文本生成、问答、翻译、摘要等。
        """,
    ]
    
    rag.add_documents(documents, metadatas=[{"source": "ai_intro"} for _ in documents])
    
    # 查询
    result = rag.query("什么是深度学习？它和机器学习有什么关系？")
    
    print("问题:", result["question"])
    print("\n回答:", result["answer"])
    print("\n来源文档:")
    for i, source in enumerate(result["sources"]):
        print(f"  [{i+1}] {source[:100]}...")

四、高级 RAG 技术

4.1 RAG 优化全景

4.2 Query 改写（Query Rewriting）

用户的问题可能不适合直接检索：

def rewrite_query(query: str, llm_client) -> List[str]:
    """改写用户问题，生成多个检索 query"""
    
    prompt = f"""请将以下用户问题改写成 3 个更适合检索的查询语句。
要求：
1. 保持原意
2. 使用不同的表达方式
3. 可以拆分成子问题

用户问题：{query}

改写后的查询（每行一个）："""
    
    response = llm_client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )
    
    queries = response.choices[0].message.content.strip().split("\n")
    return [q.strip() for q in queries if q.strip()]


# 示例
original = "深度学习和传统机器学习有什么区别，各自适用于什么场景？"
rewritten = rewrite_query(original, llm_client)
# 可能的改写结果：
# 1. "深度学习与机器学习的区别"
# 2. "深度学习适用场景"  
# 3. "传统机器学习应用领域"

4.3 混合检索（Hybrid Search）

结合向量检索和关键词检索：

from rank_bm25 import BM25Okapi
import numpy as np


class HybridRetriever:
    """混合检索：向量 + BM25"""
    
    def __init__(self, documents: List[str], embedding_model):
        self.documents = documents
        self.embedding_model = embedding_model
        
        # 向量索引
        self.embeddings = embedding_model.encode(documents)
        
        # BM25 索引
        tokenized = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)
    
    def search(
        self,
        query: str,
        top_k: int = 5,
        vector_weight: float = 0.5,
    ) -> List[tuple]:
        """混合检索"""
        
        # 向量检索
        query_embedding = self.embedding_model.encode([query])[0]
        vector_scores = np.dot(self.embeddings, query_embedding)
        
        # BM25 检索
        bm25_scores = self.bm25.get_scores(query.split())
        
        # 归一化
        vector_scores = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min() + 1e-6)
        bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-6)
        
        # 加权融合
        final_scores = vector_weight * vector_scores + (1 - vector_weight) * bm25_scores
        
        # 排序
        top_indices = np.argsort(final_scores)[::-1][:top_k]
        
        return [(self.documents[i], final_scores[i]) for i in top_indices]

4.4 重排序（Reranking）

用更精确的模型对初步检索结果重新排序：

from sentence_transformers import CrossEncoder


class Reranker:
    """重排序器"""
    
    def __init__(self, model_name: str = "BAAI/bge-reranker-large"):
        self.model = CrossEncoder(model_name)
    
    def rerank(
        self,
        query: str,
        documents: List[str],
        top_k: int = 3,
    ) -> List[tuple]:
        """重新排序"""
        
        # 构造 query-document 对
        pairs = [[query, doc] for doc in documents]
        
        # 计算相关性分数
        scores = self.model.predict(pairs)
        
        # 排序
        sorted_indices = np.argsort(scores)[::-1][:top_k]
        
        return [(documents[i], scores[i]) for i in sorted_indices]


# 使用
reranker = Reranker()

# 初步检索：召回 20 个
initial_results = retriever.search(query, top_k=20)

# 重排序：精选 5 个
final_results = reranker.rerank(
    query,
    [doc for doc, _ in initial_results],
    top_k=5
)

4.5 Self-RAG：自我反思的 RAG

class SelfRAG:
    """Self-RAG：带自我反思的 RAG"""
    
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm = llm_client
    
    def needs_retrieval(self, query: str) -> bool:
        """判断是否需要检索"""
        prompt = f"""判断以下问题是否需要外部知识来回答。
        
问题：{query}

只回答 "是" 或 "否"："""
        
        response = self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        
        return "是" in response.choices[0].message.content
    
    def is_relevant(self, query: str, document: str) -> bool:
        """判断文档是否相关"""
        prompt = f"""判断以下文档是否与问题相关。

问题：{query}
文档：{document}

只回答 "相关" 或 "不相关"："""
        
        response = self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        
        return "相关" in response.choices[0].message.content
    
    def is_supported(self, answer: str, documents: List[str]) -> bool:
        """判断回答是否有文档支撑"""
        docs_str = "\n".join(documents)
        
        prompt = f"""判断以下回答是否有文档支撑。

文档：{docs_str}

回答：{answer}

只回答 "有支撑" 或 "无支撑"："""
        
        response = self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        
        return "有支撑" in response.choices[0].message.content
    
    def query(self, question: str, max_iterations: int = 3) -> str:
        """Self-RAG 查询"""
        
        # 判断是否需要检索
        if not self.needs_retrieval(question):
            return self.generate_direct(question)
        
        for _ in range(max_iterations):
            # 检索
            documents = self.retriever.retrieve(question)
            
            # 过滤不相关文档
            relevant_docs = [doc for doc in documents if self.is_relevant(question, doc)]
            
            if not relevant_docs:
                continue
            
            # 生成回答
            answer = self.generate_with_context(question, relevant_docs)
            
            # 验证支撑
            if self.is_supported(answer, relevant_docs):
                return answer
        
        return "抱歉，我无法找到足够的信息来回答这个问题。"

五、使用 LangChain 构建 RAG

5.1 LangChain RAG 快速入门

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate


def build_rag_with_langchain():
    """使用 LangChain 构建 RAG"""
    
    # 1. 加载文档
    loader = PyPDFLoader("document.pdf")
    documents = loader.load()
    
    # 2. 分块
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
    )
    chunks = splitter.split_documents(documents)
    
    # 3. 向量化和存储
    embeddings = HuggingFaceEmbeddings(
        model_name="BAAI/bge-large-zh-v1.5"
    )
    
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory="./chroma_db",
    )
    
    # 4. 创建检索器
    retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3},
    )
    
    # 5. 创建 LLM
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    # 6. 自定义 Prompt
    prompt_template = """请基于以下上下文回答问题。如果上下文中没有相关信息，请说明。

上下文：
{context}

问题：{question}

回答："""
    
    prompt = PromptTemplate(
        template=prompt_template,
        input_variables=["context", "question"],
    )
    
    # 7. 创建 QA Chain
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
        chain_type_kwargs={"prompt": prompt},
    )
    
    return qa_chain


# 使用
qa_chain = build_rag_with_langchain()
result = qa_chain.invoke({"query": "文档的主要内容是什么？"})

print("回答:", result["result"])
print("\n来源文档:")
for doc in result["source_documents"]:
    print(f"  - {doc.page_content[:100]}...")

5.2 LangChain + 高级检索

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever


def build_advanced_retriever(documents, embeddings, llm):
    """构建高级检索器"""
    
    # 向量检索器
    vectorstore = Chroma.from_documents(documents, embeddings)
    vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
    
    # BM25 检索器
    bm25_retriever = BM25Retriever.from_documents(documents)
    bm25_retriever.k = 5
    
    # 混合检索器
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.5, 0.5],
    )
    
    # 上下文压缩（提取关键信息）
    compressor = LLMChainExtractor.from_llm(llm)
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=ensemble_retriever,
    )
    
    return compression_retriever

5.3 对话式 RAG

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory


def build_conversational_rag(vectorstore, llm):
    """构建对话式 RAG"""
    
    # 记忆
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
    )
    
    # 对话式检索链
    qa_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        memory=memory,
        return_source_documents=True,
    )
    
    return qa_chain


# 使用
qa_chain = build_conversational_rag(vectorstore, llm)

# 多轮对话
result1 = qa_chain.invoke({"question": "什么是机器学习？"})
print("回答1:", result1["answer"])

result2 = qa_chain.invoke({"question": "它和深度学习有什么关系？"})  # 会理解"它"指机器学习
print("回答2:", result2["answer"])

六、RAG 评估

6.1 评估指标

6.2 使用 RAGAS 评估

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset


def evaluate_rag(questions, ground_truths, rag_system):
    """评估 RAG 系统"""
    
    # 收集预测结果
    answers = []
    contexts = []
    
    for q in questions:
        result = rag_system.query(q)
        answers.append(result["answer"])
        contexts.append(result["sources"])
    
    # 构建数据集
    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truth": ground_truths,
    }
    dataset = Dataset.from_dict(data)
    
    # 评估
    results = evaluate(
        dataset,
        metrics=[
            faithfulness,      # 回答是否忠于上下文
            answer_relevancy,  # 回答是否相关
            context_precision, # 检索精度
            context_recall,    # 检索召回
        ],
    )
    
    return results


# 使用
questions = [
    "什么是机器学习？",
    "深度学习有哪些应用？",
]

ground_truths = [
    "机器学习是人工智能的一个分支...",
    "深度学习广泛应用于图像识别、NLP...",
]

results = evaluate_rag(questions, ground_truths, rag_system)
print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, ...}

6.3 常见问题诊断

七、生产环境最佳实践

7.1 完整 RAG 服务架构

7.2 FastAPI RAG 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import redis
import hashlib
import json

app = FastAPI()

# Redis 缓存
redis_client = redis.Redis(host='localhost', port=6379, db=0)


class QueryRequest(BaseModel):
    question: str
    top_k: int = 3
    use_cache: bool = True


class QueryResponse(BaseModel):
    answer: str
    sources: List[str]
    cached: bool = False


def get_cache_key(question: str) -> str:
    """生成缓存键"""
    return f"rag:{hashlib.md5(question.encode()).hexdigest()}"


@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    """RAG 查询接口"""
    
    # 检查缓存
    if request.use_cache:
        cache_key = get_cache_key(request.question)
        cached = redis_client.get(cache_key)
        if cached:
            data = json.loads(cached)
            return QueryResponse(**data, cached=True)
    
    try:
        # RAG 查询
        result = rag_system.query(
            request.question,
            top_k=request.top_k,
        )
        
        response = QueryResponse(
            answer=result["answer"],
            sources=result["sources"],
        )
        
        # 写入缓存
        if request.use_cache:
            redis_client.setex(
                cache_key,
                3600,  # 1 小时过期
                json.dumps(response.dict()),
            )
        
        return response
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/index")
async def index_documents(documents: List[str]):
    """索引文档接口"""
    try:
        rag_system.add_documents(documents)
        return {"status": "success", "count": len(documents)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy"}

7.3 性能优化清单

# 1. Embedding 缓存
from functools import lru_cache

@lru_cache(maxsize=10000)
def get_embedding_cached(text: str) -> list:
    return embedding_model.encode(text).tolist()


# 2. 批量处理
def batch_embed(texts: List[str], batch_size: int = 32):
    embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        embeddings.extend(embedding_model.encode(batch))
    return embeddings


# 3. 异步检索
import asyncio

async def async_retrieve(queries: List[str]):
    tasks = [retriever.aretrieve(q) for q in queries]
    return await asyncio.gather(*tasks)


# 4. 流式输出
async def stream_answer(question: str):
    context = retriever.retrieve(question)
    
    async for chunk in llm.astream(
        prompt.format(context=context, question=question)
    ):
        yield chunk

八、总结

RAG 核心要点

关键 Takeaway

1. RAG = 检索 + 生成：让大模型拥有外部知识
2. 分块策略很重要：影响检索质量和上下文完整性
3. 混合检索效果更好：向量 + BM25 互补
4. Reranking 提升精度：粗排 + 精排两阶段
5. 评估驱动优化：用 RAGAS 等工具量化效果
6. 生产环境要考虑：缓存、异步、监控

技术选型建议

下一步学习

• [ ] Agent：让大模型自己行动
• [ ] Function Calling：工具调用
• [ ] 多模态 RAG：图片、视频

参考资料

1. RAG 原论文 - Retrieval-Augmented Generation
2. Self-RAG 论文 - Self-Reflective RAG
3. LangChain 文档 - RAG 框架
4. RAGAS - RAG 评估框架
5. BGE Embedding - 中文 Embedding