文本切分:chunk_size / chunk_overlap 与 RecursiveCharacterTextSplitter
Day09 把书搬进了图书馆,但一整本《差旅制度》几万字,检索时不可能整本塞给模型。今天离开 core 包,去独立包 libs/text-splitters/langchain_text_splitters/ 看切分的艺术。弄清三件事:①chunk_size/chunk_overlap 两个旋钮在基类怎么约定和校验;②所有切分器共享的装箱算法 _merge_splits 怎么实现"重叠滑动窗口";③RecursiveCharacterTextSplitter 为什么按"段→行→词→字"递归降级、又为什么是官方默认。
chunk_size),并且每册开头重印上一册的最后 50 字(chunk_overlap),这样读者随手抽出一册也能接得上上文,不会一句话被拦腰斩断在两册之间读不懂。类比二:RecursiveCharacterTextSplitter 像一位讲究的装订师傅:能整章拆就整章拆;章太厚就按段拆;段还太长就按句、按词拆;实在不行才逐字硬切——刀口永远优先落在最自然的缝隙上。痛点:整本书塞不进上下文,切碎又怕断句
langchain-text-splitters,因为它不依赖任何模型——纯文本算法。| 类 | 位置 | 切法 |
|---|---|---|
TextSplitter | text_splitters/base.py:59 | 抽象基类:旋钮 + 装箱算法 _merge_splits |
CharacterTextSplitter | text_splitters/character.py:13 | 一刀流:按单一分隔符(默认 "\n\n") |
RecursiveCharacterTextSplitter | text_splitters/character.py:91 | 递归降级:["\n\n", "\n", " ", ""],官方默认 |
TextSplitter 基类:两个旋钮的约法三章
基类 TextSplitter(libs/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_tokenizer(base.py:212)和 from_tiktoken_encoder 两个工厂)。模型的账本按 token 记,生产上常用 token 口径。split_text 抽象基类不规定"在哪下刀"(那是子类的个性),只规定旋钮和装箱算法(下一节的 _merge_splits,所有子类共享)。模板方法模式。_merge_splits:滑动窗口装箱算法
子类先把文本按分隔符剁成小片(splits),然后统一交给基类的 _merge_splits(libs/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)。CharacterTextSplitter:一刀流
最简单的子类 CharacterTextSplitter(libs/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 出场了。RecursiveCharacterTextSplitter:段→行→词→字递归降级
官方默认切分器(D09 的 load_and_split 就用它):RecursiveCharacterTextSplitter(libs/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_language(character.py:165)按编程语言给出如 class/def 边界优先的分隔符组。create_documents / split_documents:卡片跟着走
切分器怎么和 D09 的 Document 世界接轨?看 create_documents(libs/text-splitters/langchain_text_splitters/base.py:118)和 split_documents(base.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 工厂)。串起来 + 今日小结
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
embeddings/ 与 vectorstores/:Embeddings 抽象怎么定义、InMemoryVectorStore 怎么存向量、余弦相似度检索怎么跑起来。RAG 流水线再进一站。