LangChain篇-自定义RAG加载器

SuifengFlying

公告

欢迎来到我的博客，这里记录生活与技术。

了解更多

1792 字

9 分钟

LangChain篇-自定义RAG加载器

2026-07-29

LangChain

/

RAG

一、概述#

基于 LLM 的应用程序通常涉及从数据库或文件（如 PDF）中提取数据，并将其转换为LLM可以利用的格式。在 LangChain 中，这通常涉及创建Document对象，该对象封装了提取的文本（page_content）以及元数据 - 包含有关文档的详细信息的字典，例如作者姓名或出版日期。Document 对象通常被格式化为提示，然后输入 LLM，以便 LLM 可以使用 Document 中的信息生成所需的响应（例如，对文档进行摘要）。Documents 可以立即使用，也可以索引到向量存储中以供将来检索和使用。文档加载的主要抽象为：

下面将演示如何编写自定义文档加载和文件解析逻辑；具体而言，我们将看到如何：

通过从 BaseLoader 进行子类化来创建标准文档加载器。
使用 BaseBlobParser 创建解析器，并将其与 Blob 和 BlobLoaders 结合使用。这在处理文件时非常有用。

二、标准文档加载器#

可以通过从 BaseLoader 进行子类化来实现文档加载器，BaseLoader 提供了用于加载文档的标准接口。

1. 接口#

load 方法是一个方便的方法，仅用于交互式工作 - 它只是调用 list(self.lazy_load())。
alazy_load 具有默认实现，将委托给 lazy_load。如果您使用异步操作，建议覆盖默认实现并提供本机异步实现。 {.callout-important} 在实现文档加载器时，不要通过 lazy_load 或 alazy_load 方法传递参数。所有配置都应通过初始化器（init）传递。这是 LangChain 的设计选择，以确保一旦实例化了文档加载器，它就具有加载文档所需的所有信息。

2. 实现#

让我们创建一个标准文档加载器的示例，该加载器从文件中加载数据，并从文件的每一行创建一个文档。

1
from typing import AsyncIterator, Iterator
2
from langchain_core.document_loaders import BaseLoader
3
from langchain_core.documents import Document
4
class CustomDocumentLoader(BaseLoader):
5
    """一个从文件逐行读取的示例文档加载器。"""
6
    def init(self, file_path: str) -> None:
7
        """使用文件路径初始化加载器。
8
        Args:
9
            file_path: 要加载的文件的路径。
10
        """
11
        self.file_path = file_path
12

13
    def lazy_load(self) -> Iterator[Document]:  # <-- 不接受任何参数
14
        """逐行读取文件的惰性加载器。
15
        当您实现惰性加载方法时，应使用生成器逐个生成文档。
16
        """
17
        with open(self.file_path, encoding="utf-8") as f:
18
            line_number = 0for line in f:
19
                yield Document(
20
                    page_content=line,
21
                    metadata={"line_number": line_number, "source": self.file_path},
22
                )
23
                line_number += 1
24
   # alazy_load是可选的。
25
   # 如果您省略了实现，将使用默认实现，该实现将委托给lazy_load！
26
   async def alazy_load(
27
        self,
28
   ) -> AsyncIterator[Document]:  # <-- 不接受任何参数
29
       """逐行读取文件的异步惰性加载器。"""
30
       # 需要aiofiles
31
       # 使用`pip install aiofiles`安装
32
       # https://github.com/Tinche/aiofiles
33
       import aiofiles
34
       async with aiofiles.open(self.file_path, encoding="utf-8") as f:
35
            line_number = 0
36
            async for line in f:
37
                yield Document(
38
                    page_content=line,
39
                    metadata={"line_number": line_number, "source": self.file_path},
40
                )
41
                line_number += 1

3. 测试#

为了测试文档加载器，我们需要一个包含一些优质内容的文件。

1
with open("./meow.txt", "w", encoding="utf-8") as f:
2
    quality_content = "喵喵🐱 \n 喵喵🐱 \n 喵😻😻"
3
    f.write(quality_content)
4
loader = CustomDocumentLoader("./meow.txt")

1
# 测试延迟加载接口
2
for doc in loader.lazy_load():
3
    print()
4
    print(type(doc))
5
    print(doc)

1
<class 'langchain_core.documents.base.Document'>
2
page_content='喵喵🐱
3
' metadata={'line_number': 0, 'source': './meow.txt'}
4

5
<class 'langchain_core.documents.base.Document'>
6
page_content=' 喵喵🐱
7
' metadata={'line_number': 1, 'source': './meow.txt'}
8

9
<class 'langchain_core.documents.base.Document'>
10
page_content=' 喵😻😻' metadata={'line_number': 2, 'source': './meow.txt'}

