LangChain 07：文档加载器

发表于 分类于
覆盖文本、CSV、JSON、PDF、Word、Markdown、HTML 与目录批量加载，统一产出 Document，避开已日落的 langchain-community。

前言

RAG 流水线的第一步是把磁盘上的 txt、CSV、JSON、PDF、Word 等读成统一结构。
LangChain 用 Document（page_content + metadata）承接后续切分与向量化。
旧教程常见的 langchain_community.document_loaders 会触发 sunset 弃用警告；简单类型不必再依赖它。
本文按常见格式给出可运行写法：langchain-core + 标准库 / 独立包，接口尽量贴近旧 Loader（load / lazy_load）。
目录约定：示例脚本放在代码子目录（如 examples/），样例文件放在其上级的 data/；加载路径写成 ../data/xxx。
整链概要见《LangChain 06：RAG 概要》；切分见《LangChain 08：文本切分器》。
下文需要 Python 3.12+，依赖用 uv 管理。
在代码目录下执行 uv run python xxx.py 即可（相对路径相对当前工作目录）。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明基础依赖（版本请按项目实际调整）。
JSON、PDF、Word、网页等额外包写在对应示例前的安装代码块中。

1
2
3
4
uv init langchain-document-loaders
cd langchain-document-loaders
uv venv --python 3.12
uv add langchain-core rich

若已在空目录内初始化，也可写 uv init --name langchain-document-loaders，再执行 uv venv 与 uv add。

后续运行示例脚本时，可直接 uv run python demo.py，一般不必手动 activate 虚拟环境。
若习惯激活，Windows 使用 .venv\Scripts\activate，macOS / Linux 使用 source .venv/bin/activate。

rich 用于查看 Document 的 page_content 与 metadata。
示例里常用 from rich import print as rprint。

实现

Document 结构

无论哪种格式，加载结果都应是 Document 列表（或可迭代对象）。

  1. page_content：真正进入切分与嵌入的正文（字符串）。
  2. metadata：来源路径、行号、页码等字典，便于引用与过滤。

下面手写一条确认字段含义。

1
2
3
4
5
6
7
8
9
10
from langchain_core.documents import Document
from rich import print as rprint

doc = Document(
    page_content="LangChain 用 Document 统一表示一段可检索文本。",
    metadata={"source": "demo.txt", "page": 1},
)
rprint(type(doc))
rprint(doc.metadata)
rprint(doc.page_content)

需要与生态对齐时，可继承 BaseLoader 并实现 lazy_load()；load() 会基于它收集全部结果。

文本文件

纯文本注意 编码：UTF-8 与 GBK 文件要用对应 encoding，否则乱码或报错。

创建 hello-utf8.txt

在上级目录的 ../data/hello-utf8.txt 中写入如下内容（UTF-8 保存）：

1
LangChain 是一个用于构建基于大语言模型（LLM）应用的开发框架。

加载文本

下面用小型 BaseLoader 封装，接口与旧 TextLoader 类似。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from collections.abc import Iterator
from pathlib import Path

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from rich import print as rprint


class TextLoader(BaseLoader):
    def __init__(self, file_path: str | Path, encoding: str = "utf-8") -> None:
        self.file_path = Path(file_path)
        self.encoding = encoding

    def lazy_load(self) -> Iterator[Document]:
        text = self.file_path.read_text(encoding=self.encoding)
        yield Document(
            page_content=text,
            metadata={"source": str(self.file_path)},
        )


docs = TextLoader("../data/hello-utf8.txt", encoding="utf-8").load()
rprint(docs[0].metadata)
rprint(docs[0].page_content)

若文件是 GBK，把 encoding="gbk" 即可，其它用法相同。

CSV

表格通常 一行一条 Document：各列拼进 page_content，metadata 带 source 与 row。

创建 articles.csv

在上级目录的 ../data/articles.csv 中写入如下内容：

1
2
3
id,title,content,author
1,Introduction to Python,Python is a popular programming language.,John Doe
2,Data Science Basics,Data science involves statistics and machine learning.,Jane Smith

加载 CSV

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import csv
from pathlib import Path

from langchain_core.documents import Document
from rich import print as rprint

path = Path("../data/articles.csv")
docs: list[Document] = []

with path.open(encoding="utf-8", newline="") as f:
    for i, row in enumerate(csv.DictReader(f)):
        page_content = "\n".join(f"{k}: {v}" for k, v in row.items())
        docs.append(
            Document(
                page_content=page_content,
                metadata={"source": str(path), "row": i},
            )
        )

rprint(docs)

大表行数多时，切分往往可以更宽松，甚至跳过切分直接嵌入。

