LangChain 设计与实现

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

作者 杨艺韬 · 13,603 字 · 发布于 · 更新于

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

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

本章要点

  • 理解 Document 类的设计哲学及其与消息系统的区别
  • 掌握 BaseMedia -> Blob / Document 的继承关系
  • 理解 BaseLoader 的 lazy loading 接口设计
  • 了解 Blob / BlobLoader / BaseBlobParser 的解耦架构
  • 深入理解 TextSplitter 基类的 _merge_splits 合并算法
  • 掌握 RecursiveCharacterTextSplitter 的递归分割策略
  • 了解 BaseDocumentTransformer 和 BaseDocumentCompressor 的角色
  • 掌握 indexing API 的 RecordManager 记账与三种 cleanup 模式

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 的用户代码，输出不会因为新增 id 这类字段而改变。这是一个向后兼容性的务实考量；同一段注释也说明，这个重写将来可能被更通用的 Prompt 内格式化方案取代。

9.1.4 Document / Blob / BaseMedia 字段总表

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

字段 所属类 类型 默认值 定义行 说明
id BaseMedia str | None None（coerce_numbers_to_str=True） 48 可选文档标识符，建议使用 UUID；传入整数会自动字符串化
metadata BaseMedia dict {}（default_factory） 55 通用元数据，任何 loader / splitter 都应尽量原样透传
data Blob bytes | str | None None 116 内存中的原始数据；None 表示数据尚未加载（配合 path 使用）
mimetype Blob str | None None 119 由 from_path 通过 mimetypes.guess_type 推断
encoding Blob str "utf-8" 122 仅在 data 是字节流且需要解码时生效
path Blob PathLike | None None 128 延迟加载的文件路径；as_* 方法在访问时才 read_*
source Blob str | None 只读 @property：优先 metadata["source"]，否则 str(path) 137 不是 Pydantic 字段，VectorStore 排查来源必读
page_content Document str 必填（__init__ 第一个位置参数） 306 唯一存文本内容的字段
type Document Literal["Document"] "Document" 309 固定取值的类型标签字段

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

  1. Document 没有 role / tool_call_id / response_metadata——全是消息族系（BaseMessage）才有的字段。想同时存"谁写的"+"文本"，要自己塞进 metadata["author"]，不要指望 Document 原生提供。
  2. Blob.source 根本不是字段，而是只读 @property——它不能由 __init__ 赋值，取值顺序是"先看 metadata["source"]，没有才回退到 str(self.path)"。所以两个 path 相同的 Blob，只要其中一个在 metadata 里写了 source，两者的 source 就会不同——拿它当对比键前要先确认 metadata 干净。另外 Blob 上真正的 model_validator 叫 check_blob_is_valid，作用是强制 data 与 path 至少给一个。
  3. Document.__init__ 允许 page_content 作为第一个位置参数——Pydantic 模型自动生成的 __init__ 只接受关键字参数，所以源码专门写了 def __init__(self, page_content: str, **kwargs)（docstring：Pass page_content in as positional or named arg.），Document("text", metadata={...}) 与 Document(page_content="text", ...) 两种写法都能用。

9.1.5 和消息族系的字段对照

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

维度 Document HumanMessage / AIMessage
核心文本字段 page_content: str content: str | list[ContentBlock]
元数据 metadata: dict（完全自由） response_metadata + additional_kwargs（半结构化）
角色 无（RAG 流水线不需要） 由 type 字段的固定取值区分
工具调用 无 tool_calls（AIMessage）/ tool_call_id（ToolMessage）
序列化标签 type: "Document" type: "human" / "ai" / "tool" / "system"
主要用途 喂给 VectorStore / Retriever 喂给 ChatModel.invoke / stream
多模态支持 只有文本（二进制走 Blob） content 可以是图像 / 音频块列表

一个实操含义：如果你从向量库检索到 Document 后要把它喂给 LLM，不要直接 llm.invoke(doc)——聊天模型只接受 PromptValue、字符串或消息列表，传入 Document 会抛 ValueError: Invalid input type <class 'langchain_core.documents.base.Document'>. Must be a PromptValue, str, or list of BaseMessages.（假模型实跑）。正确做法是 HumanMessage(content=f"参考资料：{doc.page_content}\n\n问题：{query}") 或用 create_stuff_documents_chain 这类专门把 Document 压扁成 Prompt 的 chain。

9.2 Blob：原始数据的抽象

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

先记一条 import 路径的坑：Blob 的类定义确实在 libs/core/langchain_core/documents/base.py:59，但它没有从 langchain_core.documents 这个包导出——libs/core/langchain_core/documents/__init__.py 的 __all__ 只有 Document / BaseDocumentCompressor / BaseDocumentTransformer 三个，_dynamic_imports 里也没有 Blob，所以 from langchain_core.documents import Blob 会失败，实跑报 ImportError: cannot import name 'Blob' from 'langchain_core.documents'。公开入口在 langchain_core.document_loaders（libs/core/langchain_core/document_loaders/__init__.py:15 的 __all__ 与 :24 的 _dynamic_imports 都有 Blob，实体从 blob_loaders 转出）。特别注意 Blob 自己的 docstring 里那三段示例写的全是 from langchain_core.documents import Blob——那是上游没跟着改的旧写法，照抄直接报错。docstring 是断言不是证据，以 __init__.py 的导出表为准。

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.data = "x" 抛 ValidationError），这是一个重要的设计选择 -- 它避免了 Blob 在流水线中传递时字段被意外改写。注意这只是浅层冻结：metadata 这个 dict 本身仍可原地修改（blob.metadata["k"] = v 照样生效）。

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 if metadata is not None else {},
    )

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

