Day 10 / 共 20 天 · 阶段3 数据与 RAG

文本切分:chunk_size / chunk_overlap 与 RecursiveCharacterTextSplitter

Day09 把书搬进了图书馆,但一整本《差旅制度》几万字,检索时不可能整本塞给模型。今天离开 core 包,去独立包 libs/text-splitters/langchain_text_splitters/ 看切分的艺术。弄清三件事:chunk_size/chunk_overlap 两个旋钮在基类怎么约定和校验;②所有切分器共享的装箱算法 _merge_splits 怎么实现"重叠滑动窗口";③RecursiveCharacterTextSplitter 为什么按"段→行→词→字"递归降级、又为什么是官方默认。

📍 你在 20 天里的位置(阶段3:数据与 RAG · D09-12)
S2 模型/消息/提示 D09 Document/加载 D10 文本切分 D11 Embedding/向量库 D12 Retriever/RAG链 S4 工具/Agent S5 进阶收官
💡 先用两个类比兜住今天 继续图书馆世界观。类比一:切分像把一整本书拆成活页装订的小册子——每册不超过 500 字(chunk_size),并且每册开头重印上一册的最后 50 字chunk_overlap),这样读者随手抽出一册也能接得上上文,不会一句话被拦腰斩断在两册之间读不懂。类比二:RecursiveCharacterTextSplitter 像一位讲究的装订师傅:能整章拆就整章拆;章太厚就按段拆;段还太长就按句、按词拆;实在不行才逐字硬切——刀口永远优先落在最自然的缝隙上。
L01

痛点:整本书塞不进上下文,切碎又怕断句

🤔 痛点为什么非切不可?三个硬约束:①上下文有限——几万字的文档塞不进(塞进也贵);②向量检索的粒度——一整本书压成一个向量,"报销上限"这种细节问题根本查不准,块越聚焦命中越准;③模型注意力——塞一堆无关内容进提示词反而拉低回答质量。但切也有切的风险:一刀下去正好把"上限为每晚 500 元"斩成两半,两个块都答不上问题。怎么切得又小又不断意思?
💡 本质:块大小是检索精度与上下文完整性的滑杆块小 → 向量更聚焦、检索更准,但单块信息可能不完整;块大 → 上下文完整,但向量"什么都有一点"、检索变糊。LangChain 的工程答案是三件套:chunk_size 限制册厚 + chunk_overlap 册间重印几行 + 优先在自然缝隙(段落/句子)下刀。这套逻辑独立成包 langchain-text-splitters,因为它不依赖任何模型——纯文本算法。
位置切法
TextSplittertext_splitters/base.py:59抽象基类:旋钮 + 装箱算法 _merge_splits
CharacterTextSplittertext_splitters/character.py:13一刀流:按单一分隔符(默认 "\n\n")
RecursiveCharacterTextSplittertext_splitters/character.py:91递归降级:["\n\n", "\n", " ", ""],官方默认
L02

TextSplitter 基类:两个旋钮的约法三章

基类 TextSplitterlibs/text-splitters/langchain_text_splitters/base.py:59)的构造函数把规矩立在最前面:

# libs/text-splitters/langchain_text_splitters/base.py:59
class TextSplitter(BaseDocumentTransformer, ABC):
    """Interface for splitting text into chunks."""

    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,   # 是否在 metadata 里记录块的起点
        strip_whitespace: bool = True,
    ) -> None:
        if chunk_size <= 0:
            raise ValueError(msg)                       # 约法一:册厚必须为正
        if chunk_overlap < 0:
            raise ValueError(msg)                       # 约法二:重叠不能为负
        if chunk_overlap > chunk_size:
            msg = (f"Got a larger chunk overlap ({chunk_overlap}) than chunk size "
                   f"({chunk_size}), should be smaller.")
            raise ValueError(msg)                       # 约法三:重叠必须小于册厚
        self._chunk_size = chunk_size
        self._chunk_overlap = chunk_overlap
        self._length_function = length_function
        ...

    @abstractmethod
    def split_text(self, text: str) -> list[str]: ...   # base.py:108 子类唯一必须实现的
