- 文档加载：多源数据的统一接入
文档加载是 RAG 离线索引的第一步，目标是从各类异构数据源中提取纯文本内容，为后续处理提供标准化的输入。常见的数据源包括本地文件（TXT、CSV、JSON、Markdown、PDF 等）、网页内容、数据库、第三方服务接口等。
在 LangChain 框架中，提供了丰富的 DocumentLoader 类，不同格式的文件经过对应加载器处理后，会统一输出为 Document 对象，该对象包含page_content（正文内容）与metadata（元数据）两个核心属性，方便后续流程统一处理。
实战：多格式文档加载实现
我们以服装产品相关数据为例，实现 TXT、CSV、JSON、Markdown、PDF 五种格式文件的加载。首先安装所需依赖：
pip install pypdf jq
完整加载代码如下：
复制代码
from langchain_community.document_loaders import (
TextLoader, CSVLoader, JSONLoader, PyPDFLoader
)
方式1: 从 TXT 文件加载服装目录
txt_loader = TextLoader(“clothing_data/clothing_catalog.txt”, encoding=“utf-8”)
txt_docs = txt_loader.load()
for i, doc in enumerate(txt_docs[:2]):
print(doc.page_content[:100])
方式2: 从 CSV 文件加载产品信息
csv_loader = CSVLoader(“clothing_data/products.csv”, encoding=“utf-8”)
csv_docs = csv_loader.load()
for i, doc in enumerate(csv_docs[:2]):
print(doc.page_content[:100])
方式3: 从 JSON 文件加载产品详情
json_loader = JSONLoader(
file_path=“clothing_data/product_details.json”,
jq_schema=“.[].description”,
text_content=False
)
json_docs = json_loader.load()
for i, doc in enumerate(json_docs[:2]):
print(doc.page_content[:100])
方式4: 从 Markdown 文件加载尺码指南
md_loader = TextLoader(“clothing_data/size_guide.md”, encoding=“utf-8”)
md_docs = md_loader.load()
for i, doc in enumerate(md_docs[:2]):
print(doc.page_content[:100])
方式5: 从 PDF 文件加载服装产品手册
pdf_loader = PyPDFLoader(“clothing_data/clothing_catalog.pdf”)
pdf_docs = pdf_loader.load()
for i, doc in enumerate(pdf_docs[:2]):
print(doc.page_content[:100])
复制代码
代码中，JSON加载相对特殊，需要通过jq_schema参数指定提取字段的路径，示例中.[].description表示提取JSON数组中每个元素的description字段；text_content=False则表示内容需要按jq规则解析，而非直接读取纯文本。
image
动画视频在：《32. 文档智能分割拆解，长文本轻量化分块》
- 文档智能分割：长文本的轻量化分块
完成原始文档的加载后，长文本并不能直接用于后续的向量嵌入与检索环节。一方面大语言模型的上下文窗口存在上限，无法一次性处理超长文本；另一方面过大的文本块会混入无关内容，稀释语义，降低检索精度；同时向量嵌入模型本身对输入长度也有限制，通常在512到8192个Token之间。因此，我们需要对长文档进行智能分割，将其拆分为合适大小的文本块。
文档分割（Text Splitting）是离线索引的关键环节，直接影响后续检索的精准度。LangChain提供了多种分割策略，适配不同的场景需求：
RecursiveCharacterTextSplitter：按递归字符分隔符切分，优先使用段落、句子等自然边界，是最推荐的通用分割方案；
TokenTextSplitter：严格按照Token数量进行切分；
CharacterTextSplitter：按用户指定的字符分隔符进行切分。
分割过程中有两个核心参数需要关注：chunk_size表示单个文本块的最大长度，chunk_overlap表示相邻文本块之间的重叠长度，用于避免语义在切块边界处断裂，保证上下文的连贯性。
实战：递归字符分割实现
我们使用RecursiveCharacterTextSplitter对加载完成的所有文档进行统一分块，设置块大小为200字符，重叠长度为40字符，并按段落、句号、逗号、空格的优先级选择分隔符。
复制代码
from langchain_community.document_loaders import (
TextLoader, CSVLoader, JSONLoader, PyPDFLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
加载 TXT 文件
txt_loader = TextLoader(“clothing_data/clothing_catalog.txt”, encoding=“utf-8”)
txt_docs = txt_loader.load()
加载 CSV 文件
csv_loader = CSVLoader(“clothing_data/products.csv”, encoding=“utf-8”)
csv_docs = csv_loader.load()
加载 JSON 文件
json_loader = JSONLoader(
file_path=“clothing_data/product_details.json”,
jq_schema=“.[].description”,
text_content=False
)
json_docs = json_loader.load()
加载 Markdown 文件
md_loader = TextLoader(“clothing_data/size_guide.md”, encoding=“utf-8”)
md_docs = md_loader.load()
加载 PDF 文件
pdf_loader = PyPDFLoader(“clothing_data/clothing_catalog.pdf”)
pdf_docs = pdf_loader.load()
合并所有文档
all_docs = txt_docs + csv_docs + json_docs + md_docs + pdf_docs
创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每个块最大200个字符
chunk_overlap=40, # 块之间重叠40个字符
separators=[“\n\n”, “。”, “，”, " ", “”] # 按优先级尝试的分隔符
)
对加载的所有文档进行分割
chunks = text_splitter.split_documents(all_docs)
print(f"原始文档数: {len(all_docs)}“)
print(f"分割后的文本块数: {len(chunks)}”)
for i, chunk in enumerate(chunks[:3]):
print(f"块 {i+1} ({len(chunk.page_content)} 字符): {chunk.page_content[:60]}…“)
print(f"元数据: {chunk.metadata}”)
复制代码
调用split_documents()方法时，会自动保留每个文本块对应的元数据（如来源文件名），方便后续检索时追溯信息来源。
image
动画视频在：《33. 向量嵌入，文本语义的数字化表达》
- 向量嵌入：文本语义的数字化表达
得到尺寸合适的文本块后，我们需要将自然语言文本转化为计算机可以计算的高维向量，这一过程就是向量嵌入（Embedding），它是实现语义检索的基础。
Embedding是将文本转换为固定维度高维向量的过程，其核心特性是：语义越相似的文本，在向量空间中的距离越近。基于这一特性，我们可以通过计算向量间的距离，来衡量两段文本的语义相似度。
向量具备三个核心特性：
维度：向量的维度由嵌入模型决定，常见为768、1536、3072维；
语义相似性：通常通过余弦相似度来衡量两段文本的语义接近程度；
可计算性：向量之间支持距离计算、加减运算等数学操作。
目前主流的嵌入模型包括：OpenAI的text-embedding-3-small、text-embedding-3-large；Cohere的embed-english-v3.0、embed-multilingual-v3.0；以及开源的BGE、GTE、E5系列，可本地部署，无需API调用费用。
实战：开源模型批量嵌入与相似度验证
我们使用开源中文嵌入模型bge-small-zh-v1.5，对分割后的文本块进行批量嵌入，并实现余弦相似度计算与Top N相似文本匹配。首先安装依赖：
pip install sentence-transformers
完整实现代码如下：
复制代码
from langchain_community.document_loaders import (
TextLoader, CSVLoader, JSONLoader, PyPDFLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np
txt_loader = TextLoader(“clothing_data/clothing_catalog.txt”, encoding=“utf-8”)
txt_docs = txt_loader.load()
csv_loader = CSVLoader(“clothing_data/products.csv”, encoding=“utf-8”)
csv_docs = csv_loader.load()
json_loader = JSONLoader(file_path=“clothing_data/product_details.json”, jq_schema=“.[].description”,
text_content=False)
json_docs = json_loader.load()
md_loader = TextLoader(“clothing_data/size_guide.md”, encoding=“utf-8”)
md_docs = md_loader.load()
pdf_loader = PyPDFLoader(“clothing_data/clothing_catalog.pdf”)
pdf_docs = pdf_loader.load()
合并所有文档
all_docs = txt_docs + csv_docs + json_docs + md_docs + pdf_docs
创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每个块最大200个字符
chunk_overlap=40, # 块之间重叠40个字符
separators=[“\n\n”, “。”, “，”, " ", “”] # 按优先级尝试的分隔符
)
对所有文档进行分割
chunks = text_splitter.split_documents(all_docs)
model = SentenceTransformer(‘BAAI/bge-small-zh-v1.5’)
提取所有文本块的内容，过滤掉空块
chunk_texts = [chunk.page_content for chunk in chunks if chunk.page_content.strip()]
批量嵌入所有文本块
vectors = model.encode(chunk_texts)
print(f"文本块数量: {len(chunks)}“)
print(f"向量数量: {len(vectors)}”)
print(f"每个向量的维度: {len(vectors[0])}")
定义余弦相似度计算函数
def cosine_similarity(v1, v2):
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
计算查询文本与所有文本块的相似度，返回 Top N 结果
def get_top_similar(query_text, vectors, chunk_texts, top_n=3):
query_vector = model.encode([query_text])[0]
similarities = []
for i, v in enumerate(vectors):
score = cosine_similarity(query_vector, v)
similarities.append((i, score))
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_n]
验证嵌入效果
for query in [“冬天穿的保暖外套”, “夏天穿的凉快T恤”]:
print(f"\n’{query}’ 与各文本块的相似度 (Top 3):“)
for idx, score in get_top_similar(query, vectors, chunk_texts):
print(f” 相似度 {score:.4f}: {chunk_texts[idx][:60]}")
复制代码
通过测试查询可以直观看到，语义相近的文本会获得更高的相似度分数，验证了嵌入模型的语义表征能力。
image
动画视频在：《34. 向量存储，RAG 架构的核心组件》
- 向量存储：RAG架构的核心组件
当文本数量较少时，我们可以像上面一样直接在内存中计算相似度，但当知识库规模扩大到十万、百万级别时，就需要专门的向量数据库来实现高效的存储与检索。
向量数据库是专门用于存储和高效检索向量数据的系统，是RAG架构中的核心组件。它通过特殊的索引算法，能够在海量向量中快速找到与查询向量最接近的结果，大幅提升检索效率。
目前主流的向量数据库包括：
Chroma：开源轻量级产品，开箱即用，非常适合本地开发和原型验证；
FAISS：Facebook开源的向量检索库，性能极高，但仅提供检索功能，不具备完整的数据库管理能力；
Pinecone：全托管云服务，无需运维，适合生产环境直接使用；
Weaviate：功能丰富的开源向量数据库；
Milvus：分布式向量数据库，适合大规模生产环境部署。
选型向量数据库时，通常需要考虑数据规模、查询性能要求、部署方式（本地/云端托管）、是否支持元数据过滤与混合检索等因素。
实战：Chroma向量数据库集成
我们选择轻量级的Chroma作为向量存储方案，将本地嵌入模型与Chroma进行集成，实现向量的持久化存储。首先安装依赖：
pip install langchain-chroma
由于Chroma默认的嵌入接口与本地SentenceTransformer模型不直接兼容，我们需要自定义一个封装类，实现embed_documents与embed_query两个方法，完整代码如下：
复制代码
from langchain_community.document_loaders import (
TextLoader, CSVLoader, JSONLoader, PyPDFLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np
from langchain_chroma import Chroma
txt_loader = TextLoader(“clothing_data/clothing_catalog.txt”, encoding=“utf-8”)
txt_docs = txt_loader.load()
csv_loader = CSVLoader(“clothing_data/products.csv”, encoding=“utf-8”)
csv_docs = csv_loader.load()
json_loader = JSONLoader(file_path=“clothing_data/product_details.json”, jq_schema=“.[].description”,
text_content=False)
json_docs = json_loader.load()
md_loader = TextLoader(“clothing_data/size_guide.md”, encoding=“utf-8”)
md_docs = md_loader.load()
pdf_loader = PyPDFLoader(“clothing_data/clothing_catalog.pdf”)
pdf_docs = pdf_loader.load()
合并所有文档
all_docs = txt_docs + csv_docs + json_docs + md_docs + pdf_docs
创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每个块最大200个字符
chunk_overlap=40, # 块之间重叠40个字符
separators=[“\n\n”, “。”, “，”, " ", “”] # 按优先级尝试的分隔符
)
对所有文档进行分割
chunks = text_splitter.split_documents(all_docs)
model = SentenceTransformer(‘bge-small-zh-v1.5’)
chunk_texts = [chunk.page_content for chunk in chunks if chunk.page_content.strip()]
vectors = model.encode(chunk_texts)
print(f"文本块数量：{len(chunks)}“)
print(f"向量数量：{len(vectors)}”)
print(f"每个向量的维度：{len(vectors[0])}")
def cosine_similarity(vec1, vec2):
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
def get_top_similar(query_text, vectors, chunk_texts, top_n=3):
query_vector = model.encode([query_text])[0]
similarities = []
for i, v in enumerate(vectors):
score = cosine_similarity(query_vector, v)
similarities.append((i, score))
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_n]
persist_directory = “chroma_clothing_db”
class LocalEmbedding:
def embed_documents(self, texts):
return model.encode(texts).tolist()
def embed_query(self, text): return model.encode([text])[0].tolist()embedding_function = LocalEmbedding()
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embedding_function,
persist_directory=persist_directory
)
print(f"向量库存储条数：{vectorstore._collection.count()}")
复制代码
调用Chroma.from_documents()方法后，系统会自动完成所有文本块的向量化与存储，并将数据持久化到指定的本地目录中，后续可直接加载复用。
image
动画视频在：《35. 向量检索，快速匹配相关文档》
- 向量检索：快速匹配相关文档
完成向量库的构建后，就进入了RAG的在线查询阶段，核心是根据用户的问题，从向量库中快速匹配出最相关的文档，这一步的检索策略直接决定了最终回答的质量。
检索策略决定了如何从向量数据库中找到与用户问题最相关的文档，是RAG流程中的关键环节。我们可以将检索方式分为基础检索与高级检索两大类。
基础检索方式
(1) 相似度检索（Similarity Search）
最基础的检索方式，直接返回与查询向量余弦距离最近的K个文档；
(2) MMR检索（Maximal Marginal Relevance）
在相关性与多样性之间取得平衡，避免返回大量内容重复的文档，适合需要覆盖更多信息维度的场景。
高级检索技术
(1) 多查询检索（Multi-Query）
由大模型从不同角度改写原始问题，生成多个查询向量后合并检索结果，提升召回率；
(2) 父子文档检索（Parent-Child）
检索时匹配细粒度的小块，但返回包含该小块的更大的父文档，保证上下文完整性；
(3) 上下文压缩（Contextual Compression）
对检索到的文档进行片段提取或摘要，只保留与问题最相关的部分，减少无效上下文；
(4) 自查询检索（Self-Query）
由大模型将用户的自然语言问题转换为结构化查询，结合元数据过滤提升精准度；
(5) 混合检索（Ensemble）
结合向量语义检索与传统关键词检索，兼顾语义匹配与关键词匹配。
实战：多种检索方式实现
基于前面构建的Chroma向量库，我们演示相似度检索、MMR检索、元数据过滤检索以及Retriever接口的使用。
复制代码
query = “有没有保暖的冬天外套？”
results = vectorstore.similarity_search(query, k=2)
for i, doc in enumerate(results):
print(f"{i+1}. {doc.page_content[:80]}“)
print(f"元数据: {doc.metadata}”)
results_mmr = vectorstore.max_marginal_relevance_search(
query,
k=2,
fetch_k=10,
lambda_mult=0.7 )
for i, doc in enumerate(results_mmr):
print(f" {i+1}. {doc.page_content[:80]}…")
results_filtered = vectorstore.similarity_search(
“保暖的外套”,
k=2,
filter={“gender”: “女”})
for doc in results_filtered:
print(f"{doc.page_content[:80]}")
retriever = vectorstore.as_retriever(
search_type=“similarity”,
search_kwargs={“k”: 3}
)
print(retriever.invoke(“有没有保暖的冬天外套？”))
复制代码
其中MMR检索的fetch_k参数表示先召回的候选文档数量，lambda_mult用于控制多样性程度，取值为0时追求最大多样性，取值为1时追求最大相关性。通过as_retriever()方法可以将向量库封装为标准的Retriever接口，方便后续与RAG链、Agent等组件集成。
image
- RAG+Agent：让检索更智能
到这里，一套基础的RAG检索流程就已经搭建完成。但基础RAG存在一个明显的问题：无论用户输入什么内容，哪怕是问候、闲聊、数学计算等与知识库无关的问题，系统都会执行一次数据库检索，造成不必要的资源浪费。为了解决这个问题，我们可以将检索能力封装为工具，交给Agent自主调度。
Agent的核心能力是根据用户问题自主思考、判断并选择合适的工具来完成任务。将Retriever包装为Agent的一个工具后，Agent会自行判断：当问题涉及知识库内容时，调用检索工具获取信息；当问题可以直接回答或需要使用其他工具时，则不会触发检索，从而大幅提升资源利用率与交互效率。
在LangChain中，我们可以通过create_retriever_tool函数快速将Retriever包装为标准工具，只需指定工具名称与描述，Agent就会根据描述来判断何时调用该工具。
实战：智能导购Agent实现
我们将检索工具与计算、文件操作等工具整合，搭建一个具备服装导购能力的智能Agent，它可以自主判断是否需要检索产品信息，也能完成数学计算、文件读写等操作。完整代码如下：
复制代码
import os
from deepagents.backends import FilesystemBackend
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, BaseTool
from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool
from langchain_community.document_loaders import (
TextLoader, CSVLoader, JSONLoader, PyPDFLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np
from langchain_chroma import Chroma
from langchain_core.tools import create_retriever_tool
txt_loader = TextLoader(“clothing_data/clothing_catalog.txt”, encoding=“utf-8”)
txt_docs = txt_loader.load()
csv_loader = CSVLoader(“clothing_data/products.csv”, encoding=“utf-8”)
csv_docs = csv_loader.load()
json_loader = JSONLoader(file_path=“clothing_data/product_details.json”, jq_schema=“.[].description”,
text_content=False)
json_docs = json_loader.load()
md_loader = TextLoader(“clothing_data/size_guide.md”, encoding=“utf-8”)
md_docs = md_loader.load()
pdf_loader = PyPDFLoader(“clothing_data/clothing_catalog.pdf”)
pdf_docs = pdf_loader.load()
合并所有文档
all_docs = txt_docs + csv_docs + json_docs + md_docs + pdf_docs
创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每个块最大200个字符
chunk_overlap=40, # 块之间重叠40个字符
separators=[“\n\n”, “。”, “，”, " ", “”] # 按优先级尝试的分隔符
)
对所有文档进行分割
chunks = text_splitter.split_documents(all_docs)
embedding_model = SentenceTransformer(‘bge-small-zh-v1.5’)
chunk_texts = [chunk.page_content for chunk in chunks if chunk.page_content.strip()]
vectors = embedding_model.encode(chunk_texts)
def cosine_similarity(vec1, vec2):
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
def get_top_similar(query_text, vectors, chunk_texts, top_n=3):
query_vector = model.encode([query_text])[0]
similarities = []
for i, v in enumerate(vectors):
score = cosine_similarity(query_vector, v)
similarities.append((i, score))
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_n]
persist_directory = “chroma_clothing_db”
class LocalEmbedding:
def embed_documents(self, texts):
return embedding_model.encode(texts).tolist()
def embed_query(self, text): return embedding_model.encode([text])[0].tolist()embedding_function = LocalEmbedding()
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embedding_function,
persist_directory=persist_directory
)
query = “有没有保暖的冬天外套？”
results = vectorstore.similarity_search(query, k=2)
for i, doc in enumerate(results):
print(f"{i + 1}.{doc.page_content[:80]}“)
print(f"元数据：{doc.metadata}”)
results_mmr = vectorstore.max_marginal_relevance_search(
query,
k=2,
fetch_k=10,
lambda_mult=0.7
)
for i, doc in enumerate(results_mmr):
print(f"{i + 1}.{doc.page_content[:80]}“)
results_filtered = vectorstore.similarity_search(
“保暖的外套”,
k=2,
filter={“gender”: “女”}
)
for doc in results_filtered:
print(f”{doc.page_content[:80]}")
retrieve = vectorstore.as_retriever(
search_type=“similarity”,
search_kwargs={“k”: 3}
)
load_dotenv()
prefix = “QWEN”
model = init_chat_model(
model_provider=“openai”,
configurable_fields=[“model”, “api_key”, “base_url”],
config_prefix=prefix
).with_config({
“configurable”: {
f"{prefix}_model": os.getenv(f"{prefix}_MODEL"),
f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"),
f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL")
}
})
class CalculateTool(BaseTool):
name: str = “calculate”
description: str = “计算数学表达式的值”
def _run(self, expression: str) -> str: try: return f"计算结果: {eval(expression)}" except Exception as e: return f"计算错误: {str(e)}" async def _arun(self, expression: str) -> str: return self._run(expression) # ========== 初始化工具 ==========calculate = CalculateTool()
write_file = WriteFileTool()
read_file = ReadFileTool()
list_dir = ListDirectoryTool()
search_tool = create_retriever_tool(
retriever=retrieve,
name=“search”,
description=“搜索服装产品信息，当用户询问服装相关问题时使用此工具”
)
agent = create_agent(
model=model,
tools=[calculate, write_file, read_file, list_dir,search_tool],
system_prompt=“你是一个服装导购助手，可以搜索产品信息、计算数值、读写文件、列出目录。”,
debug=True
)
queries = [
“你好”,
“有没有保暖的冬天外套？”,
“计算 2024*12+500，然后把结果保存到 result.txt”,
“读取 result.txt 的内容”,
“列出当前目录文件”
]
for q in queries:
print(f"\n问：{q}“)
result = agent.invoke({“messages”: [{“role”: “user”, “content”: q}]})
print(f"答：{result[‘messages’][-1].content}”)
复制代码
通过多组测试问题可以验证，面对普通问候时Agent会直接回复，面对服装相关问题时会自动调用检索工具，面对计算与文件操作需求时则会调用对应工具，实现了能力的智能调度。
总结