# libs/core/langchain_core/documents/base.py:158
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
    msg = f"Unable to get string for blob {self}"
    raise ValueError(msg)

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

9.2.2 多种数据访问方式

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

方法 返回类型 适用场景
as_string() str 文本文件处理
as_bytes() bytes 二进制文件处理
as_bytes_io() BytesIO / BufferedReader 流式处理、传递给第三方库

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

# libs/core/langchain_core/documents/base.py:194
@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 文件句柄
    else:
        msg = f"Unable to convert blob {self}"
        raise NotImplementedError(msg)

注意这里没有 str 分支：Blob.from_data("abc") 这种内存字符串 Blob 调 as_bytes_io() 会抛 NotImplementedError（实跑），需要字节流时先用 as_bytes() 或以 bytes 构造。

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())

    async def aload(self) -> list[Document]:
        """Load data into Document objects."""
        return [document async for document in self.alazy_load()]

    # ...（load_and_split 见 §9.3.2）

    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 中执行；aload 则把 alazy_load 的结果收集成列表
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[PyPDFLoader] --> 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:
        if not _HAS_TEXT_SPLITTERS:  # 没装 langchain-text-splitters
            raise ImportError(msg)   # msg 提示 pip install -U langchain-text-splitters
        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 源码账本

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

文件 行数 关键符号（行号） 责任
base.py 155 BaseLoader(26) / load(37) / aload(45) / load_and_split(53, 注释标注已 deprecated) / lazy_load(91) / alazy_load(102) / BaseBlobParser(117) / lazy_parse(128) / parse(140) 加载 / 解析抽象基类
blob_loaders.py 38 BlobLoader(19) / yield_blobs(27) 只定义一个 ABC，把 Blob 和 PathLike 从 documents.base re-export
langsmith.py 143 LangSmithLoader 从 LangSmith 数据集拉取样例转成 Document

core 里只有这三份（外加 39 行的 __init__.py，全目录 375 行）。真正的"loader 动物园"不在 core 里，而是分散在两个地方：

  1. langchain_community.document_loaders —— 第三方数据源的 loader 实现集合，发布在独立仓库、不在本专栏快照的 monorepo 里。能在快照里直接数到的是它的转发层：libs/langchain/langchain_classic/document_loaders/ 顶层 145 个 .py（另有 blob_loaders/、parsers/ 子目录），除 base.py 直接转出 core 的 BaseLoader / BaseBlobParser 外，都只是一段 create_importer(...) 的 shim，__getattr__ 时才真正 import langchain_community.document_loaders 的同名类，并在非交互环境下发出弃用警告。涵盖 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_classic/document_loaders/ 的转发表，实现在 langchain-community）：

类别 代表 loader 典型 MIME / 数据源
纯文本 TextLoader / DirectoryLoader / UnstructuredFileLoader text/plain, 整个目录递归
PDF PyPDFLoader / PyMuPDFLoader / PDFPlumberLoader / UnstructuredPDFLoader / AmazonTextractPDFLoader application/pdf
Office Docx2txtLoader / UnstructuredWordDocumentLoader / UnstructuredExcelLoader / UnstructuredPowerPointLoader .docx / .xlsx / .pptx
网页 WebBaseLoader / RecursiveUrlLoader / SeleniumURLLoader / PlaywrightURLLoader / SitemapLoader text/html
代码 GitLoader / GitHubIssuesLoader / GitbookLoader / LanguageParser 代码仓库
数据库 DuckDBLoader / MongodbLoader / SnowflakeLoader / BigQueryLoader SQL/NoSQL/数仓
SaaS NotionDBLoader / ConfluenceLoader / SlackDirectoryLoader / AirtableLoader / FigmaFileLoader 各自 SDK / API
学术 ArxivLoader / PubMedLoader / HuggingFaceDatasetLoader / WikipediaLoader 论文 / 数据集 / 百科接口
云存储 S3FileLoader / S3DirectoryLoader / GCSFileLoader / AzureBlobStorageFileLoader 对象存储
结构化 CSVLoader / JSONLoader / BSHTMLLoader / TomlLoader text/csv, application/json

Core / Community 的分层带来的效果——core 里只放"ABC + 一个 LangSmith 开箱即用的 loader"，langchain-core 的硬依赖只有 langsmith、pydantic、PyYAML、tenacity 等少数几个（libs/core/pyproject.toml）；而"每接一家就要引一个 SDK"（PyPDF、boto3、playwright、google-cloud-storage）都在 community 或专用小包里。这样下游引入 langchain-core 的用户，不需要为了一个抽象基类被迫装上一堆厂商 SDK。

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

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

from pathlib import Path
from langchain_core.document_loaders import BlobLoader, BaseBlobParser, Blob
from langchain_core.documents import 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 基类