chunk_size=4000默认每块最多 4000 个"长度单位"。注意单位由 length_function 说了算。
chunk_overlap=200相邻两册重印 200 单位。若 overlap ≥ size,每一册几乎全是上一册的重印——逻辑上荒谬,所以构造时直接拒绝(约法三)。
length_function=len★"厚度怎么量"是可注入的策略:默认数字符;换成 tokenizer 计数就变成按 token 切(基类还内置了 from_huggingface_tokenizerbase.py:212)和 from_tiktoken_encoder 两个工厂)。模型的账本按 token 记,生产上常用 token 口径。
split_text 抽象基类不规定"在哪下刀"(那是子类的个性),只规定旋钮和装箱算法(下一节的 _merge_splits,所有子类共享)。模板方法模式
L03

_merge_splits:滑动窗口装箱算法

子类先把文本按分隔符剁成小片(splits),然后统一交给基类的 _merge_splitslibs/text-splitters/langchain_text_splitters/base.py:167)装箱成册——这是全包最核心的 40 行:

# libs/text-splitters/langchain_text_splitters/base.py:167
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:
                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_ + ... > self._chunk_size and total > 0   #   只留 overlap 的量
                ):
                    total -= self._length_function(current_doc[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
① 超厚判断厚度 = 已装片 + 新片 + 片间分隔符,凑一起超过 chunk_size 就先出货。分隔符长度也算钱——细到这个程度。
③ while 弹出到只剩 overlap★重叠的实现精髓:出货后不清空 current_doc,而是从头部弹出旧片,直到剩余厚度 ≤ chunk_overlap。留下的这个"尾巴"就是下一册开头的重印内容——滑动窗口就是这么滑的。
logger.warning如果某一片单片就超过 chunk_size(比如一个 800 字符的段落、chunk_size=500),装不下也得装,打个警告继续——宁可超尺寸也不静默丢内容。这也是 Recursive 版要递归再切的动机(L05)。
_merge_splits:滑动窗口装箱(chunk_size=10,overlap=3 示意) 小片流入: A(4) B(3) C(3) D(4) E(5) 册1 出货:A+B+C(厚度10,D 装不下) 出货后弹出 A、B,留 C(3) ≤ overlap(3) 册2 接着装:C+D(C 是重印的尾巴) 再来 E 超厚 → 册2 出货 → 留 D … 依此滑动 重叠 = 出货后"不清空、只弹到剩 overlap"(base.py:195 的 while 循环) 相邻两册共享尾巴 C,被拦腰的语义在下一册还能读全 —— 活页册开头重印上一册末行
图注:窗口向前滑、尾巴留下来。overlap 不是"额外复制一段",而是装箱时自然留在箱底的旧片。
L04

CharacterTextSplitter:一刀流

最简单的子类 CharacterTextSplitterlibs/text-splitters/langchain_text_splitters/character.py:13)——只认一个分隔符:

# libs/text-splitters/langchain_text_splitters/character.py:13
class CharacterTextSplitter(TextSplitter):
    """Splitting text that looks at characters."""

    def __init__(self, separator: str = "\n\n", is_separator_regex: bool = False, **kwargs):
        super().__init__(**kwargs)
        self._separator = separator          # 默认按"空行"(段落边界)下刀

    @override
    def split_text(self, text: str) -> list[str]:    # character.py:28
        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)
        ...
        merge_sep = ""
        if not (self._keep_separator or is_lookaround):
            merge_sep = self._separator
        return self._merge_splits(splits, merge_sep)                 # ② 交给基类装箱
separator="\n\n"默认在空行(段落边界)下刀——最自然的缝隙。也可以换成正则(is_separator_regex=True)。
两步走子类的全部个性就是"①怎么剁片";"②怎么装箱"完全复用基类 _merge_splits。后面的 markdown.py、python.py 等所有切分器都是这个结构。
致命短板只有一把刀:如果某个段落本身超过 chunk_size,它不会再想别的办法,只能打警告塞出一个超大块(L03 见过)。这就轮到 Recursive 出场了。
L05

RecursiveCharacterTextSplitter:段→行→词→字递归降级

官方默认切分器(D09 的 load_and_split 就用它):RecursiveCharacterTextSplitterlibs/text-splitters/langchain_text_splitters/character.py:91):

