在构建 RAG(检索增强生成)系统或企业本地知识库时,绝大多数企业数据都以静态文件形式存在——例如 PDF 员工手册、Markdown 售后规则、TXT 产品说明书 等。

如何把这些异构文档高效读取并处理成大模型和向量数据库能理解的格式?核心第一步就是:文档加载(Document Loading)与 文本切分(Text Splitting)

本文将深入解读 LangChain 中的 Document 核心概念,带你用 Python 手把手实现一个多格式文档自动预处理管道。

1. 为什么不能把整份文件直接扔给大模型?

在开始前,很多新手会问:“现在的 LLM 支持十几万甚至上百万 Token 的上下文,为什么不直接把 200 页的 PDF 直接投给模型?”

在工程落地中,这样做存在四大硬伤:

  1. 上下文成本极高:每次提问都附带全量文档,Token 消耗呈指数级上升。

  2. “大海捞针”失准(Lost in the Middle):模型在长文本中间部分的注意力容易衰减,导致回答准确度下降。

  3. 响应延迟变大:过长的 Prompt 会大幅增加推理首字延迟(TTFT)。

  4. 检索精细度低:向量数据库无法针对整本书建立高精度的向量索引,必须“切块”才能实现准确匹配。

因此,典型的知识库处理流程第一阶段如下图所示:

2. 依赖安装与准备

首先安装 LangChain 核心扩展包与 PDF 解析工具(本阶段仅做文档处理,无需消耗 LLM API 额度):

Bash

pip install langchain-community langchain-text-splitters pypdf
  • langchain-community:提供各种文档加载器(Loaders)

  • langchain-text-splitters:提供文本切分器

  • pypdf:用于解析文本型 PDF

3. 解密 LangChain 核心数据结构:Document

在 LangChain 中,任何文本数据进入管道后都会统一封装为 Document 对象。它由两个核心属性构成:

  • page_content:文档正文文本(字符串)。

  • metadata:元数据(字典),记录文档的附加信息(如文件名、来源路径、页码、分类等)。

Python(01-document解析)

from langchain_core.documents import Document

doc = Document(
    page_content="公司员工每月享有 2 次补卡机会,需在考勤系统申请。",
    metadata={
        "source": "data/employee_handbook.txt",
        "category": "考勤制度",
        "author": "HR部门"
    }
)

print(doc)
print(doc.page_content)
print(doc.metadata)

输出结果:

page_content='公司员工每月享有 2 次补卡机会,需在考勤系统申请。' metadata={'source': 'data/employee_handbook.txt', 'category': '考勤制度', 'author': 'HR部门'}
公司员工每月享有 2 次补卡机会,需在考勤系统申请。
{'source': 'data/employee_handbook.txt', 'category': '考勤制度', 'author': 'HR部门'}

进程已结束,退出代码为 0

工程提示:元数据(metadata)在真实系统中极其关键!模型生成回答后,前端展示的**“参考资料来源”、“定位到第 X 页”**全靠元数据实现。

4. 多格式文档加载实战 (Loaders)

LangChain 的 Loader 统一提供 .load() 方法,返回包含一个或多个 Document 的列表。

4.1 加载 TXT 文件

使用 TextLoader 可以轻松读取普通文本与 Markdown 文档。建议显式指定 encoding="utf-8" 避免 Windows 系统出现中文乱码。

Python(02_load_txt)

from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    file_path="data/employee_handbook.txt",
    encoding="utf-8",
)

documents = loader.load()

print(f"文档数量:{len(documents)}")
print(f"文档内容:\n{documents[0].page_content}")
print(f"元数据:{documents[0].metadata}")

其中data/employee_handbook.txt文件是我自己编辑的,文件格式如下图:

然后可以在employee_handbook.txt文件中写以下内容:

员工考勤制度

工作时间为周一至周五,每天 9:00 至 18:00。

员工每月可以申请两次补卡。超过两次后,需要部门负责人审批。

正式员工每年享有 5 天带薪年假。工作满三年后,每年享有 10 天带薪年假。

之后运行代码输出结果:

文档数量:1
文档内容:
员工考勤制度

工作时间为周一至周五,每天 9:00 至 18:00。

