langchain递归切片器原理分析

机制分析

关键文件和类

文件路径：langchain_text_splitters/character.py

类名：RecursiveCharacterTextSplitter

核心入口函数：_split_text

解析步骤及源码分析

上述步骤的总体逻辑在_split_text函数中，关键处均有代码注释说明：

def_split_text(self,text:str,separators:list[str])->list[str]:"""Split incoming text and return chunks."""final_chunks=[]# Get appropriate separator to use# 得到当前层级的分隔符separator=separators[-1]new_separators=[]fori,_sinenumerate(separators):_separator=_sifself._is_separator_regexelsere.escape(_s)if_s=="":separator=_sbreak# 快速搜索文本中是否存在分隔符，不存在，则分隔符降级ifre.search(_separator,text):separator=_s new_separators=separators[i+1:]break_separator=separatorifself._is_separator_regexelsere.escape(separator)splits=_split_text_with_regex(text,_separator,keep_separator=self._keep_separator)# Now go merging things, recursively splitting longer texts.# 所谓的“好段”指的是长度小于chunk_size的段_good_splits=[]_separator=""ifself._keep_separatorelseseparatorforsinsplits:ifself._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=[]ifnotnew_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)returnfinal_chunks

合并与重叠的逻辑则在_merge_splits函数中，关键处均有代码注释说明：

def_merge_splits(self,splits:Iterable[str],separator:str)->list[str]:# We now want to combine these smaller pieces into medium size# chunks to send to the LLM.separator_len=self._length_function(separator)docs=[]current_doc:list[str]=[]total=0# splits是所谓的好段，我们将好段尽可能拼接为较大的块fordinsplits:_len=self._length_function(d)# 拼到再加就超chunk_size，则封口、起新块if(total+_len+(separator_leniflen(current_doc)>0else0)>self._chunk_size):iftotal>self._chunk_size:logger.warning(f"Created a chunk of size{total}, "f"which is longer than the specified{self._chunk_size}")iflen(current_doc)>0:doc=self._join_docs(current_doc,separator)ifdocisnotNone:docs.append(doc)# Keep on popping if:# - we have a larger chunk than in the chunk overlap# - or if we still have any chunks and the length is long# 这一段是overlap的核心处理逻辑# 算法回退<=chunk_overlap的长度，以便下个块A2包含前一个块A1末尾overlap的内容。特别的，若块A1的最后一段长度大于chunk_overlap，则为保证语义完整，不会生成overlap。current_doc里存的就是一个块的所有段。whiletotal>self._chunk_overlapor(total+_len+(separator_leniflen(current_doc)>0else0)>self._chunk_sizeandtotal>0):total-=self._length_function(current_doc[0])+(separator_leniflen(current_doc)>1else0)current_doc=current_doc[1:]current_doc.append(d)total+=_len+(separator_leniflen(current_doc)>1else0)doc=self._join_docs(current_doc,separator)ifdocisnotNone:docs.append(doc)returndocs