LangChain RAG 检索增强生成

RAG 解决的问题非常具体：LLM 不知道你公司内部的文档、你上传的 PDF、你的私有数据库里的内容——它的知识来自训练数据，截止到某个时间点。RAG 让 LLM 在回答前先去"查资料"，然后基于查到的内容作答。

我调试过很多 RAG 系统，其中一个教训让我印象深刻：花了两周优化 Prompt、调整 temperature、换更好的模型，系统还是经常给出错误答案。后来发现根本原因是分块太粗糙，检索出来的内容跟问题根本不相关。95% 的 RAG 问题出在检索环节，不在生成环节。所以这页会重点讲检索质量，不是 Prompt 怎么写。

---

#工作原理

阶段一：建库（一次性）

文档 → 分块 → 向量化 → 存入向量数据库
                ↑
         Embedding 模型把文字变成数字向量

阶段二：查询（每次提问时）

用户问题 → 向量化 → 在数据库里找最相似的块 → 塞进 Prompt → LLM 回答

关键理解："向量相似"等于"语义相近"——不是关键词匹配，是找意思最接近的内容。这是 RAG 比普通搜索聪明的地方。

---

#最小可用示例

5 步搭一个能跑的 RAG：

python复制
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from dotenv import load_dotenv

load_dotenv()

# 1. 加载文档
loader = TextLoader("./knowledge.txt", encoding="utf-8")
documents = loader.load()

# 2. 分块
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(documents)

# 3. 向量化 + 存入 Chroma
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 5. RAG Chain
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。如果上下文中没有相关信息，就说"我在文档中没有找到相关内容"，不要编造。

上下文：
{context}

问题：{question}

回答：
""")

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

answer = rag_chain.invoke("这个项目的主要功能是什么？")
print(answer)

---

#文档加载：各种格式

python复制
# 文本
from langchain_community.document_loaders import TextLoader
loader = TextLoader("./doc.txt", encoding="utf-8")

# PDF（常用，装 pypdf 包）
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("./doc.pdf")

# Word 文档（装 python-docx）
from langchain_community.document_loaders import Docx2txtLoader
loader = Docx2txtLoader("./doc.docx")

# 网页
from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://example.com/article")

# CSV
from langchain_community.document_loaders.csv_loader import CSVLoader
loader = CSVLoader("./data.csv")

# 批量加载整个目录
from langchain_community.document_loaders import DirectoryLoader
loader = DirectoryLoader("./docs/", glob="**/*.md", show_progress=True)

# 加载
documents = loader.load()

---

#分块：这里最影响 RAG 质量

分块策略是 RAG 里最被低估的环节，也是出问题最多的地方。

块太大：一个块里混了好几个不同主题的内容，检索出来的"相关块"会引入很多无关信息，导致 LLM 回答时被干扰。

块太小：一个完整的概念被切断了，检索出来的块缺少上下文，LLM 没法给出完整答案。

经验值参考：

python复制
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,      # 相邻块之间保留一些重叠，保证切断处的上下文
    separators=["\n\n", "\n", " ", ""]  # 按段落 → 换行 → 空格的顺序切
)

chunks = splitter.split_documents(documents)
print(f"共 {len(chunks)} 个块")

代码文档用专用 splitter：

python复制
from langchain_text_splitters import RecursiveCharacterTextSplitter, Language

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=2000,
    chunk_overlap=200,
)
# 按 Python 语法（类、函数）切，不会把一个函数从中间截断

---

#向量数据库选型

我的建议：开发用 Chroma，零配置启动，非常适合验证想法。上生产时再评估——如果数据量不大（几万个块以内），Chroma 持久化版本完全够用，不一定要花钱买 Pinecone。

python复制
# Chroma：本地持久化
from langchain_community.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db",  # 持久化路径
)

# 下次重启加载已有数据
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

python复制
# FAISS：追求速度
from langchain_community.vectorstores import FAISS

vectorstore = FAISS.from_documents(chunks, OpenAIEmbeddings())
vectorstore.save_local("./faiss_index")

# 加载
vectorstore = FAISS.load_local("./faiss_index", OpenAIEmbeddings())

---

#检索策略

默认的相似度检索够用，但有两个场景值得了解：

MMR（Maximum Marginal Relevance）：如果检索出来的 5 个块都在说同一件事，它们提供的信息是重叠的。MMR 在保证相关性的同时增加多样性——不让检索结果都是"差不多的内容"：

python复制
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,         # 最终返回 5 个
        "fetch_k": 20,  # 先取 20 个候选
        "lambda_mult": 0.7  # 越高越看重相关性，越低越看重多样性
    }
)

相似度阈值过滤：检索到的内容相关性低于阈值时直接丢掉，不传给 LLM：

python复制
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 5}
)

这个设置可以减少 LLM 拿到无关内容后编造答案的概率。实际项目里我几乎都开着这个过滤。

---

#在 Prompt 里显示来源

让 RAG 能告诉用户"这个答案来自哪个文档"：

python复制
def format_docs_with_source(docs):
    """格式化文档时包含来源信息"""
    return "\n\n---\n\n".join([
        f"[来源: {doc.metadata.get('source', '未知')}]\n{doc.page_content}"
        for doc in docs
    ])

prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。回答时注明信息来源。

上下文：
{context}

问题：{question}
""")

