🔍 RAG 检索增强生成

🤔 一、为什么需要 RAG？

大语言模型（LLM）能力强大，但存在三个根本性缺陷，严重制约了其在生产环境中的应用。RAG（Retrieval-Augmented Generation，检索增强生成）正是为解决这些问题而生。

微调 vs RAG vs 检索：三种方案对比

RAG 的核心思想

⚙️ 二、RAG 的完整流程

一个完整的 RAG 系统分为两个阶段：离线索引阶段（文档→分块→Embedding→存储）和在线查询阶段（用户查询→检索→生成）。下面逐一拆解。

# LangChain 文档加载示例
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    WebBaseLoader,
    Docx2txtLoader,
)

# 加载 PDF
pdf_loader = PyPDFLoader("company_policy.pdf")
pdf_docs = pdf_loader.load()  # 每页一个 Document

# 加载网页
web_loader = WebBaseLoader("https://docs.example.com/api")
web_docs = web_loader.load()

# 加载 Markdown
md_loader = TextLoader("README.md", encoding="utf-8")
md_docs = md_loader.load()

print(f"共加载 {len(pdf_docs)} 个文档块")
# 每个 Document: page_content + metadataPython

# LangChain 文本分块
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 推荐配置：递归分块，保持语义完整
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # 每块最大 1000 字符
    chunk_overlap=200,     # 重叠 200 字符（防止语义断裂）
    length_function=len,
    separators=["\n\n", "\n", ".", " ", ""],  # 优先按段落切分
)

chunks = text_splitter.split_documents(pdf_docs)
print(f"共分出 {len(chunks)} 个文本块")
print("平均块长度:", sum(len(c.page_content) for c in chunks) / len(chunks))Python

# Embedding 示例
from langchain_openai import OpenAIEmbeddings

# 使用 OpenAI 最新的 text-embedding-3-small
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536  # 可降至 512/1024 以节省存储
)

# 将文本向量化
vector = embeddings.embed_query("什么是机器学习？")
print(f"向量维度: {len(vector)}")  # 输出: 1536

# 批量向量化
vectors = embeddings.embed_documents([
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络处理数据",
    "今天的天气真不错",
])

# 语义相似的文本，余弦相似度更高
import numpy as np
sim = np.dot(vectors[0], vectors[1]) / (
    np.linalg.norm(vectors[0]) * np.linalg.norm(vectors[1])
)
print(f"相似度: {sim:.4f}")  # 接近 0.85（高相似度）Python

# 使用 Chroma（最简单的方案）
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建向量数据库（自动 Embedding + 存储）
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化到本地
)

# 查询：返回最相关的 3 个文档块
results = vectorstore.similarity_search(
    "公司的年假制度是什么？",
    k=3
)

for i, doc in enumerate(results):
    print(f"\n--- 结果 {i+1} ---")
    print(doc.page_content[:200])  # 打印前 200 字Python

# 多路召回 + Reranking 示例
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 1️⃣ 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 2️⃣ BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 10

# 3️⃣ 多路召回（合并去重）
ensemble = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 向量搜索权重更高
)

# 4️⃣ Cross-Encoder 重排序
reranker = CrossEncoderReranker(
    model=HuggingFaceCrossEncoder("BAAI/bge-reranker-large"),
    top_k=5  # 最终只取 top 5
)

# 5️⃣ 组合：多路召回 → 重排序
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=ensemble
)

results = compression_retriever.invoke("公司年假政策？")Python

# LangChain RAG 生成链
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

