Skip to main content
Open In ColabOpen on GitHub

如何创建自定义文档加载器

概述​

基于 LLM 的应用程序通常需要从数据库或文件（如 PDF）中提取数据，并将其转换为 LLM 可以使用的格式。在 LangChain 中，这通常涉及创建 Document 对象，这些对象封装了提取的文本（page_content）以及元数据（一个包含有关文档详细信息的字典，例如作者姓名或发布日期）。

Document 对象通常会被格式化成提示，然后输入给 LLM，使 LLM 能够利用 Document 中的信息来生成所需的响应（例如，总结文档）。 Documents 可以立即使用，也可以索引到向量存储中以供将来的检索和使用。

文档加载器的主要抽象是：

组件描述
Document包含 text 和 metadata
BaseLoader用于将原始数据转换为 Documents
Blob二进制数据的表示形式，位于文件或内存中
BaseBlobParser用于解析 Blob 以生成 Document 对象的逻辑

本指南将演示如何编写自定义文档加载和文件解析逻辑；具体来说，我们将看到如何：

  1. 通过继承 BaseLoader 来创建标准的文档加载器。
  2. 使用 BaseBlobParser 创建一个解析器，并将其与 Blob 和 BlobLoaders 结合使用。这主要在处理文件时非常有用。

标准文档加载器​

文档加载器可以通过继承 BaseLoader 来实现，BaseLoader 提供了加载文档的标准接口。

接口​

方法名解释
lazy_load用于惰性地逐个加载文档。用于生产代码。
alazy_loadlazy_load 的异步变体
load用于立即地将所有文档加载到内存中。用于原型设计或交互式工作。
aload用于立即地将所有文档加载到内存中。用于原型设计或交互式工作。 于 2024-04 添加至 LangChain。
  • load 方法是一个便捷方法，仅用于原型设计工作——它仅调用 list(self.lazy_load())。
  • alazy_load 有一个默认实现，它会委托给 lazy_load。如果您使用异步，我们建议重写默认实现并提供原生的异步实现。
important

在实现文档加载器时，请不要通过 lazy_load 或 alazy_load 方法提供参数。

所有配置都应通过初始化器（__init__）传递。这是 LangChain 做出的设计选择，以确保文档加载器在实例化后拥有加载文档所需的所有信息。

安装​

安装 langchain-core 和 langchain_community。

%pip install -qU langchain_core langchain_community

实现​

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

from typing import AsyncIterator, Iterator

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document


class CustomDocumentLoader(BaseLoader):
    """An example document loader that reads a file line by line."""

    def __init__(self, file_path: str) -> None:
        """Initialize the loader with a file path.

        Args:
            file_path: The path to the file to load.
        """
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:  # <-- Does not take any arguments
        """A lazy loader that reads a file line by line.

        When you're implementing lazy load methods, you should use a generator
        to yield documents one by one.
        """
        with open(self.file_path, encoding="utf-8") as f:
            line_number = 0
            for line in f:
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": self.file_path},
                )
                line_number += 1

    # alazy_load is OPTIONAL.
    # If you leave out the implementation, a default implementation which delegates to lazy_load will be used!
    async def alazy_load(
        self,
    ) -> AsyncIterator[Document]:  # <-- Does not take any arguments
        """An async lazy loader that reads a file line by line."""
        # Requires aiofiles
        # https://github.com/Tinche/aiofiles
        import aiofiles

        async with aiofiles.open(self.file_path, encoding="utf-8") as f:
            line_number = 0
            async for line in f:
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": self.file_path},
                )
                line_number += 1
API Reference:BaseLoader | Document

测试 🧪​

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

with open("./meow.txt", "w", encoding="utf-8") as f:
    quality_content = "meow meow🐱 \n meow meow🐱 \n meow😻😻"
    f.write(quality_content)

loader = CustomDocumentLoader("./meow.txt")
%pip install -q aiofiles
## Test out the lazy load interface
for doc in loader.lazy_load():
    print()
    print(type(doc))
    print(doc)

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

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

<class 'langchain_core.documents.base.Document'>
page_content=' meow😻😻' metadata={'line_number': 2, 'source': './meow.txt'}
## Test out the async implementation
async for doc in loader.alazy_load():
    print()
    print(type(doc))
    print(doc)

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

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

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

load() 在交互式环境（例如 jupyter notebook）中可能很有用。

请避免在生产代码中使用它，因为急切加载（eager loading）假设所有内容都可以放入内存，但这并非总是如此，尤其是对于企业级数据。

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

处理文件​

许多文档加载器涉及解析文件。这些加载器之间的区别通常源于文件的解析方式，而不是文件的加载方式。例如，你可以使用 open 来读取 PDF 或 markdown 文件的二进制内容，但你需要不同的解析逻辑才能将这些二进制数据转换为文本。

因此，将解析逻辑与加载逻辑解耦会很有帮助，这样可以更容易地重用某个解析器，而无需考虑数据是如何加载的。

BaseBlobParser​

BaseBlobParser 是一个接受 blob 并输出 Document 对象列表的接口。blob 是存在于内存或文件中的数据的表示。LangChain Python 有一个 Blob 原语，其灵感来自 Blob WebAPI 规范。

from langchain_core.document_loaders import BaseBlobParser, Blob


class MyParser(BaseBlobParser):
    """A simple parser that creates a document from each line."""

    def lazy_parse(self, blob: Blob) -> Iterator[Document]:
        """Parse a blob into a document line by line."""
        line_number = 0
        with blob.as_bytes_io() as f:
            for line in f:
                line_number += 1
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": blob.source},
                )
