第9章 文档加载与文本分割

RAG（检索增强生成）系统的第一步是将外部知识源中的数据转换为可检索的文档片段。这个过程涉及两个关键环节：文档加载（从各种数据源读取原始内容）和文本分割（将长文档切分为适合嵌入和检索的小块）。LangChain 为这两个环节构建了精心设计的抽象层，使得开发者可以统一地处理 PDF、网页、数据库、代码仓库等千差万别的数据源。

本章将从 Document 这个核心数据结构开始，逐步展开 BaseLoader、Blob/BlobLoader 的加载体系，深入剖析文本分割器的算法细节，特别是 RecursiveCharacterTextSplitter 的递归分割策略。

9.1 Document：检索流水线的核心数据结构

Document 是 LangChain 中最基础的数据结构之一，它代表一个可检索的文本单元。其设计刻意简洁：

# langchain_core/documents/base.py
class Document(BaseMedia):
    page_content: str
    """String text."""
    type: Literal["Document"] = "Document"

    def __init__(self, page_content: str, **kwargs: Any) -> None:
        super().__init__(page_content=page_content, **kwargs)

Document 只有两个核心字段：page_content（文本内容）和从 BaseMedia 继承的 metadata（元数据字典）。这种极简设计是有意为之的 -- 它使得 Document 可以作为整个检索流水线的通用数据载体，从文档加载到文本分割，再到向量存储和检索，所有组件都以 Document 为中心。

9.1.1 BaseMedia：基础媒体抽象

class BaseMedia(Serializable):
    id: str | None = Field(default=None, coerce_numbers_to_str=True)
    metadata: dict = Field(default_factory=dict)

BaseMedia 提供了 id 和 metadata 两个通用字段。id 是可选的文档标识符，设计上建议使用 UUID 格式。coerce_numbers_to_str=True 的设定允许传入数字类型的 ID 并自动转为字符串，提升了 API 的容错性。

9.1.2 Document 与消息的区别

源码中的注释特别强调了一点：

class Document(BaseMedia):
    """Class for storing a piece of text and associated metadata.

    Note:
        `Document` is for **retrieval workflows**, not chat I/O.
        For sending text to an LLM in a conversation,
        use message types from `langchain.messages`.
    """

这个区分非常重要。Document 用于数据处理流水线（加载、分割、嵌入、检索），而 BaseMessage 用于对话 I/O。虽然两者都承载文本，但它们服务于不同的架构层。

classDiagram
    class Serializable {
        +is_lc_serializable() bool
        +get_lc_namespace() list
    }

    class BaseMedia {
        +id: str | None
        +metadata: dict
    }

    class Blob {
        +data: bytes | str | None
        +mimetype: str | None
        +encoding: str
        +path: PathLike | None
        +source: str | None
        +as_string() str
        +as_bytes() bytes
        +as_bytes_io() Generator
        +from_path(path) Blob$
        +from_data(data) Blob$
    }

    class Document {
        +page_content: str
        +type: Literal["Document"]
    }

    Serializable <|-- BaseMedia
    BaseMedia <|-- Blob
    BaseMedia <|-- Document

9.1.3 str 方法的考量

Document 重写了 __str__ 方法：

def __str__(self) -> str:
    if self.metadata:
        return f"page_content='{self.page_content}' metadata={self.metadata}"
    return f"page_content='{self.page_content}'"

这个重写排除了 id 字段和其他可能在未来添加的字段。注释解释了原因：确保将 Document 直接嵌入 Prompt 的用户代码不会因为新增字段而中断。这是一个向后兼容性的务实考量。

9.1.4 Document / Blob / BaseMedia 字段总表（截至 main 2026-04-22）

langchain_core/documents/base.py 全文 354 行，三个类各自声明的字段、默认值与来源行号如下——把它集中列成一张表，便于与 BaseMessage 族系对照，避免"Document 有个 role 字段吗？"这类搜索多遍源码才能回答的问题：

三条不显眼但很重要的设计约束：

1. Document 没有 role / tool_call_id / response_metadata——全是消息族系（BaseMessage）才有的字段。想同时存"谁写的"+"文本"，要自己塞进 metadata["author"]，不要指望 Document 原生提供。
2. Blob.source 是 computed 字段——它不能由 __init__ 直接赋值，而是 Pydantic 的 model_validator 从 metadata["source"] 或 path 派生。这意味着两个 Blob，只要 path 相同、source 就一定相同——测试里可以放心用 source 做对比键。
3. Document.__init__ 把 page_content 提前到第一个位置参（而非 kwargs-only），这是 LangChain 少有的"人体工学破例"——正因为 Document(text, metadata={...}) 是一百万行用户代码里最常写的两行之一，强制关键字传参会让所有示例代码变丑；所以源码里专门写了 def __init__(self, page_content: str, **kwargs) 而非让 Pydantic 自动生成。

9.1.5 和消息族系的字段对照

Document 经常被误当成 HumanMessage 使用（反之亦然）——以下对照表可以消除混淆：

一个实操含义：如果你从向量库检索到 Document 后要把它喂给 LLM，不要直接 llm.invoke(doc)——它的 content 字段不兼容。正确做法是 HumanMessage(content=f"参考资料：{doc.page_content}\n\n问题：{query}") 或用 create_stuff_documents_chain 这类专门把 Document 压扁成 Prompt 的 chain。

9.2 Blob：原始数据的抽象

Blob 是文档加载过程中的中间表示，它代表一段原始数据 -- 可以是内存中的字节流，也可以是文件系统中的路径引用。

