LangChain自定义RAG加载器

示例代码

注：将ppt后缀名改为zip后解压

python版本：3.12.4

概述

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

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

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

标准文档加载器

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

接口

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

实现

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

#示例：doc_loader_custom.py
from typing import AsyncIterator, Iterator
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
class CustomDocumentLoader(BaseLoader):
    """一个从文件逐行读取的示例文档加载器。"""
    def __init__(self, file_path: str) -> None:
        """使用文件路径初始化加载器。
        Args:
            file_path: 要加载的文件的路径。
        """
        self.file_path = file_path
    def lazy_load(self) -> Iterator[Document]:  # <-- 不接受任何参数
        """逐行读取文件的惰性加载器。
        当您实现惰性加载方法时，应使用生成器逐个生成文档。
        """
        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是可选的。
    # 如果您省略了实现，将使用默认实现，该实现将委托给lazy_load！
    async def alazy_load(
        self,
    ) -> AsyncIterator[Document]:  # <-- 不接受任何参数
        """逐行读取文件的异步惰性加载器。"""
        # 需要aiofiles
        # 使用`pip install 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