员工每月可以申请两次补卡。超过两次后,需要部门负责人审批。

正式员工每年享有 5 天带薪年假。工作满三年后,每年享有 10 天带薪年假。
元数据:{'source': 'data/employee_handbook.txt'}

进程已结束,退出代码为 0

4.2 加载 Markdown 文件

这次我们可以先编写数据文件,文件结构data/refund_policy.md,如图

然后我们可以在refund_policy.md文件中编写以下内容用于测试:

# 退款规则

## 未发货订单

订单未发货时,用户可以直接申请退款。

## 已发货订单

订单已经发货时,需要等待商品送达后申请退货退款。

## 退款到账时间

审核通过后,退款通常在 1 至 3 个工作日内原路返回。

Python(03_load_markdown)

from langchain_community.document_loaders import TextLoader

loader=TextLoader(
	file_path="data/refund_policy.md",
	encoding="utf-8"
)

documents=loader.load()

print("切割成了几个文档:"+str(len(documents)))
# 文档的内容
print(documents[0].page_content)
# 文档的元数据
print(documents[0].metadata)

输出结果:

切割成了几个文档:1
# 退款规则

## 未发货订单

订单未发货时,用户可以直接申请退款。

## 已发货订单

订单已经发货时,需要等待商品送达后申请退货退款。

## 退款到账时间

审核通过后,退款通常在 1 至 3 个工作日内原路返回。
{'source': 'data/refund_policy.md'}

进程已结束,退出代码为 0

4.3 加载 PDF 文件

企业场景中最常用的是 PDF。使用 PyPDFLoader 时,它会按页切分,将 PDF 的每一页加载为一个独立的 Document 对象。

首先我们应该准备一些PDF文件,由于这里我无法上传PDF文件,还请读者自行准备一下。如果想要笔者同样的文件,可以在后台私信一下我,看到就会回复。这是我准备好的文件结构:

Python(04_load_pdf)

from langchain_community.document_loaders import PyPDFLoader

loader=PyPDFLoader(
	file_path="data/XX销售有限公司员工守则.pdf"
)

documents=loader.load()
print("切割成了几个文档:"+str(len(documents)))
# 文档的内容
for document in documents:
	print(document.page_content[-20:])
	# 文档的元数据
	print(document.metadata)
	print("当前页码:",document.metadata['page'])

输出结果:

切割成了几个文档:3
周五可着休闲装,但须保持⼲净、得体。有重
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1'}
当前页码: 0
的违纪⾏为或重复违纪,并记录在个⼈档案。
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2'}
当前页码: 1
⾃发布之⽇起⽣效,旧版相关规定同时废⽌。
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 2, 'page_label': '3'}
当前页码: 2

进程已结束,退出代码为 0

注意事项

  1. PyPDFLoader 提取的 page 元数据通常从 0 开始

  2. 该组件仅支持文本型 PDF。如果是扫描件(图片 PDF),需要接入外部 OCR 工具。

5. 核心原理:RecursiveCharacterTextSplitter 递归切分

把长文档切分成合适的数据块(Chunks)是 RAG 的核心功底。

处理通用文本时,推荐使用 RecursiveCharacterTextSplitter(递归字符文本切分器)。

5.1 工作机制

与硬生生按固定字符数截断不同,递归切分器会优先寻找自然的语义断句标识符,按照以下顺序尝试切分:

这样可以最大程度保障段落和句子的完整性,不会把一句话从中劈成两半。

5.2 核心参数:chunk_sizechunk_overlap

  • chunk_size:每个文档块允许的最大字符数。

  • chunk_overlap:相邻两个文档块之间重复保留的重叠字符数

为什么必须设置重叠(Overlap)?

若不设置重叠,当关键信息正好处于切分边界时(例如:“员工每月最高可报销 [切断] 500 元”),语义就会遭到破坏。保留适当重叠能有效减少上下文丢失。

5.3 切分并保留 Metadata 最佳实践

在实际项目中,我们必须使用 split_documents() 而不是单纯切字符串,这样才能保障原始文档的源路径、页码等元数据无损继承到每一个 Chunk 中。

这里仍旧需要读者像上面一样准备一个PDF文件。

