算法核心特点重叠滑窗相邻chunk之间有重叠token避免信息在切分边界丢失句子完整性按句子边界切分保证语义完整标题保留每个chunk保留对应的章节标题提供上下文动态调整处理超长句子时强制拆分避免单一chunk过大Token精确控制支持中文分词精确控制每个chunk的大小importrefromtypingimportList,Dict,Any,OptionalfromtransformersimportAutoTokenizerimportlogging# 配置日志logging.basicConfig(levellogging.INFO)loggerlogging.getLogger(__name__)classDocumentChunker: 文档切分器 - 基于重叠滑窗分句算法 将大文档切分成适合检索的小段落chunks def__init__(self,chunk_size:int512,overlap_size:int50,tokenizer_name:strbert-base-chinese,min_chunk_size:int50,separator:str\n): 初始化文档切分器 Args: chunk_size: 每个切分块的最大token长度K overlap_size: 相邻切分块之间的重叠token长度L tokenizer_name: tokenizer模型名称 min_chunk_size: 最小切分块token长度低于此值将被合并 separator: 段落分隔符 self.chunk_sizechunk_size self.overlap_sizeoverlap_size self.min_chunk_sizemin_chunk_size self.separatorseparator# 加载tokenizertry:self.tokenizerAutoTokenizer.from_pretrained(tokenizer_name)exceptExceptionase:logger.warning(f加载tokenizer失败使用简单计数:{e})self.tokenizerNonedef_count_tokens_simple(self,text:str)-int: 简单token计数当没有tokenizer时使用 # 中英文混合的简单统计中文按字英文按空格分割chinese_charslen(re.findall(r[\u4e00-\u9fa5],text))english_wordslen(re.findall(r[a-zA-Z],text))numberslen(re.findall(r\d,text))punctuationslen(re.findall(r[。、\.,!?;:\\()],text))returnchinese_charsenglish_wordsnumberspunctuationsdef_count_tokens(self,text:str)-int: 计算文本的token数量 ifself.tokenizer:returnlen(self.tokenizer.encode(text,add_special_tokensFalse))else:returnself._count_tokens_simple(text)def_split_into_sentences(self,text:str)-List[str]: 将文本分割成句子 # 中文句子分割符。等sentence_patternr([^。\n][。\n])sentencesre.findall(sentence_pattern,text)ifnotsentences:# 如果没有匹配到句子分割符按段落分割sentences[s.strip()forsintext.split(\n)ifs.strip()]# 如果还是没有句子将整个文本作为一个句子ifnotsentences:sentences[text.strip()]returnsentencesdef_merge_sentences_to_chunks(self,sentences:List[str],title:str)-List[Dict[str,Any]]: 将句子合并成chunks使用重叠滑窗策略 Args: sentences: 句子列表 title: 章节标题 Returns: 切分后的chunk列表 chunks[]current_chunk[]current_tokens0# 为每个句子计算token数sentence_tokens[]forsentenceinsentences:token_countself._count_tokens(sentence)sentence_tokens.append((sentence,token_count))i0total_sentenceslen(sentence_tokens)whileitotal_sentences:# 开始构建一个新的chunkcurrent_chunk[]current_tokens0start_indexi# 添加句子到当前chunk直到达到chunk_size限制whileitotal_sentences:sentence,token_countsentence_tokens[i]ifcurrent_tokenstoken_countself.chunk_size:current_chunk.append(sentence)current_tokenstoken_count i1else:# 如果当前chunk为空强制加入这个长句子ifnotcurrent_chunk:current_chunk.append(sentence)current_tokenstoken_count i1break# 构建chunk文本添加标题chunk_text .join(current_chunk)iftitleandchunk_text:# 检查标题是否已经在文本中iftitlenotinchunk_text:chunk_textf{title}{chunk_text}# 记录chunk信息ifchunk_text.strip():chunks.append({text:chunk_text.strip(),token_count:current_tokens,start_index:start_index,end_index:i-1,sentences:current_chunk.copy()})# 应用重叠策略回退到重叠位置ifitotal_sentences:# 计算应该回退到的位置overlap_tokens0overlap_count0# 从当前位置向前遍历累积重叠tokenforjinrange(i-1,start_index-1,-1):sentence,token_countsentence_tokens[j]ifoverlap_tokenstoken_countself.overlap_size:overlap_tokenstoken_count overlap_count1else:break# 如果有重叠内容回退到重叠位置ifoverlap_count0:ii-overlap_countreturnchunksdef_extract_chapter_titles(self,text:str)-Dict[int,str]: 提取章节标题及其在文本中的位置 # 匹配常见章节标题模式第X章、第X节、1.、一、等title_patterns[r第[一二三四五六七八九十百千万]章\s*[^\n],r第[一二三四五六七八九十百千万]节\s*[^\n],r[0-9]\.\s*[^\n],r[一二三四五六七八九十]、\s*[^\n]]titles{}linestext.split(\n)foridx,lineinenumerate(lines):forpatternintitle_patterns:ifre.match(pattern,line.strip()):titles[idx]line.strip()breakreturntitlesdef_map_title_to_sentences(self,sentences:List[str],titles:Dict[int,str],original_text:str)-List[str]: 将章节标题映射到对应的句子列表 # 简化实现在句子列表中查找标题并添加result_sentences[]current_title# 将原始文本按行分割linesoriginal_text.split(\n)line_index0# 构建行号到标题的映射line_to_title{}forline_idx,titleintitles.items():line_to_title[line_idx]title# 遍历句子为每个句子分配对应的标题forsentenceinsentences:# 检查当前句子是否包含标题matched_titleNonefortitleintitles.values():iftitleinsentence:matched_titletitlebreakifmatched_title:current_titlematched_title# 如果当前句子以标题开头跳过标题本身因为已经记录了ifcurrent_titleandsentence.startswith(current_title):continue# 将句子添加到结果中并携带标题信息ifcurrent_title:result_sentences.append(f{current_title}{sentence})else:result_sentences.append(sentence)returnresult_sentencesdefchunk_document(self,document:str,document_id:str,extract_titles:boolTrue)-List[Dict[str,Any]]: 将文档切分成chunks Args: document: 文档文本内容 document_id: 文档ID extract_titles: 是否提取章节标题 Returns: 切分后的chunk列表 logger.info(f开始切分文档:{document_id}, 文本长度:{len(document)}字符)# 1. 提取章节标题titles{}ifextract_titles:titlesself._extract_chapter_titles(document)logger.info(f提取到{len(titles)}个章节标题)# 2. 分割成句子sentencesself._split_into_sentences(document)logger.info(f分割成{len(sentences)}个句子)# 3. 为句子添加标题信息ifextract_titlesandtitles:sentences_with_titlesself._map_title_to_sentences(sentences,titles,document)else:sentences_with_titlessentences# 4. 合并句子成chunkschunksself._merge_sentences_to_chunks(sentences_with_titles)# 5. 添加元数据fori,chunkinenumerate(chunks):chunk[chunk_id]f{document_id}_chunk_{i:04d}ifdocument_idelsefchunk_{i:04d}chunk[document_id]document_id chunk[chunk_index]i chunk[total_chunks]len(chunks)logger.info(f切分完成生成{len(chunks)}个chunks)self._print_chunk_statistics(chunks)returnchunksdef_print_chunk_statistics(self,chunks:List[Dict[str,Any]]): 打印chunk统计信息 ifnotchunks:returntoken_counts[chunk[token_count]forchunkinchunks]avg_tokenssum(token_counts)/len(token_counts)min_tokensmin(token_counts)max_tokensmax(token_counts)logger.info(fChunk统计: 平均token{avg_tokens:.1f}, 最小{min_tokens}, 最大{max_tokens})defbatch_chunk_documents(self,documents:List[Dict[str,str]])-List[Dict[str,Any]]: 批量处理多个文档 Args: documents: 文档列表每个文档包含 id 和 content Returns: 所有文档的chunk列表 all_chunks[]fordocindocuments:doc_iddoc.get(id,)contentdoc.get(content,)ifnotcontent:logger.warning(f文档{doc_id}内容为空跳过)continuechunksself.chunk_document(content,doc_id)all_chunks.extend(chunks)logger.info(f批量切分完成总共生成{len(all_chunks)}个chunks)returnall_chunksdeftest_document_chunker(): 测试文档切分器 # 测试文档test_document 第一章 系统概述 本装备维修保障升级技术方案是面向装备维修业务打造的一套全离线、本地化、可溯源、3D可视化作业平台。 主要服务一线维修人员与资料管理员用于统一管理各类装备维修资料、辅助工作人员快速处置装备故障。 系统具备三大核心功能维修资料统一数字化管理、故障智能查询与维修方案自动生成、本地模型自主优化升级。 第二章 技术路线 2.1 文档上传与处理 支持批量导入装备维修手册、故障案例、设备规范、扫描图纸等各类文件。 系统自动完成资料规整、分类存储与离线知识库归档。 2.2 文档切分 通过重叠滑窗分句算法将维修手册、技术手册等大块文档分解成更小段落切分。 确保搜索结果与用户查询相匹配极为关键。 2.3 文档向量化 文档向量化将文档块进行向量化编码用于后续的向量检索。 本方案采用通用向量化模型BGE对文档分块进行向量化处理。 # 创建切分器chunkerDocumentChunker(chunk_size256,overlap_size30,tokenizer_namebert-base-chinese,min_chunk_size20)# 切分文档chunkschunker.chunk_document(test_document,document_idtest_001)# 打印结果print(\n*60)print(切分结果:)print(*60)forchunkinchunks:print(f\nChunk ID:{chunk[chunk_id]})print(fToken数:{chunk[token_count]})print(f包含句子数:{len(chunk[sentences])})print(f内容预览:{chunk[text][:150]}...)print(-*40)returnchunksif__name____main__:test_document_chunker()使用示例# 1. 基本使用chunkerDocumentChunker(chunk_size512,overlap_size50)chunkschunker.chunk_document(document_text,document_iddoc_001)# 2. 批量处理documents[{id:doc_001,content:文档1内容...},{id:doc_002,content:文档2内容...}]all_chunkschunker.batch_chunk_documents(documents)# 3. 自定义配置chunkerDocumentChunker(chunk_size256,# 每个chunk最多256个tokenoverlap_size30,# 相邻chunk重叠30个tokentokenizer_namebert-base-chinese,min_chunk_size20,# 最小chunk大小separator\n# 段落分隔符)# 4. 保存chunks到文件importjsonwithopen(chunks.json,w,encodingutf-8)asf:json.dump(all_chunks,f,ensure_asciiFalse,indent2)