# 定义 Prompt 模板
system_prompt = """你是一个专业的知识库助手。请仅基于以下提供的上下文来回答用户问题。

规则：
1. 优先使用上下文中的信息回答
2. 如果上下文中没有相关信息，明确说"根据现有知识库，我无法回答这个问题"
3. 回答时引用来源（标注文档标题和章节）
4. 用清晰、专业的语言回答

上下文：
{context}"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
])

# 创建 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 构建 RAG 链：检索 → 文档注入 → 生成
question_answer_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, question_answer_chain)

# 执行查询
response = rag_chain.invoke({"input": "公司年假制度是什么？"})
print(response["answer"])Python

🏗️ 三、RAG 架构图

下图展示了完整的 RAG 系统架构，分为离线索引阶段（上方）和在线查询阶段（下方）。

🚀 四、Advanced RAG 技巧

基础的 RAG 系统已经能解决很多问题，但面对复杂场景，还需要一些进阶技巧来提升质量。

# Parent-Child 分块策略
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 父块分块器（大块，提供完整上下文）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

# 子块分块器（小块，用于精准检索）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# 父块存储（普通内存存储即可）
store = InMemoryStore()

# 创建 Parent-Child 检索器
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

retriever.add_documents(docs)

# 检索：匹配子块，但返回的是父块（更完整的上下文）
results = retriever.invoke("年假政策")
print(len(results[0].page_content))  # 比子块大很多Python

# HyDE 示例
from langchain_community.llms import HuggingFaceHub
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Step 1: 让 LLM 生成假设性文档
hyde_prompt = ChatPromptTemplate.from_template("""请为以下问题写一个详细的假设性回答，
即使你不确定答案也没关系，只需要写一个看起来合理的段落：
问题：{question}
假设性回答：""")

hyde_chain = hyde_prompt | llm | StrOutputParser()

# Step 2: 用假设性回答进行检索
question = "Transformer 的自注意力机制如何工作？"
hypothetical_doc = hyde_chain.invoke({"question": question})

# Step 3: 用假设性文档的 Embedding 进行向量检索
results = vectorstore.similarity_search(hypothetical_doc, k=3)Python

# Agentic RAG 概念示例（使用 LangGraph 实现）
from langgraph.prebuilt import create_react_agent

# 定义 Agent 可用的工具
tools = [
    search_hr_docs_tool,      # 搜索 HR 知识库
    search_tech_docs_tool,    # 搜索技术文档库
    search_product_docs_tool, # 搜索产品文档库
]

# 创建 Agentic RAG
agent = create_react_agent(llm, tools)

# Agent 会自主决定：需要检索吗？检索哪个库？
result = agent.invoke({
    "messages": ["公司新员工的社保缴纳流程是什么？"]
})

# Agent 的决策过程：
# 1. 思考：这是一个 HR 相关问题 → 选择 search_hr_docs_tool
# 2. 执行检索："新员工社保缴纳流程"
# 3. 判断检索结果是否充分
# 4. 如果充分 → 基于检索结果生成回答
# 5. 如果不充分 → 换个关键词再次检索Python

📊 五、RAG 的评估

RAG 系统是一个流水线，每个环节都可能出现问题。只有通过系统化的评估，才能找到瓶颈并持续优化。

RAGAS 框架

RAGAS（RAG Assessment）是专门用于评估 RAG 系统的开源框架^[3]，它从三个维度评估系统质量：

# RAGAS 评估示例
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)

# 准备评估数据
eval_data = {
    "question": ["公司年假政策？", "如何申请调休？"],
    "answer": [answer_1, answer_2],         # RAG 系统生成的回答
    "contexts": [[ctx_1a, ctx_1b], [ctx_2a]], # 检索到的上下文
    "ground_truth": [gt_1, gt_2],           # 标准答案（可选）
}

# 执行评估
results = evaluate(
    eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.92  ✅ 高忠实度
# answer_relevancy: 0.88  ✅ 回答相关
# context_precision: 0.75  ⚠️ 检索精度可优化Python

LangSmith 评估

LangSmith 是 LangChain 官方的可观测性平台，提供 RAG 系统的端到端追踪和评估能力。

💻 六、Python 完整代码示例

下面使用 LangChain + Chroma 实现一个完整的 RAG 系统，涵盖从文档加载到生成回答的全流程。

pip install langchain langchain-openai langchain-chroma chromadb pypdf

"""
完整的 RAG 系统 — LangChain + Chroma
功能：文档加载 → 分块 → Embedding → 存储 → 检索 → 生成
"""

import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, TextLoader
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_core.messages import HumanMessage, AIMessage

# ═══════════════════════════════════════════════
# 1. 配置
# ═══════════════════════════════════════════════
os.environ["OPENAI_API_KEY"] = "your-api-key"

CHROMA_DIR = "./chroma_rag_db"       # 向量数据库持久化目录
CHUNK_SIZE = 1000                     # 分块大小
CHUNK_OVERLAP = 200                  # 分块重叠
TOP_K = 5                            # 检索返回的文档数量

# ═══════════════════════════════════════════════
# 2. 初始化 Embedding 和 LLM
# ═══════════════════════════════════════════════
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536,
)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ═══════════════════════════════════════════════
# 3. 文档加载与分块
# ═══════════════════════════════════════════════
def load_and_split(file_paths):
    """加载多个文件并分块"""
    all_docs = []

    for path in file_paths:
        if path.endswith(".pdf"):
            loader = PyPDFLoader(path)
        else:
            loader = TextLoader(path, encoding="utf-8")

        docs = loader.load()
        # 给每个文档块添加来源元数据
        for doc in docs:
            doc.metadata["source"] = path

        all_docs.extend(docs)

    # 递归分块：优先按段落 → 换行 → 句子 → 单词
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", ".", " ", ""],
    )

    chunks = splitter.split_documents(all_docs)
    print(f"✅ 加载了 {len(file_paths)} 个文件，"
               "生成 {len(chunks)} 个文本块")
    return chunks

# ═══════════════════════════════════════════════
# 4. 构建向量数据库
# ═══════════════════════════════════════════════
def build_vectorstore(chunks):
    """将文本块 Embedding 并存入 Chroma"""
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CHROMA_DIR,
    )
    print(f"✅ 向量数据库已构建，共 {len(chunks)} 条记录")
    return vectorstore

# ═══════════════════════════════════════════════
# 5. 构建 RAG 链
# ═══════════════════════════════════════════════
def create_rag_chain(vectorstore):
    """创建检索增强生成链"""
    # 定义检索器
    retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": TOP_K},
    )

    # 定义 Prompt 模板
    system_prompt = """你是一个专业的知识库问答助手。