class Blob(BaseMedia):
    data: bytes | str | None = None
    mimetype: str | None = None
    encoding: str = "utf-8"
    path: PathLike | None = None

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        frozen=True,  # Blob 是不可变的
    )

Blob 的 frozen=True 配置使其成为不可变对象，这是一个重要的设计选择 -- 它确保了 Blob 在流水线中传递时不会被意外修改。

9.2.1 延迟加载策略

Blob 采用延迟加载模式。当通过 from_path 创建时，它不会立即读取文件内容：

@classmethod
def from_path(cls, path, *, encoding="utf-8", mime_type=None, guess_type=True, metadata=None):
    if mime_type is None and guess_type:
        mimetype = mimetypes.guess_type(path)[0]
    else:
        mimetype = mime_type
    # 不加载数据，只保存路径引用
    return cls(data=None, mimetype=mimetype, encoding=encoding, path=path, metadata=metadata or {})

数据只在实际需要时才被读取：

def as_string(self) -> str:
    if self.data is None and self.path:
        return Path(self.path).read_text(encoding=self.encoding)
    if isinstance(self.data, bytes):
        return self.data.decode(self.encoding)
    if isinstance(self.data, str):
        return self.data

这种延迟加载模式在处理大量文件时非常重要 -- 它避免了将所有文件内容一次性加载到内存中。

9.2.2 多种数据访问方式

Blob 提供了三种数据访问接口，适应不同的使用场景：

as_bytes_io() 是一个上下文管理器，当数据来自文件路径时，它直接返回文件句柄而不是将整个文件加载到内存：

@contextlib.contextmanager
def as_bytes_io(self) -> Generator[BytesIO | BufferedReader, None, None]:
    if isinstance(self.data, bytes):
        yield BytesIO(self.data)
    elif self.data is None and self.path:
        with Path(self.path).open("rb") as f:
            yield f  # 直接 yield 文件句柄

9.3 文档加载体系

LangChain 的文档加载体系由三个核心抽象组成：BaseLoader、BlobLoader 和 BaseBlobParser。

9.3.1 BaseLoader：统一加载接口

# langchain_core/document_loaders/base.py
class BaseLoader(ABC):
    def load(self) -> list[Document]:
        """Load data into Document objects."""
        return list(self.lazy_load())

    def lazy_load(self) -> Iterator[Document]:
        """A lazy loader for Document."""
        if type(self).load != BaseLoader.load:
            return iter(self.load())
        raise NotImplementedError(...)

    async def alazy_load(self) -> AsyncIterator[Document]:
        """A lazy loader for Document."""
        iterator = await run_in_executor(None, self.lazy_load)
        done = object()
        while True:
            doc = await run_in_executor(None, next, iterator, done)
            if doc is done:
                break
            yield doc

这段代码有几个值得注意的设计决策：

1. lazy_load 优先：load 方法的默认实现调用 lazy_load 并收集为列表。鼓励子类实现 lazy_load 而非 load，以支持惰性加载
2. 向后兼容：如果子类覆盖了 load 但没有覆盖 lazy_load，通过检测 type(self).load != BaseLoader.load 来回退
3. 异步桥接：alazy_load 的默认实现将同步迭代器包装到线程池中，每次调用 next 都在 executor 中执行

flowchart TD
    subgraph BaseLoader
        A[load] -->|默认实现| B[list of lazy_load]
        C[lazy_load] -->|子类实现| D[yield Document]
        E[alazy_load] -->|默认实现| F[run_in_executor of lazy_load]
    end

    subgraph 子类实现
        G[TextLoader] --> C
        H[PDFLoader] --> C
        I[WebBaseLoader] --> C
    end

    C --> J[load_and_split]
    J --> K[TextSplitter.split_documents]

9.3.2 load_and_split：便捷但不推荐

BaseLoader 提供了一个 load_and_split 方法，将加载和分割合并为一步操作：

def load_and_split(self, text_splitter: TextSplitter | None = None) -> list[Document]:
    if text_splitter is None:
        text_splitter_ = RecursiveCharacterTextSplitter()
    else:
        text_splitter_ = text_splitter
    docs = self.load()
    return text_splitter_.split_documents(docs)

注释中标记了 Do not override this method. It should be considered to be deprecated!。这暗示了 LangChain 更倾向于让开发者显式地分开调用加载和分割步骤，以获得更好的控制粒度。

9.3.3 BlobLoader 与 BaseBlobParser：解耦加载与解析

Blob 加载体系将文档加载分解为两个独立的步骤：

# langchain_core/document_loaders/blob_loaders.py
class BlobLoader(ABC):
    @abstractmethod
    def yield_blobs(self) -> Iterator[Blob]:
        """A lazy loader for raw data represented by Blob."""

# langchain_core/document_loaders/base.py
class BaseBlobParser(ABC):
    @abstractmethod
    def lazy_parse(self, blob: Blob) -> Iterator[Document]:
        """Lazy parsing interface."""

    def parse(self, blob: Blob) -> list[Document]:
        """Eagerly parse the blob."""
        return list(self.lazy_parse(blob))

flowchart LR
    A[数据源] --> B[BlobLoader]
    B -->|yield_blobs| C[Blob 流]
    C --> D[BaseBlobParser]
    D -->|lazy_parse| E[Document 流]
    E --> F[TextSplitter]
    F --> G[分割后的 Document]
    G --> H[VectorStore]

这种解耦的好处是：

• 复用：同一个 BlobLoader（如文件系统 loader）可以搭配不同的 BaseBlobParser（如 PDF parser、CSV parser）
• 组合灵活性：可以构建管道，如先过滤特定 MIME 类型的 Blob，再传递给对应的 parser
• 测试友好：可以独立测试加载和解析逻辑

