lang

1. 本章目标 #

第 11 章解决的是「这场对话说过什么」；本章转向另一个问题：「公司的资料里写了什么」。第三个问题——「这个用户是谁、有什么偏好」——放在第 16 章长期记忆；它依赖本阶段的向量检索，所以排在 RAG 之后。

这两类信息常被混在一起。看一个具体场景：

用户问：「我买的这台空气净化器，滤芯多久换一次？」

Agent 需要三种来源完全不同的信息：

需要什么	从哪来	属于
用户上一轮说的是哪台机器	会话历史（`messages`）	短期记忆，第 11 章
这个用户家里装的是哪个型号	用户档案（`store`）	长期记忆，第 16 章（本阶段最后一章）
滤芯更换周期写在产品手册第 12 页	公司的 PDF 手册	知识，本章起

这类信息既不在模型训练数据里，也不该塞进系统提示——几百页手册塞不进去，成本也极高。正确做法是把资料加载 → 切分 → 向量化 → 检索，让模型每次只读相关几段。这条链路就是 RAG（Retrieval-Augmented Generation，检索增强生成），也是第 12～15 章的主线：

第 12 章 Load     把 PDF / Word / 网页读成统一的 Document   ← 本章
   │
   ▼
第 13 章 Split    切成大小合适、语义完整的 chunk
   │
   ▼
第 14 章 Embed    转成向量存进向量库
   │
   ▼
第 15 章 Retrieve 按问题检索相关片段，交给 Agent 回答

本章只做第一步，但这一步的质量决定了后面三步的天花板——加载时丢掉的内容，检索时找不回来。

本章目标：

用统一的 Document Loader 接口，把各种格式的企业资料读成带来源信息的 Document 列表，并做加载质量自检。

学完你应能：

解释 Document 的 page_content 与 metadata 分别装什么，为什么 metadata 决定了 RAG 能不能「说出处」
用 TextLoader / PyPDFLoader / Docx2txtLoader / CSVLoader 加载常见格式
处理 Windows 上最常见的中文编码报错
用 DirectoryLoader 批量加载目录，并知道它的局限
区分 load() 与 lazy_load()，知道大批量文档该用哪个
给 Document 补齐统一的 metadata（来源、页码、doc_id、更新时间）
识别加载环节的三类事故：空文档、乱码、扫描件无文字层
产出企业文档摄入脚本，输出可交给第 13 章的 jsonl

加载这一层的麻烦，几乎都属于「不报错，但数据已经错了」。下面是本章实测、和直觉不一样的结论，先扫一眼：

你可能以为	实际情况	见
`Document` 只有 `page_content` 和 `metadata`	还有 `id` 和 `type` 两个字段，`id` 在第 14 章决定能不能增量更新	§3.1
不传 `encoding` 会用 UTF-8	中文 Windows 上用的是 GBK（cp936），读 UTF-8 中文文件直接报错	§4.2
`autodetect_encoding=True` 能救编码问题	它需要 `chardet`，而装了新版 chardet 反而会崩（`TypeError`）	§4.2
`DirectoryLoader` 的 `glob` 只能给一种模式	`glob` 可以传列表，一次匹配多种扩展名	§6.1
扫描件 PDF 加载会失败	不失败，返回一堆 `page_content=''` 的 Document，静默通过	§5.1
练习里的 `JSONLoader` 装上就能用	它依赖 `jq`，这个包在 Windows 上通常装不上	§12

参考文档：

2. 为什么需要 Document Loaders #

2.1. 「用 Python 直接读文件」不行吗 #

对 .txt 来说确实可以：

# 纯 Python 读文本：能拿到内容，但仅此而已
# 注意这里必须显式写 encoding，否则中文 Windows 会用 GBK 去读（§4.2）
text = open("faq.txt", encoding="utf-8").read()

但真实企业资料长这样：制度是 PDF，合同是 Word，客户名单是 Excel/CSV，产品说明在官网，历史工单在数据库。格式各异，解析方式也不同，最后会堆出一堆互不兼容的函数。

更麻烦的是，光有文本还不够用。等到第 15 章 RAG 上线，产品经理一定会提这个要求：

「回答里要标明这句话是从哪份文件、第几页来的，不然客服不敢用。」

到这一步你会发现：读文件时只留下了字符串，来源信息全丢了，只能重做。

2.2. 统一抽象带来什么 #

Document Loader 就是把「五花八门的数据源」收敛成一种数据结构 + 两个方法：

PDF   ─┐
Word  ─┤                        ┌─ load()       一次全部读出
CSV   ─┼─► Document Loader ─────┤
网页  ─┤     统一接口            └─ lazy_load()  逐个流式读出
数据库─┘
                 │
                 ▼
        list[Document]  ← 后面三章只认这个类型

好处很直接：

好处	说明
换数据源不改下游	从「本地 PDF」换成「网页」，切分、向量化、检索代码一行不用动
来源信息不丢	每个 Document 自带 `metadata`，天然支持溯源与过滤
大文件不炸内存	`lazy_load()` 流式产出，不必一次性全读进来

一句话：

Loader 的价值不只是「读出文字」，更是「读出文字和它的出处」。

3. Document：RAG 的最小数据单位 #

3.1. 两个主字段，外加两个容易被忽略的 #

Document 是 langchain_core 里的简单数据类，第 12～15 章全程都在处理它，务必先弄清结构：

# Document 定义在 langchain_core 里，所有 loader 的产出都是它
from langchain_core.documents import Document

# 手动构造一个 Document，感受它的结构
doc = Document(
    # page_content：正文，纯字符串，这部分将来会被切分、向量化
    page_content="签收 7 日内可无理由退货。质量问题 15 日内可换货。",
    # metadata：附带信息，字典，随你放什么；检索时用来溯源和过滤
    metadata={
        # source 是最重要的一个键，几乎所有 loader 都会给
        "source": "policy/refund.md",
        # 标题这类字段要自己补（§8.1）
        "title": "退换货政策",
        # 页码，PDF 之类的 loader 会自动带上
        "page": 1,
    },
)

# 正文
print("正文：", doc.page_content)
# 元数据
print("元数据：", doc.metadata)
# 打印整个对象，能看到 LangChain 的标准表示
print("对象：", doc)

运行输出：

正文： 签收 7 日内可无理由退货。质量问题 15 日内可换货。
元数据： {'source': 'policy/refund.md', 'title': '退换货政策', 'page': 1}
对象： page_content='签收 7 日内可无理由退货。质量问题 15 日内可换货。' metadata={'source': 'policy/refund.md', 'title': '退换货政策', 'page': 1}

日常用到的就是这两个，分工非常明确：

字段	类型	装什么	谁会用它
`page_content`	`str`	正文文本	第 13 章切分、第 14 章向量化
`metadata`	`dict`	来源、页码、标题、时间等	第 15 章溯源引用与条件过滤

其实 Document 有 4 个字段，打印一下就清楚：

# 列出 Document 的全部字段（它是一个 Pydantic 模型）
print("全部字段：", list(Document.model_fields.keys()))

# 新建一个不指定 id 的 Document
d = Document(page_content="正文", metadata={"source": "a.md"})
# id 默认是 None
print("默认 id：", repr(d.id))
# type 是一个固定标识，用于序列化时区分对象种类
print("默认 type：", repr(d.type))

全部字段： ['id', 'metadata', 'page_content', 'type']
默认 id： None
默认 type： 'Document'

type 基本不用管（永远是 'Document'），id 现在就该记住：

字段	说明
`id`	文档的唯一标识，默认 `None`；loader 不会自动生成
`type`	固定为 `'Document'`，序列化时用于区分类型

id 为什么重要？第 14 章往向量库写数据时，很多向量库支持「按 id 覆盖写入」（upsert）。每个 Document 都有稳定 id，制度更新后就能精准替换对应条目；id 为空，每次重新摄入就只能全库重建。

本章 §8.1 会在 metadata 里放 doc_id 承担这个职责。为什么不直接赋给 Document.id？第 13 章切分后，一个 Document 会变成多个 chunk，chunk 的 id 要重新生成（如 doc_id-0、doc_id-1）。metadata 里的 doc_id 会一路继承，正好表达「这些 chunk 来自同一份原文」。

3.2. metadata 决定了 RAG 能不能「说出处」 #

新手最常犯的错，是把 metadata 当可选项。看看它在第 15 章会撑起哪些需求：

产品需求	靠哪个 metadata 字段
回答后面附「来源：员工手册 P12」	`source` + `page`
「只查 2026 年之后的制度」	`updated_at`
「财务的文档不给普通员工检索」	`department` / `acl`
资料更新后删掉旧版本切片	`doc_id`

这些字段只能在加载阶段补齐——切分之后只剩碎片，再也无从知道它来自哪一页。所以本章一半工作量在经营 metadata（§8 专讲）。

记住这句话：

page_content 决定模型答得准不准，metadata 决定这个答案敢不敢用。

4. 第一个加载器：TextLoader #

4.1. 最小可运行示例 #

先从最简单的纯文本开始。下面这段代码自己创建示例文件再加载。

# Path 用于跨平台处理文件路径
from pathlib import Path

# 文本加载器，来自 langchain_community
from langchain_community.document_loaders import TextLoader

# 创建演示用的文本文件，encoding 显式写 utf-8（原因见 §4.2）
sample = Path("faq.txt")
# write_text 直接把字符串写进文件
sample.write_text(
    # \n 是换行，模拟一问一答两行
    "问：发货要多久？\n答：一般 48 小时内发出。\n",
    # 写入时也要显式指定编码，保证和下面读取时一致
    encoding="utf-8",
)

# 构造加载器：第一个参数是文件路径，encoding 一定要显式指定
loader = TextLoader(str(sample), encoding="utf-8")

# load() 返回 list[Document]；TextLoader 是「整个文件 → 1 个 Document」
docs = loader.load()

# 一个文件产出几个 Document，是各 loader 最重要的差异（见 §5.6）
print("Document 数量：", len(docs))
# repr 能显示出 \n 这类不可见字符，调试时比直接 print 有用
print("正文：", repr(docs[0].page_content))
# TextLoader 只自动带一个 source 字段
print("元数据：", docs[0].metadata)

运行输出：

Document 数量： 1
正文： '问：发货要多久？\n答：一般 48 小时内发出。\n'
元数据： {'source': 'faq.txt'}

