RAG入门：检索增强生成实战

大模型很强但有两个硬伤：知识有截止日期、容易编造事实。RAG（Retrieval-Augmented Generation，检索增强生成）是目前最实用的解决方案——先从知识库里检索相关文档，再让模型基于检索结果生成回答。这篇用LangChain从零搭建一个RAG系统。

RAG的核心思路

RAG的流程分两步：

用户提问 → 检索（Retrieval）→ 找到相关文档片段
                                    ↓
              生成（Generation）← 把文档片段 + 问题一起喂给LLM
                   ↓
              最终回答（基于真实文档，而非模型记忆）

相比直接问LLM，RAG的优势：

知识可更新：换一批文档就行，不需要重新训练模型
减少幻觉：回答有据可查，模型基于提供的文档生成
可溯源：可以告诉用户答案来自哪份文档的哪个段落
成本低：不需要微调模型，开箱即用

第一步：文档切分

文档不能整篇扔给模型（超出上下文长度），需要切成小块。LangChain提供了多种切分器：

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader
)

# 加载文档
loader = DirectoryLoader(
    "./docs",
    glob="**/*.md",
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"}
)
documents = loader.load()
print(f"Loaded {len(documents)} documents")

# 切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大500字符
    chunk_overlap=50,      # 相邻块重叠50字符
    length_function=len,
    separators=["

", "
", "。", "！", "？", ".", " "]
)
chunks = text_splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks")

RecursiveCharacterTextSplitter 是最常用的切分器。它会按照 separators 的优先级来切分——优先在段落之间切，其次句子之间，最后才按字符硬切。chunk_overlap 保证上下文不会在切分处断裂。

chunk_size的选择：太小会丢失上下文，太大会引入噪声。500-1000个字符是常见的范围。中文场景下500字符大约是200-250个汉字，差不多一两个自然段。

第二步：Embedding向量化

把文档块转成向量，才能做语义检索。有两种选择：

方案一：OpenAI Embedding

from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002",
    openai_api_key="sk-xxx"
)

优点是效果好，缺点是要联网且按token计费。

方案二：本地Embedding模型

from langchain.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="shibing624/text2vec-base-chinese",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True}
)

text2vec-base-chinese 是中文场景效果不错的小模型，跑在本地完全免费。对于中文文档我倾向用这个。

第三步：向量存储

把向量存起来，用于后续检索。ChromaDB是轻量级的向量数据库，本地使用很方便：

from langchain.vectorstores import Chroma

# 创建向量存储（首次会比较慢，需要计算所有chunk的embedding）
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",   # 持久化到磁盘
    collection_name="my_docs"
)
vectorstore.persist()

# 后续加载（不用重新计算embedding）
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_docs"
)

检索测试：

# 相似度搜索
results = vectorstore.similarity_search("什么是微服务", k=3)
for doc in results:
    print(f"[{doc.metadata['source']}]")
    print(doc.page_content[:100])
    print("---")

# 带分数的搜索
results = vectorstore.similarity_search_with_score("什么是微服务", k=3)
for doc, score in results:
    print(f"Score: {score:.4f} | {doc.page_content[:80]}")

第四步：LangChain实现RAG

把检索和生成串起来：

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

# LLM
llm = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    temperature=0,
    openai_api_key="sk-xxx"
)

# 自定义Prompt
prompt_template = (
    "基于以下已知信息来回答用户的问题。\n"
    "如果已知信息中没有足够的内容来回答，请直接说\"根据已有资料无法回答该问题\"，不要编造答案。\n\n"
    "已知信息:\n{context}\n\n"
    "用户问题: {question}\n\n"
    "回答:"
)

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 构建RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",    # stuff=把所有检索结果塞进prompt
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 4}   # 检索4个最相关的文档块
    ),
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# 提问
result = qa_chain({"query": "支付系统中如何处理对账差异？"})
print("回答:", result["result"])
print("
来源文档:")
for doc in result["source_documents"]:
    print(f"  - {doc.metadata['source']}")

chain_type 有几种模式：

stuff：最简单，把所有检索结果拼接到prompt里。适合文档块少的场景
map_reduce：对每个文档块分别生成回答，再汇总。适合文档块多的场景
refine：逐步精炼答案。质量好但慢

大多数场景下 stuff 就够了。

效果评估

RAG的效果取决于两个环节：检索质量和生成质量。

检索质量评估：

# 准备测试问题和预期答案来源
test_cases = [
    {"query": "Redis分布式锁怎么实现", "expected_source": "redis-lock.md"},
    {"query": "对账差异如何处理", "expected_source": "reconciliation.md"},
]

hits = 0
for tc in test_cases:
    results = vectorstore.similarity_search(tc["query"], k=3)
    sources = [doc.metadata["source"] for doc in results]
    if tc["expected_source"] in sources:
        hits += 1
recall = hits / len(test_cases)
print(f"Recall@3: {recall:.2%}")

常见问题及调优：

检索不到相关文档 → 调整chunk_size，尝试不同的embedding模型
检索到但排序不对 → 增大k值，或用MMR（最大边际相关性）替代纯相似度搜索
生成答案不准确 → 改prompt模板，强调"基于文档回答"
回答太泛泛 → 减小chunk_size，让检索结果更精确

完整可运行代码

把上面的步骤整合：

#!/usr/bin/env python3
\"\"\"最简RAG示例\"\"\"
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 1. 加载 & 切分
loader = DirectoryLoader("./docs", glob="**/*.md", loader_cls=TextLoader)
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(loader.load())

# 2. Embedding & 存储
embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db")

# 3. RAG Chain
qa = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0),
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

# 4. 使用
result = qa({"query": "你的问题"})
print(result["result"])

RAG是当前LLM应用最实用的模式之一。理解了核心流程之后，可以在这个基础上做很多扩展——多轮对话、混合检索（向量+关键词）、文档权限控制、对话历史引用等。