9.3.4 document_loaders 源码账本（截至 main 2026-04-22）

把 langchain_core/document_loaders/ 三个核心文件的行数与职责一次性列清，方便读者定位和二次开发：

core 里只有这三份 —— 加起来才 ~250 行。真正的"loader 动物园"不在 core 里，而是分散在两个地方：

1. langchain_community.document_loaders —— 第三方数据源的 loader 实现集合（约 180+ 个文件）。涵盖 PDF、Word、Excel、HTML、Notion、Confluence、Slack、Discord、Google Drive、S3、Airbyte、Figma、Airtable、Arxiv、PubMed、Wikipedia 等。
2. 各专用包 —— 如 langchain_unstructured 专门走 Unstructured.io API、langchain_pymupdf4llm 专门走 PyMuPDF、langchain_docling 专门走 Docling。

按数据源类型划分的主流 loader 子模块清单（来自 langchain-community main）：

Core / Community 的分层不是随便切的——把"ABC + 一个 LangSmith 开箱即用的 loader"放 core，让 langchain-core 的依赖收敛到 pydantic / httpx 几个；而"每接一家就要引一个 SDK"（PyPDF、boto3、playwright、google-cloud-storage）都被推进 community 或专用小包。这样下游引入 langchain-core 的用户，不需要为了一个抽象基类被迫装 100 MB 的 SDK 依赖。

9.3.5 自己实现一个 BlobLoader 的最小骨架

为了让"解耦"这件事具象化，给出最小可用的 FileSystem BlobLoader + PDF Parser 骨架：

from pathlib import Path
from langchain_core.document_loaders import BlobLoader, BaseBlobParser
from langchain_core.documents import Blob, Document

class GlobBlobLoader(BlobLoader):
    def __init__(self, root: str, pattern: str = "**/*"):
        self.root = Path(root)
        self.pattern = pattern

    def yield_blobs(self):  # 唯一抽象方法
        for p in self.root.glob(self.pattern):
            if p.is_file():
                yield Blob.from_path(p)  # 不读内容，只记路径

class MyPDFParser(BaseBlobParser):
    def lazy_parse(self, blob: Blob):
        import pypdf
        with blob.as_bytes_io() as f:  # 拿文件句柄、不全量加载
            reader = pypdf.PdfReader(f)
            for i, page in enumerate(reader.pages):
                yield Document(
                    page_content=page.extract_text() or "",
                    metadata={"source": str(blob.path), "page": i},
                )

# 组合
loader = GlobBlobLoader("./docs", "**/*.pdf")
parser = MyPDFParser()
docs = [d for blob in loader.yield_blobs() for d in parser.lazy_parse(blob)]

注意三处细节——Blob.from_path 不读文件；blob.as_bytes_io() 才触发 open；Document.metadata["source"] 约定俗成地存文件路径，许多 Retriever / re-ranker 会默认从这里取来源。

9.4 文本分割器体系

文本分割是 RAG 系统中最关键也最容易被低估的环节。分割策略直接影响检索质量：块太大，嵌入向量可能丢失精确语义；块太小，又可能丢失上下文。LangChain 的文本分割器位于独立的 langchain-text-splitters 包中，以 TextSplitter 为基类构建了丰富的分割器体系。

9.4.1 TextSplitter 基类

# langchain_text_splitters/base.py
class TextSplitter(BaseDocumentTransformer, ABC):
    def __init__(
        self,
        chunk_size: int = 4000,
        chunk_overlap: int = 200,
        length_function: Callable[[str], int] = len,
        keep_separator: bool | Literal["start", "end"] = False,
        add_start_index: bool = False,
        strip_whitespace: bool = True,
    ) -> None:
        if chunk_size <= 0:
            raise ValueError(f"chunk_size must be > 0, got {chunk_size}")
        if chunk_overlap > chunk_size:
            raise ValueError("chunk_overlap must be <= chunk_size")
        self._chunk_size = chunk_size
        self._chunk_overlap = chunk_overlap
        self._length_function = length_function
        ...

核心参数解析：

length_function 的可替换性是一个关键设计点。默认使用 Python 的 len（按字符数计算），但在实际应用中，通常需要按 token 数计算长度（因为模型的上下文窗口以 token 为单位）。from_tiktoken_encoder 和 from_huggingface_tokenizer 类方法就是为此提供的便捷构造器。

9.4.2 核心方法链

@abstractmethod
def split_text(self, text: str) -> list[str]:
    """Split text into multiple components."""

def create_documents(self, texts, metadatas=None) -> list[Document]:
    """Create Document objects from texts."""
    documents = []
    for i, text in enumerate(texts):
        index = 0
        previous_chunk_len = 0
        for chunk in self.split_text(text):
            metadata = copy.deepcopy(metadatas_[i])
            if self._add_start_index:
                offset = index + previous_chunk_len - self._chunk_overlap
                index = text.find(chunk, max(0, offset))
                metadata["start_index"] = index
                previous_chunk_len = len(chunk)
            documents.append(Document(page_content=chunk, metadata=metadata))
    return documents

def split_documents(self, documents: Iterable[Document]) -> list[Document]:
    texts, metadatas = [], []
    for doc in documents:
        texts.append(doc.page_content)
        metadatas.append(doc.metadata)
    return self.create_documents(texts, metadatas=metadatas)

add_start_index 的实现值得仔细看：它使用 text.find(chunk, max(0, offset)) 来定位每个 chunk 在原文中的位置。offset 的计算考虑了 chunk 重叠，确保搜索起点不会跳过目标位置。

