RAG管线四件套：从原始文档到可检索知识

一、概念：什么是 RAG 管线四件套？

把原始文档变成可检索知识，需要经过四步：

原始文档 ──①──→ 标准文档 ──②──→ 文档片段 ──③──→ 嵌入向量 ──④──→ 向量索引 Loader Splitter Embedding VectorStore

每一步对应一个组件：

#	组件	职责	输入	输出
①	Document Loader	从数据源加载文档	文件路径 / URL	list[Document]
②	Text Splitter	把长文档切成小片段	list[Document]	list[Document] （更短）
③	Embedding Model	把文本转成向量	str	list[float]
④	Vector Store	存储向量并支持搜索	文档 + 向量	相似文档列表

四步串起来就是 RAG 的离线索引管线。

一句话总结：四件套是"原始文档 → 可检索知识"的 ETL 管线——加载、切分、嵌入、存储，每一步都有标准抽象。

---

二、价值：为什么需要标准化管线？

1. 数据源千差万别

PDF、Word、Markdown、HTML、Notion、Slack、数据库……格式各异。Document Loader 把所有来源统一成Document(page_content, metadata)。

2. 长文档必须切分

一个 42K 字符的 PDF 超出多数模型的上下文窗口。即使塞进去，模型也会"注意力涣散"。切分让每个片段独立可检索、大小可控。

3. 语义搜索需要向量

关键词搜索无法理解同义词。"退货"和"退换"意思相近但词不同。嵌入向量把语义映射到空间距离，近义词自然靠近。

4. 向量库是检索的基石

有了向量索引，才能毫秒级返回与查询最相似的文档片段。不同向量库（Chroma、FAISS、Pinecone）各有适用场景，但接口统一。

核心价值：四件套把"数据接入"变成标准化 ETL 管线，让你专注业务而非数据格式。

---

三、用法：四件套逐一实战

① Document Loader：加载数据

# PDF from langchain_community.document_loaders import PyPDFLoader docs = PyPDFLoader("report.pdf").load() # Markdown from langchain_community.document_loaders import UnstructuredMarkdownLoader docs = UnstructuredMarkdownLoader("guide.md").load() # 网页 from langchain_community.document_loaders import WebBaseLoader docs = WebBaseLoader("https://example.com/faq").load() # 纯文本 from langchain_community.document_loaders import TextLoader docs = TextLoader("notes.txt", encoding="utf-8").load() # CSV（每行一个文档） from langchain_community.document_loaders import CSVLoader docs = CSVLoader("data.csv").load()

每个 Loader 返回list[Document]，Document 有两个核心字段：

• page_content: str — 文本内容
• metadata: dict — 元数据（来源文件名、页码、URL 等）

② Text Splitter：切分文档

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个片段最大 1000 字符 chunk_overlap=200, # 相邻片段重叠 200 字符 separators=["\n\n", "\n", "。", "！", "？", " ", ""], # 中文友好 add_start_index=True, # 记录每个片段在原文的起始位置 ) chunks = splitter.split_documents(docs) print(f"原文档: {len(docs)} 个, 切分后: {len(chunks)} 个片段")

关键参数：

• chunk_size：片段上限。太大→噪声多，太小→上下文断裂。经验值 500-1500
• chunk_overlap：重叠区。保证片段边界不丢信息。经验值 chunk_size 的 10-20%
• separators：切分优先级。RecursiveCharacterTextSplitter依次尝试每个分隔符，直到片段足够小

其他 Splitter：

Splitter	适用场景
RecursiveCharacterTextSplitter	通用文本（推荐默认）
MarkdownHeaderTextSplitter	按标题层级切分 Markdown
PythonCodeTextSplitter	按函数/类切分 Python 代码
TokenTextSplitter	按 token 数切分

③ Embedding Model：文本转向量

from langchain_huggingface import HuggingFaceEmbeddings # 推荐：多语言 + 中文优化 embeddings = HuggingFaceEmbeddings( model="BAAI/bge-m3", ) # 嵌入单条文本 vector = embeddings.embed_query("退货政策") print(f"维度: {len(vector)}") # 通常 768 或 1024 # 批量嵌入 vectors = embeddings.embed_documents(["退货政策", "换货流程"])

选型指南：

模型	维度	特点	安装
BAAI/bge-m3	1024	多语言，中文优秀	langchain-huggingface
BAAI/bge-small-zh-v1.5	512	中文专用，轻量	langchain-huggingface
text-embedding-3-small	1536	OpenAI，需联网	langchain-openai

国内用户推荐bge-m3：中文效果好、离线可用、免费。

