RAG 系统链路解析与 Document Loaders 多案例实战

2.2 架构核心优势

1. 知识实时更新：无需重新训练 LLM，只需更新向量库中的数据（如新增 2024 年 LangChain 新特性），即可让 LLM 获取最新知识。
2. 结果可追溯：生成的回答可对应到具体的原始数据（如'该结论来自 XX PDF 第 5 页'），便于验证准确性。
3. 降低训练成本：无需训练千亿参数的大模型，用中小模型（如 7B 的 Llama3）+ RAG，即可实现高精度回答。

三、Document Loaders 核心概念：LangChain 如何统一数据入口？

LangChain 为了解决'不同数据源适配'问题，设计了统一的加载器接口，无论原始数据格式如何，最终都能转为 Document 对象。

3.1 LangChain Loaders 的核心设计：BaseLoader 接口

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。

3.2 Document 对象：数据的'统一载体'

所有 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 页
)

3.3 Loaders 分类：覆盖所有常见数据源

LangChain 的 langchain_community.document_loaders 模块提供了上百种 Loader，按数据源类型可分为三大类：

四、Document Loaders 多案例实战：从理论到代码

光说不练假把式，下面通过 3 个最常用的 Loader 案例（TextLoader、CSVLoader、JSONLoader），带你掌握实战技巧，所有代码可直接复制运行。

前置准备：安装依赖

首先安装 LangChain 及 Loader 所需的额外依赖：

# 核心依赖：LangChain
pip install langchain langchain-community

# JSONLoader 需要 jq 命令行工具 (Linux: sudo apt install jq / Mac: brew install jq)

案例 1：TextLoader——加载纯文本文件

纯文本文件（.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 字符

案例 2：CSVLoader——加载 Excel 表格数据

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 开始）

关键说明

• 自定义分隔符：如果 CSV 是用制表符（\t）分隔，需将 delimiter 设为 "\t"。
• 字段筛选：如果只需加载'产品名称'和'销售数量'，可在 csv_args 中添加 "usecols": ["产品名称", "销售数量"]。

案例 3：JSONLoader——加载 JSON 文件（复杂结构解析）

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 语法常用模式

jq_schema 是 JSONLoader 的核心，用于定义'从 JSON 中提取哪些数据'，常用语法如下：

五、Loaders 实战常见问题与解决方案

实战中难免遇到各种问题，这里整理 3 个高频问题及解决方案：

六、总结

本文我们掌握了 RAG 系统的核心链路，以及作为'数据入口'的 Document Loaders 实战技巧。