文档加载器 API

from langchain_core.document_loaders import BaseLoader

class BaseLoader(ABC):
    """文档加载器基类"""

    @abstractmethod
    def load(self) -> List[Document]:
        """
        加载文档

        Returns:
            Document 对象列表
        """

    def load_and_split(
        self,
        text_splitter: Optional[TextSplitter] = None,
    ) -> List[Document]:
        """
        加载并分割文档

        Args:
            text_splitter: 文本分割器

        Returns:
            分割后的 Document 列表
        """

    async def aload(self) -> List[Document]:
        """异步加载文档"""
        return await asyncio.to_thread(self.load)

    def lazy_load(self) -> Iterator[Document]:
        """
        懒加载文档

        Yields:
            Document 对象
        """

    async def alazy_load(self) -> AsyncIterator[Document]:
        """
        异步懒加载

        Yields:
            Document 对象
        """

from langchain_core.documents import Document

class Document(Serializable):
    """文档对象"""

    page_content: str
    """文档内容"""

    metadata: Dict[str, Any] = {}
    """元数据字典"""

    type: Literal["Document"] = "Document"
    """类型标识"""

    def __init__(
        self,
        page_content: str,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ):
        """
        初始化文档

        Args:
            page_content: 文档内容
            metadata: 元数据
            **kwargs: 额外参数
        """

    def to_json(self) -> str:
        """转换为 JSON"""

    @classmethod
    def from_json(cls, json_str: str) -> "Document":
        """从 JSON 创建文档"""

from langchain_community.document_loaders import TextLoader

class TextLoader(BaseLoader):
    """文本文件加载器"""

    def __init__(
        self,
        file_path: str,
        encoding: Optional[str] = None,
        autodetect_encoding: bool = False,
    ):
        """
        初始化

        Args:
            file_path: 文件路径
            encoding: 文件编码
            autodetect_encoding: 是否自动检测编码
        """

    def load(self) -> List[Document]:
        """加载文本文件"""

from langchain_community.document_loaders import PyPDFLoader

class PyPDFLoader(BaseLoader):
    """PDF 文件加载器"""

    def __init__(
        self,
        file_path: str,
        password: Optional[str] = None,
        extract_images: bool = False,
    ):
        """
        初始化

        Args:
            file_path: PDF 文件路径
            password: PDF 密码（如有）
            extract_images: 是否提取图片
        """

    def load(self) -> List[Document]:
        """加载 PDF（每页一个 Document）"""

from langchain_community.document_loaders import WebBaseLoader

class WebBaseLoader(BaseLoader):
    """网页加载器"""

    def __init__(
        self,
        web_path: Union[str, List[str]],
        header_template: Optional[Dict[str, str]] = None,
        verify_ssl: bool = True,
        requests_per_second: Optional[int] = None,
        continue_on_failure: bool = False,
    ):
        """
        初始化

        Args:
            web_path: URL 或 URL 列表
            header_template: HTTP 请求头
            verify_ssl: 是否验证 SSL
            requests_per_second: 请求速率限制
            continue_on_failure: 失败时是否继续
        """

    def load(self) -> List[Document]:
        """加载网页"""

from langchain_community.document_loaders import DirectoryLoader

class DirectoryLoader(BaseLoader):
    """目录加载器"""

    def __init__(
        self,
        path: str,
        glob: str = "**/[!.]*",
        silent_errors: bool = False,
        load_hidden: bool = False,
        loader_cls: LoaderType = TextLoader,
        loader_kwargs: Optional[Dict[str, Any]] = None,
        recursive: bool = False,
        show_progress: bool = False,
        use_multithreading: bool = False,
        max_concurrency: int = 4,
    ):
        """
        初始化

        Args:
            path: 目录路径
            glob: 文件匹配模式
            silent_errors: 是否忽略错误
            load_hidden: 是否加载隐藏文件
            loader_cls: 使用的加载器类
            loader_kwargs: 传递给加载器的参数
            recursive: 是否递归加载
            show_progress: 是否显示进度
            use_multithreading: 是否使用多线程
            max_concurrency: 最大并发数
        """

    def load(self) -> List[Document]:
        """加载目录中的所有文件"""

from langchain_community.document_loaders import JSONLoader

class JSONLoader(BaseLoader):
    """JSON 文件加载器"""

    def __init__(
        self,
        file_path: str,
        jq_schema: Optional[str] = None,
        content_key: Optional[str] = None,
        metadata_func: Optional[Callable[[Dict, Dict], Dict]] = None,
        text_content: bool = True,
        json_lines: bool = false,
    ):
        """
        初始化

        Args:
            file_path: JSON 文件路径
            jq_schema: jq 表达式（用于选择数据）
            content_key: 内容字段名
            metadata_func: 元数据生成函数
            text_content: 是否提取文本内容
            json_lines: 是否为 JSON Lines 格式
        """

    def load(self) -> List[Document]:
        """加载 JSON 文件"""