Python(05_split_text)

from idlelib.iomenu import encoding
from turtledemo.penrose import start

from langchain_community.document_loaders import PyPDFLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader=PyPDFLoader(
	file_path="data/XX销售有限公司员工守则.pdf"
)
documents=loader.load()

print("切割后的文档数据量是:",len(documents))

splitter=RecursiveCharacterTextSplitter(
	chunk_size=200,
	chunk_overlap=30,
	add_start_index=True,# 每一个大文档的起始位置,假如一个文件被切割成了多个document,它指的是一个document
	separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""]
)

chunks=splitter.split_documents(documents)
print("再次切割后的文档数量是:",len(chunks))
for index,chunk in enumerate(chunks,start=1):
	print(isinstance(chunk,Document))
	print(f"第{index}文档的内容如下:")
	print(chunk.page_content[:20])
	print(chunk.metadata)

enumerate() 遍历可迭代对象(这里 chunks 是分割好的文本块列表),同时返回【序号 + 元素】

  • 第一个参数:chunks → 需要遍历的列表
  • start=1:序号从 1 开始计数;默认不写 start 时,序号从 0 开始

chunk:列表里每一段文本块

输出结果:

切割后的文档数据量是: 3
再次切割后的文档数量是: 15
True
第1文档的内容如下:
XX销售有限公司员⼯守则  
第⼀章 总
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 0}
True
第2文档的内容如下:
况,特制定本守则。
第2条 适⽤范围
本
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 133}
True
第3文档的内容如下:
第⼆章 员⼯权利与义务  
第4条 员⼯
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 311}
True
第4文档的内容如下:
5. 培训发展权: 员⼯有权获得公司提供
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 508}
True
第5文档的内容如下:
3. 保密义务: 严格保守公司的商业秘密
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 676}
True
第6文档的内容如下:
第三章 ⼯作规范与纪律  
第6条 考勤
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 852}
True
第7文档的内容如下:
4. 请假流程: 所有请假(事假、病假等
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 0, 'page_label': '1', 'start_index': 1014}
True
第8文档的内容如下:
要客户接待或会议时,须按通知要求着装。

{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 0}
True
第9文档的内容如下:
第9条 信息安全
1. 账户安全: 每位
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 184}
True
第10文档的内容如下:
第四章 销售⾏为准则  
第10条 客户
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 338}
True
第11文档的内容如下:
抢客户。
第11条 报价与合同
1. 统
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 484}
True
第12文档的内容如下:
2. 费⽤报销: 业务活动中产⽣的费⽤,
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 666}
True
第13文档的内容如下:
3. 在维护公司利益和声誉⽅⾯有重⼤贡献
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 1, 'page_label': '2', 'start_index': 824}
True
第14文档的内容如下:
3. 最后警告: 情节严重,留司察看。

{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 2, 'page_label': '3', 'start_index': 0}
True
第15文档的内容如下:
第六章 附则  
第15条 守则的修订与
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'data/XX销售有限公司员工守则.pdf', 'total_pages': 3, 'page': 2, 'page_label': '3', 'start_index': 172}

进程已结束,退出代码为 0

6. 实战工程:企业多类型知识库文档预处理流水线

下面我们编写一个支持自动化扫描文件夹、智能识别文件类型、提取元数据并自动化切块的完整脚本。

项目目录结构

Plaintext,学习阶段可以不用在乎这个结构

knowledge_base_project/
├── knowledge_base/
│   ├── employee_handbook.txt
│   ├── refund_policy.md
│   └── product_manual.pdf
└── document_processor.py

完整代码实现:document_processor.py

Python

7. 切分参数设置避坑指南 (FAQ)

在真实业务落地中,切分参数(chunk_size / chunk_overlap)没有放之四海而皆准的固定数值,建议根据场景调优:

业务场景建议 chunk_size建议 chunk_overlap优化调优思路
简短 FAQ / 问答对100 - 250 字符10 - 20 字符尽量让单个问答在同一个 Chunk 内保持完整
公司规章制度 / 合同300 - 500 字符50 - 80 字符按条款/段落切分,保留适度重叠以防语境割裂
技术文档 / API 手册500 - 800 字符80 - 100 字符块可适当加大,避免函数说明与代码示例被切开