两个细节留意一下：

metadata 里只有 source，标题、更新时间要自己补——这正是 §8 要做的事。
source 是你传入的路径原样。上面传相对路径 faq.txt，存的就是 faq.txt；传绝对路径，存下来就是 D:\forever\...\faq.txt 这样带盘符的字符串。§8.1 会把它收成硬性约定。

关于一条警告： 首次从 langchain_community 导入时，终端可能出现这样一行提示：

DeprecationWarning: `langchain-community` is being sunset and is no longer actively
maintained. See https://github.com/langchain-ai/langchain-community/issues/674 for
details and migration guidance toward standalone integration packages.

含义是：官方正在把「大杂烩包」拆成独立集成包（如 PDF 有 langchain-pymupdf4llm 等）。当前仍能正常工作，本章示例照常跑即可；项目要长期维护、或需要更强 PDF 解析时，再按官方集成列表换独立包。接口（load() / lazy_load() / 返回 Document）稳定，换包不影响下游。

4.2. Windows 上的中文编码坑 #

这是中文用户在本章遇到的头号问题，下面讲透。

根因：你的 Windows 默认编码不是 UTF-8

先在自己机器上跑这两行，看清事实：

# locale 模块能告诉我们 Python 打开文件时默认用什么编码
import locale

# 这就是 open() 在不指定 encoding 时实际采用的编码
print("open() 默认编码：", locale.getpreferredencoding(False))
# 这个是「源码字符串」的编码，和读文件无关，别混淆
import sys
print("sys.getdefaultencoding()：", sys.getdefaultencoding())

中文 Windows 上的实测结果：

open() 默认编码： cp936
sys.getdefaultencoding()： utf-8

cp936 就是 GBK。这两个值不一样——很多人看到 sys.getdefaultencoding() 是 utf-8，就以为读文件也是 UTF-8，这是常见误会。管读文件的是前者。

现在项目里的文件（Git 仓库、VS Code 新建、Linux 服务器导出）几乎都是 UTF-8。中文 Windows 默认却是 GBK——两套编码一碰，就会冲突。

TextLoader 的编码回退逻辑

TextLoader 内部只有十几行，把它的逻辑理顺，后面所有现象都能解释：

# 这是 TextLoader.lazy_load 的核心逻辑（简化示意）
try:
    # 第一步：用 self.encoding 打开文件
    # 如果你没传 encoding，这里就是 None，Python 会用系统默认（cp936）
    with open(self.file_path, encoding=self.encoding) as f:
        text = f.read()
except UnicodeDecodeError as e:
    # 第二步：解码失败了，看有没有开自动检测
    if self.autodetect_encoding:
        # 开了就去猜编码，返回一串按置信度排序的候选（这里需要 chardet）
        detected_encodings = detect_file_encodings(self.file_path)
        # 逐个候选尝试
        for encoding in detected_encodings:
            try:
                # 用猜出来的编码再打开一次
                with open(self.file_path, encoding=encoding.encoding) as f:
                    text = f.read()
                # 有一个成功就跳出循环
                break
            except UnicodeDecodeError:
                # 这个候选也不行，试下一个
                continue
    else:
        # 没开自动检测：把原始异常包成 RuntimeError 抛出
        raise RuntimeError(f"Error loading {self.file_path}") from e

三个要点：

不传 encoding 不等于用 UTF-8，等于「用系统默认」，在中文 Windows 上就是 GBK。
抛出来的是 RuntimeError，真正的原因藏在 __cause__ 里。所以只看异常类型会一头雾水，得往里挖。
autodetect_encoding 的候选如果全部失败，循环正常结束、text 保持空字符串，函数会产出一个内容为空的 Document 而不报错——这是个典型的静默失败，正是 §9 要拦的。

第 2 点的正确排查姿势：

# Path 用于构造路径
from pathlib import Path

# 文本加载器
from langchain_community.document_loaders import TextLoader

# 造一个 UTF-8 编码的中文文件，这是真实项目里最常见的情况
utf8_file = Path("policy_utf8.txt")
# 用 write_bytes + encode 精确控制字节，避免受系统默认影响
utf8_file.write_bytes("退换货政策：签收七日内可退。\n".encode("utf-8"))

# 故意不传 encoding，让它走系统默认（中文 Windows 上是 GBK）
try:
    TextLoader(str(utf8_file)).load()
except Exception as exc:
    # 外层异常：只说"加载出错"，看不出原因
    print("外层异常：", type(exc).__name__, exc)
    # __cause__ 才是真正的元凶，会指出是哪个编码、哪个字节出的问题
    print("真正原因：", type(exc.__cause__).__name__, exc.__cause__)

实测输出，最后一行把问题说得非常清楚：

外层异常： RuntimeError Error loading policy_utf8.txt
真正原因： UnicodeDecodeError 'gbk' codec can't decode byte 0x80 in position 2: illegal multibyte sequence

看到 'gbk' codec 就知道：代码在用 GBK 读一个 UTF-8 文件。以后遇到 UnicodeDecodeError，第一件事就是看它报的是哪个 codec，那就是当前实际用的编码。

六种组合的实测结果

把「文件真实编码 × 传什么 encoding」的组合全跑一遍，结论一目了然：

# Path 用于处理路径
from pathlib import Path

# 文本加载器
from langchain_community.document_loaders import TextLoader

# 造两个内容相同但编码不同的文件
utf8_file = Path("policy_utf8.txt")
# UTF-8 版本，代表现在的项目文件
utf8_file.write_bytes("退换货政策：签收七日内可退。\n".encode("utf-8"))
gbk_file = Path("legacy_gbk.txt")
# GBK 版本，模拟老系统 / 旧 Excel 导出的中文文本
gbk_file.write_bytes("退换货政策：签收七日内可退。\n".encode("gbk"))

# 依次尝试各种组合
for label, path, kwargs in [
    # 老文件碰上系统默认，碰巧对上了
    ("GBK 文件 + 不传 encoding", gbk_file, {}),
    # 编码写错，报错
    ("GBK 文件 + encoding=utf-8", gbk_file, {"encoding": "utf-8"}),
    # 编码写对，成功
    ("GBK 文件 + encoding=gbk", gbk_file, {"encoding": "gbk"}),
    # 最常见的真实场景：新文件碰上 GBK 默认
    ("UTF-8 文件 + 不传 encoding", utf8_file, {}),
    # 正确做法
    ("UTF-8 文件 + encoding=utf-8", utf8_file, {"encoding": "utf-8"}),
    # 编码写错，报错
    ("UTF-8 文件 + encoding=gbk", utf8_file, {"encoding": "gbk"}),
]:
    try:
        # ** 把字典展开成关键字参数
        docs = TextLoader(str(path), **kwargs).load()
        # 检查关键词是否还在，用来判断是否乱码
        ok = "退换货政策" in docs[0].page_content
        # :<28 是左对齐补空格，让输出成一列好对比
        print(f"{label:<28} -> {'内容正确' if ok else '乱码'}")
    except Exception as exc:
        # 解码失败会走到这里，只打印异常类型名
        print(f"{label:<28} -> {type(exc).__name__}")

GBK 文件 + 不传 encoding         -> 内容正确
GBK 文件 + encoding=utf-8      -> RuntimeError
GBK 文件 + encoding=gbk        -> 内容正确
UTF-8 文件 + 不传 encoding       -> RuntimeError
UTF-8 文件 + encoding=utf-8    -> 内容正确
UTF-8 文件 + encoding=gbk      -> RuntimeError

第一行最容易误导人：GBK 文件不传 encoding 居然读对了。这只是系统默认恰好等于文件编码。在这台机器上跑通的代码，换到 macOS / Linux（默认 UTF-8）或服务器上，同一份 GBK 文件立刻就崩。

结论：能读对不代表写对了。任何时候都显式写 encoding=，别把「系统默认恰好匹配」当成代码正确。

autodetect_encoding=True 为什么不推荐

官方提供了这个参数让它自动猜编码。听起来很省事，但实测有两个问题。

问题一：它依赖 chardet，不装就报错。 而且报的是原始的 ModuleNotFoundError，没有被包成友好提示：

ModuleNotFoundError: No module named 'chardet'

问题二（更糟）：装了新版 chardet 反而会崩。 用 chardet 7.5.1 实测：

TypeError: FileEncoding.__new__() got an unexpected keyword argument 'mime_type'

原因是新版 chardet 的检测结果多返回了 mime_type 字段，而 langchain_community 内部的 FileEncoding 只定义了三个字段（encoding、confidence、language），多出来的键直接把构造函数打崩。这是版本不兼容，不是你用错了。

问题三：就算装对了版本，猜测本身也不可靠。 让 chardet 去猜那个 GBK 文件，它返回 32 个候选，排第一的是：

{'encoding': 'GB18030', 'confidence': 0.195, 'language': 'zh', ...}

置信度只有 0.195。中文短文本的编码特征本来就弱，猜错完全正常（GB18030 这次刚好兼容 GBK 才没出问题）。

推荐做法：自己写一个候选列表兜底

与其依赖不可控的自动检测，不如显式列出项目里可能出现的编码，依次尝试。行为可预测，全部失败时明确报错，不会悄悄给你一个空文档：

# Path 用于处理路径
from pathlib import Path

# Document 是我们要返回的标准类型
from langchain_core.documents import Document


def load_text_robust(
    path: Path,
    # 候选编码按可能性从高到低排；改成你项目的实际情况
    encodings: tuple[str, ...] = ("utf-8", "gbk"),
) -> Document:
    """按候选编码依次尝试读取，全部失败就明确报错。"""
    # 一次性读成字节，后面反复尝试解码，避免重复读盘
    raw = path.read_bytes()
    # 逐个候选尝试
    for enc in encodings:
        try:
            # decode 成功就说明这个编码是对的
            return Document(
                # 解码后的正文
                page_content=raw.decode(enc),
                # 把实际用的编码记进 metadata，便于日后排查
                metadata={"source": path.as_posix(), "encoding": enc},
            )
        except UnicodeDecodeError:
            # 这个编码不对，试下一个
            continue
    # 所有候选都失败：明确抛错，绝不返回空内容
    raise ValueError(f"无法解码 {path}，尝试过：{encodings}")


