PythonAI算法
RAG 系统链路解析与 Document Loaders 多案例实战
在之前的 AI 大模型系列中，我们从基础概念逐步深入到实战，并在前期初步提及 RAG 技术的重要性。 RAG（Retrieval-Augmented Generation，检索增强生成）是解决大模型'幻觉'的核心方案——通过'检索外部知识'为 LLM 提供精准上下文，让生成结果更可靠。但 RAG 系统并非单一技术，而是一条完整的'数据处理→存储→检索→生成'链路，其中**数据加载（Document…
在之前的 AI 大模型系列中，我们从基础概念逐步深入到实战，并在前期初步提及 RAG 技术的重要性。 RAG（Retrieval-Augmented Generation，检索增强生成）是解决大模型'幻觉'的核心方案——通过'检索外部知识'为 LLM 提供精准上下文，让生成结果更可靠。但 RAG 系统并非单一技术，而是一条完整的'数据处理→存储→检索→生成'链路，其中**数据加载（Document…
在之前的 AI 大模型系列中，我们从基础概念逐步深入到实战，并在前期初步提及 RAG 技术的重要性。
RAG（Retrieval-Augmented Generation，检索增强生成）是解决大模型'幻觉'的核心方案——通过'检索外部知识'为 LLM 提供精准上下文，让生成结果更可靠。但 RAG 系统并非单一技术，而是一条完整的'数据处理→存储→检索→生成'链路，其中数据加载（Document Loaders）是整个链路的'入口' ——没有高质量的原始数据加载，后续的向量化、检索都无从谈起。
本文将聚焦 RAG 系统的完整链路解析，并通过多案例实战，带你掌握 LangChain 中 Document Loaders 的核心用法，为后续 RAG 项目打下坚实基础。
RAG 的本质是'让 LLM 带着外部知识回答问题'，其核心链路可拆解为数据准备'和'检索生成'两大阶段，每个阶段包含多个关键技术环节。理解这条链路，是后续实战的前提。
原始数据需要经过一系列处理，才能最终为 LLM 提供有效支撑，完整流程如下：
链路中每个环节都有明确的职责，缺一不可：
Document 对象，解决'数据入口不统一'问题。Document 进行清洗（去除乱码、空白行）和分块（如按 500 字符/块分割），避免长文本向量化时的语义失真。text-embedding-3-small、开源的 BERT，负责将文本转为高维向量（如 1536 维），让计算机能通过'向量距离'判断文本相似度。传统 LLM 直接生成回答时，依赖的是训练时的'记忆'，但存在'知识过时''虚构信息'问题；而 RAG 通过'检索 + 生成'的交互模式，让 LLM'带着参考资料答题'，从根本上减少幻觉。
LangChain 为了解决'不同数据源适配'问题，设计了统一的加载器接口，无论原始数据格式如何，最终都能转为 Document 对象。
LangChain 所有加载器都继承自抽象类 BaseLoader，确保统一的调用方式。核心接口定义如下（简化版）：
from abc import ABC, abstractmethod
from typing import Iterable
from langchain_core.documents import Document
class BaseLoader(ABC):
# 懒加载（生成器模式，避免大文件加载时内存溢出）
@abstractmethod
def lazy_load(self) -> Iterable[Document]:
pass
# 直接加载（返回 Document 列表，内部调用 lazy_load）
def load(self) -> list[Document]:
return list(self.lazy_load())
load() 方法：最常用，直接返回所有 Document 组成的列表，适合小文件。lazy_load() 方法：生成器模式，逐个返回 Document，适合大文件（如 1000 页的 PDF），避免一次性加载到内存导致 OOM。所有 Loader 最终输出的都是 Document 对象，其结构非常简单，包含两个核心字段：
class Document(BaseMedia):
page_content: str # 文本内容（如 PDF 某一页的文字、CSV 某一行的数据）
metadata: dict # 元数据（如来源路径、页码、行号，便于后续追溯）
type: Literal["Document"] = "Document"
示例 Document 对象：
Document(
page_content="LangChain Loaders 支持 PDF、CSV、网页等多种数据源",
metadata={"source": "data/rag_intro.pdf", "page": 3} # 来源 PDF 第 3 页
)
LangChain 的 langchain_community.document_loaders 模块提供了上百种 Loader，按数据源类型可分为三大类：
| 分类 | Loader 类型 | 功能描述 | 适用场景 |
|---|---|---|---|
| 文件加载器 | TextLoader | 加载纯文本文件（.txt） | 本地日志文件、纯文本笔记 |
| PyPDFLoader | 加载 PDF 文件，支持提取页码元数据 | 技术文档、论文 | |
| Docx2txtLoader | 加载 Word 文档（.docx） | 工作报告、需求文档 | |
| CSVLoader | 加载 CSV 文件，按行生成 Document | 数据报表、用户列表 | |
| 网页加载器 | WebBaseLoader | 抓取静态网页文本（无需 JS 渲染） | 博客文章、百科页面 |
| SeleniumURLLoader | 加载动态网页（需 JS 渲染，如 Vue/React 页面） | 电商商品页、登录后页面 | |
| 数据库加载器 | SQLDatabaseLoader | 执行 SQL 查询，加载结果为 Document | MySQL、PostgreSQL 等关系库 |
| MongoDBLoader | 从 MongoDB 集合中加载文档 | NoSQL 数据库数据 |
光说不练假把式，下面通过 3 个最常用的 Loader 案例（TextLoader、CSVLoader、JSONLoader），带你掌握实战技巧，所有代码可直接复制运行。
首先安装 LangChain 及 Loader 所需的额外依赖：
# 核心依赖：LangChain
pip install langchain langchain-community
# JSONLoader 需要 jq 命令行工具 (Linux: sudo apt install jq / Mac: brew install jq)
纯文本文件（.txt）是最基础的数据源，TextLoader 支持自定义编码，解决中文乱码问题。
from langchain_community.document_loaders import TextLoader
# 1. 初始化 Loader（处理中文文件时，建议开启 autodetect_encoding）
loader = TextLoader(
file_path="data/test.txt", # 本地文本文件路径
encoding="utf-8", # 编码格式（默认 utf-8）
autodetect_encoding=True # 自动检测编码（解决中文乱码）
)
# 2. 加载数据（返回 Document 列表）
documents = loader.load()
# 3. 查看结果
print(f"加载的 Document 数量：{len(documents)}") # 输出：1（纯文本文件默认 1 个 Document）
print(f"\n前 100 个字符内容：{documents[0].page_content[:100]}")
print(f"\n元数据：{documents[0].metadata}") # 输出：{'source': 'data/test.txt'}
test.txt 是 GBK 编码，直接用 encoding="utf-8" 会乱码，开启 autodetect_encoding=True 后，Loader 会自动识别编码。test.txt 有 10 万行，用 lazy_load() 逐个获取 Document：for doc in loader.lazy_load():
print(doc.page_content[:50]) # 逐个打印每行前 50 字符
CSV 文件（.csv）常用于存储结构化数据（如销售报表），CSVLoader 支持按行生成 Document，且可指定字段名。
假设 data/sales.csv 文件内容如下（销售数据）：
产品名称，销售数量，客户名称
手机，100，张三
电脑，50，李四
平板，30，王五
from langchain_community.document_loaders import CSVLoader
# 1. 初始化 Loader（指定 CSV 分隔符和字段名）
loader = CSVLoader(
file_path="data/sales.csv",
csv_args={
"delimiter": ",", # CSV 分隔符（默认逗号，Excel 导出的 CSV 常用）
"fieldnames": ["产品名称", "销售数量", "客户名称"] # 字段名（可选，默认用首行）
},
source_column="产品名称" # 将'产品名称'作为 metadata 的 source 字段（便于追溯）
)
# 2. 加载数据
documents = loader.load()
# 3. 查看结果
print(f"加载的 Document 数量：{len(documents)}") # 输出：3（3 行数据，不含表头）
print(f"\n第 1 条数据内容：{documents[0].page_content}") # 输出：产品名称：手机，销售数量：100，客户名称：张三
print(f"\n第 1 条数据元数据：{documents[0].metadata}") # 输出：{'source': '手机', 'row': 0}（row 为行号，从 0 开始）
delimiter 设为 "\t"。csv_args 中添加 "usecols": ["产品名称", "销售数量"]。JSON 文件常用于存储半结构化数据（如接口返回、日志），JSONLoader 的核心是通过 jq_schema 指定数据提取规则，支持复杂结构解析。
假设 data/articles.json 文件内容如下（文章列表）：
{
"status": "success",
"data": {
"articles": [
{"id": 1, "title": "RAG 系统链路解析", "content": "RAG 包含加载、预处理、向量化、存储、检索五大环节", "author": "工藤学编程", "date": "2024-05-20"},
{"id": 2, "title": "Document Loaders 实战", "content": "TextLoader 适合纯文本，CSVLoader 适合表格数据", "author": "工藤学编程", "date": "2024-05-21"}
]
}
}
from langchain_community.document_loaders import JSONLoader
# 1. 初始化 Loader（核心：jq_schema 指定提取规则）
loader = JSONLoader(
file_path="data/articles.json",
jq_schema=".data.articles[]", # 提取 data.articles 数组中的每个元素
content_key="content", # 将'content'字段作为 Document 的 page_content
metadata_func=lambda record: {
"article_id": record["id"],
"title": record["title"],
"author": record["author"],
"date": record["date"]
}
)
# 2. 加载数据
documents = loader.load()
# 3. 查看结果
print(f"加载的 Document 数量：{len(documents)}") # 输出：2（articles 数组有 2 个元素）
print(f"\n第 1 篇文章内容：{documents[0].page_content}") # 输出：RAG 包含加载、预处理、向量化、存储、检索五大环节
print(f"\n第 1 篇文章元数据：{documents[0].metadata}") # 输出：{'article_id': 1, 'title': 'RAG 系统链路解析', 'author': '工藤学编程', 'date': '2024-05-20'}
jq_schema 是 JSONLoader 的核心，用于定义'从 JSON 中提取哪些数据'，常用语法如下：
| 需求场景 | jq_schema 示例 | 说明 |
|---|---|---|
| 提取根级数组 | .[] | 适合 JSON 本身是数组（如 [{"id":1},{"id":2}]） |
| 提取嵌套数组 | .data.articles[] | 提取深层数组元素（如本文案例） |
| 条件过滤 | `.data.articles[] | select(.id > 1)` |
| 多字段合并 | `.data.articles[] | {c: .content, t: .title}` |
实战中难免遇到各种问题，这里整理 3 个高频问题及解决方案：
| 常见问题 | 原因分析 | 解决方案 |
|---|---|---|
| 中文乱码（如'��'） | 文本编码与 Loader 指定的 encoding 不匹配 | 1. 开启 autodetect_encoding=True；2. 手动指定编码（如 encoding="gbk"） |
| JSONLoader 报错'jq: error' | jq_schema 语法错误，或 JSON 结构与预期不符 | 1. 用 jq 在线验证工具测试语法；2. 打印原始 JSON 确认结构 |
本文我们掌握了 RAG 系统的核心链路，以及作为'数据入口'的 Document Loaders 实战技巧。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online