生产环境常见坑点:

  1. chunk_size 是字符数还是 Token 数?

    RecursiveCharacterTextSplitter 默认基于 Python 的 len() 计算字符长度。如果需要精准按 Token 截断,可后续替换使用 from_tiktoken_encoder

  2. chunk_overlap 是否设置越大越好?

    不是。重叠面积过大会导致向量数据库产生大量的重复冗余数据,并且最终召回多个包含高度重复内容的 Chunk,挤占模型有限的上下文窗口。

  3. 加载大型文档卡顿如何处理?

    在处理海量文档(几万张 PDF)时,避免使用一次性加载内存的 .load(),应改用 .lazy_load() 迭代生成器模式,控制内存占用。

8. 总结

本文完成了 RAG 系统预处理的第一阶段:

Plaintext

原始文档 (TXT/MD/PDF) 
  ↳ Document Loaders (解析文件)
  ↳ Standard Documents (统一结构与元数据)
  ↳ Text Splitters (语义递归切分)
  ↳ Clean Document Chunks (待向量化的文档块)

完整代码实现:document_processor.py

Python

from idlelib.iomenu import encoding
from pathlib import Path

from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter


def split_documents(ducoments:list[Document])->list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=200,
        chunk_overlap=30,
        add_start_index=True,  # 每一个大文档的起始位置,假如一个文件被切割成了多个document,它指的是一个document
        separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""]
    )

    chunks = splitter.split_documents(ducoments)
    for index,chunk in enumerate(chunks,start=1):
        chunk.metadata['chunk_id']=index
    return chunks

def load_knowledge_base(dictory)->list[Document]:
    all_documents=[]
    for f in dictory.rglob("*"):
        if f.is_dir():
            continue
        file_suffix=f.suffix
        file_path=dictory.name+"/"+f.name
        if file_suffix in ['.txt','.md']:
            loader=TextLoader(
                file_path=file_path,
                encoding="utf-8"
            )

            docs=loader.load()

            for document in docs:
                document.metadata["file_name"] = file_path
                document.metadata["file_type"] = file_suffix

            all_documents.extend(docs)
        elif file_suffix=='.pdf':
            loader=PyPDFLoader(
                file_path=file_path
            )

            docs = loader.load()

            for document in docs:
                document.metadata["file_name"] = file_path
                document.metadata["file_type"] = file_suffix

            all_documents.extend(docs)




    return all_documents

def main():
    path=Path("data")
    # 先将文档切分
    documents=load_knowledge_base(path)
    print("文档切分的个数:",len(documents))
    #再将文档继续切分
    chunks=split_documents(documents)
    for chunk in chunks:
        print("~"*30)
        metadata=chunk.metadata
        print(f"当前是第{metadata['chunk_id']}片段,内容是{chunk.page_content[:20]}")
        print("当前的片段来自于"+chunk.metadata.get("file_name"))


if __name__ == '__main__':
    main()

Path("knowledge_base")

使用 Python 内置 pathlib 创建路径对象,代表当前程序目录下名叫 knowledge_base 的文件夹。

等价路径字符串:"./knowledge_base"

directory.rglob("*")

  • rglob = recursive glob:递归遍历
  • glob():只遍历当前文件夹(不进子文件夹)
  • rglob():递归遍历所有子文件夹

通配符说明:

"*" → 匹配所有文件和文件夹

"*.txt" → 递归查找所有后缀 txt 文件

输出结果:

已加载:employee_handbook.txt,生成 1 个原始文档
已加载:product_manual.pdf,生成 6 个原始文档
已加载:refund_policy.md,生成 1 个原始文档

===== 处理结果 =====
原始文档数量:8
文档块数量:16
文件数量:3

===== 文档块预览 =====
------------------------------------------------------------
chunk_id:0
文件:employee_handbook.txt
页码:无
起始位置:0
内容:员工考勤制度……

下一阶段,我们将把预处理好的 Document Chunks 通过 Embedding 模型转化为高维向量,并存储到 向量数据库(Vector DB) 中,敬请关注!

更多推荐