9.4.3 _merge_splits：块合并的核心算法

_merge_splits 是所有文本分割器共享的核心方法，它将初始的小段文本合并为目标大小的块：

def _merge_splits(self, splits: Iterable[str], separator: str) -> list[str]:
    separator_len = self._length_function(separator)
    docs = []
    current_doc: list[str] = []
    total = 0
    for d in splits:
        len_ = self._length_function(d)
        if (total + len_ + (separator_len if len(current_doc) > 0 else 0) > self._chunk_size):
            if len(current_doc) > 0:
                doc = self._join_docs(current_doc, separator)
                if doc is not None:
                    docs.append(doc)
                # 回退以保持重叠
                while total > self._chunk_overlap or (
                    total + len_ + (separator_len if len(current_doc) > 0 else 0)
                    > self._chunk_size and total > 0
                ):
                    total -= self._length_function(current_doc[0]) + (
                        separator_len if len(current_doc) > 1 else 0
                    )
                    current_doc = current_doc[1:]
        current_doc.append(d)
        total += len_ + (separator_len if len(current_doc) > 1 else 0)
    # 处理最后一个块
    doc = self._join_docs(current_doc, separator)
    if doc is not None:
        docs.append(doc)
    return docs

这个算法的核心逻辑可以用以下图示说明：

flowchart TD
    A[输入 splits 序列] --> B[初始化 current_doc 和 total]
    B --> C{遍历每个 split}
    C --> D{current_doc + split 超过 chunk_size?}
    D -->|否| E[添加 split 到 current_doc]
    D -->|是| F[输出 current_doc 为一个 chunk]
    F --> G[回退 current_doc 保留 overlap 部分]
    G --> H{total <= chunk_overlap?}
    H -->|否| I[移除 current_doc 首元素]
    I --> G
    H -->|是| E
    E --> J[更新 total]
    J --> C
    C -->|遍历结束| K[输出最后的 current_doc]

算法的关键在于回退机制：当需要开始一个新的 chunk 时，不是完全清空 current_doc，而是从头部逐步移除元素，直到剩余部分的长度不超过 chunk_overlap。这确保了相邻 chunk 之间有适当的重叠，避免语义断裂。

9.4.4 keep_separator 的三种模式

keep_separator 控制分隔符的处理方式：

def _split_text_with_regex(text, separator, *, keep_separator):
    if separator:
        if keep_separator:
            splits_ = re.split(f"({separator})", text)
            if keep_separator == "end":
                # 分隔符追加到前一个块的末尾
                splits = [splits_[i] + splits_[i + 1] for i in range(0, len(splits_) - 1, 2)]
            else:
                # 分隔符追加到后一个块的开头 (keep_separator == "start" 或 True)
                splits = [splits_[i] + splits_[i + 1] for i in range(1, len(splits_), 2)]
        else:
            splits = re.split(separator, text)
    else:
        splits = list(text)  # 逐字符分割
    return [s for s in splits if s]

三种模式的效果：

原文: "Hello\n\nWorld\n\nFoo"
分隔符: "\n\n"

keep_separator=False:  ["Hello", "World", "Foo"]
keep_separator="start": ["\n\nHello", "\n\nWorld", "\n\nFoo"]  # 首段前无分隔符
keep_separator="end":   ["Hello\n\n", "World\n\n", "Foo"]

9.5 RecursiveCharacterTextSplitter：递归分割的艺术

RecursiveCharacterTextSplitter 是 LangChain 中最重要也最常用的文本分割器。它的核心思想是：尝试用最大粒度的分隔符分割文本，如果某个片段仍然超过 chunk_size，则递归使用更细粒度的分隔符继续分割。

9.5.1 默认分隔符层级

class RecursiveCharacterTextSplitter(TextSplitter):
    def __init__(self, separators=None, keep_separator=True, is_separator_regex=False, **kwargs):
        super().__init__(keep_separator=keep_separator, **kwargs)
        self._separators = separators or ["\n\n", "\n", " ", ""]
        self._is_separator_regex = is_separator_regex

默认的分隔符序列 ["\n\n", "\n", " ", ""] 代表了从粗到细的四个粒度层级：

9.5.2 递归分割算法

def _split_text(self, text: str, separators: list[str]) -> list[str]:
    final_chunks = []
    # 找到第一个能匹配到的分隔符
    separator = separators[-1]
    new_separators = []
    for i, s_ in enumerate(separators):
        separator_ = s_ if self._is_separator_regex else re.escape(s_)
        if not s_:
            separator = s_
            break
        if re.search(separator_, text):
            separator = s_
            new_separators = separators[i + 1:]
            break

    # 使用选定的分隔符分割
    separator_ = separator if self._is_separator_regex else re.escape(separator)
    splits = _split_text_with_regex(text, separator_, keep_separator=self._keep_separator)

    # 合并小段，递归处理大段
    good_splits = []
    separator_ = "" if self._keep_separator else separator
    for s in splits:
        if self._length_function(s) < self._chunk_size:
            good_splits.append(s)
        else:
            if good_splits:
                merged_text = self._merge_splits(good_splits, separator_)
                final_chunks.extend(merged_text)
                good_splits = []
            if not new_separators:
                final_chunks.append(s)  # 无法继续分割，直接添加
            else:
                # 递归：使用更细粒度的分隔符继续分割
                other_info = self._split_text(s, new_separators)
                final_chunks.extend(other_info)
    if good_splits:
        merged_text = self._merge_splits(good_splits, separator_)
        final_chunks.extend(merged_text)
    return final_chunks