# 验证：UTF-8 和 GBK 两种文件都能正确读出
for enc in ["utf-8", "gbk"]:
    # 每种编码写一个文件
    p = Path(f"robust_{enc}.txt")
    # 用指定编码写入
    p.write_bytes("退换货政策：签收七日内可退。\n".encode(enc))
    # 不告诉函数编码是什么，让它自己试
    d = load_text_robust(p)
    # 打印内容和实际命中的编码，确认识别正确
    print(f"{enc} 文件 -> 内容={d.page_content.strip()!r} 实际编码={d.metadata['encoding']}")

# 验证：无法解码时会明确报错，而不是返回空文档
bad = Path("robust_bad.bin")
# 写一串不属于任何常见中文编码的字节
bad.write_bytes(b"\xff\xfe\x81\x8f" * 30)
try:
    # 所有候选都会解码失败
    load_text_robust(bad)
except ValueError as exc:
    # 拿到明确的错误信息，而不是一个空 Document
    print("全部失败时：", exc)

实测输出：

utf-8 文件 -> 内容='退换货政策：签收七日内可退。' 实际编码=utf-8
gbk 文件 -> 内容='退换货政策：签收七日内可退。' 实际编码=gbk
全部失败时： 无法解码 robust_bad.bin，尝试过：('utf-8', 'gbk')

小结

做法	评价
统一把资料转成 UTF-8，代码里写死 `encoding="utf-8"`	首选，一次性解决
已知是 GBK 就写 `encoding="gbk"`	可行，但混合目录里会顾此失彼
自己写候选编码列表兜底（上面的 `load_text_robust`）	混合编码目录的推荐方案，行为可预测
用 `autodetect_encoding=True`	不推荐：依赖 `chardet`、与新版不兼容、置信度低、失败时可能静默产出空文档
不传 `encoding` 靠系统默认	不要，代码在别人机器上就崩

如果编码问题和 DirectoryLoader 的 silent_errors=True 撞在一起，后果会被放大：失败文件被静默跳过——你以为加载了 100 个，实际只进来 60 个，终端却一片祥和。这类事故正是 §9 质量自检要拦的。

5. 常用文件类型怎么加载 #

各 loader 用法高度一致（构造 → load()），真正要记的差异只有一个：一个文件会被切成几个 Document。

5.1. PDF：PyPDFLoader #

PDF 是企业资料里占比最高的格式。PyPDFLoader 一页产出一个 Document，并自动记录页码——后面「来源：手册 P12」就靠它。

依赖：pip install pypdf

# PDF 加载器
from langchain_community.document_loaders import PyPDFLoader

# 把路径换成你自己的 PDF 文件
loader = PyPDFLoader("handbook.pdf")

# 一页一个 Document，所以 len(docs) == PDF 页数
docs = loader.load()

# 用这个数字和 PDF 实际页数对比，是最快的完整性校验
print("Document 数量（= 页数）：", len(docs))
# 逐页查看正文开头与元数据
for doc in docs:
    # page 是从 0 开始的索引，所以显示给人看时要 +1
    print(f"--- 第 {doc.metadata['page'] + 1} 页 ---")
    # 只看前 60 字；把换行替换成空格，避免刷屏
    print(doc.page_content[:60].replace("\n", " "))
    # 打印完整元数据，看看 loader 自动带了什么
    print(doc.metadata)

用一个handbook.pdf 实测，输出如下：

Document 数量（= 页数）： 2
--- 第 1 页 ---
Page one: refund policy
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'handbook.pdf', 'total_pages': 2, 'page': 0, 'page_label': '1'}
--- 第 2 页 ---
Page two: shipping policy
{'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': '', 'source': 'handbook.pdf', 'total_pages': 2, 'page': 1, 'page_label': '2'}

三个字段要分清：

字段	含义	注意
`page`	页索引，从 0 开始	显示给用户时记得 `+1`
`page_label`	PDF 里印的页号，字符串	有前言的书里可能是 `"iii"`
`total_pages`	总页数	可用来做完整性校验

PDF 最大的陷阱：扫描件

扫描件 PDF 只有图片、没有文字层时，PyPDFLoader 会返回一堆 page_content 为空的 Document——不报错，就是空的。

不用真找扫描件也能复现。用 pypdf 造两个纯空白页，效果等同「没有文字层」：

# PdfWriter 用来生成 PDF 文件
from pypdf import PdfWriter

# PDF 加载器
from langchain_community.document_loaders import PyPDFLoader

# 新建一个空的 PDF 写入器
writer = PdfWriter()
# 加两个空白页（尺寸是 A4 的点数），模拟"没有文字层"的扫描件
writer.add_blank_page(width=595, height=842)
writer.add_blank_page(width=595, height=842)
# 以二进制模式写入磁盘
with open("scan_like.pdf", "wb") as f:
    writer.write(f)

# 用正常方式加载这个"扫描件"
docs = PyPDFLoader("scan_like.pdf").load()

# 注意：没有任何异常抛出
print("Document 数量：", len(docs))
# 逐页检查内容长度
for i, doc in enumerate(docs):
    # 正文是空字符串，长度为 0
    print(f"第 {i} 页：content={doc.page_content!r} 长度={len(doc.page_content)}")
# 元数据一应俱全，看起来"加载成功"了
print("元数据：", docs[0].metadata)

实测输出：

Document 数量： 2
第 0 页：content='' 长度=0
第 1 页：content='' 长度=0
元数据： {'producer': 'pypdf', 'creator': 'PyPDF', 'creationdate': '', 'source': 'scan_like.pdf', 'total_pages': 2, 'page': 0, 'page_label': '1'}

这就是本章最需要警惕的失败模式：页数对、元数据齐全、没有异常，正文却全是空的。不做 §9 自检，这两条空记录会一路进向量库——到第 15 章你会百思不得其解：「手册明明摄入了，为什么检索不到滤芯更换周期？」

扫描件必须走 OCR（光学字符识别）才能提取文字，PyPDFLoader 对它无能为力。常见处理路径：

方案	说明
让业务方提供电子原件	首选，最省事也最准确
用支持 OCR 的解析服务	如 `unstructured` 的 hi-res 模式、云厂商的文档识别 API
本地 OCR	`pytesseract` + `pdf2image`，需额外装 Tesseract 引擎，中文还要下载中文语言包

无论走哪条路，先用自检把扫描件揪出来，再决定怎么处理，比闷头全量摄入靠谱。

5.2. Word：Docx2txtLoader #

Word 用 Docx2txtLoader，整篇读成 1 个 Document（不像 PDF 分页）。

依赖：pip install docx2txt

contract.docx

# Word 加载器
from langchain_community.document_loaders import Docx2txtLoader

# 只支持 .docx（新格式）；老的 .doc 需先用 Word 另存为 .docx
loader = Docx2txtLoader("contract.docx")

# 整个文档 → 1 个 Document
docs = loader.load()

# 无论文档多长，这里永远是 1
print("Document 数量：", len(docs))
# 看前 60 字确认解析正常、没有乱码
print("正文前 60 字：", docs[0].page_content[:60])
# metadata 只有 source，没有页码（Word 的分页由排版决定，解析阶段拿不到）
print("元数据：", docs[0].metadata)

实测输出（示例文档只有一行字）：

Document 数量： 1
正文前 60 字： Word 文档正文测试
元数据： {'source': 'contract.docx'}

两点提醒：.doc 不支持（要先转 .docx）；表格、页眉页脚、批注的提取效果有限，重要表格建议单独导出成 CSV。

「整篇 1 个 Document」对下游也有影响：Word 往往很长（几十页合同、制度），metadata 里没有页码。第 15 章溯源时只能说「来自《劳动合同模板.docx》」，没法精确到页。

溯源精度要求高，有两条路：让业务方导出 PDF 再摄入（有页码）；或在第 13 章切分时给每个 chunk 记序号，用「第 N 段」代替页码。

5.3. CSV：CSVLoader #

CSV 的行为和前面都不同：一行数据产出一个 Document，并把每列渲染成 列名: 值 的文本。

# Path 用于写演示文件
from pathlib import Path

# CSV 加载器
from langchain_community.document_loaders import CSVLoader

# 造一个演示 CSV：一行表头 + 两行数据
Path("orders.csv").write_text(
    # 第一行是表头，后两行是数据
    "order_id,status,eta\nA1001,已发货,明天\nA1002,运输中,后天\n",
    # 中文 CSV 必须显式指定编码
    encoding="utf-8",
)

# CSV 也要显式指定编码，中文 CSV 尤其容易踩 GBK 坑
loader = CSVLoader("orders.csv", encoding="utf-8")

# 两行数据 → 两个 Document
docs = loader.load()

# 表头不算一行数据，所以这里是 2 而不是 3
print("Document 数量（= 数据行数）：", len(docs))
# 正文是「列名: 值」逐行拼接，方便模型理解字段含义
print("第一行正文：", repr(docs[0].page_content))
# metadata 多了一个 row，记录这是第几行（从 0 开始）
print("第一行元数据：", docs[0].metadata)

运行输出：

Document 数量（= 数据行数）： 2
第一行正文： 'order_id: A1001\nstatus: 已发货\neta: 明天'
第一行元数据： {'source': 'orders.csv', 'row': 0}

注意正文格式是 列名: 值 逐行排列，不是原始逗号分隔。向量化时模型看到的是 status: 已发货，字段名和值绑在一起，语义完整；直接喂 A1001,已发货,明天，模型根本不知道每个值是什么。

什么时候该用 CSVLoader，先想清楚：

场景	建议
每行是一条独立知识（FAQ、产品条目）	适合，一行一个 Document 正好
几万行的交易流水	不适合，这类精确查询该用 SQL 工具（第 8 章），不要塞进向量库

第二行要多说一句：RAG 擅长「语义相似」，不擅长「精确匹配和聚合」。「查订单 A1002 的状态」「统计上月退货总额」这类问题，用 SQL 一步到位；塞进向量库反而可能检索出 A1003（文本相似）。判断标准：这条数据是用来「读」的，还是用来「算」的。 政策、说明、FAQ 进 RAG；流水、库存、报表走数据库工具。

5.4. Excel：.xlsx 怎么办 #

§2.1 提过「客户名单是 Excel」——中文企业里这几乎是常态，不少业务知识就躺在 .xlsx 里。LangChain 内置的 UnstructuredExcelLoader 依赖庞大的 unstructured 生态，对新手不太友好。

