LangChain RAG 系统实战——文档加载、向量化、检索增强生成完整方案
大约 10 分钟
LangChain RAG 系统实战——文档加载、向量化、检索增强生成完整方案
适读人群：想构建企业知识库的工程师、RAG 系统入门者 | 阅读时长：约20分钟 | 核心价值：从零搭建一套生产可用的 RAG 系统，彻底解决"AI 一本正经说废话"的问题
小李是个做了六年的 Java 后端，公司让他负责一个内部客服机器人项目。需求很简单：把公司的产品手册、FAQ、操作文档全部喂给 AI，让 AI 代替客服回答用户问题。
他兴冲冲地把所有文档扔给 ChatGPT，让它直接回答。起初效果不错，但很快发现问题：文档一多，ChatGPT 的 context window 装不下；更麻烦的是，AI 开始"创造"答案，把产品里没有的功能描述得绘声绘色，客服组每天要处理大量用户投诉"AI 说有这个功能但实际没有"。
他来找我的时候很委屈："老张，我都把文档给它了，为什么还是瞎说？"
我说：你的问题是没用 RAG。直接把文档塞进 prompt 和做 RAG 系统，是两件完全不同的事情。RAG 的核心不是让模型记住文档，而是在回答每个问题之前，先精准检索出最相关的片段，只让模型看这几段话。
这样做的好处：不超 context，回答有依据，幻觉大幅减少。
今天我们把 RAG 的每个环节都讲透，给你一套真实可用的方案。
一、RAG 的完整流程
RAG（Retrieval-Augmented Generation）分两个阶段：
离线构建阶段（Indexing）： 文档加载 → 文本分割 → 向量化 → 存入向量数据库
在线查询阶段（Retrieval + Generation）： 用户提问 → 问题向量化 → 向量相似度检索 → 找到相关片段 → 组合成 Prompt → LLM 生成回答
每个环节的质量都直接影响最终效果，我们逐一拆解。
二、文档加载——让各种格式的文档变成文本
2.1 LangChain 内置的文档加载器
import os
from langchain_community.document_loaders import (
PyPDFLoader,
Docx2txtLoader,
UnstructuredMarkdownLoader,
CSVLoader,
WebBaseLoader,
DirectoryLoader
)
def load_documents_from_directory(docs_dir: str) -> list:
"""
从目录中加载多种格式的文档
支持 PDF、Word、Markdown、CSV
"""
all_docs = []
# 加载 PDF
pdf_loader = DirectoryLoader(
docs_dir,
glob="**/*.pdf",
loader_cls=PyPDFLoader,
show_progress=True
)
all_docs.extend(pdf_loader.load())
# 加载 Word 文档
docx_loader = DirectoryLoader(
docs_dir,
glob="**/*.docx",
loader_cls=Docx2txtLoader
)
all_docs.extend(docx_loader.load())
# 加载 Markdown
md_loader = DirectoryLoader(
docs_dir,
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader
)
all_docs.extend(md_loader.load())
print(f"共加载 {len(all_docs)} 个文档片段")
return all_docs
# 也可以直接加载网页
def load_from_urls(urls: list) -> list:
loader = WebBaseLoader(urls)
docs = loader.load()
# 清洗网页内容，去掉导航栏等干扰
for doc in docs:
# 简单的文本清洗
doc.page_content = " ".join(doc.page_content.split())
return docs
# 使用示例
docs = load_documents_from_directory("/data/company_docs")
print(f"第一个文档内容预览：{docs[0].page_content[:200]}")
print(f"文档来源：{docs[0].metadata}")三、文本分割——最被忽视的环节
文本分割的质量是整个 RAG 系统效果好坏的最关键因素，没有之一。切得太短，每个片段缺少上下文；切得太长，检索到的片段含有大量无关信息。
3.1 递归字符分割器（推荐首选）
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(docs: list) -> list:
"""
使用递归字符分割器处理文档
按照 paragraph -> sentence -> word 的优先级分割
"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每个块的目标字符数（中文约250字）
chunk_overlap=100, # 相邻块之间的重叠字符数
length_function=len,
separators=[
"\n\n", # 先按段落分
"\n", # 再按换行分
"。", # 再按句号分
"！",
"？",
"；",
" ", # 最后按空格分
"" # 兜底按字符分
],
is_separator_regex=False
)
split_docs = splitter.split_documents(docs)
# 过滤太短的片段（通常是页眉页脚、目录等）
filtered_docs = [
doc for doc in split_docs
if len(doc.page_content.strip()) > 50
]
print(f"分割后共 {len(filtered_docs)} 个文本块")
print(f"平均块长度：{sum(len(d.page_content) for d in filtered_docs) / len(filtered_docs):.0f} 字符")
return filtered_docs
chunks = split_documents(docs)3.2 踩坑实录一：chunk_overlap 设置不当导致关键信息断裂
现象：明明文档里有"退款流程：第一步提交申请，第二步等待审核（3-5个工作日），第三步到账"，但 AI 回答退款时间时说"不清楚具体时间"。
原因：chunk 在"第二步等待审核"后被截断，"3-5个工作日"落在了下一个 chunk 里，但下一个 chunk 没有被检索到。
解法：
# 对于包含步骤、流程的文档，加大 chunk_overlap
splitter = RecursiveCharacterTextSplitter(
chunk_size=600,
chunk_overlap=200, # overlap 设为 chunk_size 的 1/3
)
# 对于 Markdown 文档，用专门的 MarkdownHeaderTextSplitter
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
# 按标题分割能保留更完整的语义单元
md_splits = md_splitter.split_text(markdown_content)四、向量化与向量数据库
4.1 选择 Embedding 模型
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
# 方案一：OpenAI Embeddings（效果好，有成本）
openai_embeddings = OpenAIEmbeddings(
model="text-embedding-3-small", # 新模型，性价比高
openai_api_key=os.environ.get("OPENAI_API_KEY")
)
# 成本：约 $0.02/1M tokens，处理 100 万汉字约 $0.03
# 方案二：本地开源模型（免费，适合隐私敏感场景）
local_embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-large-zh-v1.5", # 中文效果最好的开源模型之一
model_kwargs={"device": "cpu"}, # Mac 上用 cpu，有 GPU 用 cuda
encode_kwargs={"normalize_embeddings": True}
)4.2 构建 Chroma 向量数据库
from langchain_community.vectorstores import Chroma
import os
def build_vectorstore(chunks: list, embeddings, persist_dir: str = "./chroma_db") -> Chroma:
"""
构建并持久化向量数据库
如果已存在则直接加载，避免重复计算
"""
if os.path.exists(persist_dir) and os.listdir(persist_dir):
print("从磁盘加载已有向量数据库...")
vectorstore = Chroma(
persist_directory=persist_dir,
embedding_function=embeddings
)
print(f"加载完成，共 {vectorstore._collection.count()} 个向量")
else:
print("开始构建向量数据库（首次运行，需要一些时间）...")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=persist_dir,
collection_metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
print(f"构建完成，共 {vectorstore._collection.count()} 个向量")
return vectorstore
# 构建向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = build_vectorstore(chunks, embeddings)
# 测试检索
test_query = "退款需要多少个工作日？"
retrieved_docs = vectorstore.similarity_search(test_query, k=4)
for i, doc in enumerate(retrieved_docs):
print(f"\n--- 检索结果 {i+1} ---")
print(f"来源：{doc.metadata.get('source', 'unknown')}")
print(f"内容：{doc.page_content[:200]}...")五、检索策略——提升召回质量
默认的相似度检索有时候效果不够好，有几种增强策略：
5.1 MMR 检索（最大边际相关性）
# MMR 在保证相关性的同时，提高结果多样性
# 避免返回5个意思相近的片段，而是覆盖多个角度
retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={
"k": 5, # 返回5个结果
"fetch_k": 20, # 先召回20个候选，再从中选5个最多样的
"lambda_mult": 0.5 # 0=最多样化，1=最相关，0.5是平衡点
}
)5.2 混合检索（BM25 + 向量）
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
# BM25：关键词精确匹配
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5
# 向量检索：语义相似
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 混合：兼顾关键词匹配和语义理解
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6] # 向量检索权重稍高
)
# 测试
results = ensemble_retriever.get_relevant_documents("如何申请产品退款")
print(f"混合检索返回 {len(results)} 个结果")六、完整的 RAG Chain 实现
6.1 标准 RAG Chain
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 精心设计的 RAG Prompt
RAG_PROMPT = ChatPromptTemplate.from_template("""
你是一个专业的客服助手，只根据提供的参考资料回答问题。
参考资料：
{context}
用户问题：{question}
回答要求：
1. 只基于参考资料中的信息回答，不要添加资料中没有的内容
2. 如果参考资料中没有相关信息，直接说"根据现有资料，我暂时没有这方面的信息，建议联系人工客服"
3. 回答要简洁，关键信息要突出
回答：
""")
def format_docs(docs: list) -> str:
"""将检索到的文档格式化为字符串"""
formatted = []
for i, doc in enumerate(docs):
source = doc.metadata.get("source", "未知来源")
formatted.append(f"[资料{i+1}] 来源：{source}\n{doc.page_content}")
return "\n\n".join(formatted)
# 使用 LCEL 构建 RAG Chain
rag_chain = (
{
"context": ensemble_retriever | format_docs,
"question": RunnablePassthrough()
}
| RAG_PROMPT
| llm
| StrOutputParser()
)
# 带来源引用的版本
from langchain.schema.runnable import RunnableParallel
rag_chain_with_source = RunnableParallel(
{
"answer": rag_chain,
"source_documents": ensemble_retriever
}
)
# 测试
result = rag_chain_with_source.invoke("退款申请在哪里提交？需要多少时间？")
print("回答：", result["answer"])
print("\n引用来源：")
for doc in result["source_documents"]:
print(f" - {doc.metadata.get('source', 'unknown')}: {doc.page_content[:100]}...")6.2 带对话历史的 RAG（Conversational RAG）
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(
k=5,
memory_key="chat_history",
return_messages=True,
output_key="answer"
)
# 对话式 RAG：会考虑历史对话来改写查询
conv_rag_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=ensemble_retriever,
memory=memory,
return_source_documents=True,
verbose=False,
# 改写问题的提示词（把"它"、"这个"等指代词展开）
condense_question_prompt=ChatPromptTemplate.from_template(
"根据以下对话历史，将用户的最新问题改写为完整、独立的问题（不含指代词）：\n\n"
"对话历史：{chat_history}\n"
"最新问题：{question}\n"
"改写后的问题："
)
)
# 多轮对话测试
questions = [
"你们产品支持退款吗？",
"一般需要多久？", # 这里"需要多久"依赖上文，需要改写查询
"退款到哪个账户？"
]
for q in questions:
result = conv_rag_chain({"question": q})
print(f"\nQ: {q}")
print(f"A: {result['answer']}")七、踩坑实录二：向量检索"相关但答非所问"
现象：用户问"如何重置密码"，检索到的是"密码强度要求"相关的片段，AI 回答了密码复杂度规则，而不是重置流程。
原因：向量相似度基于语义，"密码"这个词让两类文档的向量距离很近，但用户的真实意图（操作步骤）没有被正确捕获。
解法：
# 方案一：Query 改写——让模型先理解用户意图，再生成更精准的检索词
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
query_rewrite_prompt = PromptTemplate.from_template(
"请将以下用户问题改写为3个不同角度的检索查询，用于在文档库中检索相关信息：\n"
"原始问题：{question}\n"
"改写后的查询（每行一个）："
)
rewrite_chain = query_rewrite_prompt | ChatOpenAI(temperature=0) | StrOutputParser()
def multi_query_retrieval(question: str, retriever) -> list:
"""多查询检索：用多个角度搜索，合并去重"""
rewritten = rewrite_chain.invoke({"question": question})
queries = [question] + [q.strip() for q in rewritten.split("\n") if q.strip()]
all_docs = []
seen_content = set()
for query in queries:
docs = retriever.get_relevant_documents(query)
for doc in docs:
content_hash = hash(doc.page_content[:100])
if content_hash not in seen_content:
all_docs.append(doc)
seen_content.add(content_hash)
return all_docs[:6] # 最多返回6个不重复片段八、踩坑实录三：文档更新后忘记重建索引
现象：文档更新了价格，但用户问到还是得到旧价格，客户投诉。
原因：向量数据库是静态的，文档更新后必须重新向量化并更新到数据库。
解法：给文档加上版本戳，实现增量更新：
import hashlib
from datetime import datetime
def compute_doc_hash(content: str) -> str:
"""计算文档内容的哈希值"""
return hashlib.md5(content.encode()).hexdigest()
def incremental_update_vectorstore(
new_docs: list,
vectorstore: Chroma,
hash_store_path: str = "./doc_hashes.json"
) -> int:
"""
增量更新向量数据库
只处理内容有变化的文档
返回更新的文档数量
"""
import json
# 加载已存储的哈希值
if os.path.exists(hash_store_path):
with open(hash_store_path, "r") as f:
stored_hashes = json.load(f)
else:
stored_hashes = {}
updated_count = 0
new_hashes = {}
docs_to_add = []
for doc in new_docs:
source = doc.metadata.get("source", "unknown")
current_hash = compute_doc_hash(doc.page_content)
new_hashes[source] = current_hash
if stored_hashes.get(source) != current_hash:
# 文档有更新，先删除旧版本
vectorstore._collection.delete(where={"source": source})
docs_to_add.append(doc)
updated_count += 1
# 批量添加更新的文档
if docs_to_add:
vectorstore.add_documents(docs_to_add)
# 保存新的哈希值
stored_hashes.update(new_hashes)
with open(hash_store_path, "w") as f:
json.dump(stored_hashes, f)
return updated_count九、效果评估与监控
上线前一定要有评估，不然不知道效果好不好：
from langchain.evaluation import load_evaluator
# 使用 LangChain 内置评估器
faithfulness_evaluator = load_evaluator(
"labeled_criteria",
llm=ChatOpenAI(model="gpt-4o-mini"),
criteria="faithfulness"
)
def evaluate_rag_answer(question: str, answer: str, context: str) -> dict:
"""评估 RAG 回答的忠实度（是否只基于检索到的内容）"""
result = faithfulness_evaluator.evaluate_strings(
input=question,
prediction=answer,
reference=context
)
return result
# 实际测量数据（我在某客服项目上的真实数据）：
# - text-embedding-3-small 向量化速度：约 500 tokens/s
# - 检索延迟：平均 80ms（Chroma 本地，5万文档块）
# - 端到端延迟：gpt-3.5-turbo 约 2.5s，gpt-4o-mini 约 3.2s
# - 幻觉率：纯 LLM 约 23%，加 RAG 后降至 4%
# - 成本（万次查询）：embedding $0.8 + LLM $12 = $12.8十、生产部署建议
- 向量数据库选型：开发用 Chroma（本地），生产用 Qdrant 或 Milvus（支持高并发）
- Embedding 模型：中文场景首选
BAAI/bge-large-zh-v1.5（开源免费）或text-embedding-3-small（效果稍好但有成本） - chunk 策略：先用 500/100（chunk_size/overlap）跑基线，再根据实际召回效果调整
- 监控：记录每次查询的检索分数，低于阈值（如 0.7）时触发人工审核队列