# libs/text-splitters/langchain_text_splitters/character.py:91
class RecursiveCharacterTextSplitter(TextSplitter):
    """Splitting text by recursively look at characters.
    Recursively tries to split by different characters to find one that works."""

    def __init__(self, separators=None, keep_separator=True, ...):
        super().__init__(keep_separator=keep_separator, **kwargs)
        self._separators = separators or ["\n\n", "\n", " ", ""]   # ★刀谱:段→行→词→字

    def _split_text(self, text: str, separators: list[str]) -> list[str]:  # character.py:110
        final_chunks = []
        separator = separators[-1]
        new_separators = []
        for i, s_ in enumerate(separators):
            if re.search(separator_, text):        # ① 选出当前文本里能用的最粗的刀
                separator = s_
                new_separators = separators[i + 1 :]   #    剩下的细刀留给递归
                break
        splits = _split_text_with_regex(text, separator_, ...)   # ② 用选中的刀剁片
        good_splits = []
        for s in splits:
            if self._length_function(s) < self._chunk_size:
                good_splits.append(s)              # ③ 片够小 → 攒着待装箱
            else:
                if good_splits:
                    final_chunks.extend(self._merge_splits(good_splits, separator_))
                    good_splits = []
                if not new_separators:
                    final_chunks.append(s)         # ④ 刀用完了,认了
                else:
                    final_chunks.extend(self._split_text(s, new_separators))  # ⑤ ★换细刀递归
        if good_splits:
            final_chunks.extend(self._merge_splits(good_splits, separator_))
        return final_chunks
刀谱 ["\n\n","\n"," ",""]从粗到细四把刀:空行(段)→ 换行(行)→ 空格(词)→ 空字符串(逐字符)。最后一把 "" 保证任何文本最终都能切开——哪怕是没有空格的中文长句。
① 选刀不是死板从第一把开始:挑当前文本里实际存在的最粗一把(全文没有空行就直接从换行开始),并记下更细的刀(new_separators)备用。
⑤ 递归降级★灵魂所在:某片用粗刀切完还是超厚 → 对这一片单独换细刀重切(递归调用),细刀还不够就再细。对比一刀流的"认命打警告",Recursive 只有刀谱全用完(④)才认命。
③ good_splits 分批装箱够小的片攒着,遇到超厚片时先把攒的装箱出货、再处理超厚片——保证输出顺序与原文一致
💡 本质:语义边界的优先级搜索为什么"段→行→词→字"?因为语义完整性递减:整段是完整论述,整行是完整句子,整词至少不切碎单词,逐字是最后的无奈。装订师傅的原则——刀口尽量落在最自然的缝隙,实在不行才硬切。这也解释了为什么它是官方默认:对任意未知文本,它给出"尽可能不破坏语义"的最稳妥切法。代码场景还有专门刀谱:from_languagecharacter.py:165)按编程语言给出如 class/def 边界优先的分隔符组。
L06

create_documents / split_documents:卡片跟着走

切分器怎么和 D09 的 Document 世界接轨?看 create_documentslibs/text-splitters/langchain_text_splitters/base.py:118)和 split_documentsbase.py:146):

# libs/text-splitters/langchain_text_splitters/base.py:118
def create_documents(self, texts, metadatas=None) -> list[Document]:
    metadatas_ = metadatas or [{}] * len(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

# base.py:146
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)
copy.deepcopy(metadata)★每个子块拿到母文档卡片的深拷贝——source、page 等信息全继承。为什么深拷贝?如果 100 个块共享同一个 dict,改一个块的卡片会污染其他 99 个。
start_index 的寻找起点开了 add_start_index 后用 text.find(chunk, offset) 定位,起点扣掉了 overlap——因为重叠意味着同样的文字出现两次,直接从头 find 可能定位到上一块的重印区。答案引用要标"原文第几个字符起"就靠它。
split_documentsDocument 进 → 拆成(正文, 卡片) → 切正文、继承卡片 → Document 出。D09 说的"通货换零钱、卡片跟着走"就是这三行。
⚠️ 坑:中文文本 + 默认刀谱默认刀谱里的"词刀"是空格——中文没有空格!于是中文常从"行刀"直接跌到"字刀",切在词中间(如"报销上|限")。中文实践通常自定义刀谱:separators=["\n\n", "\n", "。", ",", ""],把句号、逗号加进优先级。另一个常见坑:chunk_size 数的是"length_function 的单位",默认按字符数——中文 1 字符 ≈ 1 token 甚至更多,按英文经验设 4000 可能超预算,按 token 计数更稳(L02 的 tokenizer 工厂)。
L07