更轻的做法是自己写十几行，用 openpyxl 直接读，行为可控。思路和 CSVLoader 一致：首行当表头，之后每行产出一个 Document。

依赖：pip install openpyxl

# Path 用于处理路径
from pathlib import Path

# openpyxl 是读写 .xlsx 的标准库
import openpyxl

# Document 是我们要返回的标准类型
from langchain_core.documents import Document


def load_excel(path: Path) -> list[Document]:
    """把 Excel 每一行读成一个 Document，首行当表头。"""
    # read_only=True 让大文件也能低内存读取
    # data_only=True 表示读公式的计算结果，而不是公式本身（很重要）
    wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
    # 收集所有产出的 Document
    out: list[Document] = []
    # 一个 Excel 可能有多个工作表，都要处理
    for ws in wb.worksheets:
        # values_only=True 表示只取单元格的值，不要样式对象
        rows = ws.iter_rows(values_only=True)
        # 取第一行当表头；next 的第二个参数是"取不到时的默认值"
        header = next(rows, None)
        # 空工作表没有表头，直接跳过
        if not header:
            continue
        # 遍历剩下的数据行，i 从 0 开始计数
        for i, row in enumerate(rows):
            # 整行都是空的就跳过（Excel 里常有大量空白行）
            if all(cell is None for cell in row):
                continue
            # 拼成「列名: 值」的格式，和 CSVLoader 保持一致
            body = "\n".join(
                # zip 把表头和数据一一配对
                f"{h}: {v}" for h, v in zip(header, row) if h is not None
            )
            out.append(
                Document(
                    # 正文
                    page_content=body,
                    metadata={
                        # as_posix 统一成正斜杠（§8.1 的约定）
                        "source": path.as_posix(),
                        # 记下来自哪个工作表，多表文件溯源时必需
                        "sheet": ws.title,
                        # 行号，作用和 CSVLoader 的 row 一样
                        "row": i,
                    },
                )
            )
    # read_only 模式打开的文件需要显式关闭
    wb.close()
    return out


# 造一个演示 Excel 来验证
from openpyxl import Workbook

# 新建工作簿
wb = Workbook()
# 取默认工作表并改名
ws = wb.active
ws.title = "订单"
# 写表头
ws.append(["order_id", "status", "eta"])
# 写两行数据
ws.append(["A1001", "已发货", "明天"])
ws.append(["A1002", "运输中", "后天"])
# 再建一个空表，验证它会被跳过
wb.create_sheet("空表")
# 保存到磁盘
wb.save("orders.xlsx")

# 加载并打印结果
docs = load_excel(Path("orders.xlsx"))
# 应该是 2（空表被跳过、表头不算数据）
print("产出 Document 数：", len(docs))
# 逐条检查
for d in docs:
    # 确认是「列名: 值」格式
    print("正文：", repr(d.page_content))
    # 确认 sheet 和 row 都记上了
    print("元数据：", d.metadata)

实测输出，空工作表被正确跳过：

产出 Document 数： 2
正文： 'order_id: A1001\nstatus: 已发货\neta: 明天'
元数据： {'source': 'orders.xlsx', 'sheet': '订单', 'row': 0}
正文： 'order_id: A1002\nstatus: 运输中\neta: 后天'
元数据： {'source': 'orders.xlsx', 'sheet': '订单', 'row': 1}

Excel 的几个特有注意点：

注意点	说明
`data_only=True`	不加这个参数，公式单元格读出来是 `"=SUM(A1:A9)"` 而不是数值
多工作表	一个文件常有多个 sheet，务必都遍历，并把 `sheet` 记进 metadata
合并单元格	读出来只有左上角那格有值，其余是 `None`，中文报表里很常见
大量空行	Excel 的「已用区域」常比实际数据大，必须过滤全空行
`.xls`（老格式）	`openpyxl` 不支持，需先用 Excel 另存为 `.xlsx`

合并单元格最容易出问题。表里大量用合并单元格做分类表头，读出来会残缺——这种表建议先请业务方拉平成「一行一条记录」，比在代码里猜合并逻辑可靠。

5.5. 网页：WebBaseLoader #

抓取网页正文用 WebBaseLoader。它比前面几个多两个前提，单独说明。

依赖：pip install beautifulsoup4；另外需要联网，且建议设置 USER_AGENT 环境变量（不设会有一条提醒，说明你的请求缺少身份标识）。

# os 用于设置环境变量
import os

# 设置 USER_AGENT，表明请求来源；不设只是警告，但礼貌的爬取应该带上
# setdefault 表示"已经设过就不覆盖"，必须在导入 loader 之前执行
os.environ.setdefault("USER_AGENT", "langrag-tutorial/1.0")

# 网页加载器，内部用 requests + BeautifulSoup
from langchain_community.document_loaders import WebBaseLoader

# 可以传单个 URL，也可以传 URL 列表批量抓取
loader = WebBaseLoader(["https://example.com/"])

# 一个 URL → 1 个 Document，正文是去掉标签后的文本
docs = loader.load()

# 传了几个 URL 就有几个 Document
print("Document 数量：", len(docs))
# strip 去掉网页文本常见的大量首尾空白
print("正文前 80 字：", docs[0].page_content.strip()[:80])
# metadata 里的 source 是 URL，天然可点击溯源
print("元数据：", docs[0].metadata)

网页加载的现实问题，提前知道能省很多时间：

问题	说明
导航栏 / 页脚 / 广告混进正文	会污染检索质量，需要清洗（§8.2）
前端渲染的页面抓到空内容	内容由 JS 生成，静态抓取拿不到，需要专门的抓取服务
反爬、限流	尊重 `robots.txt`，控制频率

企业内部知识库通常有更可靠的路径：让业务方导出文件，比爬自家官网稳定。

5.6. 加载器与依赖速查 #

把本节压成一张表，日常照着查。「一个文件产出几个 Document」这一列最该记，它直接决定第 13 章切分时的心理预期：

格式	Loader	额外依赖	一个文件 → 几个 Document	自带 metadata
`.txt` / `.md`	`TextLoader`	无	1	`source`
`.pdf`	`PyPDFLoader`	`pypdf`	每页 1 个	`source`, `page`, `page_label`, `total_pages`
`.docx`	`Docx2txtLoader`	`docx2txt`	1	`source`
`.csv`	`CSVLoader`	无	每行 1 个	`source`, `row`
`.xlsx`	自己写（§5.4）	`openpyxl`	每行 1 个	自定义：`source`, `sheet`, `row`
网页	`WebBaseLoader`	`beautifulsoup4`	每个 URL 1 个	`source`(URL), `title` 等

粒度差异会带来一个直觉陷阱：Document 数量多不代表内容多。5000 行 CSV 产出 5000 个 Document，80 页制度 Word 只产出 1 个。§9 自检要看字符数分布，不是 Document 条数。

一次装齐本章依赖：

# pypdf 解析 PDF；docx2txt 解析 Word；beautifulsoup4 解析网页；openpyxl 解析 Excel
# 若用 uv 管理项目，把 pip install 换成 uv add
pip install pypdf docx2txt beautifulsoup4 openpyxl

更多格式（Notion、飞书、S3、数据库等）查官方集成列表。注意官方对社区加载器有一句提示：社区集成由用户贡献、未经官方审核，接入前建议先用少量样本验证效果。

6. 批量加载整个目录 #

6.1. DirectoryLoader #

实际项目不会一个个手点文件，而是「把 docs/ 整个吃进来」。DirectoryLoader 遍历目录，把匹配到的文件交给指定 loader。

# Path 用于创建演示目录与文件
from pathlib import Path

# DirectoryLoader 负责遍历目录，TextLoader 负责读单个文件
from langchain_community.document_loaders import DirectoryLoader, TextLoader

# 准备一个有子目录的演示知识库
root = Path("kb_demo")
# parents=True 表示连父目录一起建，exist_ok=True 表示已存在不报错
(root / "policy").mkdir(parents=True, exist_ok=True)
# 根目录放一个 txt
(root / "faq.txt").write_text("发货一般 48 小时内。\n", encoding="utf-8")
# 子目录放一个 md，用来验证递归扫描
(root / "policy" / "refund.md").write_text("签收 7 日内可退货。\n", encoding="utf-8")

# glob 可以直接传列表，一次匹配多种扩展名
loader = DirectoryLoader(
    # 要扫描的根目录
    str(root),
    # 文件匹配模式列表，开头的 ** 表示递归所有子目录
    glob=["**/*.txt", "**/*.md"],
    # 用哪个 loader 处理匹配到的文件
    loader_cls=TextLoader,
    # 传给 loader 的参数，中文场景务必显式指定编码
    loader_kwargs={"encoding": "utf-8"},
    # 显示进度条（大目录时很有用，需要 tqdm）
    show_progress=False,
    # 单个文件失败时是否跳过；设 True 一定要配合 §9 的自检
    silent_errors=True,
)

# 一次调用就把两种扩展名都加载进来了
docs = loader.load()

print("合计 Document 数：", len(docs))
# source 是带子目录的路径，Windows 下用反斜杠
for doc in docs:
    print("-", doc.metadata["source"])

运行输出：

合计 Document 数： 2
- kb_demo\faq.txt
- kb_demo\policy\refund.md

两个细节：

glob 支持列表。 不少资料（含本文旧版）说「一个 DirectoryLoader 只能吃一种 glob，多扩展名要分多次加载再合并」。实测传列表可以，上面那段就是一次搞定 .txt 和 .md。分多次也行，只是没必要：

root = Path("kb_demo")
# 也可以分多次加载再合并，但没必要这么写
docs = []
# 逐个模式构造 loader
for pattern in ["**/*.txt", "**/*.md"]:
    # 每次只匹配一种扩展名
    batch = DirectoryLoader(
        # 根目录
        str(root),
        # 本轮的单个 glob 模式
        glob=pattern,
        # 仍然是同一个 loader
        loader_cls=TextLoader,
        # 编码照旧要传
        loader_kwargs={"encoding": "utf-8"},
    ).load()
    # 打印本次匹配数量，便于定位是哪种格式没扫到
    print(f"{pattern} 匹配到 {len(batch)} 个文件")
    # 累加到总列表
    docs.extend(batch)

