from __future__ import annotations
from pathlib import Path
from typing import (
TYPE_CHECKING,
Any,
Iterator,
List,
Literal,
Optional,
Sequence,
Union,
)
from langchain_core.documents import Document
from langchain_community.document_loaders.base import BaseBlobParser, BaseLoader
from langchain_community.document_loaders.blob_loaders import (
BlobLoader,
FileSystemBlobLoader,
)
from langchain_community.document_loaders.parsers.registry import get_parser
if TYPE_CHECKING:
from langchain_text_splitters import TextSplitter
_PathLike = Union[str, Path]
DEFAULT = Literal["default"]
[docs]class GenericLoader(BaseLoader):
"""通用文档加载器。
一个通用的文档加载器，允许将任意的blob加载器与blob解析器结合在一起。
示例：
解析特定的PDF文件：
.. code-block:: python
from langchain_community.document_loaders import GenericLoader
from langchain_community.document_loaders.parsers.pdf import PyPDFParser
# 递归加载目录中的所有文本文件。
loader = GenericLoader.from_filesystem(
"my_lovely_pdf.pdf",
parser=PyPDFParser()
)
.. code-block:: python
from langchain_community.document_loaders import GenericLoader
from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader
loader = GenericLoader.from_filesystem(
path="path/to/directory",
glob="**/[!.]*",
suffixes=[".pdf"],
show_progress=True,
)
docs = loader.lazy_load()
next(docs)
更改要加载的文件示例实例：
.. code-block:: python
# 递归加载目录中的所有文本文件。
loader = GenericLoader.from_filesystem("/path/to/dir", glob="**/*.txt")
# 递归加载目录中所有非隐藏文件。
loader = GenericLoader.from_filesystem("/path/to/dir", glob="**/[!.]*")
# 在不递归的情况下加载目录中的所有文件。
loader = GenericLoader.from_filesystem("/path/to/dir", glob="*")
更改要使用的解析器的示例实例：
.. code-block:: python
from langchain_community.document_loaders.parsers.pdf import PyPDFParser
# 递归加载目录中的所有文本文件。
loader = GenericLoader.from_filesystem(
"/path/to/dir",
glob="**/*.pdf",
parser=PyPDFParser()
)""" # noqa: E501
[docs] def __init__(
self,
blob_loader: BlobLoader, # type: ignore[valid-type]
blob_parser: BaseBlobParser,
) -> None:
"""一个通用的文档加载器。
参数：
blob_loader：一个知道如何生成blob的blob加载器
blob_parser：一个知道如何将blob解析为文档的blob解析器
"""
self.blob_loader = blob_loader
self.blob_parser = blob_parser
[docs] def lazy_load(
self,
) -> Iterator[Document]:
"""懒加载文档。在大规模工作时使用。"""
for blob in self.blob_loader.yield_blobs(): # type: ignore[attr-defined]
yield from self.blob_parser.lazy_parse(blob)
[docs] def load_and_split(
self, text_splitter: Optional[TextSplitter] = None
) -> List[Document]:
"""加载所有文档并将它们分割成句子。"""
raise NotImplementedError(
"Loading and splitting is not yet implemented for generic loaders. "
"When they will be implemented they will be added via the initializer. "
"This method should not be used going forward."
)
[docs] @classmethod
def from_filesystem(
cls,
path: _PathLike,
*,
glob: str = "**/[!.]*",
exclude: Sequence[str] = (),
suffixes: Optional[Sequence[str]] = None,
show_progress: bool = False,
parser: Union[DEFAULT, BaseBlobParser] = "default",
parser_kwargs: Optional[dict] = None,
) -> GenericLoader:
"""创建一个使用文件系统blob加载器的通用文档加载器。
参数:
path: 要加载文档的目录路径或要加载的单个文件的路径。如果这是一个文件，则glob、exclude、suffixes将被忽略。
glob: 用于查找文档的glob模式。
suffixes: 用于过滤文档的后缀。如果为None，则将加载与glob匹配的所有文件。
exclude: 要从加载器中排除的模式列表。
show_progress: 是否显示进度条（需要tqdm）。代理到文件系统加载器。
parser: 一个知道如何将blob解析为文档的blob解析器，如果未提供，则将实例化一个默认解析器。默认值可以通过传递解析器或设置类属性`blob_parser`来覆盖（后者应该与继承一起使用）。
parser_kwargs: 传递给解析器的关键字参数。
返回:
一个通用的文档加载器。
"""
blob_loader = FileSystemBlobLoader( # type: ignore[attr-defined, misc]
path,
glob=glob,
exclude=exclude,
suffixes=suffixes,
show_progress=show_progress,
)
if isinstance(parser, str):
if parser == "default":
try:
# If there is an implementation of get_parser on the class, use it.
blob_parser = cls.get_parser(**(parser_kwargs or {}))
except NotImplementedError:
# if not then use the global registry.
blob_parser = get_parser(parser)
else:
blob_parser = get_parser(parser)
else:
blob_parser = parser
return cls(blob_loader, blob_parser)
[docs] @staticmethod
def get_parser(**kwargs: Any) -> BaseBlobParser:
"""重写此方法以将默认解析器与类关联。"""
raise NotImplementedError()