flowchart TD
    A["_split_text(text, separators)"] --> B[找到文本中存在的最粗分隔符]
    B --> C["用该分隔符分割文本"]
    C --> D{遍历每个片段}
    D --> E{片段长度 < chunk_size?}
    E -->|是| F[加入 good_splits]
    E -->|否| G{还有更细的分隔符?}
    G -->|是| H["递归: _split_text(片段, 更细分隔符)"]
    G -->|否| I[直接作为 chunk 但可能超长]
    H --> J[合并递归结果到 final_chunks]
    F --> K{下一个片段超长?}
    K -->|是| L[先 merge good_splits]
    L --> G
    K -->|否| D
    D -->|遍历结束| M[merge 剩余的 good_splits]
    M --> N[返回 final_chunks]

让我们用一个具体例子来理解这个算法：

假设 chunk_size=50，chunk_overlap=10，输入文本为：

这是第一段内容。

这是第二段内容，比较长比较长比较长比较长比较长比较长比较长比较长比较长。

短段。

1. 首先用 "\n\n" 分割，得到三段
2. 第一段和第三段长度小于 50，进入 good_splits
3. 第二段长度超过 50，递归用 "\n" 分割
4. 如果仍然超长，继续递归用 " " 分割
5. 最终用 " " 分割后的片段通过 _merge_splits 合并为不超过 50 的 chunk

9.5.3 语言感知分割

RecursiveCharacterTextSplitter 提供了 from_language 类方法，为不同编程语言预定义了分隔符序列：

@classmethod
def from_language(cls, language: Language, **kwargs):
    separators = cls.get_separators_for_language(language)
    return cls(separators=separators, is_separator_regex=True, **kwargs)

以 Python 为例：

if language == Language.PYTHON:
    return [
        "\nclass ",     # 类定义边界
        "\ndef ",       # 函数定义边界
        "\n\tdef ",     # 方法定义边界
        "\n\n",         # 段落边界
        "\n",           # 行边界
        " ",            # 单词边界
        "",             # 字符边界
    ]

这种语言感知的分隔符设计确保代码在语义边界处被分割，而不是在函数或类的中间断开。LangChain 目前支持 Python、JavaScript、TypeScript、Java、Go、Rust、C/C++、Markdown、HTML、LaTeX 等近 30 种语言。

9.5.3 from_language：28 种编程语言的语法感知分割

默认 ["\n\n", "\n", " ", ""] 对自然文本很好，但对代码就不合适——在一个函数中间断开会让 chunk 失去语义完整性。LangChain 提供 RecursiveCharacterTextSplitter.from_language(Language.PYTHON) 让分隔符按语言语法定制（langchain_text_splitters/character.py:161）：

@classmethod
def from_language(
    cls, language: Language, **kwargs: Any
) -> RecursiveCharacterTextSplitter:
    separators = cls.get_separators_for_language(language)
    return cls(separators=separators, is_separator_regex=True, **kwargs)

关键细节：is_separator_regex=True 默认开启——因为语言分隔符需要用 regex 表达"行首匹配"。例如 \nclass 要求分隔符前必须是换行——否则字符串里出现的 class= 会被误切。

Language 枚举（base.py:380）列出了 28 种支持的语言：

class Language(str, Enum):
    CPP, GO, JAVA, KOTLIN, JS, TS, PHP, PROTO, PYTHON,
    R, RST, RUBY, RUST, SCALA, SWIFT, MARKDOWN, LATEX,
    HTML, SOL, CSHARP, COBOL, C, LUA, PERL, HASKELL,
    ELIXIR, POWERSHELL, VISUALBASIC6

从主流（Python/Go/JS）到冷门（COBOL/Elixir/VisualBasic6）都有。把语言 enum 公开作为 API 参数——让用户代码可以写 Language.PYTHON 而不是字符串 "python"、拼错能在编译期抓住。

各语言的分隔符设计有明确逻辑分层，以 Python 为例（character.py:363-374）：

if language == Language.PYTHON:
    return [
        # First, try to split along class definitions
        "\nclass ",
        "\ndef ",
        "\n\tdef ",          # ← 方法（缩进过的 def）
        # Now split by the normal type of lines
        "\n\n",
        "\n",
        " ",
        "",
    ]

三段式：

1. 语法级分隔符：\nclass / \ndef —— 顶层类和函数定义。放最前面是因为这是最有语义完整性的切点（一个类/函数内尽量不被切开）。
2. \n\tdef ——嵌套方法（一级缩进的 def）。Python 里 class 里的方法缩进一级、这个分隔符专门切 class body。实际只切 tab 缩进的 method、不切 space 缩进的——这是 LangChain 一个已知的简化（和标准 PEP 8 的 4-space 缩进不完全匹配）。
3. 通用分隔符：\n\n / \n / / "" 兜底，代码过长强制切到字符级。

对照 C/C++ 的分隔符（character.py:191-211）更丰富：

"\nclass ", "\nvoid ", "\nint ", "\nfloat ", "\ndouble ",   # ← 函数签名
"\nif ", "\nfor ", "\nwhile ", "\nswitch ", "\ncase ",       # ← 控制流
"\n\n", "\n", " ", "",                                       # ← 通用

C/C++ 没有 def 关键字、函数通过返回类型 + 名字声明——所以要列 void/int/float/double 这些常见返回类型。switch/case 作为分隔符是因为 C/C++ 的 switch 块经常独立成逻辑单元。