API Reference:BaseBlobParser | Blob
blob = Blob.from_path("./meow.txt")
parser = MyParser()
list(parser.lazy_parse(blob))
[Document(metadata={'line_number': 1, 'source': './meow.txt'}, page_content='meow meow🐱 \n'),
 Document(metadata={'line_number': 2, 'source': './meow.txt'}, page_content=' meow meow🐱 \n'),
 Document(metadata={'line_number': 3, 'source': './meow.txt'}, page_content=' meow😻😻')]

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

blob = Blob(data=b"some data from memory\nmeow")
list(parser.lazy_parse(blob))
[Document(metadata={'line_number': 1, 'source': None}, page_content='some data from memory\n'),
 Document(metadata={'line_number': 2, 'source': None}, page_content='meow')]

Blob​

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

blob = Blob.from_path("./meow.txt", metadata={"foo": "bar"})
blob.encoding
'utf-8'
blob.as_bytes()
b'meow meow\xf0\x9f\x90\xb1 \n meow meow\xf0\x9f\x90\xb1 \n meow\xf0\x9f\x98\xbb\xf0\x9f\x98\xbb'
blob.as_string()
'meow meow🐱 \n meow meow🐱 \n meow😻😻'
blob.as_bytes_io()
<contextlib._GeneratorContextManager at 0x74b8d42e9940>
blob.metadata
{'foo': 'bar'}
blob.source
'./meow.txt'

Blob 加载器​

解析器封装了将二进制数据解析成文档所需的逻辑，而 blob 加载器 则封装了从给定存储位置加载 blob 所必需的逻辑。

目前，LangChain 支持 FileSystemBlobLoader 和 CloudBlobLoader。

您可以使用 FileSystemBlobLoader 加载 blob，然后使用解析器解析它们。

from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader

filesystem_blob_loader = FileSystemBlobLoader(
    path=".", glob="*.mdx", show_progress=True
)
API Reference:FileSystemBlobLoader
%pip install -q tqdm
parser = MyParser()
for blob in filesystem_blob_loader.yield_blobs():
    for doc in parser.lazy_parse(blob):
        print(doc)
        break

或者，您可以使用 CloudBlobLoader 从云存储位置加载 Blob（支持 s3://, az://, gs://, file:// 协议）。

%pip install -q 'cloudpathlib[s3]'
from cloudpathlib import S3Client, S3Path
from langchain_community.document_loaders.blob_loaders import CloudBlobLoader

client = S3Client(no_sign_request=True)
client.set_as_default_client()

path = S3Path(
    "s3://bucket-01", client=client
)  # 支持 s3://, az://, gs://, file:// 协议。

cloud_loader = CloudBlobLoader(path, glob="**/*.pdf", show_progress=True)

for blob in cloud_loader.yield_blobs():
    print(blob)
API Reference:CloudBlobLoader

通用加载器​

LangChain 拥有一个 GenericLoader 抽象，它组合了 BlobLoader 和 BaseBlobParser。

GenericLoader 旨在提供标准化的类方法，方便使用现有的 BlobLoader 实现。目前支持 FileSystemBlobLoader 和 CloudBlobLoader。请参阅下面的示例：

from langchain_community.document_loaders.generic import GenericLoader

generic_loader_filesystem = GenericLoader(
    blob_loader=filesystem_blob_loader, blob_parser=parser
)
for idx, doc in enumerate(generic_loader_filesystem.lazy_load()):
    if idx < 5:
        print(doc)

print("... output truncated for demo purposes")
API Reference:GenericLoader
100%|██████████| 7/7 [00:00<00:00, 1224.82it/s]
``````output
page_content='# Text embedding models
' metadata={'line_number': 1, 'source': 'embed_text.mdx'}
page_content='
' metadata={'line_number': 2, 'source': 'embed_text.mdx'}
page_content=':::info
' metadata={'line_number': 3, 'source': 'embed_text.mdx'}
page_content='Head to [Integrations](/docs/integrations/text_embedding/) for documentation on built-in integrations with text embedding model providers.
' metadata={'line_number': 4, 'source': 'embed_text.mdx'}
page_content=':::
' metadata={'line_number': 5, 'source': 'embed_text.mdx'}
... output truncated for demo purposes

自定义通用加载器​

如果你喜欢创建类，可以进行子类继承并创建一个类来封装逻辑。

你可以从此类进行子类继承，以使用现有的加载器加载内容。

from typing import Any


class MyCustomLoader(GenericLoader):
    @staticmethod
    def get_parser(**kwargs: Any) -> BaseBlobParser:
        """Override this method to associate a default parser with the class."""
        return MyParser()
loader = MyCustomLoader.from_filesystem(path=".", glob="*.mdx", show_progress=True)

for idx, doc in enumerate(loader.lazy_load()):
    if idx < 5:
        print(doc)

print("... output truncated for demo purposes")
100%|██████████| 7/7 [00:00<00:00, 814.86it/s]
``````output
page_content='# Text embedding models
' metadata={'line_number': 1, 'source': 'embed_text.mdx'}
page_content='
' metadata={'line_number': 2, 'source': 'embed_text.mdx'}
page_content=':::info
' metadata={'line_number': 3, 'source': 'embed_text.mdx'}
page_content='Head to [Integrations](/docs/integrations/text_embedding/) for documentation on built-in integrations with text embedding model providers.
' metadata={'line_number': 4, 'source': 'embed_text.mdx'}
page_content=':::
' metadata={'line_number': 5, 'source': 'embed_text.mdx'}
... output truncated for demo purposes