# 同时返回检索到的文档和最终答案
from langchain_core.runnables import RunnableParallel

rag_chain_with_source = RunnableParallel(
    answer=(
        {"context": retriever | format_docs_with_source, "question": RunnablePassthrough()}
        | prompt
        | ChatOpenAI(model="gpt-4o-mini", temperature=0)
        | StrOutputParser()
    ),
    sources=(retriever),  # 同时返回检索到的文档对象
)

result = rag_chain_with_source.invoke("这个项目的主要功能是什么？")
print(result["answer"])
print("\n来源文件:")
for doc in result["sources"]:
    print(f"  - {doc.metadata.get('source')}")

---

#RAG 常见失败原因

用 RAG 不等于 AI 就不会说错话了。以下是最常见的失败场景：

检索质量差：找不到正确的内容，或者找到了一堆无关内容。这是 RAG 最常见的失败点，95% 的 RAG 问题都在检索环节，不在生成环节。排查方法：把 retriever.invoke(问题) 单独跑一遍，看返回的内容是否和问题相关。这一步我每次都做，不确认检索质量不往下走。

分块切坏了：一个答案被切到了两个块里，检索时只找到了一半。换分块策略，加大 chunk_overlap。

问题和文档语言不匹配：用户用中文问，文档是英文，Embedding 跨语言相似度会变差。要么在检索前把问题翻译成文档语言，要么用多语言 Embedding 模型（如 text-embedding-3-large）。

Prompt 没有"不知道就说不知道"：LLM 在没有相关上下文时会编造。System Prompt 里必须明确写"如果上下文没有相关信息，直接说不知道，不要编造"。

---

#动手练习

把你自己的一份 PDF 或 Markdown 文档跑成一个问答系统：

python复制
# TODO 1：加载你的文档（PDF 或 .txt 或 .md）

# TODO 2：调整分块参数（从 chunk_size=500, chunk_overlap=50 开始）

# TODO 3：创建向量数据库并存储

# TODO 4：测试检索效果（retriever.invoke("你的问题") 看返回什么）

# TODO 5：组装完整的 RAG Chain 并运行

# 进阶：尝试把检索策略从 similarity 换成 mmr，对比效果差异

---

#小结

1. RAG 的核心是"先检索后生成"——LLM 基于检索到的文档内容作答，而不是靠记忆。
2. 分块质量决定 RAG 质量——90% 的 RAG 失败是检索问题，不是模型问题。先把 retriever.invoke() 单独跑通，确认能找到正确内容。
3. 开发用 Chroma（零配置），生产再评估是否需要 Pinecone 等托管服务。
4. MMR 检索比默认相似度搜索更实用——避免返回一堆内容重叠的块。
5. Prompt 里必须明确写"上下文没有相关内容时说不知道"——否则 LLM 会在检索失败时编造答案，RAG 反而让错误更严重。

---

下一步：Agents 代理系统 — 让 AI 不只是"查文档回答"，而是自主调用工具完成任务

向量数据库参考：Chroma 文档↗ | FAISS↗ | Pinecone↗