R 语言（line 375-397）甚至把包导入（library(/require(）作为分隔符——R 代码文件经常开头一堆 library() 声明、这些作为 import 块被隔开合理。

这些分隔符列表是 LangChain 社区多年积累的"每门语言的切块直觉"。不完美（比如没覆盖 JS 的 => arrow function），但覆盖了大多数常见模式。如果你为某语言做 RAG、先用 from_language(Language.YOUR_LANG)、实测效果后再考虑自定义 separators。

9.6 CharacterTextSplitter：简单字符分割

与递归分割器相比，CharacterTextSplitter 使用单一分隔符进行分割：

class CharacterTextSplitter(TextSplitter):
    def __init__(self, separator="\n\n", is_separator_regex=False, **kwargs):
        super().__init__(**kwargs)
        self._separator = separator
        self._is_separator_regex = is_separator_regex

    def split_text(self, text: str) -> list[str]:
        sep_pattern = self._separator if self._is_separator_regex else re.escape(self._separator)
        splits = _split_text_with_regex(text, sep_pattern, keep_separator=self._keep_separator)

        # 检测零宽断言，避免重复插入分隔符
        lookaround_prefixes = ("(?=", "(?<!", "(?<=", "(?!")
        is_lookaround = self._is_separator_regex and any(
            self._separator.startswith(p) for p in lookaround_prefixes
        )
        merge_sep = ""
        if not (self._keep_separator or is_lookaround):
            merge_sep = self._separator
        return self._merge_splits(splits, merge_sep)

值得注意的是对正则零宽断言（lookaround）的特殊处理。当分隔符是零宽断言时（如 (?<=\.)，在句号后分割），分隔符本身不消耗字符，因此合并时不应再次插入分隔符。

9.7 TokenTextSplitter：基于 Token 的分割

当需要精确控制每个 chunk 的 token 数量时，TokenTextSplitter 直接在 token 层面进行分割：

class TokenTextSplitter(TextSplitter):
    def split_text(self, text: str) -> list[str]:
        def _encode(_text):
            return self._tokenizer.encode(_text, allowed_special=..., disallowed_special=...)

        tokenizer = Tokenizer(
            chunk_overlap=self._chunk_overlap,
            tokens_per_chunk=self._chunk_size,
            decode=self._tokenizer.decode,
            encode=_encode,
        )
        return split_text_on_tokens(text=text, tokenizer=tokenizer)

split_text_on_tokens 的实现是直接的滑动窗口：

def split_text_on_tokens(*, text, tokenizer):
    splits = []
    input_ids = tokenizer.encode(text)
    start_idx = 0
    while start_idx < len(input_ids):
        cur_idx = min(start_idx + tokenizer.tokens_per_chunk, len(input_ids))
        chunk_ids = input_ids[start_idx:cur_idx]
        decoded = tokenizer.decode(chunk_ids)
        if decoded:
            splits.append(decoded)
        if cur_idx == len(input_ids):
            break
        start_idx += tokenizer.tokens_per_chunk - tokenizer.chunk_overlap
    return splits

这种方法保证了每个 chunk 的 token 数精确符合要求，但可能在单词或语义边界的中间断开文本。在实践中，通常推荐使用 RecursiveCharacterTextSplitter.from_tiktoken_encoder，它结合了语义感知分割和 token 计数。

9.7b 九种切分策略全景对比（截至 main 2026-04-22）

langchain-text-splitters 包里实际暴露的 splitter 有 13 个，来源文件分布在 13 个模块。把它们对照成一张矩阵，能一眼看出每一种的"切点语义"、"长度度量"、"典型 RAG 场景"：

三条可以直接拿来做决策的经验法则：

1. 不知道切什么格式 → 默认用 RecursiveCharacterTextSplitter.from_tiktoken_encoder(chunk_size=512, chunk_overlap=64)。它是"懒人万能方案"：递归语义感知 + token 精确计数。
2. 原文有天然大纲（Markdown / HTML） → 优先用对应的 Header splitter 把大纲信息写进 metadata，再在 retrieval 阶段把 header 作为 prefix 拼回 prompt。没有这一步，chunk 容易失去层级上下文（"这一段属于哪一节"检索后无法还原）。
3. 英文长 academic 文本 → 试 SpacyTextSplitter 或 NLTKTextSplitter，句子边界通常比字符边界对嵌入模型更友好；但对中文它们退化为按标点切，效果不一定好，中文场景仍然推荐递归字符切分。

9.7c _merge_splits 的最小可跑例子

很多人卡在"chunk_overlap 到底是字符重叠还是块列表重叠"上。用 RecursiveCharacterTextSplitter 跑一个最短可观察例子即可定性：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=20,
    chunk_overlap=5,
    separators=[" "],       # 强制只按空格切，方便观察
    keep_separator=False,
)
text = "the quick brown fox jumps over the lazy dog again and again"
for i, c in enumerate(splitter.split_text(text)):
    print(i, repr(c))

观察 chunk_overlap=5（字符长度 ≤5）是如何让相邻块共享"dog"或"the"这种单词的——这是 _merge_splits 的回退循环在工作，而不是"最后 5 个 split 元素重叠"。把 chunk_overlap 从 5 调到 10，你会看到相邻块重叠的单词数增加——这是 RAG 调参里最频繁改动的一个值。

9.7d 结构感知切分：Markdown / HTML / JSON 三类特殊 splitter 的内部分工

前面对比表里三类"结构感知"splitter 值得展开讲——因为它们和前述字符 / token splitter 在返回值的 metadata 结构上完全不同，用错了会导致 RAG 链里"拿不到 header 信息"。

MarkdownHeaderTextSplitter

定义在 markdown.py:21，它不是 TextSplitter 的子类（没有 _merge_splits 行为），而是一个独立类：

class MarkdownHeaderTextSplitter:
    def __init__(self, headers_to_split_on, strip_headers=True, ...):
        # headers_to_split_on 形如 [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3")]
        ...

它的切分产物是 list[Document]，每个 Document.metadata 会额外带上当前层级之上所有 header 的路径——比如处在 # 第一章 下 ## 1.1 节 下 ### 实验 段的那块文字，metadata 就是 {"Header 1": "第一章", "Header 2": "1.1 节", "Header 3": "实验"}。

实操技巧：把 MarkdownHeaderTextSplitter 的输出再过一遍 RecursiveCharacterTextSplitter，就能得到"大纲 metadata + 精确长度控制"的组合：先按 header 切保留结构、再按字符 / token 切控制长度——两级流水线是生产级 Markdown RAG 的标准做法。

HTMLHeaderTextSplitter / HTMLSectionSplitter / HTMLSemanticPreservingSplitter

html.py 全文 978 行，容纳三个类——各自对应一种 HTML 切分假设：

三者的共同陷阱：如果上游是 JS 渲染的 SPA（React / Vue），爬下来的 HTML 经常只有一个 <div id="app"></div> + 一堆 script，header 和 section 都是空的，这三个 splitter 都会退化——必须先用 SeleniumURLLoader / PlaywrightURLLoader 走真浏览器渲染。

HTMLSemanticPreservingSplitter 的**"语义保留"**含义是：碰到 <table> 时它不会把表格从中间切断，而是作为一个整体放进一个 chunk（即使这个 chunk 超过 max_chunk_size 也优先保完整性）。这在爬财经 / 百科类网站时非常重要——割裂的半张表格在 LLM 看来是噪声，语义价值为零。

RecursiveJsonSplitter

json.py 全文 153 行，类定义在第 10 行。它不继承 TextSplitter，而是自己实现了一套深度优先拆分 + 序列化长度度量的算法：

class RecursiveJsonSplitter:
    def __init__(self, max_chunk_size=2000, min_chunk_size=None):
        ...

    def split_json(self, json_data: dict, convert_lists=False) -> list[dict]:
        """返回一组子 dict，每个序列化后 <= max_chunk_size。"""

切分算法的核心是：能保持 dict 结构就保持——不会把 {"a": 1, "b": 2} 拆成 "a: 1" 和 "b: 2" 两段字符串，而是尽量拆成 {"a": 1} 和 {"b": 2} 两个仍是合法 JSON 的子对象。

为什么 JSON 要单独一个 splitter——OpenAPI spec、结构化 API 响应、知识图谱导出都是 JSON，字符级切分会把它们切成语义无效的碎片（半个 key、半个 value、不闭合的 {）。保持 JSON 闭合 + 结构，让下游 LLM 可以直接读 / 继续发 JSON。

9.7e 选型决策树

把以上所有内容压缩成一张决策树，方便实操：

flowchart TD
    A[开始: 要切分什么数据?] --> B{原文格式?}
    B -->|纯文本/未知| C{需要精确 token 预算?}
    B -->|Markdown| D[先 MarkdownHeaderTextSplitter 保 header 结构]
    B -->|HTML| E{HTML 有 table/list 要保全?}
    B -->|JSON/API spec| F[RecursiveJsonSplitter]
    B -->|源代码| G[RecursiveCharacterTextSplitter.from_language]
    B -->|LaTeX/arXiv| H[from_language Language.LATEX]

    C -->|是| I[RecursiveCharacterTextSplitter.from_tiktoken_encoder]
    C -->|否| J[RecursiveCharacterTextSplitter]

    D --> K[再串联 RecursiveCharacterTextSplitter 控长度]

    E -->|是| L[HTMLSemanticPreservingSplitter]
    E -->|无特殊要求| M[HTMLHeaderTextSplitter]

    I --> N[产出 Document list]
    J --> N
    K --> N
    L --> N
    M --> N
    F --> N
    G --> N
    H --> N

唯一需要记住的原则：能用结构信息就用结构信息——把 Markdown header、HTML section、JSON key 路径写进 Document.metadata，后面 Retriever + Re-ranker 随时可以拿回来；字符串一旦切碎就再也拼不回层级了。

9.8 BaseDocumentTransformer 与 BaseDocumentCompressor

除了文本分割，LangChain 还定义了两个文档处理抽象。

9.8.1 BaseDocumentTransformer

class BaseDocumentTransformer(ABC):
    @abstractmethod
    def transform_documents(self, documents: Sequence[Document], **kwargs) -> Sequence[Document]:
        """Transform a list of documents."""

    async def atransform_documents(self, documents: Sequence[Document], **kwargs) -> Sequence[Document]:
        return await run_in_executor(None, self.transform_documents, documents, **kwargs)

TextSplitter 继承了这个接口，其 transform_documents 的实现就是调用 split_documents。但这个接口的设计更加通用 -- 它可以表示任何文档转换操作，如去重、翻译、摘要等。

9.8.2 BaseDocumentCompressor

class BaseDocumentCompressor(BaseModel, ABC):
    @abstractmethod
    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Compress retrieved documents given the query context."""

BaseDocumentCompressor 与 BaseDocumentTransformer 的关键区别在于它接受一个 query 参数。这使得压缩操作可以是查询感知的 -- 例如，根据查询相关性对检索结果进行重排序（reranking），或者提取文档中与查询最相关的段落。

flowchart TD
    subgraph 文档处理流水线
        A[原始数据源] --> B[BaseLoader.lazy_load]
        B --> C[Document 流]
        C --> D[TextSplitter.split_documents]
        D --> E[分割后的 Document]
        E --> F[Embeddings.embed_documents]
        F --> G[VectorStore.add_documents]
    end

    subgraph 检索后处理
        H[用户查询] --> I[Retriever.invoke]
        I --> J[检索到的 Document]
        J --> K[BaseDocumentCompressor.compress_documents]
        K --> L[压缩/重排后的 Document]
        L --> M[送入 LLM]
    end

9.8.3 langchain_text_splitters/ 13 文件 3531 行的模块拆分

独立成 langchain-text-splitters 包后、这个 crate 的物理拆分也值得分析——

三点看得出设计取舍——

1. HTML 为什么比 markdown 大 2 倍——HTML 的 tag 嵌套 + 属性 + 注释 + CDATA 比 markdown 复杂太多；markdown 只需处理 #/```/--- 几种边界；HTML 要维护 tag stack
2. latex.py / python.py 各 17 行只定义一个类别别名——这是 LangChain 对"向后兼容"的体现——早期 PythonCodeTextSplitter 是独立类、现在统一走 from_language、但老代码 import PythonCodeTextSplitter 必须能继续 work——薄壳保留 API 稳定性
3. sentence_transformers/spacy/nltk/konlpy 四份 tokenizer 包装都在 100 行内——因为重活（token 化）由依赖完成、splitter 只做"按 tokenizer 切分界限做文本切片"的胶水——再次印证 LangChain"把 tokenizer 当外部组件注入、splitter 保持纯逻辑"的抽象纪律

9.9 设计决策分析

为什么文本分割器是独立包？

LangChain 将文本分割器放在了独立的 langchain-text-splitters 包中。这个决策的原因有二：

1. 依赖隔离：某些分割器依赖 tiktoken、transformers 等重型包，将其独立可以避免核心包的依赖膨胀
2. 复用性：文本分割不仅用于 RAG，还用于长文档处理、代码分析等场景，独立包便于在非 LangChain 项目中使用

Blob 的不可变设计

Blob 被设计为 frozen=True 的不可变对象。这个选择在并发场景中尤其重要 -- 当多个 parser 同时处理同一个 Blob 时，不可变性确保了线程安全。同时，不可变对象可以安全地被缓存。

chunk_overlap 的作用

chunk_overlap 是 RAG 质量的一个关键杠杆。重叠区域确保了跨 chunk 边界的信息不会完全丢失。但过大的重叠会增加存储和计算成本。经验上，重叠比例通常设为 chunk_size 的 5%-20%。

为什么 load_and_split 被标记为弃用？

将加载和分割耦合在一个方法中违反了单一职责原则。分开调用的好处是：

• 可以在分割前对文档进行过滤或预处理
• 可以选择不同的分割器处理不同类型的文档
• 更容易调试和测试

为什么 BaseLoader.lazy_load 不直接返回 async iterator？

BaseLoader 的 lazy_load 返回 Iterator[Document]（同步），而 alazy_load 是一个独立的异步方法——两者默认实现会互相桥接。这种"双接口"设计是 LangChain 整个架构的缩影：

1. 历史包袱：早期 loader（2022 - 2023 年）大量走同步文件 I/O、PDF parser 里 pypdf / pdfplumber 也都是同步阻塞 API；强行改 async 会破坏所有已有 loader。
2. 桥接成本低：run_in_executor 把同步迭代器包进线程池，每次 next() 切一次线程——开销可控。对于 PDF / DOCX 这种 CPU-bound 的解析任务，线程池反而可以并行多文件。
3. 并发粒度：真正需要 async 的是网络型 loader（WebBaseLoader / AirbyteLoader / NotionDBLoader）——这些 loader 的子类会显式覆盖 alazy_load 直接走 httpx.AsyncClient，而不是让默认桥接去包装同步版。

查代码的捷径：给任意 loader 判断"有没有真 async"——搜索 async def alazy_load 或 async def aload；只搜到基类 BaseLoader 一处就是没真 async、两处以上才有独立实现。

9.10 事实校对与跨章索引

为了避免以讹传讹，本章对现存的事实做一次集中校对（截至 main 2026-04-22）：

跨章参照索引——本章概念在其他章节继续出现的位置：

有了这张索引，下一次读者遇到"Document 在哪个章节讲得最细"、"Retriever 返回什么类型"类问题时，可以从本表反查回源章节，不必全文重搜。

9.11 小结

LangChain 的文档加载与文本分割系统是 RAG 流水线的基石。Document 以其极简的 page_content + metadata 结构，成为整个流水线的通用数据载体。Blob 通过延迟加载和不可变设计，优雅地处理了原始数据的多样性。BaseLoader 的 lazy loading 接口确保了处理大规模数据时的内存效率。

文本分割器体系以 TextSplitter 为基类，提供了从简单字符分割到递归语义分割的完整方案。RecursiveCharacterTextSplitter 的递归算法是这个体系的精华 -- 它通过多层级分隔符序列，尽可能在语义边界处分割文本，同时通过 _merge_splits 的回退机制保持了块间的重叠连续性。语言感知分割则将这种递归策略与编程语言的语法结构相结合，为代码检索提供了优质的分割方案。

BaseDocumentTransformer 和 BaseDocumentCompressor 分别代表了文档处理的两个方向：通用转换和查询感知压缩。它们与加载器、分割器一起，构成了从原始数据到可检索知识库的完整处理链。

一个值得记住的物理事实：整个 langchain_text_splitters 包仅 3531 行——但 html.py 一个文件占了 1068 行、接近 30%——说明 HTML 的 DOM 感知切分是整个包里最重的基础建设、而不是代码分割或 markdown。