④ Vector Store：存储 + 搜索

from langchain_chroma import Chroma # 创建 + 索引（一次性） vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, collection_name="my_docs", persist_directory="./chroma_db", # 持久化到磁盘 ) # 后续加载（不用重新索引） vectorstore = Chroma( collection_name="my_docs", embedding_function=embeddings, persist_directory="./chroma_db", ) # 相似度搜索 docs = vectorstore.similarity_search("退货政策", k=3) # 带分数 docs_with_scores = vectorstore.similarity_search_with_score("退货政策", k=3) for doc, score in docs_with_scores: print(f"分数: {score:.4f} | {doc.page_content[:80]}")

向量库选型：

向量库	特点	适用场景
Chroma	嵌入式，零配置，支持持久化	开发/小规模部署
FAISS	Facebook 开源，纯内存，极快	大规模内存检索
Pinecone	全托管云服务	生产环境、免运维
Milvus	开源分布式，支持 GPU	超大规模生产
Qdrant	Rust 实现，性能好	中大规模生产

完整管线：从文件到可检索 Agent

from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_huggingface import HuggingFaceEmbeddings from langchain_chroma import Chroma from langchain.tools.retriever import create_retriever_tool from langchain.agents import create_agent # ① 加载 docs = PyPDFLoader("company_policy.pdf").load() # ② 切分 chunks = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ).split_documents(docs) # ③ 嵌入 + ④ 存储 embeddings = HuggingFaceEmbeddings(model="BAAI/bge-m3") vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db") # 接入 Agent retriever_tool = create_retriever_tool( vectorstore.as_retriever(search_kwargs={"k": 3}), name="search_docs", description="搜索公司政策文档", ) agent = create_agent("deepseek:deepseek-chat", tools=[retriever_tool]) result = agent.invoke({ "messages": [{"role": "user", "content": "公司年假多少天？"}] })

---

四、原理：管线各步的内部机制

1. Document Loader 的统一接口

所有 Loader 继承BaseLoader，实现load()→list[Document]：

class BaseLoader: def load(self) -> list[Document]: ... def lazy_load(self) -> Iterator[Document]: ... # 懒加载，省内存

大文件用lazy_load()避免一次性加载到内存。

2. RecursiveCharacterTextSplitter 的递归策略

原文: "第一章\n\n第一节\n\n内容A...\n\n第二节\n\n内容B..." ↓ 尝试用 "\n\n" 切分 ["第一章", "第一节\n\n内容A...", "第二节\n\n内容B..."] ↓ 某段超过 chunk_size? 继续用 "\n" 切 ↓ 还超过? 用 "。" 切 ↓ 还超过? 用 " " 切 ↓ 还超过? 按字符强制切

递归保证优先在语义边界切分，只在必要时才暴力切断。

3. Embedding 的向量空间

嵌入模型把文本映射到高维空间（768/1024 维），语义相近的文本在空间中距离近：

"退货政策" ──→ [0.12, -0.34, 0.56, ...] "退换货流程" ──→ [0.11, -0.33, 0.55, ...] ← 距离很近 "天气预报" ──→ [-0.45, 0.78, -0.23, ...] ← 距离很远

相似度用余弦距离或欧氏距离计算。搜索时就是"找查询向量最近的 K 个文档向量"。

4. Vector Store 的索引结构

向量库用近似最近邻（ANN）算法加速搜索：

算法	原理	特点
HNSW	层级导航小世界图	速度快、精度高，内存占用大
IVF	倒排索引 + 聚类	可控精度，适合大规模
PQ	乘积量化压缩	省内存，精度略降

Chroma 默认用 HNSW，FAISS 支持 IVF + PQ，Pinecone 用自研索引。

5. 管线性能优化

瓶颈	优化方向
嵌入计算慢	批量嵌入 + GPU 加速
搜索不够准	调整 chunk_size / 换嵌入模型 / 加元数据过滤
结果太相似	用 MMR 搜索 / 调大 fetch_k
中文效果差	换中文优化的嵌入模型（bge-m3）
索引更新频繁	增量 add_documents()，不全量重建

---

小结

视角	一句话
概念	"原始文档→可检索知识"的 ETL 管线：加载→切分→嵌入→存储
价值	标准化数据接入，让你专注业务而非格式
用法	Loader 统一来源、Splitter 控制粒度、Embedding 选中文模型、VectorStore 选合适引擎
原理	递归切分保语义边界、向量空间做语义搜索、ANN 算法加速检索

记住一件事：四件套中切分（Splitter）对最终检索质量影响最大——切得不好，后面嵌入再好、向量库再快也白搭。