跳到主要内容
Open In ColabOpen on GitHub

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

概述​

基于大型语言模型（LLM）的应用通常需要从数据库或文件（如 PDF）中提取数据，并将其转换为 LLM 可以利用的格式。在 LangChain 中，这通常涉及创建 Document 对象，该对象封装了提取的文本（page_content）以及元数据——一个包含文档详细信息（如作者姓名或出版日期）的字典。

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

文档加载（Document Loading）的主要抽象包括

组件描述
Document包含 text 和 metadata
BaseLoader用于将原始数据转换为 Document 对象
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。如果你正在使用异步，我们建议重写默认实现并提供一个原生的异步实现。
重要

在实现文档加载器时，请**不要**通过 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 参考: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'}
提示

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

避免将其用于生产代码，因为急切加载假定所有内容都可以载入内存，但对于企业数据而言，情况并非总是如此。

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 参考: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
)
%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
)  # Supports s3://, az://, gs://, file:// schemes.

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

for blob in cloud_loader.yield_blobs():
    print(blob)
API 参考: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 参考: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