JSON

业务接口、会话日志大量是 JSON；关键是用 jq 表达式 抽出要进 page_content 的字段。
常见 schema 对照如下。

  1. [{"text": ...}, ...] → .[].text
  2. {"key": [{"text": ...}, ...]} → .key[].text
  3. ["...", "..."] → .[]

先安装 jq（Python 绑定）。

1
uv add jq

创建 chat.json

在上级目录的 ../data/chat.json 中写入如下内容：

1
2
3
4
5
6
7
{
  "messages": [
    {"content": "Hello, how are you today?"},
    {"content": "I am doing well, thanks!"},
    {"content": "Would you like to meet for lunch?"}
  ]
}

加载 JSON

用 jq_schema=".messages[].content" 抽出每条消息正文（等价于旧 JSONLoader 的常见用法）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import json
from pathlib import Path

import jq
from langchain_core.documents import Document
from rich import print as rprint

path = Path("../data/chat.json")
data = json.loads(path.read_text(encoding="utf-8"))
contents = jq.compile(".messages[].content").input_value(data).all()

docs = [
    Document(
        page_content=str(content),
        metadata={"source": str(path), "seq_num": i},
    )
    for i, content in enumerate(contents, start=1)
]
rprint(docs)

若要拼多个字段，可在 jq 里构造对象，例如：

1
.data.items[] | { author, created_at, content: (.title + "\n" + .content) }

此时得到的是结构化对象；写入 page_content 前可 json.dumps(..., ensure_ascii=False)，并设语义上等价于旧参数 text_content=False。

PDF

PDF 有文本版、扫描版、多栏布局等差异；简单文本版用 pypdf 按页提取即可。
复杂版式（公式、表格、OCR）可再评估 MinerU、Docling 等专用方案。

1
uv add pypdf

准备 sample.pdf

将任意可检索的文本型 PDF 保存为 ../data/sample.pdf（二进制文件，此处不贴正文）。

加载 PDF

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from pathlib import Path

from langchain_core.documents import Document
from pypdf import PdfReader
from rich import print as rprint

pdf_path = Path("../data/sample.pdf")
if not pdf_path.exists():
    raise SystemExit("请先把 PDF 放到 ../data/sample.pdf 再运行。")

reader = PdfReader(str(pdf_path))
docs = [
    Document(
        page_content=page.extract_text() or "",
        metadata={
            "source": str(pdf_path),
            "page": i,
            "total_pages": len(reader.pages),
        },
    )
    for i, page in enumerate(reader.pages)
]

rprint(len(docs))
rprint(docs[0].metadata)
rprint(docs[0].page_content[:400])

一页一个 Document 便于按页引用；空页可在入库前过滤。
扫描件或强版式需求时，可使用 MinerU 等在线/本地解析服务，把结果 Markdown 再转成 Document。

Word

.docx 可用 python-docx 抽取段落文本，拼成一条或多条 Document。

1
uv add python-docx

准备 report.docx

将待测 Word 保存为 ../data/report.docx（二进制文件，此处不贴正文）。

加载 Word

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from pathlib import Path

from docx import Document as DocxDocument
from langchain_core.documents import Document
from rich import print as rprint

path = Path("../data/report.docx")
if not path.exists():
    raise SystemExit("请先把 Word 放到 ../data/report.docx 再运行。")

docx = DocxDocument(str(path))
text = "\n".join(p.text for p in docx.paragraphs if p.text.strip())
docs = [Document(page_content=text, metadata={"source": str(path)})]

rprint(len(docs))
rprint(docs[0].page_content[:500])

若需按标题拆成多条，可在遍历段落时根据样式名（如 Heading 1）分桶，再分别构造 Document。

Markdown

Markdown 可整文件读成单条，也可按标题先拆。

创建 nlp.md

在上级目录的 ../data/nlp.md 中写入如下内容：

1
2
3
4
5
6
7
8
9
10
11
# 自然语言处理技术文档

本文档用于测试 Markdown 加载。

## 第一章：简介

自然语言处理(NLP)是人工智能的重要分支。

## 第二章：关键技术

BERT、GPT、T5 等预训练模型。

整文件加载

类似旧 Loader 的 mode="single"。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pathlib import Path

from langchain_core.documents import Document
from rich import print as rprint

path = Path("../data/nlp.md")
docs = [
    Document(
        page_content=path.read_text(encoding="utf-8"),
        metadata={"source": str(path)},
    )
]
rprint(len(docs))
rprint(docs[0].page_content[:200])

按标题拆分

需要「按标题拆成多条」时，直接用切分器包（不必先上 Unstructured）。