source 里是 Windows 反斜杠。 上面输出 kb_demo\policy\refund.md——和 §4.1 一样，loader 原样保留你传入的路径。反斜杠会带来两个麻烦：写进 JSON 会被转义成 kb_demo\\policy\\refund.md；换到 Linux 服务器路径也对不上。这正是 §8.1 要统一成相对路径 + 正斜杠的原因。

顺便说一下 **/ 的语义：表示「递归零层或多层子目录」，所以 **/*.txt 既能匹配根目录 faq.txt，也能匹配 policy/ 里的文件。只要根目录那一层，写 *.txt（不带 **/）。

参数里最需要留意的是 loader_cls 和 silent_errors：

参数	作用	建议
`glob`	文件匹配模式，`**/` 表示递归	可传字符串或列表；列表更省事
`loader_cls`	用哪个 loader	一次只能指定一种，这是它最大的局限
`loader_kwargs`	透传给 loader 的参数	中文场景必传 `encoding`
`silent_errors`	失败是否静默跳过	方便，但会掩盖问题，必须配自检
`show_progress`	进度条	文件多时打开，便于判断是否卡住

注意 glob 能传列表不能解决 loader_cls 的问题：glob=["**/*.txt", "**/*.pdf"] 会把 PDF 也交给 TextLoader 去读，结果必然出错。列表里的所有模式必须能用同一个 loader 处理，这就是下一节要讲的局限。

6.2. 为什么本章实战不用它 #

DirectoryLoader 的硬伤是 loader_cls 只能给一种：真实 docs/ 里 PDF、Word、Markdown、CSV 混放，它没法一次搞定。常见两种应对：

做法	说明
按格式分多次调用	每种格式一个 `DirectoryLoader`，结果合并；写法简洁
按扩展名手写分派	自己遍历文件、按后缀选 loader；能逐文件捕获异常、逐文件记录质检结果

§10 的本章实战选第二种。前者不是不好，企业摄入真正的难点是「哪个文件失败了、为什么失败」——把每个文件的成败单独记下来，比少写几行代码值钱。

具体说，silent_errors=True 只给你一个「总数」：加载到了 87 个 Document。但你真正要知道的是：

那份新上传的《2026 年报销制度.pdf》进来了没有？
财务/ 目录下 12 个文件为什么只出来 9 个？
失败的那 3 个是编码问题，还是扫描件，还是文件损坏？

DirectoryLoader 回答不了这些——异常在它内部就被吞掉了。手写分派多几十行，但能给出「文件级成败清单」，这才是运维知识库真正需要的。

7. load() 与 lazy_load() #

所有 loader 都实现了这两个方法，差别只在什么时候占内存：

方法	返回	行为	适合
`load()`	`list[Document]`	一次全读进内存	几十到几百个文件，调试方便
`lazy_load()`	生成器（generator）	边读边产出，用完即弃	大批量 / 大文件 / 边加载边入库

# Path 用于写演示文件
from pathlib import Path

# 文本加载器
from langchain_community.document_loaders import TextLoader

# 造一个演示文件
Path("faq.txt").write_text("发货一般 48 小时内。\n", encoding="utf-8")

# 同一个 loader 对象，两个方法都能调
loader = TextLoader("faq.txt", encoding="utf-8")

# load()：立刻返回完整列表，可以取 len、可以随机索引
docs = loader.load()
# 类型是 list，长度立即可知
print("load 返回类型：", type(docs).__name__, "数量：", len(docs))

# lazy_load()：返回生成器，此刻还没真正读取
stream = loader.lazy_load()
# 类型是 generator，这一步几乎不耗时、不占内存
print("lazy_load 返回类型：", type(stream).__name__)

# 只有开始迭代时才逐个产出 Document
for doc in stream:
    print("流式拿到一个 Document：", doc.metadata)

# 生成器已被消耗完，再迭代一次什么都没有
print("再次迭代同一个生成器：", list(stream))

运行输出：

load 返回类型： list 数量： 1
lazy_load 返回类型： generator
流式拿到一个 Document： {'source': 'faq.txt'}
再次迭代同一个生成器： []

最后一行是用生成器时最容易踩的坑：只能遍历一次，第二次拿到空列表，还不报错。下面这种写法是错的：

# 反例：先数条数再遍历，结果第二步什么都读不到
stream = loader.lazy_load()
# 这一步已经把生成器消耗完了
print("总数：", len(list(stream)))
# 这个循环一次都不会执行
for doc in stream:
    process(doc)

想边遍历边计数，应该在循环里自己累加：

# 正确写法：单次遍历，顺便计数
count = 0
# 只迭代一次
for doc in loader.lazy_load():
    # 处理这一个 Document（切分、入库等）
    process(doc)
    # 计数器加一
    count += 1
# 循环结束后再打印总数
print("总数：", count)

选择原则很简单：

学习和调试用 load()（能随时打印、能数条数）；生产上文档量大时用 lazy_load()，边加载边切分入库，内存占用与文档总量无关。

lazy_load() 省的是加载侧内存。写 docs = list(loader.lazy_load()) 就和 load() 完全等价，一点内存也省不了——收益只在「读一个、处理一个、扔一个」的流水线里。下游也得是流式的：后面若 vector_store.add_documents(all_docs) 一次性入库，前面用 lazy_load() 意义不大，要改成分批入库（第 14 章会用到）。

8. 元数据治理 #

8.1. 该补哪些字段 #

loader 自带的 metadata 很薄（多数只有 source），§3.2 列的产品需求都要靠它。摄入阶段应统一补齐一批字段。推荐最小集合：

字段	值	为什么需要
`source`	相对路径（如 `policy/refund.md`）	绝对路径含本机盘符，换机器就失效
`file_name`	文件名	展示引用时比长路径友好
`file_type`	扩展名（`pdf` / `md`）	支持「只搜 PDF」这类过滤
`doc_id`	稳定哈希	资料更新时能精准删旧切片
`loaded_at`	摄入时间（ISO 格式）	排查「知识库是不是太旧了」
`page` / `row`	loader 给的（若有）	溯源到页 / 行

# hashlib 用于计算稳定的 doc_id
import hashlib
# datetime / timezone 用于生成带时区的时间戳
from datetime import datetime, timezone
# Path 用于处理路径
from pathlib import Path

# Document 是要加工的对象
from langchain_core.documents import Document