串起来 + 今日小结

📝 真实值:切一段制度文档 输入(假设 chunk_size=60, chunk_overlap=12,字符计):
text = "第一条 差旅住宿标准。\n\n一线城市每晚上限 500 元,二线城市每晚上限 350 元。\n\n第二条 报销流程。\n\n出差结束后 7 日内提交行程单与发票。"
RecursiveCharacterTextSplitter(chunk_size=60, chunk_overlap=12).split_text(text) 的行为:先用"段刀"(\n\n)剁成 4 片(12/27/10/19 字符),再由 _merge_splits 装箱:片1+片2 = 41 字符,再加片3 要超 60?41+2+10=53 还装得下;再加片4 超了 → 册1出货("第一条…\n\n一线城市…\n\n第二条 报销流程。"),弹旧片留尾巴 ≤12("第二条 报销流程。"10 字符留下)→ 册2 = "第二条 报销流程。\n\n出差结束后 7 日内…"。两册,重叠的正是"第二条 报销流程。"这句标题——两边都能接上下文。create_documents([text], metadatas=[{"source":"差旅制度.txt"}]),两册各自带上同一张 source 卡片。

👶 小白:chunk_overlap 白白多存了重复内容,不浪费吗?到底设多大?

👨‍🏫 老师:是的,overlap 就是用存储换语义完整——多存 5%-15% 的重复文字,换"关键句子不被拦腰斩断在两块之间"的保险。设多大没有铁律,常见经验是 chunk_size 的 10%-20%(比如 1000/150)。设 0 的后果:正好切在"上限为每晚 | 500 元"处时,问"上限多少"两块都检索不出完整答案。设太大(基类约法三禁止 ≥ chunk_size)则块间大量雷同,向量库里全是重复内容,检索结果一片近亲。从 10% 起步,用你自己的问答对实测调优,比背数字有用。

🧠 今天你应该能回答

  • 基类的约法三章?(size>0、overlap≥0、overlap<size,base.py:88-99)
  • "长度"一定是字符数吗?(不,length_function 可注入,可换 token 计数,base.py:67)
  • overlap 是怎么实现的?(_merge_splits 出货后不清空、从头弹片到剩 ≤overlap,留作下册开头,base.py:167/195)
  • 单片超过 chunk_size 时一刀流和递归流各怎么办?(一刀流打警告硬塞;Recursive 换细刀递归重切,character.py:145)
  • 默认刀谱及其含义?(["\n\n","\n"," ",""] 段→行→词→字,语义完整性递减,character.py:107)
  • 切出来的块为什么还带 source?(create_documents 深拷贝母卡片给每个子块,base.py:135)

✋ 10 分钟动手

cd /Users/bitmart/work/codes/github/AI_WORK/langchain/libs/text-splitters/langchain_text_splitters

# 1. 基类:旋钮 + 装箱
sed -n '59,107p'  base.py    # 构造函数:约法三章 + length_function
sed -n '167,210p' base.py    # _merge_splits:滑动窗口装箱

# 2. 两个切分器
sed -n '13,62p'   character.py   # 一刀流
sed -n '91,152p'  character.py   # Recursive:选刀 + 递归降级

# 3. 与 Document 接轨
sed -n '118,160p' base.py    # create_documents / split_documents(卡片深拷贝)
grep -n "def from_language\|def get_separators_for_language" character.py
明日预告 · Day 11:书拆成小册子了,接下来给每册编"语义索引"——明天看 embeddings/vectorstores/:Embeddings 抽象怎么定义、InMemoryVectorStore 怎么存向量、余弦相似度检索怎么跑起来。RAG 流水线再进一站。
← Day 09 Document 与加载 Day 11 · Embedding 与向量库 →