请基于以下提供的【上下文】来回答用户的问题。

规则：
1. 仅使用上下文中的信息，不要编造答案
2. 如果上下文中没有相关信息，请明确说明
3. 在回答中引用信息来源
4. 使用清晰、结构化的语言

【上下文】
{context}"""

    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "{input}"),
    ])

    # 构建 RAG 链
    qa_chain = create_stuff_documents_chain(llm, prompt)
    rag_chain = create_retrieval_chain(retriever, qa_chain)

    return rag_chain

# ═══════════════════════════════════════════════
# 6. 交互式问答
# ═══════════════════════════════════════════════
def chat(rag_chain):
    """交互式 RAG 问答"""
    print("\n🔍 RAG 问答系统已就绪！（输入 'quit' 退出）\n")
    while True:
        question = input("👤 你: ").strip()
        if question.lower() in ("quit", "exit", "q"):
            break
        if not question:
            continue

        response = rag_chain.invoke({"input": question})
        print(f"\n🤖 助手: {response['answer']}")

        # 显示来源
        if response.get("context"):
            sources = set()
            for doc in response["context"]:
                sources.add(doc.metadata.get("source", "unknown"))
            print(f"📚 来源: {', '.join(sources)}")
        print()

# ═══════════════════════════════════════════════
# 主程序
# ═══════════════════════════════════════════════
if __name__ == "__main__":
    # Step 1: 加载文档
    files = ["company_handbook.pdf", "hr_policy.md", "tech_docs.txt"]
    chunks = load_and_split(files)

    # Step 2: 构建向量数据库
    vectorstore = build_vectorstore(chunks)

    # Step 3: 创建 RAG 链
    rag_chain = create_rag_chain(vectorstore)

    # Step 4: 开始交互
    chat(rag_chain)Python · complete_rag.py

🏢 七、案例：企业知识库问答系统

场景描述

某中型科技公司（500+ 员工）面临知识管理困境：

• 问题：HR 政策、技术文档、产品规范散落在 Confluence、Google Docs、本地文件中
• 痛点：新员工入职时不知道在哪找信息，反复询问同事；老员工也常记不清政策细节
• 目标：构建一个统一的 AI 问答系统，员工可以直接提问并获得准确答案

实现步骤

效果对比