1
# 测试异步实现
2
async for doc in loader.alazy_load():
3
    print()
4
    print(type(doc))
5
    print(doc)

1
<class 'langchain_core.documents.base.Document'>
2
page_content='喵喵🐱
3
' metadata={'line_number': 0, 'source': './meow.txt'}
4

5
<class 'langchain_core.documents.base.Document'>
6
page_content=' 喵喵🐱
7
' metadata={'line_number': 1, 'source': './meow.txt'}
8

9
<class 'langchain_core.documents.base.Document'>
10
page_content=' 喵😻😻' metadata={'line_number': 2, 'source': './meow.txt'}

{.callout-tip} load() 在诸如 Jupyter Notebook 之类的交互式环境中很有用。在生产代码中避免使用它，因为急切加载假定所有内容都可以放入内存中，而这并不总是成立，特别是对于企业数据而言。

1
loader.load()

1
[Document(metadata={'line_number': 0, 'source': './meow.txt'}, page_content='喵喵🐱 \n'), Document(metadata={'line_number': 1, 'source': './meow.txt'}, page_content=' 喵喵🐱 \n'), Document(metadata={'line_number': 2, 'source': './meow.txt'}, page_content=' 喵😻😻')]

三、文件处理#

许多文档加载器涉及解析文件。这些加载器之间的区别通常在于文件的解析方式，而不是文件的加载方式。例如，您可以使用 <font style="color:rgb(28, 30, 33);">open</font> 来读取 PDF 或 markdown 文件的二进制内容，但您需要不同的解析逻辑来将该二进制数据转换为文本。因此，将解析逻辑与加载逻辑分离可能会很有帮助，这样无论数据如何加载，都更容易重用给定的解析器。

1. BaseBlobParser#

BaseBlobParser 是一个接口，接受一个 blob 并输出一个 Document 对象列表。blob 是一个表示数据的对象，可以存在于内存中或文件中。LangChain Python 具有受 Blob WebAPI 规范启发的 Blob 原语。

1
from langchain_core.document_loaders import BaseBlobParser, Blob
2
class MyParser(BaseBlobParser):
3
    """一个简单的解析器，每行创建一个文档。"""
4
    def lazy_parse(self, blob: Blob) -> Iterator[Document]:
5
        """逐行将 blob 解析为文档。"""
6
        line_number = 0
7
        with blob.as_bytes_io() as f:
8
            for line in f:
9
                line_number += 1yield Document(
10
                    page_content=line,
11
                    metadata={"line_number": line_number, "source": blob.source},
12
                )

1
blob = Blob.from_path("./meow.txt")
2
parser = MyParser()

1
list(parser.lazy_parse(blob))

1
[Document(page_content='喵喵🐱 \n', metadata={'line_number': 1, 'source': './meow.txt'}),
2
 Document(page_content=' 喵喵🐱 \n', metadata={'line_number': 2, 'source': './meow.txt'}),
3
 Document(page_content=' 喵😻😻', metadata={'line_number': 3, 'source': './meow.txt'})]

使用 blob API 还允许直接从内存加载内容，而无需从文件中读取！

1
blob = Blob(data=b"来自内存的一些数据\n喵")
2
list(parser.lazy_parse(blob))

1
[Document(page_content='来自内存的一些数据\n', metadata={'line_number': 1, 'source': None}),
2
 Document(page_content='喵', metadata={'line_number': 2, 'source': None})]

2. Blob#

让我们快速浏览一下 Blob API 的一些内容。

1
blob = Blob.from_path("./meow.txt", metadata={"foo": "bar"})

1
blob.encoding

1
'utf-8'

1
blob.as_bytes()

1
b'\xe5\x96\xb5\xe5\x96\xb5\xf0\x9f\x90\xb1 \r\n \xe5\x96\xb5\xe5\x96\xb5\xf0\x9f\x90\xb1 \r\n \xe5\x96\xb5\xf0\x9f\x98\xbb\xf0\x9f\x98\xbb'

1
blob.as_string()

1
喵喵🐱
2
 喵喵🐱
3
 喵😻😻

1
blob.as_bytes_io()

1
<contextlib._GeneratorContextManager object at 0x0000012E064CC2F0>

3. Blob 元数据#

1
blob.metadata

1
{'foo': 'bar'}

1
blob.source

1
./meow.txt

4. Blob 加载器#

在解析器中封装了将二进制数据解析为文档所需的逻辑，blob 加载器封装了从给定存储位置加载 blob 所需的逻辑。目前，LangChain 仅支持 FileSystemBlobLoader。您可以使用 FileSystemBlobLoader 加载 blob，然后使用解析器对其进行解析。

