langchain从入门到精通（二十）——自定义文档加载器使用技巧及Blob 方案介绍
1. 自定义加载器使用技巧
对于一些企业的内部数据，例如数据库、API接口等定制化非常强的数据，如果使用通用的文档加载器进行提取，虽然可以提取记录到相应的信息，但是加载的数据格式或者样式大概率没法满足我们的需求，这个时候就可以考虑实现自定义文档加载器。
例如使用的 WebBaseLoader 文档加载器加载网页首页的信息，会提取得到很多空白数据（空格、换行、Tab 等），将这类数据通过分割存储到向量数据库中，会极大降低检索与生成的效率和正确性。
在 LangChain 中实现自定义文档加载器非常简单，只需要继承 BaseLoader 基类，然后实现 lazy_load() 方法即可，如果该文档加载器有异步使用的场景，还需要实现 alazy_load() 方法。
假设有一个这样的需求，加载对应的文本信息，其中每行数据都作为一个 Document 组件，该文档加载器实现示例:
from typing import Iterator, AsyncIteratorfrom langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Documentclass CustomDocumentLoader(BaseLoader):"""自定义文档加载器"""def __init__(self, file_path: str) -> None:self.file_path = file_pathdef lazy_load(self) -> Iterator[Document]:"""逐行读取文件的数据并使用yield返回文档"""with open(self.file_path, encoding="utf-8") as f:line_number = 0for line in f:yield Document(page_content=line,metadata={"line_number": line_number, "source": self.file_path})line_number += 1async def alazy_load(self) -> AsyncIterator[Document]:"""lazy_load的异步方法，如果不实现，将委托lazy_load实现"""import aiofilesasync with aiofiles.open(self.file_path, encoding="utf-8") as f:line_number = 0async for line in f:yield Document(page_content=line,metadata={"line_number": line_number, "source": self.file_path})line_number += 1loader = CustomDocumentLoader("./喵喵.txt")
documents = loader.load()print(documents)
print(len(documents))
print(documents[0].metadata)
输出示例:
[Document(page_content='喵喵🐱\n', metadata={'line_number': 0, 'source': './喵喵.txt'}), Document(page_content='喵喵🐱\n', metadata={'line_number': 1, 'source': './喵喵.txt'}), Document(page_content='喵😻😻', metadata={'line_number': 2, 'source': './喵喵.txt'})]
3
{'line_number': 0, 'source': './喵喵.txt'}
2. 文档加载器扩展思考
在上面的自定义文档加载器中，可以看到 lazy_load() 方法的两个核心步骤就是：读取文件数据、将文件数据解析成Document，并且绝大部分文档加载器都有着两个核心步骤，而且 读取文件数据 这个步骤大家都大差不差。
就像 .md、.txt、*.py 这类文本文件，甚至是 .pdf、.doc 等这类非文本文件，都可以使用同一个 读取文件数据 步骤将文件读取为 二进制内容，然后在使用不同的解析逻辑来解析对应的二进制内容，所以很容易可以得出：
文档加载器 = 二进制数据读取 + 解析逻辑
因此，在项目开发中，如果大量配置自定义文档解析器的话，将解析逻辑与加载逻辑分离，维护起来会更容易，而且也更容易复用相应的逻辑（具体使用哪种方式取决于开发）。
这样原先的 DocumentLoader 运行流程就变成了如下:
这样所有 DocumentLoader 就变成了共用 Blob(数据读取)，每个加载器内部只实现不同的 parse 即可，这也是 LangChain 目前正在设计的新方案，由于目前 LangChain 在这块尚不完善，并且篇幅较长.
3. LangChain 中的 Blob 方案
许多文档加载器都涉及到解析文件，此类加载器之间的差异通常源于文件解析方式，而不是文件加载方式。例如，你可以使用 open() 函数来读取 PDF 或 Markdown 文件的二进制内容，但是需要不同的解析逻辑来将二进制数据转换为文本。
在 LangChain 中也提供了一个类似的解决方案 Blob，其灵感来源于 Blob WebAPI规范（这是前端 Web 浏览器中定义的相关规范）。
Blob WebAPI 规范文档链接
该方案下有 Blob、BlobLoader 和 BaseBlobParser 三个类，含义如下：
Blob：LangChain 封装的数据对象，通过引用或值表示原始数据，该类提供一个接口，以表示不同形式具体化的二进制数据，使用该类可以有助于将数据加载器的开发与解析器耦合。BlobLoader：Blob 数据加载器，类似 DocumentLoader，不过 BlobLoader 被设计成可以加载任何数据（未来的规划）。BlobParser：Blob 数据解析器，用于将传入的 Blob 数据转换成文档列表。
在这个方案下，文档加载器的运行流程变成如下:
例如加载对应的文本信息，其中每行数据都作为一个 Document 组件，使用 Blob 的方案来实现，只需自定义一个解析器并实现 lazy_parser() 方法即可，示例代码如下:
from typing import Iteratorfrom langchain_core.document_loaders import Blob
from langchain_core.document_loaders.base import BaseBlobParser
from langchain_core.documents import Documentclass CustomParser(BaseBlobParser):"""自定义解析器，提取文本文件的每一行为一个Document元素"""def lazy_parse(self, blob: Blob) -> Iterator[Document]:line_number = 0with blob.as_bytes_io() as f:for line in f:yield Document(page_content=line,metadata={"source": blob.source, "line_number": line_number})line_number += 1blob = Blob.from_path("./喵喵.txt")
parser = CustomParser()
documents = list(parser.lazy_parse(blob))print(documents)
print(len(documents))
print(documents[0].metadata)
输出内容：
[Document(page_content='喵喵🐱\r\n', metadata={'source': './喵喵.txt', 'line_number': 0}), Document(page_content='喵喵🐱\r\n', metadata={'source': './喵喵.txt', 'line_number': 1}), Document(page_content='喵😻😻', metadata={'source': './喵喵.txt', 'line_number': 2})]
3
{'source': './喵喵.txt', 'line_number': 0}
除了 from_path() 函数，Blob 类还可以在构造时传递 data 参数，直接从内存中加载内容，而无需从文件中获取，示例如下
blob = Blob(data="喵喵🐱\r\n喵喵🐱\r\n喵😻😻")
4. Blob 数据存储类
LangChain 中设计的 Blob 数据存储类和 Blob WebAPI规范 定义的类非常接近，拥有以下方法和属性：
- data：原始数据，支持存储字节、字符串数据。
- mimetype：文件的 mimetype 类型。
- encoding：文件的编码，默认值为 utf-8。
- path：文件的原始路径，支持传递字符串路径或者 Path 类。
- metadata：存储的元数据，一般都有 source 字段。
- source()：只读函数/属性，用于返回数据的来源。
- as_string()：将数据转换成字符串。
- as_bytes()：将数据转换成字节数据。
- as_bytes_io()：将数据转换成缓冲流字节数据。
- from_path()：从对应的路径中加载 Blob 数据（文件）。
- from_data()：从对应的原始数据中加载 Blob 数据（非文件）。
- Blob 加载器
解析器封装了将二进制数据解析为 Document 组件所需的逻辑，而 Blob 加载器则封装了从给定存储位置加载 Blob 所需的逻辑。不过目前在 LangChain 中，只集成了一个 FileSystemBlobLoader，即文件系统二进制数据加载器。
这个加载器可以加载传入文件夹下的特定文件，代码示例:
from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoaderloader = FileSystemBlobLoader(".", show_progress=True)
for blob in loader.yield_blobs():
print(blob.source)
输出内容
100%|██████████| 3/3 [00:00<00:00, 58.15it/s]
1.Blob解析器示例.py
2.FileSystemBlobLoader示例.py
喵喵.txt
如果要实现一个 Blob 加载器，只需要继承 BlobLoader 类，并实现 yield_blobs() 方法`即可，BlobLoader 类的代码如下
# langchain_core/document_loaders/blob_loaders::BlobLoader->yield_blobs
class BlobLoader(ABC):"""Abstract interface for blob loaders implementation.Implementer should be able to load raw content from a storage system accordingto some criteria and return the raw content lazily as a stream of blobs."""@abstractmethoddef yield_blobs(self,) -> Iterable[Blob]:"""A lazy loader for raw data represented by LangChain's Blob object.Returns:A generator over blobs"""
04. Blob 通用加载器
在上面的示例中，Blob 加载器与解析器是分开使用的，其实在 LangChain 中还封装了一个由 BlobLoader 与 BaseBlobParser 组成的类——GenericLoader，这个类旨在提供标准化的方法，让 BlobLoader 使用更简单，不过目前也仅支持 FileSystemBlobLoader。
使用示例如下
from langchain_community.document_loaders.generic import GenericLoaderloader = GenericLoader.from_filesystem(".", glob="*.txt", show_progress=True)for idx, doc in enumerate(loader.lazy_load()):print(f"当前加载第{idx + 1}个文件，文件信息:{doc.metadata}")
输出内容：
100%|██████████| 1/1 [00:00<00:00, 19.76it/s]
当前加载第1个文件，文件信息:{'source': '喵喵.txt'}
整体来说，Blob 解决方案目前 LangChain 封装与集成得非常少，如果需要使用 Blob 的形式来加载文件，目前还需要大量编写加载文件与解析数据的逻辑，效率比较低，不过随着未来 LangChain 团队封装的 Blob 解析逻辑越来越多，会逐渐代替 DocumentLoader 的方案。对于目前的版本来说，大家只需要知道有这个东西即可