# libs/text-splitters/langchain_text_splitters/base.py:44
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 < 0:
            raise ValueError(f"chunk_overlap must be >= 0, got {chunk_overlap}")
        if chunk_overlap > chunk_size:
            raise ValueError(
                f"Got a larger chunk overlap ({chunk_overlap}) than chunk size "
                f"({chunk_size}), should be smaller."
            )
        self._chunk_size = chunk_size
        self._chunk_overlap = chunk_overlap
        self._length_function = length_function
        ...

核心参数解析：

参数 默认值 含义
chunk_size 4000 每个块的最大长度
chunk_overlap 200 相邻块之间的重叠长度
length_function len 计算文本长度的函数
keep_separator False 是否保留分隔符及其位置
add_start_index False 是否在 metadata 中记录起始位置
strip_whitespace True 是否去除块首尾空白

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

9.4.2 核心方法链

# libs/text-splitters/langchain_text_splitters/base.py:93
@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."""
    metadatas_ = metadatas or [{}] * len(texts)   # :115——没给 metadata 就补空字典
    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 是所有文本分割器共享的核心方法，它将初始的小段文本合并为目标大小的块：

# libs/text-splitters/langchain_text_splitters/base.py:152
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 total > self._chunk_size:
                # 已攒的块超长（通常是某个 split 本身就超过 chunk_size）：只打 warning，不再切
                logger.warning(
                    "Created a chunk of size %d, which is longer than the specified %d",
                    total, 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、并且再放入当前 split 也不超过 chunk_size（图中只画了前一个条件）。这确保了相邻 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)]
            # 补回配对时被落下的那一段
            if len(splits_) % 2 == 0:
                splits += splits_[-1:]
            splits = (
                [*splits, splits_[-1]] if keep_separator == "end"
                else [splits_[0], *splits]
            )
        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": ["Hello", "\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", " ", ""] 代表了从粗到细的四个粒度层级：

层级 分隔符 含义
1 "\n\n" 段落边界
2 "\n" 行边界
3 " " 单词边界
4 "" 字符边界（逐字分割）

9.5.2 递归分割算法

# libs/text-splitters/langchain_text_splitters/character.py:107
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（其余用默认值，keep_separator=True），输入文本为（第二段是"这是第二段内容，"后接 20 个"比较长"再加句号，共 69 字）：

这是第一段内容。

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

短段。
  1. "\n\n" 在文本中存在，于是用它分割，得到三段（keep_separator=True，后两段开头带着 "\n\n"）
  2. 第一段长度小于 50，进入 good_splits
  3. 第二段超过 50：先把 good_splits 里的第一段合并输出为一个 chunk，再对第二段递归。递归时从剩余分隔符 ["\n", " ", ""] 里挑第一个在该段中出现的——段首保留下来的 "\n\n" 里恰好有 "\n"，于是先按 "\n" 切出一个孤立的换行（合并后 strip 成空串，被丢弃）和 "\n这是第二段……"（70 个字符，仍超长）；后者再递归，文本里没有空格，落到 ""，逐字切开，由 _merge_splits 合并成不超过 50 的块，相邻块重叠 10 个字符
  4. 第三段进入 good_splits，遍历结束后单独合并输出

实跑输出（len 与内容）：

8  '这是第一段内容。'
49 '这是第二段内容，比较长比较长比较长比较长比较长比较长比较长比较长比较长比较长比较长比较长比较长比较'
30 '较长比较长比较长比较长比较长比较长比较长比较长比较长比较长。'
3  '短段。'

第二块只有 49 字，是因为逐字切出的第一个元素是段首残留的那个换行：合并出的 50 个字符里含它，strip_whitespace 去掉后剩 49。这个例子也说明，keep_separator=True 时留在片段开头的分隔符会参与下一层的分隔符选择。

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",           # 行边界
        " ",            # 单词边界
        "",             # 字符边界
    ]

这种语言感知的分隔符设计让代码优先在类、函数等语义边界处被分割，尽量不在函数或类的中间断开。Language 枚举有 28 个成员，其中 27 种（Python、JavaScript、TypeScript、Java、Go、Rust、C/C++、Markdown、HTML、LaTeX 等）预置了分隔符表；Language.PERL 没有对应分支，from_language(Language.PERL) 实跑抛 ValueError: Language perl is not implemented yet!。

9.5.4 from_language：按编程语言语法感知分割

默认 ["\n\n", "\n", " ", ""] 对自然文本很好，但对代码就不合适——在一个函数中间断开会让 chunk 失去语义完整性。LangChain 提供 RecursiveCharacterTextSplitter.from_language(Language.PYTHON) 让分隔符按语言语法定制（libs/text-splitters/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)

关键细节：from_language 固定传 is_separator_regex=True——因为不少语言的分隔符本身就写成了正则：R 的 "\nlibrary\\(" 转义了括号，RST 的 "\n=+\n"、Markdown 的 "\n#{1,6} " 用了量词，VisualBasic6 的分隔符还带 (?!End\s) 这样的前瞻。若按字面量匹配，这些分隔符永远命中不了。（"行首"约束来自分隔符开头的 \n，与是否按正则解释无关。）

Language 枚举（libs/text-splitters/langchain_text_splitters/base.py:380）共 28 个成员（其中 PERL 尚无分隔符表，见上）：

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"，拼错会立即抛 AttributeError，类型检查器也能提前发现。（Language 继承自 str，直接传 "python" 实跑同样可用，只是失去了这层保护。）

各语言的分隔符设计有明确逻辑分层，以 Python 为例（libs/text-splitters/langchain_text_splitters/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 缩进的 "\n\tdef "，按 PEP 8 用 4 个空格缩进的方法不会在这一级被切开，而是落到后面的 "\n\n" / "\n"。
  3. 通用分隔符：\n\n / \n / / "" 兜底，代码过长强制切到字符级。

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

        if language in {Language.C, Language.CPP}:
            return [
                # Split along class definitions
                "\nclass ",
                # Split along function definitions
                "\nvoid ",
                "\nint ",
                "\nfloat ",
                "\ndouble ",
                # Split along control flow statements
                "\nif ",
                "\nfor ",
                "\nwhile ",
                "\nswitch ",
                "\ncase ",
                # Split by the normal type of lines
                "\n\n",
                "\n",
                " ",
                "",
            ]

C/C++ 没有 def 关键字、函数通过返回类型 + 名字声明——所以要列 void/int/float/double 这些常见返回类型。控制流关键字 if/for/while/switch/case 也被列为分隔符，排在函数级分隔符之后。

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

这些分隔符列表覆盖了各语言最常见的顶层结构，但并不完美（比如 JS 列表里没有针对 => 箭头函数的分隔符）。如果你为某语言做 RAG、先用 from_language(Language.YOUR_LANG)、实测效果后再考虑自定义 separators。

9.6 CharacterTextSplitter：简单字符分割

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

# libs/text-splitters/langchain_text_splitters/character.py:11
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 层面进行分割：

# libs/text-splitters/langchain_text_splitters/base.py:304
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 的实现是直接的滑动窗口：

# libs/text-splitters/langchain_text_splitters/base.py:430（源码签名带类型标注）
def split_text_on_tokens(*, text, tokenizer):
    splits = []
    input_ids = tokenizer.encode(text)
    start_idx = 0
    if tokenizer.tokens_per_chunk <= tokenizer.chunk_overlap:
        raise ValueError("tokens_per_chunk must be greater than chunk_overlap")

    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]
        if not chunk_ids:
            break
        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 id 计不超过 chunk_size（相邻窗口重叠 chunk_overlap 个 token），但可能在单词或语义边界的中间断开文本。在实践中，通常推荐使用 RecursiveCharacterTextSplitter.from_tiktoken_encoder，它结合了语义感知分割和 token 计数。

9.8 全部切分策略全景对比

langchain-text-splitters 1.1.1（本快照 libs/text-splitters/，13 个 .py 含 __init__.py 共 3531 行）的 __init__.py 导出了 17 个 splitter 类（另有 TextSplitter 抽象基类、Language 枚举、Tokenizer 等辅助导出）。把它们连同两个最常用的类方法变体对照成一张矩阵，能一眼看出每一种的"切点语义"、"长度度量"、"典型 RAG 场景"：

Splitter 来源文件 长度度量 切点语义 是否保留结构 metadata 典型用例
CharacterTextSplitter libs/text-splitters/langchain_text_splitters/character.py:11 len / 自定义 单一分隔符（默认 "\n\n"）+ _merge_splits 否 极短文档、边界已经足够清晰
RecursiveCharacterTextSplitter libs/text-splitters/langchain_text_splitters/character.py:88 len / 自定义 多级分隔符逐层递归 否（可选 add_start_index） RAG 默认首选、结构不可知的自然文本
RecursiveCharacterTextSplitter.from_language libs/text-splitters/langchain_text_splitters/character.py:161 同上 按 28 种编程语言的语法 token 切 否 代码库 RAG、Copilot 风格检索
RecursiveCharacterTextSplitter.from_tiktoken_encoder base.py（类方法继承） tiktoken.encode 长度 递归切 + token 计数 否 需要精确匹配模型 token 窗口的生产环境
TokenTextSplitter libs/text-splitters/langchain_text_splitters/base.py:304 tiktoken token 数 纯滑动窗口 + token 否 只关心"每块 N token"的最简场景
SentenceTransformersTokenTextSplitter libs/text-splitters/langchain_text_splitters/sentence_transformers.py:20 SentenceTransformer tokenizer 按 ST 模型最大上下文切 否 下游是 SentenceTransformer 嵌入模型
MarkdownHeaderTextSplitter libs/text-splitters/langchain_text_splitters/markdown.py:23 无（不控制长度） 按 headers_to_split_on 指定的标题层级切 是（每块附调用方命名的键，如 {"Header 1": "..."}） 技术文档、API 文档、带大纲的 Markdown
MarkdownTextSplitter libs/text-splitters/langchain_text_splitters/markdown.py:14 len 取 Language.MARKDOWN 的分隔符表，但不开 is_separator_regex，"\n#{1,6} " 这类正则分隔符按字面量处理，切法与 from_language(MARKDOWN) 不同 否 Markdown 无大纲、只想照段落切
ExperimentalMarkdownSyntaxTextSplitter libs/text-splitters/langchain_text_splitters/markdown.py:298 无 逐行正则识别 header、代码块、水平线 是（代码块语言记入 "Code" 键） 保留 code fence 不被拦腰切断
HTMLHeaderTextSplitter libs/text-splitters/langchain_text_splitters/html.py:83 无 BeautifulSoup 显式遍历 DOM，按 header 层级切 是（附层级 header） 爬虫来的带嵌套 h1/h2/h3 的网页
HTMLSectionSplitter libs/text-splitters/langchain_text_splitters/html.py:353 split_text 不控长度；split_documents 会再过 RecursiveCharacterTextSplitter(**kwargs) 按 headers_to_split_on 的 h1–h6 分节；先用 XSLT 把 font-size 大于 20px 的元素改写成 <h1> 是 标题靠字号而非标签表达的 HTML
HTMLSemanticPreservingSplitter libs/text-splitters/langchain_text_splitters/html.py:562 len（max_chunk_size） 按 header 切；elements_to_preserve 里列出的元素（如 table、ul）整块保留 是 不能让表格 / 列表被切断的网页
RecursiveJsonSplitter libs/text-splitters/langchain_text_splitters/json.py:12 序列化后 len 深度优先拆嵌套 JSON、保持结构 否（键路径保留在子 JSON 的嵌套结构里，metadata 只复制输入） 大型 OpenAPI spec、结构化配置
JSFrameworkTextSplitter libs/text-splitters/langchain_text_splitters/jsx.py:9 len React（JSX）/ Vue / Svelte：组件标签 + JS 语法边界 否 前端代码库 RAG
NLTKTextSplitter libs/text-splitters/langchain_text_splitters/nltk.py:19 len sent_tokenize 句子边界 否 英文学术文本、句级检索
SpacyTextSplitter libs/text-splitters/langchain_text_splitters/spacy.py:26 len spaCy 句子分割 pipeline（默认 en_core_web_sm，可换 "sentencizer" 快速模式） 否 需要比标点规则更准的句子切分
KonlpyTextSplitter libs/text-splitters/langchain_text_splitters/konlpy.py:19 len KoNLPy 韩语形态素 否 韩语 RAG
LatexTextSplitter libs/text-splitters/langchain_text_splitters/latex.py:11 len 取 Language.LATEX 的分隔符表但不开 is_separator_regex，"\n\\\\section{" 等按字面量匹配、实际命中不了 \section{，切法与 from_language(LATEX) 不同 否 论文、arXiv 源码（建议改用 from_language）
PythonCodeTextSplitter libs/text-splitters/langchain_text_splitters/python.py:11 len 取 Language.PYTHON 的分隔符表（其中没有正则元字符，效果与 from_language(PYTHON) 相同） 否 向后兼容老 API

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

  1. 不知道切什么格式 → 首选 RecursiveCharacterTextSplitter.from_tiktoken_encoder(chunk_size=512, chunk_overlap=64)（这两个数是推荐起点，需自己传；TextSplitter 不传参的默认是 chunk_size=4000, chunk_overlap=200）。它是"懒人万能方案"：递归语义感知 + token 精确计数。
  2. 原文有天然大纲（Markdown / HTML） → 优先用对应的 Header splitter 把大纲信息写进 metadata，再在 retrieval 阶段把 header 作为 prefix 拼回 prompt。没有这一步，chunk 容易失去层级上下文（"这一段属于哪一节"检索后无法还原）。
  3. 英文长 academic 文本 → 试 SpacyTextSplitter 或 NLTKTextSplitter，句子边界通常比字符边界对嵌入模型更友好；但两者默认都面向英文（NLTK 默认 language="english"，spaCy 默认 en_core_web_sm），直接用于中文未必能正确断句，中文场景仍然推荐递归字符切分（可把 "。" 等中文标点加进 separators）。

9.8.1 _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))

实跑输出：

0 'the quick brown fox'
1 'fox jumps over the'
2 'the lazy dog again'
3 'again and again'

chunk_overlap=5（按字符长度 ≤5）让相邻块共享了 fox、the、again 这样的单个单词——这是 _merge_splits 的回退循环在工作，而不是"最后 5 个 split 元素重叠"。把 chunk_overlap 调到 10，实跑变成 5 块（'the quick brown fox'、'brown fox jumps over'、'jumps over the lazy'、'the lazy dog again'、'dog again and again'），相邻块重叠两个单词——这是 RAG 调参里常改的一个值。

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

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

MarkdownHeaderTextSplitter

定义在 libs/text-splitters/langchain_text_splitters/markdown.py:23，它不是 TextSplitter 的子类（没有 _merge_splits 行为），而是一个独立类：

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

它的切分产物是 list[Document]，每个 Document.metadata 会额外带上当前层级之上所有 header 的路径（键名就是 headers_to_split_on 里给的第二个元素）——比如处在 # 第一章 下 ## 1.1 节 下 ### 实验 段的那块文字，实跑得到的 metadata 就是 {"Header 1": "第一章", "Header 2": "1.1 节", "Header 3": "实验"}。代码围栏（``` / ~~~）里以 # 开头的行不会被当成标题。

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

HTMLHeaderTextSplitter / HTMLSectionSplitter / HTMLSemanticPreservingSplitter

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

类 定义行 依赖 适合的 HTML
HTMLHeaderTextSplitter 83 bs4（BeautifulSoup，html.parser） 有明确 <h1>...<h6> 嵌套的文档类网页
HTMLSectionSplitter 353 lxml + XSLT，外加 bs4 标题靠 h1–h6 或大字号（font-size > 20px）表达的网页
HTMLSemanticPreservingSplitter 562 BeautifulSoup 需要把 <table> / <ul> / <pre> 整块保留的混排网页

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

HTMLSemanticPreservingSplitter 的"语义保留"含义是：只要把 table 列进 elements_to_preserve（默认为空），碰到 <table> 时它会先用 PRESERVED_n 占位符替换整张表，切完再原样填回，表格不会被从中间切断（即使这个 chunk 因此超过 max_chunk_size 也优先保完整性）。这在爬财经 / 百科类网站时非常重要——割裂的半张表格在 LLM 看来是噪声，语义价值为零。

RecursiveJsonSplitter

json.py 全文 190 行，类定义在第 12 行。它不继承 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 的子对象，每个子对象保留从根到该值的完整嵌套路径。它只向下拆 dict：单个过大的叶子值（长字符串）不会再切，list 默认也整体保留（convert_lists=True 才先把 list 转成以下标为键的 dict），所以个别块仍可能超过 max_chunk_size——实跑 max_chunk_size=60 时，一个含 100 字符字符串的叶子照样单独成块。

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

9.10 选型决策树

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

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 / HTML 的 header splitter 会把标题层级写进 Document.metadata，RecursiveJsonSplitter 则把 key 路径留在每个子 JSON 的嵌套结构里，后面 Retriever + Re-ranker 随时可以拿回来；字符串一旦切碎就再也拼不回层级了。

9.11 BaseDocumentTransformer 与 BaseDocumentCompressor

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

9.11.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.11.2 BaseDocumentCompressor

# libs/core/langchain_core/documents/compressor.py:19
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.11.3 langchain_text_splitters/ 13 文件 3531 行的模块拆分

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

文件 行 角色
html.py 1068 最大——HTML splitter 自带 DOM 感知、section header 切分
character.py 803 CharacterTextSplitter + RecursiveCharacterTextSplitter + 27 种语言的 separator 表（Language 28 个成员中 PERL 未实现）
markdown.py 481 MarkdownTextSplitter / MarkdownHeaderTextSplitter / ExperimentalMarkdownSyntaxTextSplitter——header 层级 + code fence 保护
base.py 458 TextSplitter ABC、Language Enum、Tokenizer dataclass、split_text_on_tokens
json.py 190 RecursiveJsonSplitter——按深度优先拆分 JSON 保持结构
sentence_transformers.py 126 按 SentenceTransformer tokenizer 切
jsx.py 106 React（JSX）/ Vue / Svelte——组件标签边界切分
spacy.py 73 走 spaCy 的句子 tokenizer
nltk.py 72 走 NLTK 的 sent_tokenize
__init__.py 69 public 导出
konlpy.py 51 韩语形态素切分（KoNLPy）
latex.py 17 只定义 LatexTextSplitter：RecursiveCharacterTextSplitter 子类，__init__ 里取 Language.LATEX 的分隔符表（不开 is_separator_regex）
python.py 17 同上，PythonCodeTextSplitter 取 Language.PYTHON 的分隔符表

三点看得出设计取舍——

  1. HTML 为什么是 markdown 的两倍多——html.py 装了三个类，且 HTML 的 tag 嵌套远比 markdown 复杂；仅 HTMLSemanticPreservingSplitter 一个类就有十几个选项（链接 / 图片 / 音视频转 Markdown、停用词、allow / deny 标签、自定义 tag 处理器……）；markdown 主要处理标题、代码围栏、水平线几种边界
  2. latex.py / python.py 各 17 行只定义一个薄子类——from_language 已经能覆盖同样的用途，这两个类让老代码 import PythonCodeTextSplitter 继续可用——薄壳保留 API 稳定性；但要注意它们没开 is_separator_regex，LatexTextSplitter 的切法因此与 from_language(Language.LATEX) 不同（见 §9.8 对比表）
  3. sentence_transformers/spacy/nltk/konlpy 四份 tokenizer 包装都很薄（51–126 行）——因为重活（token 化）由依赖完成、splitter 只做"按 tokenizer 切分界限做文本切片"的胶水——再次印证 LangChain"把 tokenizer 当外部组件注入、splitter 保持纯逻辑"的抽象纪律

9.12 indexing API：让向量库跟着数据源增量同步

切好的 Document 最终要写进向量库。直接 vector_store.add_documents(docs) 的问题是：重跑一次就重复写一次，源文件删改后旧块还留在库里。langchain_core.indexing 用一个 RecordManager 记账来解决这件事（libs/core/langchain_core/indexing/，api.py 948 行、base.py 661 行）。

# libs/core/langchain_core/indexing/api.py:290
def index(
    docs_source: BaseLoader | Iterable[Document],
    record_manager: RecordManager,
    vector_store: VectorStore | DocumentIndex,
    *,
    batch_size: int = 100,
    cleanup: Literal["incremental", "full", "scoped_full"] | None = None,
    source_id_key: str | Callable[[Document], str] | None = None,
    cleanup_batch_size: int = 1_000,
    force_update: bool = False,
    key_encoder: Literal["sha1", "sha256", "sha512", "blake2b"]
    | Callable[[Document], str] = "sha1",
    upsert_kwargs: dict[str, Any] | None = None,
) -> IndexingResult:

主流程（异步版 aindex 同构）：

  1. 算指纹：对每个文档分别哈希 page_content 与 json.dumps(metadata, sort_keys=True)，再合并哈希，结果写进 Document.id（_get_document_with_hash，libs/core/langchain_core/indexing/api.py:169）。所以调用方自己给的 id 会被覆盖；metadata 改一个字段也算新文档。默认 key_encoder="sha1" 会在进程内发一次 UserWarning，提示 SHA-1 不抗碰撞、建议换 blake2b / sha256 / sha512。
  2. 按批去重：每 batch_size 个一批，批内同指纹只留一个（被去掉的计入 num_skipped）；再用 record_manager.exists 查哪些已写过——已写过的只刷新时间戳、计入 num_skipped，除非 force_update=True。
  3. 先写向量库，再记账：新文档以指纹为 ids 调 add_documents（DocumentIndex 则调 upsert），然后把本批所有指纹连同 source id 以不早于本轮开始时间的时间戳写进 RecordManager。
  4. 清理：删除 RecordManager 里时间戳早于本轮开始、且"在清理范围内"的记录及对应向量。

cleanup 决定清理范围：

模式 清理范围 时机 要求
None 不删 — —
"incremental" 本轮见过的 source id 下未被刷新的旧记录 每批写完立即清 必须给 source_id_key
"full" 本轮没返回的全部记录 全部写完后 数据源必须返回完整数据集，否则会误删
"scoped_full" 本轮见过的 source id 下未被刷新的旧记录 全部写完后 必须给 source_id_key；langchain-core 0.3.25 引入

source_id_key 可以是 metadata 的键名（如 "source"），也可以是 Callable[[Document], str]。传 cleanup="incremental" 却不给它，实跑抛 ValueError: Source id key is required when cleanup mode is incremental or scoped_full.。

用 InMemoryRecordManager + InMemoryVectorStore（嵌入换成一个返回固定向量的最小 Embeddings 子类）实跑四种模式：第一轮写入 a.md 两块和 b.md 一块，第二轮只交 a.md 的两块，其中一块内容改过：

None         第二轮 {'num_added': 1, 'num_updated': 0, 'num_skipped': 1, 'num_deleted': 0}  库内 4
incremental  第二轮 {'num_added': 1, 'num_updated': 0, 'num_skipped': 1, 'num_deleted': 1}  库内 3
full         第二轮 {'num_added': 1, 'num_updated': 0, 'num_skipped': 1, 'num_deleted': 2}  库内 2
scoped_full  第二轮 {'num_added': 1, 'num_updated': 0, 'num_skipped': 1, 'num_deleted': 1}  库内 3

None 留下了 a.md 改前的旧块；incremental 与 scoped_full 只清 a.md 的旧块、保留本轮没出现的 b.md；full 把 b.md 也当成"已删除的源"清掉了。incremental 与 scoped_full 的差别在时机：前者每批就清，同一 source 的块分散在多个批次时会做重复工作（源码注释建议把 batch_size 调大）；后者在内存里收集本轮出现过的 source id，最后统一清理，适合数据源无法一次返回全集、又不好控制批大小的场景。

两点边界：

  • 向量库要能按 id 删除：index() 会检查 VectorStore 子类是否覆盖了 delete，没覆盖直接抛 ValueError。
  • RecordManager 与向量库是两套存储：RecordManager 的 docstring 明说两者分开实现、整体成了分布式系统，可能出现一边写成功另一边失败的一致性问题；它还依赖单调递增的时间戳，建议用服务端时间。core 里的 InMemoryRecordManager 注明仅供测试，持久化实现如 SQLRecordManager 在 langchain_classic.indexes（libs/langchain/langchain_classic/indexes/_sql_record_manager.py）。

9.13 设计决策分析

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

LangChain 将文本分割器放在了独立的 langchain-text-splitters 包中（本专栏基准快照里是 1.1.1 版，源码在 langchain-core-1.2.26/libs/text-splitters/langchain_text_splitters/）。从包结构看，这样拆分有两点好处：

  1. 依赖隔离：langchain-text-splitters 的硬依赖只有 langchain-core（libs/text-splitters/pyproject.toml），tiktoken、transformers、spacy、nltk、bs4、lxml 等都在用到时才 try-import，分割器的演进不会给核心包带来依赖膨胀
  2. 复用性：文本分割不仅用于 RAG，还用于长文档处理、代码分析等场景，独立包便于在非 LangChain 项目中使用

Blob 的不可变设计

Blob 被设计为 frozen=True 的不可变对象（langchain-core-1.2.26/libs/core/langchain_core/documents/base.py:59，model_config 里的 frozen=True 在同文件 :133）。这个选择在并发场景中有帮助 -- 多个 parser 同时处理同一个 Blob 时，谁都不能改写它的 data / path 等字段。但这只是浅层冻结：metadata dict 仍可原地修改，Blob 也因含 dict 字段而不可哈希（实跑 hash(blob) 抛 TypeError: unhashable type: 'dict'），不能直接当缓存键。

chunk_overlap 的作用

chunk_overlap 是 RAG 质量的一个关键杠杆。重叠区域确保了跨 chunk 边界的信息不会完全丢失。但过大的重叠会增加存储和计算成本——同一段文字会被写进多个 chunk。合理取值远小于 chunk_size（TextSplitter.__init__ 的默认值就是 chunk_size=4000 配 chunk_overlap=200，见 langchain-core-1.2.26/libs/text-splitters/langchain_text_splitters/base.py:49），具体比例没有普适答案，要按语料和检索效果实测。

为什么 load_and_split 被标记为弃用？

load_and_split（langchain-core-1.2.26/libs/core/langchain_core/document_loaders/base.py:53）把加载和分割耦合在一个方法中，违反了单一职责原则。分开调用的好处是：

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

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

BaseLoader（langchain-core-1.2.26/libs/core/langchain_core/document_loaders/base.py:26）的 lazy_load（同文件 :91）返回 Iterator[Document]（同步），而 alazy_load（同文件 :102）是一个独立的异步方法——两者默认实现会互相桥接。这种"双接口"设计是 LangChain 整个架构的缩影：

  1. 历史包袱：早期 loader 多基于同步文件 I/O，PDF parser 里 pypdf / pdfplumber 也都是同步阻塞 API；把 lazy_load 改成 async 会破坏已有 loader 的接口。
  2. 桥接成本低：run_in_executor 把同步迭代器包进线程池，每次 next() 切一次线程——开销可控，且阻塞 I/O 不会卡住事件循环。但对 PDF / DOCX 这种 CPU 密集的解析，线程受 GIL 限制，默认桥接并不会带来并行加速。
  3. 并发粒度：真正受益于原生 async 的是网络型 loader（如 WebBaseLoader）——这类 loader 可以显式覆盖 alazy_load，直接用异步 HTTP 客户端，而不是让默认桥接去包装同步版（具体哪些 loader 覆盖了，要到 langchain-community 源码里确认，本专栏快照不含该仓库）。

查代码的捷径：给任意 loader 判断"有没有真 async"——搜索 async def alazy_load 或 async def aload；只搜到基类 BaseLoader 一处（即 langchain-core-1.2.26/libs/core/langchain_core/document_loaders/base.py:102 与 :45）就是没真 async、两处以上才有独立实现。

9.14 事实校对与跨章索引

为了避免以讹传讹，本章对现存的事实做一次集中校对（基准：langchain-core 1.2.26 / langchain-text-splitters 1.1.1，commit 0a1d290a）：

陈述 真实情况 源
Language 枚举有 28 个成员 确认 28（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） libs/text-splitters/langchain_text_splitters/base.py:380
langchain_text_splitters 包约 3500 行 本快照实测 13 个 .py（含 __init__.py）共 3531 行，html.py 一个文件就占 1068 行 langchain_text_splitters/**.py
BaseLoader.load_and_split 已 deprecated 源码 !!! danger 注记写着 "Do not override this method. It should be considered to be deprecated!" libs/core/langchain_core/document_loaders/base.py:53
Blob 是不可变对象 model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True) libs/core/langchain_core/documents/base.py:131
RecursiveCharacterTextSplitter 默认分隔符 ["\n\n", "\n", " ", ""] 确认，见 __init__（separators or ["\n\n", "\n", " ", ""]） libs/text-splitters/langchain_text_splitters/character.py:95
CharacterTextSplitter 默认分隔符 "\n\n" 确认 libs/text-splitters/langchain_text_splitters/character.py:11
TokenTextSplitter 默认用 tiktoken，编码名 encoding_name="gpt2"（传 model_name 可覆盖） 确认 libs/text-splitters/langchain_text_splitters/base.py:304
BaseDocumentCompressor 额外接 query 参数 确认，与 BaseDocumentTransformer 的区别 libs/core/langchain_core/documents/compressor.py

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

概念 本章小节 其他章节
Document 数据结构 9.1 第 10 章向量存储与检索器（既是存入 VectorStore 的基本单位，也是 BaseRetriever 的返回类型）、第 11 章 Chain 组合（combine_documents 系列链的输入上下文）
BaseLoader 9.3.1 第 8 章 §8.9 create_retriever_tool（把检索链路包成 Agent 工具）
TextSplitter._merge_splits 9.4.3 第 10 章 §10.30 切块策略（chunk 大小直接决定向量库里的检索单元）
RecursiveCharacterTextSplitter.from_language 9.5.3 第 10 章 §10.30"结构 chunk"策略（代码按语言语法切）
BaseDocumentCompressor 9.11.2 第 10 章 §10.6.3 / §10.16 ContextualCompressionRetriever 与 re-ranking
indexing API（index() / RecordManager） 9.12 第 10 章 §10.35 只嵌入变化文档的成本优化
MarkdownHeaderTextSplitter 9.9 第 10 章 §10.30"结构 chunk"策略（Markdown 按 header 切）

9.15 小结

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。