def enrich(doc: Document, path: Path, root: Path) -> Document:
    """给一个 Document 补齐统一的 metadata。"""
    # 统一用相对路径，且转成正斜杠，避免 Windows 反斜杠在 JSON 里被转义
    # relative_to 把绝对路径削成相对于知识库根目录的路径
    rel = path.relative_to(root).as_posix()

    # doc_id 由「相对路径 + 页码 / 行号」计算，内容更新但位置不变时 id 保持稳定
    # 用 get 而不是 [] 是因为 txt 没有 page、pdf 没有 row
    raw_key = f"{rel}|{doc.metadata.get('page', '')}|{doc.metadata.get('row', '')}"
    # 取 md5 前 12 位，够短好读，碰撞概率在知识库规模下可忽略
    doc_id = hashlib.md5(raw_key.encode()).hexdigest()[:12]

    # update 是合并而非覆盖，loader 原有的 page / total_pages 都会保留
    doc.metadata.update(
        {
            # 用相对路径覆盖掉 loader 给的原始路径
            "source": rel,
            # 只要文件名，展示引用时比长路径友好
            "file_name": path.name,
            # suffix 带点（".md"），lstrip 去掉点得到 "md"
            "file_type": path.suffix.lstrip("."),
            # 稳定标识，供第 14 章增量更新用
            "doc_id": doc_id,
            # 带时区的 ISO 时间串，跨机器不歧义
            # timespec="seconds" 去掉microseconds，让时间串更短
            "loaded_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }
    )
    return doc


# 演示：手造一个 Document 走一遍 enrich
root = Path("kb_demo")
# 建好子目录
(root / "policy").mkdir(parents=True, exist_ok=True)
# 目标文件路径
target = root / "policy" / "refund.md"
# 写入内容
target.write_text("签收 7 日内可退货。\n", encoding="utf-8")

# 模拟 loader 的产出：metadata 里只有一个原始路径
demo = Document(page_content="签收 7 日内可退货。", metadata={"source": str(target)})
# 加工后打印，观察字段变化
print(enrich(demo, target, root).metadata)

输出（doc_id 与时间会不同）：

{'source': 'policy/refund.md', 'file_name': 'refund.md', 'file_type': 'md', 'doc_id': '87fb91c016df', 'loaded_at': '2026-08-12T13:58:56+00:00'}

有一条约定现在就该立下：metadata 键名全项目必须统一。第 15 章做元数据过滤时，source 和 file_path 混用会让过滤条件全部失灵——那时你已经灌了几万条向量，返工代价很高。

doc_id 的取法（路径 + 页码 / 行号）决定将来能不能增量更新：

取法	后果
按位置算（本文的做法）	文件内容改了但位置不变，`doc_id` 不变 → 能用「按 id 覆盖」精准替换那一页
按内容算（对正文取哈希）	内容改一个字 `doc_id` 就变 → 旧记录变成孤儿，必须先按 `source` 批量删除

两种都能用，差别在更新策略：按位置算配 upsert，按内容算配先删后插。全项目只选一种，混用会让向量库里新旧两份并存，检索时两份都命中，答案自相矛盾。

loaded_at 用 UTC 是故意的。开发机在北京、服务器在新加坡，本地时间就没法比较；统一存 UTC，展示时再转本地时区。

8.2. 文本清洗的边界 #

从 PDF、网页解析出来的文本常有噪声：连续空行、全角空格、每页重复的页眉页脚。适度清洗能提升后续检索质量。

# re 是 Python 内置的正则表达式模块
import re


def clean_text(text: str) -> str:
    """轻量清洗：统一换行、压缩空白，但不改动实质内容。"""
    # Windows 换行统一成 \n，避免 \r 混进正文
    text = text.replace("\r\n", "\n")
    # 全角空格（U+3000）在中文 PDF 里很常见，换成半角
    text = text.replace("\u3000", " ")
    # 多个空格 / 制表符压成一个空格；[ \t]+ 表示"一个或多个空格或制表符"
    text = re.sub(r"[ \t]+", " ", text)
    # 三个以上连续换行压成两个，保留段落感；\n{3,} 表示"3 个及以上换行"
    text = re.sub(r"\n{3,}", "\n\n", text)
    # 去掉首尾空白
    return text.strip()


# 构造一段包含各种噪声的文本：全角空格、多余空行、制表符、Windows 换行
messy = "第一段　　内容\n\n\n\n第二段\t\t内容   \r\n"
# 用 repr 打印才能看清不可见字符的变化
print(repr(clean_text(messy)))

输出：

'第一段 内容\n\n第二段 内容'

清洗要有节制，下面这张表是分界线：

建议做	不建议做
统一换行、压缩连续空白	删掉所有换行（段落结构对第 13 章切分很重要）
去掉每页重复的页眉页脚	用正则大改正文措辞
丢弃完全空白的 Document	把表格拍平成一行（数字会串位）

原则是：

清洗只该删「格式噪声」，不该动「信息内容」。拿不准就先留着——第 13 章切分和第 15 章检索还有机会调，但删掉的内容找不回来。

9. 加载质量自检 #

RAG 领域有句老话：Garbage in, garbage out。 后面三章切分、检索再精妙，也救不了加载阶段就错了的数据。加载事故有个共同特征——不报错，很容易一路带到线上。

三类高频事故：

事故	现象	根因
空文档	`page_content` 为空或只有几个字	扫描件无文字层、空文件、`autodetect` 全部候选失败
乱码	内容是 `锟斤拷`、问号或方块	编码猜错（§4.2）
静默漏文件	Document 数明显少于文件数	`silent_errors=True` 吞掉了异常

关于「乱码」补一句：中文场景下更常见的是直接报错，不是乱码（§4.2 真相表里，三种错配组合全都抛了 RuntimeError）。GBK 和 UTF-8 的中文字节序列大多互相不合法，解码器一撞上就报错。

真正会读出乱码的，是那些「碰巧永远合法」的编码，典型是 latin-1——它把任意单字节都当有效字符，永远不报错：

# Path 用于写测试文件
from pathlib import Path
# 文本加载器
from langchain_community.document_loaders import TextLoader

# 写一个 GBK 编码的中文文件
p = Path("mojibake_demo.txt")
p.write_bytes("退换货政策".encode("gbk"))

# 用 latin-1 去读：不会抛任何异常
docs = TextLoader(str(p), encoding="latin-1").load()
# 但内容已经完全是乱码了
print("不报错，内容 =", repr(docs[0].page_content))

不报错，内容 = 'ÍË»»»õÕþ²ß'

所以千万不要为了「让它别报错」而随手改成 encoding="latin-1"——那是把响亮的失败换成静默失败，数据照样废，只是再也看不见了。

所以摄入脚本必须自带一份体检报告：

# Document 是要检查的对象类型
from langchain_core.documents import Document

# 少于这个字符数就认为「基本没内容」，需人工确认
MIN_CHARS = 10


def inspect(docs: list[Document], expected_files: int) -> None:
    """打印加载质量报告，把可疑项摆到台面上。"""
    # 第一个指标：总数，和预期对比
    print(f"Document 总数：{len(docs)}")

    # 覆盖率：有多少个源文件真的产出了内容
    # 用集合去重，因为一个文件可能产出多个 Document（PDF 按页、CSV 按行）
    loaded_files = {d.metadata.get("source") for d in docs}
    print(f"产出内容的文件数：{len(loaded_files)} / 预期 {expected_files}")

    # 空 / 过短的 Document，通常是扫描件或解析失败
    # strip() 先去掉空白，防止"只有几个换行"被当成有内容
    too_short = [d for d in docs if len(d.page_content.strip()) < MIN_CHARS]
    print(f"内容过短的 Document：{len(too_short)}")
    # 只打印前 5 条，避免问题文件太多时刷屏
    for d in too_short[:5]:
        # page 可能不存在（txt 就没有），用 '-' 占位
        print(f"  - {d.metadata.get('source')} 第 {d.metadata.get('page', '-')} 页")

    # 字符数分布：平均值过小往往意味着解析有问题
    # 先判空，否则 min/max 会在空列表上抛异常
    if docs:
        # 收集每个 Document 的字符数
        lengths = [len(d.page_content) for d in docs]
        # 三个统计量一起看，比只看平均值更能发现异常
        print(f"字符数：最小 {min(lengths)} / 平均 {sum(lengths) // len(lengths)} / 最大 {max(lengths)}")


# 演示：混入一个空 Document，看它是否被抓出来
demo_docs = [
    # 正常的一条
    Document(page_content="签收 7 日内可无理由退货。", metadata={"source": "refund.md"}),
    # 模拟扫描件产出的空 Document
    Document(page_content="", metadata={"source": "scan.pdf", "page": 0}),
]
# 预期有 2 个源文件
inspect(demo_docs, expected_files=2)

输出：

Document 总数：2
产出内容的文件数：2 / 预期 2
内容过短的 Document：1
  - scan.pdf 第 0 页
字符数：最小 0 / 平均 7 / 最大 14

这几行数字要连起来看才有意义：

产出内容的文件数：2 / 预期 2 看着完美，但只说明「两个文件都产出了 Document」，不代表都有内容。必须配合下一行一起看。
内容过短的 Document：1 才暴露真问题，还精确指出 scan.pdf 第 0 页。
字符数：最小 0 是另一个红灯。最小值是 0，就说明有完全空白的记录。

MIN_CHARS = 10 不是定律，要按资料调：FAQ 里「答：是的。」可能只有 5 个字但完全有效；制度文档某页只有 8 个字，基本就是解析出问题。建议先用 0 跑一遍，看真实字符数分布，再定阈值。

体检清单（建立知识库时逐条过一遍）：

数量对不对：Document 数与「文件数 / PDF 总页数」是否吻合
抽样读内容：随机挑 3～5 个打印前 200 字，确认不是乱码、不是页眉重复
空文档为零：有空的就去查是不是扫描件，需要 OCR
metadata 齐整：source 是相对路径、doc_id 无重复
失败清单为空：silent_errors 跳过的文件必须有记录并被处理

在灌向量库之前把这五条走完，能省掉第 15 章一半的「为什么检索不到」。

10. 企业文档摄入脚本 #

把本章内容合成一个能真正投入使用的脚本。它是 RAG 流水线的第一个工序：

docs/ 目录（PDF / Word / Markdown / CSV 混放）
        │
        ▼  ingest.py
① 按扩展名分派 loader，逐文件捕获异常
② 清洗正文（§8.2）
③ 补齐统一 metadata（§8.1）
④ 质量自检 + 失败清单（§9）
        │
        ▼
ingested.jsonl  ← 第 13 章切分的输入

10.1. 准备演示文件 #

为了让脚本开箱可跑，先用这段代码生成一个演示知识库。它故意放了一个空文件，用来验证质检能不能发现问题。

# Path 用于创建目录与文件
from pathlib import Path

# PdfWriter 用来造一个"扫描件式"的空白 PDF
from pypdf import PdfWriter

# 演示知识库根目录
root = Path("docs")
# 建一个子目录，用来验证递归扫描
(root / "policy").mkdir(parents=True, exist_ok=True)

# Markdown：制度类文档，放在子目录里，用来验证递归扫描
(root / "policy" / "refund.md").write_text(
    # 带标题的 Markdown 正文
    "# 退换货政策\n\n签收 7 日内可无理由退货。\n质量问题 15 日内可换货。\n",
    # 始终显式指定编码
    encoding="utf-8",
)

# 纯文本：FAQ
(root / "faq.txt").write_text(
    # 一问一答
    "问：发货要多久？\n答：一般 48 小时内发出。\n",
    encoding="utf-8",
)

# CSV：一行一条订单，会变成多个 Document
(root / "orders.csv").write_text(
    # 表头 + 两行数据
    "order_id,status,eta\nA1001,已发货,明天\nA1002,运输中,后天\n",
    encoding="utf-8",
)

# 故意留一个空文件，测试质检能否抓出「内容过短」
(root / "empty.txt").write_text("", encoding="utf-8")

# 故意造一个没有文字层的 PDF，模拟扫描件（§5.1）
writer = PdfWriter()
# 一个空白 A4 页
writer.add_blank_page(width=595, height=842)
# 写入 docs/ 目录
with (root / "scan.pdf").open("wb") as f:
    writer.write(f)

# 故意放一个脚本不认识的格式，测试"不支持的类型"分支
(root / "notes.xlsx").write_bytes(b"fake")

# resolve() 打印绝对路径，方便你去文件管理器里确认
print("演示知识库已生成：", root.resolve())
print("提示：把你自己的 PDF / .docx 也丢进 docs/ 就能一起摄入")

这里故意埋了三个坏文件，对应 §9 的三类事故，用来验证质检是否有效：

文件	模拟什么	期望脚本怎么处理
`empty.txt`	空文件	进「需要人工确认」，不进结果
`scan.pdf`	扫描件无文字层	同上，且能看出是第几页
`notes.xlsx`	不支持的格式	报「不支持的类型」，而不是静默忽略

如果你自己写摄入脚本，建议在测试目录长期保留这类「坏样本」。它们是回归测试：哪天重构加载逻辑却忘了异常分支，这几个文件会立刻提醒你。

10.2. ingest.py #

完整脚本，可直接运行：

"""企业文档摄入脚本：docs/ → ingested.jsonl（第 13 章的输入）"""

# 让类型注解里可以直接写 list[Document] 这种新语法（兼容旧 Python）
from __future__ import annotations

# hashlib 用于计算 doc_id
import hashlib
# json 用于写 jsonl
import json
# re 用于文本清洗
import re
# 生成带时区的摄入时间
from datetime import datetime, timezone
# Path 统一处理路径
from pathlib import Path

# 四个内置 loader，覆盖最常见的企业格式
from langchain_community.document_loaders import (
    # CSV，按行产出
    CSVLoader,
    # Word，整篇 1 个
    Docx2txtLoader,
    # PDF，按页产出
    PyPDFLoader,
    # 纯文本 / Markdown
    TextLoader,
)
# Document 是全流程的通用数据类型
from langchain_core.documents import Document

# 要扫描的知识库目录
DOCS_DIR = Path("docs")
# 摄入结果输出文件，一行一个 Document 的 JSON
OUTPUT_FILE = Path("ingested.jsonl")
# 少于这么多字符就认为「基本没内容」，进人工确认清单
MIN_CHARS = 10

# 扩展名 → loader 工厂。新增格式只要在这里加一行
# 用 lambda 而不是直接放类，是因为不同 loader 需要的参数不一样
LOADERS = {
    # txt 和 md 都是纯文本，用同一个 loader
    ".txt": lambda p: TextLoader(str(p), encoding="utf-8"),
    # Markdown 当纯文本读，结构信息留给第 13 章处理
    ".md": lambda p: TextLoader(str(p), encoding="utf-8"),
    # PDF 按页产出，不需要 encoding 参数
    ".pdf": lambda p: PyPDFLoader(str(p)),
    # Word 只支持 .docx
    ".docx": lambda p: Docx2txtLoader(str(p)),
    # CSV 按行产出，中文必须指定编码
    ".csv": lambda p: CSVLoader(str(p), encoding="utf-8"),
}


def clean_text(text: str) -> str:
    """轻量清洗：只去格式噪声，不改信息内容（§8.2）。"""
    # 统一换行符，去掉 \r
    text = text.replace("\r\n", "\n")
    # 全角空格转半角
    text = text.replace("\u3000", " ")
    # 压缩连续空格与制表符
    text = re.sub(r"[ \t]+", " ", text)
    # 三个以上换行压成两个，保留段落结构
    text = re.sub(r"\n{3,}", "\n\n", text)
    # 去掉首尾空白后返回
    return text.strip()


def enrich(doc: Document, path: Path, root: Path) -> Document:
    """清洗正文并补齐统一 metadata（§8.1）。"""
    # 相对路径 + 正斜杠，换机器也不失效
    rel = path.relative_to(root).as_posix()

    # 就地清洗正文（注意这会修改传进来的对象）
    doc.page_content = clean_text(doc.page_content)

    # doc_id 由位置信息决定，便于将来精准替换某一页 / 某一行
    raw_key = f"{rel}|{doc.metadata.get('page', '')}|{doc.metadata.get('row', '')}"
    # 取 md5 前 12 位作为短 id
    doc_id = hashlib.md5(raw_key.encode()).hexdigest()[:12]

    # update 是合并，loader 自带的 page / total_pages / row 都会保留
    doc.metadata.update(
        {
            # 相对路径，覆盖 loader 给的原始路径
            "source": rel,
            # 纯文件名，便于展示
            "file_name": path.name,
            # 扩展名去掉点
            "file_type": path.suffix.lstrip("."),
            # 稳定标识
            "doc_id": doc_id,
            # UTC 摄入时间
            "loaded_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }
    )
    return doc


def load_all(root: Path) -> tuple[list[Document], list[dict]]:
    """遍历目录逐文件加载，返回（合格文档, 待人工确认清单）。"""
    # 通过质检的 Document
    good: list[Document] = []
    # 有问题的文件清单，每项是一个 dict
    problems: list[dict] = []

    # rglob("*") 递归列出所有条目，sorted 保证每次运行顺序一致
    for path in sorted(root.rglob("*")):
        # 跳过目录，只处理文件
        if not path.is_file():
            continue

        # 按扩展名找 loader；lower() 兼容 .PDF 这种大写后缀
        factory = LOADERS.get(path.suffix.lower())
        # 不认识的格式记录下来而不是悄悄忽略
        if factory is None:
            problems.append({"file": path.name, "issue": "不支持的类型"})
            continue

        # 逐文件捕获异常：一个坏文件不该让整批摄入失败
        try:
            # 先用工厂造出 loader，再调用 load()
            raw_docs = factory(path).load()
        except Exception as exc:
            # 把异常信息一起记下来，便于事后定位
            problems.append({"file": path.name, "issue": f"加载失败：{exc}"})
            continue

        # 加载成功但没产出内容，通常是空文件
        if not raw_docs:
            problems.append({"file": path.name, "issue": "加载结果为空"})
            continue

        # 一个文件可能产出多个 Document（PDF 按页、CSV 按行）
        for doc in raw_docs:
            # 清洗 + 补 metadata
            doc = enrich(doc, path, root)
            # 过短的单独归档，多数是扫描件无文字层（§9）
            if len(doc.page_content) < MIN_CHARS:
                # 把实际字数写进清单，便于判断是完全空还是只是很短
                problems.append(
                    {"file": path.name, "issue": f"内容过短（{len(doc.page_content)} 字）"}
                )
                # 不收进结果，避免空内容污染向量库
                continue
            # 通过质检，收进结果
            good.append(doc)

    # 返回两个列表：合格的和有问题的
    return good, problems


def save_jsonl(docs: list[Document], out: Path) -> None:
    """存成 jsonl：一行一个 Document，方便第 13 章直接读取。"""
    # 用 with 保证文件正确关闭；写文件也要显式指定编码
    with out.open("w", encoding="utf-8") as f:
        # 逐个 Document 写一行
        for d in docs:
            # 只保留下游需要的两个字段
            record = {"page_content": d.page_content, "metadata": d.metadata}
            # ensure_ascii=False 让中文按原样写入，便于人工检查
            # 每行结尾加 \n，这是 jsonl 格式的要求
            f.write(json.dumps(record, ensure_ascii=False) + "\n")


def report(docs: list[Document], problems: list[dict]) -> None:
    """打印质量报告（§9）。"""
    # 总数
    print(f"共产出 {len(docs)} 个 Document")

    # 总字符数比 Document 条数更能反映"摄入了多少内容"
    total = sum(len(d.page_content) for d in docs)
    # max(..., 1) 防止一个文档都没有时除零
    print(f"总字符数：{total}，平均 {total // max(len(docs), 1)} 字/Document")

    # 按来源统计，能直观看出哪个文件贡献了多少内容
    print("\n按来源统计：")
    # key 是 source，value 是该文件产出的 Document 数
    by_source: dict[str, int] = {}
    for d in docs:
        # get(key, 0) + 1 是计数的惯用写法
        by_source[d.metadata["source"]] = by_source.get(d.metadata["source"], 0) + 1
    # 排序后输出，保证每次运行结果顺序一致，便于 diff
    for src, count in sorted(by_source.items()):
        print(f"  {src}: {count}")

    # 待人工确认清单：这一节绝不能省，它是防止「静默漏数据」的最后一道闸
    if problems:
        print("\n需要人工确认：")
        # 逐条打印文件名和问题描述
        for p in problems:
            print(f"  [{p['file']}] {p['issue']}")
    else:
        # 明确说"没问题"，比什么都不打印更让人放心
        print("\n没有发现问题文件。")


# 只有直接运行本文件时才执行，被 import 时不执行
if __name__ == "__main__":
    # 目录不存在时给出明确提示，而不是抛一个难懂的异常
    if not DOCS_DIR.exists():
        raise SystemExit(f"目录不存在：{DOCS_DIR.resolve()}，请先运行 §10.1 生成演示文件")

    # 第一步：加载，同时收集问题清单
    docs, problems = load_all(DOCS_DIR)
    # 第二步：把合格文档写成 jsonl
    save_jsonl(docs, OUTPUT_FILE)

    # 第三步：打印质量报告
    report(docs, problems)

    print(f"\n已写入 {OUTPUT_FILE}")
    # 抽一条出来看，确认内容和 metadata 都正常
    if docs:
        print("\n示例 Document：")
        # 只看前 60 字
        print(docs[0].page_content[:60])
        # 完整 metadata
        print(docs[0].metadata)

10.3. 运行结果 #

用 §10.1 生成的演示文件实际运行，输出如下：

共产出 4 个 Document
总字符数：131，平均 32 字/Document

按来源统计：
  faq.txt: 1
  orders.csv: 2
  policy/refund.md: 1

需要人工确认：
  [empty.txt] 内容过短（0 字）
  [notes.xlsx] 不支持的类型
  [scan.pdf] 内容过短（0 字）

已写入 ingested.jsonl

示例 Document：
问：发货要多久？
答：一般 48 小时内发出。
{'source': 'faq.txt', 'file_name': 'faq.txt', 'file_type': 'txt', 'doc_id': 'b3b4f59f2504', 'loaded_at': '2026-08-13T09:42:44+00:00'}

几处对照着读：

orders.csv 一个文件产出 2 个 Document（每行一个），refund.md 只产出 1 个——印证 §5.6 那张表
三个坏文件全被抓住，分类准确：两个空内容、一个不支持的格式。它们没进结果，出现在「需要人工确认」里——脚本没有假装成功，这正是 §9 要的行为
特别注意 scan.pdf：「加载成功、页数正确、元数据齐全，正文却为空」的典型。没有 MIN_CHARS 检查，它会带着空记录一路进向量库
source 是 policy/refund.md 这样的相对路径，换台机器跑也不会失效

ingested.jsonl 的第一行长这样，可以看出它是个自解释的格式，下一章直接按行 json.loads 就能用：

{"page_content": "问：发货要多久？\n答：一般 48 小时内发出。", "metadata": {"source": "faq.txt", "file_name": "faq.txt", "file_type": "txt", "doc_id": "b3b4f59f2504", "loaded_at": "2026-08-13T09:42:44+00:00"}}

选 jsonl（每行一个独立 JSON）而不是大 JSON 数组，因为可以流式逐行读，一亿行也不会撑爆内存；追加新内容只需在末尾加行，不用重写整个文件。这和 §7 的 lazy_load() 是同一思路。

10.4. 验收清单 #

递归扫描：子目录 policy/ 里的文件被加载到了
多格式共存：.md / .txt / .csv 一次摄入完成；额外丢一个 PDF 进去也能正常处理
粒度正确：CSV 按行、PDF 按页产出 Document
metadata 齐整：每条都有 source（相对路径）、file_type、doc_id、loaded_at
问题可见：空文件出现在人工确认清单里，而不是被静默丢弃
输出可用：ingested.jsonl 能被下一章直接按行读取

第 13 章会接着这份 ingested.jsonl，把每个 Document 切成更适合检索的 chunk。

11. 实用约定与坑 #

上线前当作 Code Review 清单：

约定	说明
`encoding` 显式写 `utf-8`	别依赖系统默认，中文 Windows 默认是 GBK（§4.2）
不用 `autodetect_encoding`	依赖 `chardet`、与新版不兼容、置信度低；改用候选编码列表（§4.2）
`source` 存相对路径 + 正斜杠	绝对路径含盘符，反斜杠在 JSON 里会被转义
metadata 键名全项目统一	第 15 章过滤才不会失灵；`source` 与 `file_path` 不要混用
`doc_id` 的算法全项目统一	按位置算配 upsert，按内容算配先删后插，别混用（§8.1）
加载后必做自检	数量、抽样、空文档、失败清单（§9）
逐文件捕获异常	一个坏文件不该让整批摄入中断
`silent_errors=True` 必配报告	否则漏数据毫无声息
大批量用 `lazy_load()`	内存占用与文档总量无关；但下游也得是流式的（§7）
精确数据别进向量库	几万行流水该用 SQL 工具（第 8 章），不是 RAG

坑分两类看，静默的那类才危险。

静默类（不报错，但数据已经错了）：

现象	原因	处理
PDF 加载出来全是空的	扫描件没有文字层	需要 OCR，`PyPDFLoader` 无能为力（§5.1）
GBK 文件在你机器上读得好好的，换机器就崩	没传 `encoding`，靠系统默认碰巧对上了	任何时候都显式写 `encoding=`（§4.2）
Document 数量远少于文件数	`silent_errors` 吞了异常	逐文件捕获异常并输出失败清单
`autodetect_encoding` 读出空内容	所有候选编码都解码失败，函数返回空字符串而不报错	改用自己写的候选列表，全失败时明确抛错（§4.2）
检索结果没法标出处	加载时没存 `source` / `page`	回到摄入阶段补 metadata 重跑
Excel 公式列读出来是 `=SUM(...)`	忘了 `data_only=True`	加上这个参数（§5.4）
Excel 少了一半数据	只读了第一个工作表	遍历 `wb.worksheets`（§5.4）
`lazy_load()` 的循环一次都没执行	生成器已被 `len(list(...))` 消耗完	单次遍历时自行计数（§7）
向量库里同一份文件有新旧两份内容	`doc_id` 算法变了，旧记录成了孤儿	统一 `doc_id` 算法；更新前先按 `source` 删净

响亮类（会抛异常，好定位）：

现象	原因	处理
`RuntimeError: Error loading xxx`	编码不匹配；真正原因在 `__cause__` 里	打印 `exc.__cause__` 看是哪个 codec（§4.2）
`ModuleNotFoundError: No module named 'chardet'`	开了 `autodetect_encoding` 但没装 `chardet`	别用这个参数（§4.2）
`TypeError: FileEncoding.__new__() got an unexpected keyword argument 'mime_type'`	`chardet` 版本太新，与 `langchain_community` 不兼容	同上，别用这个参数（§4.2）
`.doc` 加载失败	`Docx2txtLoader` 只支持 `.docx`	用 Word 另存为 `.docx`
`.xls` 加载失败	`openpyxl` 只支持 `.xlsx`	用 Excel 另存为 `.xlsx`
`ImportError: jq package not found`	`JSONLoader` 依赖 `jq`，Windows 上常装不上	用标准库 `json` 自己读（§12 末尾）

设计类（不是 bug，是要想清楚的取舍）：

问题	取舍
网页正文里混着导航和广告	静态抓取无法区分主体内容；清洗，或改由业务方导出文件
网页抓到空内容	内容由前端 JS 渲染，静态抓取拿不到，需专门抓取服务
Word 没有页码，溯源不够精确	转成 PDF 摄入，或用「第 N 段」代替页码（§5.2）
合并单元格读出来残缺	请业务方拉平成规范表格，比在代码里猜可靠（§5.4）
该清洗到什么程度	只删格式噪声，不动信息内容；拿不准就先留着（§8.2）

口诀：

Loader 管「读得全、读得对、记得住出处」；读进来的质量，就是 RAG 的上限。

12. 练习 #

机制验证类（跑一遍就能纠正直觉，建议全做）：

默认编码是什么：在自己机器上打印 locale.getpreferredencoding(False)，确认它不是 utf-8；再跑 §4.2 那张六组合真相表，亲眼看到「GBK 文件不传 encoding 反而对」。
挖出真正的异常：故意用错误编码加载文件，打印 exc.__cause__，确认能看到 'gbk' codec can't decode byte ... 这类具体信息。
扫描件的静默失败：照 §5.1 用 pypdf 造一个纯空白页 PDF 加载，确认不报错且 page_content 为空。
glob 传列表：给 DirectoryLoader 传 glob=["**/*.txt", "**/*.md"]，确认一次就能加载两种扩展名。
生成器只能遍历一次：先 len(list(loader.lazy_load())) 再用 for 遍历同一个生成器，确认循环体一次都没执行。

能力构建类：

换格式：把自己的一份 PDF 放进 docs/，跑 §10.2 脚本，确认 Document 数等于 PDF 页数。
编码实验：用记事本把一个中文 txt 另存为 ANSI（即 GBK），跑脚本观察报错，再改 encoding 修好它。
加一种格式：给 LOADERS 加上 .xlsx，直接复用 §5.4 的 load_excel（注意它返回的是 list[Document]，要包一层让接口和其他 loader 一致），并补一条验收用例。
metadata 扩展：在 enrich 里增加 department 字段，值取自一级子目录名（如 policy/ → policy），为第 15 章的权限过滤打底。
lazy 改造：把 load_all 改成生成器版本（yield 每个 Document），对比摄入 100 个文件时的内存占用。
（扩展）增量摄入：给脚本加上「文件 mtime 没变就跳过」的逻辑，避免每次全量重跑。

关于 JSON 格式的提醒： 你可能想用官方 JSONLoader，但它依赖 jq，Windows 上通常装不上（缺预编译版本，需要 C 编译环境）。不装就会报：

ImportError: jq package not found, please install it with `pip install jq`

对绝大多数场景，用标准库 json 自己读更省事，也更好控制：

# 标准库，无需任何额外依赖
import json
# Path 用于读文件
from pathlib import Path
# Document 是目标类型
from langchain_core.documents import Document


def load_json_faq(path: Path) -> list[Document]:
    """读形如 [{"q": ..., "a": ...}, ...] 的 JSON，一条问答一个 Document。"""
    # 显式指定编码后读取并解析
    items = json.loads(path.read_text(encoding="utf-8"))
    # 收集结果
    out: list[Document] = []
    # enumerate 顺便拿到序号，用作行号
    for i, item in enumerate(items):
        out.append(
            Document(
                # 把问和答拼成一段自然文本，便于向量化
                page_content=f"问：{item['q']}\n答：{item['a']}",
                # 记下来源和序号，保持和其他 loader 一致的 metadata 风格
                metadata={"source": path.as_posix(), "row": i},
            )
        )
    # 返回和其他 loader 相同的类型：list[Document]
    return out

这也是本章想传达的态度：loader 只是个约定（读进来 → 返回 list[Document]），不必死守官方实现。自己写十几行，往往比引入一个装不上的依赖更划算。

13. 本章小结 #

本章看起来只是「读文件」，却是 RAG 里返工代价最高的一环：切分参数随时能调、向量库可以重建，加载时丢掉的内容和出处，后面无论如何补不回来。

记忆 ≠ 知识：第 11 章管「这场对话说过什么」，第 12～15 章管「公司资料里写了什么」。
Document 是 RAG 全链路的通用货币：page_content 决定答得准不准，metadata 决定答案敢不敢用。它还有 id / type 两个字段，id 关系到第 14 章能否增量更新。
每个 loader 的关键差异是产出粒度：TextLoader / Docx2txtLoader 一个文件 1 个，PyPDFLoader 按页，CSVLoader 与 Excel 按行。
中文 Windows 默认编码是 GBK（cp936），不是 UTF-8，必须显式写 encoding="utf-8"；「不传也能读对」只是碰巧匹配，换机器就崩。
autodetect_encoding 不值得用：依赖 chardet、与新版 chardet 不兼容（TypeError）、中文短文本置信度极低、全部失败时还会静默产出空文档。用自己写的候选编码列表替代。
DirectoryLoader 的 glob 可以传列表，但 loader_cls 只能一种；多格式混放时，按扩展名手写分派更可控，能给出文件级成败清单。
load() 调试友好，lazy_load() 省内存（生成器只能遍历一次，下游也得流式才有收益）。
metadata 要在摄入阶段统一补齐：source（相对路径 + 正斜杠）、file_type、doc_id、loaded_at、page / row；键名和 doc_id 算法全项目统一。
加载完必须自检：数量、抽样、空文档、失败清单——加载事故大多不报错，扫描件就是典型：页数对、元数据全、正文空。
本章产出：企业文档摄入脚本，输出 ingested.jsonl。

一句话记住这章的验收方法：

别问「脚本跑通了吗」，问「有几个文件没进来、为什么」。 加载阶段最大的风险不是崩，而是安静地少读了一半。

下一章：Text Splitters 文本切分——把这些 Document 切成大小合适、语义完整的 chunk，并解决「切在句子中间导致检索答非所问」的问题。

导航菜单

1. 本章目标 #

2. 为什么需要 Document Loaders #

2.1. 「用 Python 直接读文件」不行吗 #

2.2. 统一抽象带来什么 #

3. Document：RAG 的最小数据单位 #

3.1. 两个主字段，外加两个容易被忽略的 #

3.2. metadata 决定了 RAG 能不能「说出处」 #

4. 第一个加载器：TextLoader #

4.1. 最小可运行示例 #

4.2. Windows 上的中文编码坑 #

5. 常用文件类型怎么加载 #

5.1. PDF：PyPDFLoader #

5.2. Word：Docx2txtLoader #

5.3. CSV：CSVLoader #

5.4. Excel：.xlsx 怎么办 #

5.5. 网页：WebBaseLoader #

5.6. 加载器与依赖速查 #

6. 批量加载整个目录 #

6.1. DirectoryLoader #

6.2. 为什么本章实战不用它 #

7. load() 与 lazy_load() #

8. 元数据治理 #

8.1. 该补哪些字段 #

8.2. 文本清洗的边界 #

9. 加载质量自检 #

10. 企业文档摄入脚本 #

10.1. 准备演示文件 #

10.2. ingest.py #

10.3. 运行结果 #

10.4. 验收清单 #

11. 实用约定与坑 #

12. 练习 #

13. 本章小结 #