1
from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader
2
blob_loader = FileSystemBlobLoader(path=".", glob="*.mdx", show_progress=True)

1
parser = MyParser()
2
for blob in blob_loader.yield_blobs():for doc in parser.lazy_parse(blob):print(doc)break

1
100%|██████████| 8/8 [00:00<00:00, 8087.35it/s]

1
page_content='# CSV
2
' metadata={'line_number': 1, 'source': '..\\resource\\csv.mdx'}
3
page_content='# File Directory
4
' metadata={'line_number': 1, 'source': '..\\resource\\file_directory.mdx'}
5
page_content='# HTML
6
' metadata={'line_number': 1, 'source': '..\\resource\\html.mdx'}
7
page_content='---
8
' metadata={'line_number': 1, 'source': '..\\resource\\index.mdx'}
9
page_content='# JSON
10
' metadata={'line_number': 1, 'source': '..\\resource\\json.mdx'}
11
page_content='# Markdown
12
' metadata={'line_number': 1, 'source': '..\\resource\\markdown.mdx'}
13
page_content='# Microsoft Office
14
' metadata={'line_number': 1, 'source': '..\\resource\\office_file.mdx'}
15
page_content='---
16
' metadata={'line_number': 1, 'source': '..\\resource\\pdf.mdx'}

5. 通用加载器#

LangChain 拥有一个 GenericLoader 抽象，它将 BlobLoader 与 BaseBlobParser 结合在一起。GenericLoader 旨在提供标准化的类方法，使现有的 BlobLoader 实现易于使用。目前，仅支持 FileSystemBlobLoader。

1
from langchain_community.document_loaders.generic import GenericLoader
2
loader = GenericLoader.from_filesystem(
3
    path=".", glob="*.mdx", show_progress=True, parser=MyParser()
4
)
5
for idx, doc in enumerate(loader.lazy_load()):if idx < 5:print(doc)
6
print("... output truncated for demo purposes")

1
100%|██████████| 8/8 [00:00<00:00, 78.69it/s]

1
page_content='# CSV
2
' metadata={'line_number': 1, 'source': '..\\resource\\csv.mdx'}
3
page_content='# File Directory
4
' metadata={'line_number': 1, 'source': '..\\resource\\file_directory.mdx'}
5
page_content='# HTML
6
' metadata={'line_number': 1, 'source': '..\\resource\\html.mdx'}
7
page_content='---
8
' metadata={'line_number': 1, 'source': '..\\resource\\index.mdx'}
9
page_content='# JSON
10
' metadata={'line_number': 1, 'source': '..\\resource\\json.mdx'}
11
... output truncated for demo purposes

6. 自定义通用加载器#

如果您喜欢创建类，您可以子类化并创建一个类来封装逻辑。您可以从这个类中子类化以使用现有的加载器加载内容。

1
from typing import Any
2
class MyCustomLoader(GenericLoader):
3
    @staticmethod
4
    def get_parser(**kwargs: Any) -> BaseBlobParser:
5
        """Override this method to associate a default parser with the class."""
6
        return MyParser()

1
loader = MyCustomLoader.from_filesystem(path=".", glob="*.mdx", show_progress=True)
2
for idx, doc in enumerate(loader.lazy_load()):if idx < 5:print(doc)
3
print("... output truncated for demo purposes")

1
100%|██████████| 8/8 [00:00<00:00, 80.28it/s]

1
page_content='# CSV
2
' metadata={'line_number': 1, 'source': '..\\resource\\csv.mdx'}
3
page_content='# File Directory
4
' metadata={'line_number': 1, 'source': '..\\resource\\file_directory.mdx'}
5
page_content='# HTML
6
' metadata={'line_number': 1, 'source': '..\\resource\\html.mdx'}
7
page_content='---
8
' metadata={'line_number': 1, 'source': '..\\resource\\index.mdx'}
9
page_content='# JSON
10
' metadata={'line_number': 1, 'source': '..\\resource\\json.mdx'}
11
... output truncated for demo purposes

LangChain篇-自定义RAG加载器

https://mizuki.mysqil.com/posts/langchain篇-自定义rag加载器/

作者

SuifengFlying

发布于

2026-07-29

许可协议

CC BY-NC-SA 4.0

欢迎来到 SuifengFlying!

一、概述#

二、标准文档加载器#

1. 接口#

2. 实现#

3. 测试#

三、文件处理#

1. BaseBlobParser#

2. Blob#

3. Blob 元数据#

4. Blob 加载器#

5. 通用加载器#

6. 自定义通用加载器#

Related articles