LangChain 文档加载与文本切分:从零搭建企业知识库预处理流水线
手把手教你读取 PDF、TXT、Markdown,并用智能切分让 AI 精准检索
一、前言:为什么需要文档加载和切分?
前几章我们处理的数据都来自代码里的字符串。但真实的企业知识库——产品说明书、员工手册、售后规则、技术文档——都静静地躺在各类文件里。要让大模型“读懂”这些资料,第一步就是把这些文件加载成程序可处理的结构化数据。
然而,一个文件动辄几十页甚至上百页,直接喂给模型会遇到三大拦路虎:
上下文长度限制:模型的输入窗口有限,放不下整本书
成本高昂:每次请求都带全量文档,token 消耗飙升
检索精度低:用户只问一个小问题,却要扫描整本书,既慢又容易漏掉关键信息
<div align="center"> <img src="https://img-blog.csdnimg.cn/direct/placeholder-doc-pipeline.png" alt="文档处理流程概览" width="700"/> <br/> <em>图:文档预处理 → 切分 → 向量化 → 检索回答的完整链条</em> </div>
本章聚焦在“文档加载”与“文本切分”这两个前置环节,为后续的向量检索和 RAG 打下坚实基础。
二、核心概念:Document 对象
LangChain 用统一的Document类来表示一段文档,它有两个核心属性:
| 属性 | 类型 | 说明 |
|---|---|---|
page_content | str | 文档的正文内容 |
metadata | dict | 附加信息,如文件路径、页码、分类等 |
python
# 01_document_basic.py from langchain_core.documents import Document doc = Document( page_content="这是员工手册的正文内容……", metadata={ "source": "data/employee_handbook.txt", "file_type": "txt", "page": 1 } ) print(doc.page_content) print(doc.metadata)元数据为什么重要?当模型根据某个文档块生成答案后,我们可以通过元数据告诉用户“这个答案来自《员工手册》第 3 页”,提升可信度。
三、加载各类文件
3.1 安装依赖
在项目环境中安装所需包(国内可换清华源):
bash
pip install langchain-community langchain-text-splitters pypdf
| 依赖 | 作用 |
|---|---|
langchain-community | 提供文档加载器 |
langchain-text-splitters | 提供文本切分器 |
pypdf | 读取 PDF 文件 |
3.2 加载 TXT 文件
使用TextLoader,它会自动读取文本文件并包装成Document。
准备测试文件data/employee_handbook.txt:
text
员工考勤制度 1. 工作时间:上午 9:00 - 下午 18:00,午休 1 小时。 2. 迟到早退:每月累计迟到超过 3 次,扣除绩效奖金。 3. 请假流程:需提前一天在 OA 系统提交申请。
加载代码02_load_text.py:
python
from langchain_community.document_loaders import TextLoader loader = TextLoader("data/employee_handbook.txt", encoding="utf-8") docs = loader.load() # 返回 List[Document] for doc in docs: print(doc.page_content) print(doc.metadata)⚠️ 注意:即使单个文件,
load()依然返回列表,方便统一处理。
3.3 加载 Markdown 文件
Markdown 本质也是文本,直接用TextLoader即可。
准备文件data/refund_policy.md:
markdown
# 售后退款政策 ## 适用条件 - 商品未拆封,7 天内可无理由退货。 - 质量问题,15 天内可换货或退款。 ## 退款流程 1. 联系客服提交申请。 2. 寄回商品(运费由我方承担)。 3. 审核通过后 3 个工作日内原路退款。
加载代码03_load_markdown.py:
python
loader = TextLoader("data/refund_policy.md", encoding="utf-8") docs = loader.load() print(docs[0].page_content)3.4 加载 PDF 文件(文本型)
对于文字型 PDF(非扫描版),使用PyPDFLoader,它会按页拆分文档,每页生成一个Document。
准备文件:将任意文本型 PDF 放到data/product_manual.pdf。
加载代码04_load_pdf.py:
python
from langchain_community.document_loaders import PyPDFLoader loader = PyPDFLoader("data/product_manual.pdf") docs = loader.load() for doc in docs: print(f"页码: {doc.metadata['page']}") print(doc.page_content[:100]) # 只打印每页前100字符 print("-" * 30)每个Document的元数据包含source(文件路径)和page(页码,从 0 开始)。
📌局限性:扫描版 PDF 需要 OCR(如
pypdfocr或unstructured),复杂表格可能需要专用解析器,本章不展开。
四、为什么需要切分长文档?
假设员工手册有 200 页,直接喂给模型会带来:
超长上下文:超过模型窗口限制
高成本:每次请求都传 200 页内容
低召回:用户问“考勤制度”,你却要扫描整本手册,检索效率低下
所以必须把长文档切成语义独立的文档块(chunks):
<div align="center"> <img src="https://img-blog.csdnimg.cn/direct/placeholder-split-concept.png" alt="长文档切分示意图" width="600"/> <br/> <em>图:长文档 → 分割成若干语义块 → 每个块可独立检索</em> </div>
五、RecursiveCharacterTextSplitter:智能切分器
LangChain 推荐使用RecursiveCharacterTextSplitter,它会按一组分隔符递归切分,优先保留段落和句子的完整性。
5.1 核心参数
| 参数 | 作用 |
|---|---|
chunk_size | 每个文档块的最大字符数(不是 token) |
chunk_overlap | 相邻块之间重叠的字符数,用于保留上下文 |
separators | 分隔符列表,默认["\n\n", "\n", " ", ""] |
5.2 理解chunk_size与chunk_overlap
举个例子,文档内容为:
text
A B C D E F G H I J K L M N O P Q R S T U V W X Y Z
chunk_size=10,chunk_overlap=2
切分结果:
text
块1: A B C D E F G H I J 块2: I J K L M N O P Q R S (重叠 I J) 块3: R S T U V W X Y Z (重叠 R S)
重叠部分可以让被切断的句子在相邻块中都能保留完整语义,避免“话说到一半”导致信息丢失。
5.3 切分普通文本(示例)
05_split_text.py:
python
from langchain_text_splitters import RecursiveCharacterTextSplitter text = """ 员工考勤制度 1. 工作时间:上午 9:00 - 下午 18:00,午休 1 小时。 2. 迟到早退:每月累计迟到超过 3 次,扣除绩效奖金。 3. 请假流程:需提前一天在 OA 系统提交申请。 员工福利 1. 五险一金:按国家规定缴纳。 2. 年度体检:每年 10 月统一安排。 3. 团建活动:每季度一次。 """ splitter = RecursiveCharacterTextSplitter( chunk_size=50, chunk_overlap=10, separators=["\n\n", "\n", "。", ";", ",", " ", ""] ) chunks = splitter.split_text(text) for i, chunk in enumerate(chunks, start=1): print(f"块 {i}:\n{chunk}\n")5.4 切分 Document 并保留元数据
真实项目中,我们更常用split_documents(),它不仅切分文本,还会保留原 Document 的元数据,并自动添加_start_index记录每个块在原文中的起始位置。
06_split_documents.py:
python
from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter loader = TextLoader("data/employee_handbook.txt", encoding="utf-8") docs = loader.load() splitter = RecursiveCharacterTextSplitter( chunk_size=100, chunk_overlap=20 ) chunks = splitter.split_documents(docs) for i, chunk in enumerate(chunks, start=1): print(f"块 {i} (来源: {chunk.metadata['source']})") print(chunk.page_content) print("-" * 40)输出中会看到每个块的元数据都包含了原始文件路径,以及_start_index等额外信息。
六、企业级案例:知识库文档预处理
6.1 需求描述
公司内部知识库目录包含三种文件:
data/employee_handbook.txt(员工制度)
data/refund_policy.md(售后规则)
data/product_manual.pdf(产品手册)
需要编写一个预处理程序,完成:
递归扫描目录,按扩展名加载文件
为每个文档块添加
filename、file_type等元数据用
RecursiveCharacterTextSplitter切分所有文档输出文档块总数及前 3 块预览
6.2 完整代码document_processor.py
python
from pathlib import Path from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter def load_documents_from_directory(directory: str): """递归加载目录下所有 .txt, .md, .pdf 文件""" base_path = Path(directory) all_docs = [] # 递归遍历所有文件 for file_path in base_path.rglob("*"): if not file_path.is_file(): continue suffix = file_path.suffix.lower() try: if suffix in [".txt", ".md"]: loader = TextLoader(str(file_path), encoding="utf-8") elif suffix == ".pdf": loader = PyPDFLoader(str(file_path)) else: continue # 跳过其他格式 docs = loader.load() # 追加文件级别的元数据 for doc in docs: doc.metadata["filename"] = file_path.name doc.metadata["file_type"] = suffix[1:] # 去掉点号 doc.metadata["source_path"] = str(file_path) all_docs.extend(docs) except Exception as e: print(f"加载 {file_path} 失败: {e}") return all_docs def split_documents(docs, chunk_size=300, chunk_overlap=50): """使用递归字符切分器拆分文档列表""" splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", ";", ",", " ", ""] ) return splitter.split_documents(docs) if __name__ == "__main__": # 1. 加载所有文档 raw_docs = load_documents_from_directory("knowledge_base") print(f"共加载 {len(raw_docs)} 个原始文档段(如 PDF 按页拆分)") # 2. 切分 chunks = split_documents(raw_docs) print(f"切分后共有 {len(chunks)} 个文档块") # 3. 预览前3块 print("\n--- 文档块预览 ---") for i, chunk in enumerate(chunks[:3], start=1): print(f"块 {i}") print(f"来源文件: {chunk.metadata.get('filename')}") print(f"内容预览: {chunk.page_content[:80]}...") print("-" * 40)6.3 运行效果
text
共加载 5 个原始文档段(如 PDF 按页拆分) 切分后共有 23 个文档块 --- 文档块预览 --- 块 1 来源文件: employee_handbook.txt 内容预览: 员工考勤制度 1. 工作时间:上午 9:00 - 下午 18:00,午休 1 小时。 2. 迟到早退:每月累计迟到超过 3 次... ---------------------------------------- 块 2 来源文件: employee_handbook.txt 内容预览: 2. 迟到早退:每月累计迟到超过 3 次,扣除绩效奖金。 3. 请假流程:需提前一天在 OA 系统提交申请。 ...
七、如何选择合适的切分参数?
没有放之四海而皆准的参数,需要根据文档类型和检索效果调整。
| 文档类型 | 建议策略 |
|---|---|
| 短 FAQ | chunk_size=200~300,尽量让每个问答独立 |
| 规章制度 | 按段落切分,overlap可稍大(50~100) |
| 产品手册 | 可稍大500~800,避免操作步骤被拆散 |
| 技术文档 | 保留代码块和注释的完整性,适当增大分隔符列表 |
验证方法:切分后人工抽查几个块,看语义是否完整;再结合后续检索测试,观察召回准确率。
八、常见问题
Q:chunk_size是字符数还是 token 数?
A:RecursiveCharacterTextSplitter默认按字符数(len()),不是 token。若需按 token 切分,可使用TokenTextSplitter(需安装 tiktoken)。
Q:chunk_overlap越大越好吗?
A: 不是。适当重叠(10%~20%)能保留上下文,但过大会增加重复存储和检索噪音。
Q: PDF 提取出来是空白或乱码怎么办?
A: 很可能是扫描版 PDF(图片)。需要 OCR 工具(如pytesseract)或使用UnstructuredPDFLoader等更强大的加载器。
Q: 为什么不直接整篇文档交给模型?
A: 成本、窗口限制、检索精度三座大山。切分是 RAG 系统的标准前置步骤。
九、本章总结
| 知识点 | 要点 |
|---|---|
Document | page_content+metadata |
| 加载文件 | TextLoader(TXT/MD),PyPDFLoader(PDF) |
| 切分必要性 | 解决超长文本、成本、检索精度问题 |
| 推荐切分器 | RecursiveCharacterTextSplitter |
| 关键参数 | chunk_size和chunk_overlap |
| 保留元数据 | 用split_documents()而不是split_text() |
| 完整流程 | 文件 → Loader → Document → Splitter → 文档块(供后续向量化) |
下一章我们将把这些切分好的文档块转化为向量,并存入向量数据库,正式构建 RAG 检索系统,敬请期待!🚀