1
uv add langchain-text-splitters

仍读取 ../data/nlp.md。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pathlib import Path

from langchain_text_splitters import MarkdownHeaderTextSplitter
from rich import print as rprint

md = Path("../data/nlp.md").read_text(encoding="utf-8")
splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "h1"), ("##", "h2")],
)
docs = splitter.split_text(md)

rprint(len(docs))
for d in docs:
    rprint(d.metadata, d.page_content[:80])

HTML

本地 HTML 或抓取结果可用 BeautifulSoup 按标签拆成多条，metadata 里记下标签名，便于过滤。

1
uv add beautifulsoup4

创建 rag.html

在上级目录的 ../data/rag.html 中写入如下内容：

1
2
3
4
5
6
7
8
<html>
<body>
  <h1>RAG 概要</h1>
  <p>检索增强生成把检索与生成结合。</p>
  <h2>步骤</h2>
  <p>加载、切分、向量化、检索、生成。</p>
</body>
</html>

加载 HTML

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from pathlib import Path

from bs4 import BeautifulSoup
from langchain_core.documents import Document
from rich import print as rprint

path = Path("../data/rag.html")
soup = BeautifulSoup(path.read_text(encoding="utf-8"), "html.parser")

docs: list[Document] = []
for tag in soup.find_all(["h1", "h2", "h3", "p"]):
    text = tag.get_text(strip=True)
    if not text:
        continue
    docs.append(
        Document(
            page_content=text,
            metadata={"source": str(path), "tag": tag.name},
        )
    )

rprint(len(docs))
for d in docs:
    rprint(d.metadata, d.page_content)

在线页面可先用 httpx 下载 HTML，再套同一解析逻辑（遵守站点条款）。

1
uv add httpx

目录批量

批量入库用 Path.glob：按后缀选解析方式，适合首批灌库。

创建 kb 样例

在上级目录的 ../data/kb/ 下分别创建下列文件。

../data/kb/a.txt：

1
知识库文档 A：关于文档加载器。

../data/kb/b.txt：

1
知识库文档 B：关于文本切分器。

../data/kb/demo.py：

1
print("hello")

批量加载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from pathlib import Path

from langchain_core.documents import Document
from rich import print as rprint

root = Path("../data/kb")
docs: list[Document] = []

for path in sorted(root.glob("**/*")):
    if not path.is_file():
        continue
    if path.suffix.lower() in {".txt", ".md", ".py"}:
        docs.append(
            Document(
                page_content=path.read_text(encoding="utf-8"),
                metadata={"source": str(path)},
            )
        )

rprint(len(docs))
for d in docs:
    rprint(d.metadata["source"], "=>", d.page_content[:40])

文件很多时可对 glob 结果做线程池并发读取；混排 PDF/docx 时按后缀分支调用上文各节解析函数。

懒加载

大文件不要一次 load() 进内存，用 lazy_load() 边读边切分、边写入向量库。
请先按「文本文件」小节创建好 ../data/hello-utf8.txt。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from collections.abc import Iterator
from pathlib import Path

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from rich import print as rprint


class TextLoader(BaseLoader):
    def __init__(self, file_path: str | Path, encoding: str = "utf-8") -> None:
        self.file_path = Path(file_path)
        self.encoding = encoding

    def lazy_load(self) -> Iterator[Document]:
        text = self.file_path.read_text(encoding=self.encoding)
        yield Document(
            page_content=text,
            metadata={"source": str(self.file_path)},
        )


for i, doc in enumerate(TextLoader("../data/hello-utf8.txt").lazy_load()):
    rprint(i, doc.metadata.get("source"), len(doc.page_content))

验证

  1. 按各节创建样例到上级 data/，在代码目录下跑对应加载脚本。
  2. 跑「CSV」，行数与数据行一致，且带 row。
  3. 跑「JSON」，jq 抽出的消息条数与样例一致。
  4. （可选）放入真实 PDF / Word 后检查页数或正文预览。
  5. 跑「目录批量」，应覆盖 ../data/kb 下文本与 .py。

总结

  1. 加载目标是统一产出 Document，而不是直接调模型。
  2. 避开已日落的 langchain-community；用 langchain-core + 标准库 / jq / pypdf / python-docx 等独立包。
  3. 样例在上级 data/，加载用相对路径如 ../data/hello-utf8.txt。
  4. JSON 重点在 jq 表达式；PDF 简单场景用 pypdf，复杂版式再上 MinerU 等。
  5. 下一步用《LangChain 08：文本切分器》控制块大小，再进入嵌入与入库专文。