Source code for langchain_core.document_loaders.base

"""文档加载器实现的抽象接口。"""
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, AsyncIterator, Iterator, List, Optional

from langchain_core.documents import Document
from langchain_core.runnables import run_in_executor

if TYPE_CHECKING:
    from langchain_text_splitters import TextSplitter

from langchain_core.document_loaders.blob_loaders import Blob


class BaseLoader(ABC):
    """文档加载器的接口。

实现应该使用生成器实现惰性加载方法，以避免一次性将所有文档加载到内存中。

`load` 仅供用户方便使用，不应被覆盖。
"""

    # Sub-classes should not implement this method directly. Instead, they
    # should implement the lazy load method.
    def load(self) -> List[Document]:
        """将数据加载到文档对象中。"""
        return list(self.lazy_load())

    async def aload(self) -> List[Document]:
        """将数据加载到文档对象中。"""
        return [document async for document in self.alazy_load()]

    def load_and_split(
        self, text_splitter: Optional[TextSplitter] = None
    ) -> List[Document]:
        """加载文档并分割成块。块作为文档返回。

不要覆盖此方法。应该被视为已弃用！

参数：
    text_splitter: 用于分割文档的TextSplitter实例。
      默认为RecursiveCharacterTextSplitter。

返回：
    文档列表。
"""

        if text_splitter is None:
            try:
                from langchain_text_splitters import RecursiveCharacterTextSplitter
            except ImportError as e:
                raise ImportError(
                    "Unable to import from langchain_text_splitters. Please specify "
                    "text_splitter or install langchain_text_splitters with "
                    "`pip install -U langchain-text-splitters`."
                ) from e

            _text_splitter: TextSplitter = RecursiveCharacterTextSplitter()
        else:
            _text_splitter = text_splitter
        docs = self.load()
        return _text_splitter.split_documents(docs)

    # Attention: This method will be upgraded into an abstractmethod once it's
    #            implemented in all the existing subclasses.
    def lazy_load(self) -> Iterator[Document]:
        """一个用于文档的惰性加载器。"""
        if type(self).load != BaseLoader.load:
            return iter(self.load())
        raise NotImplementedError(
            f"{self.__class__.__name__} does not implement lazy_load()"
        )

    async def alazy_load(self) -> AsyncIterator[Document]:
        """一个用于文档的惰性加载器。"""
        iterator = await run_in_executor(None, self.lazy_load)
        done = object()
        while True:
            doc = await run_in_executor(None, next, iterator, done)  # type: ignore[call-arg, arg-type]
            if doc is done:
                break
            yield doc  # type: ignore[misc]


class BaseBlobParser(ABC):
    """用于blob解析器的抽象接口。

Blob解析器提供了一种将存储在blob中的原始数据解析为一个或多个文档的方法。

解析器可以与blob加载器组合，使得可以轻松地重用解析器，而不受blob最初加载方式的影响。
"""

    @abstractmethod
    def lazy_parse(self, blob: Blob) -> Iterator[Document]:
        """懒解析接口。

需要子类实现此方法。

参数：
    blob: Blob实例

返回：
    文档的生成器
"""

    def parse(self, blob: Blob) -> List[Document]:
        """将blob急切地解析为一个文档或多个文档。

这是一个用于交互式开发环境的便利方法。

生产应用程序应该更倾向于使用lazy_parse方法。

子类通常不应该覆盖这个解析方法。

参数：
    blob：Blob实例

返回：
    文档列表
"""
        return list(self.lazy_parse(blob))