2.3 文档分割策略 本节导读:学习如何科学地分割文档以优化检索效果,掌握不同分割策略的适用场景和最佳实践。 学习目标 掌握文档分割的基本原理和重要性 理解不同分割策略的特点和适用场景 能够根据业务需求选择合适的分割方法 实现高质量的文档分割和预处理 核心概念 文档分割是将大型文档切分成适合处理的小片段的过程。良好的分割策略能够显著提升RAG系统的检索准确性和生成质量。 分割的基本原理 分割的关键考虑因素 语义完整性:保持文档的逻辑连贯性 信息密度:每个片段应包含足够的信息量 检索效率:便于快速定位相关信息 处理效率:平衡粒度和性能 上下文长度:适应LLM的上下文窗口限制 分割策略分类 固定长度分割 特点:按照预设的字数或token数量进行分割 优点: 实现简单,计算效率高
本节导读:学习如何科学地分割文档以优化检索效果,掌握不同分割策略的适用场景和最佳实践。
文档分割是将大型文档切分成适合处理的小片段的过程。良好的分割策略能够显著提升RAG系统的检索准确性和生成质量。
特点:按照预设的字数或token数量进行分割
from langchain.text_splitter import RecursiveCharacterTextSplitter def fixed_length_split(text, chunk_size=1000, chunk_overlap=200): """固定长度分割""" splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, ) chunks = splitter.split_text(text) return chunks # 使用示例 text = "这是一个很长的文档内容..." * 100 # 模拟长文档 chunks = fixed_length_split(text, chunk_size=800, chunk_overlap=150) print(f"分割成 {len(chunks)} 个片段")
优点:
缺点:
特点:基于语义相似性和文档结构进行分割
from langchain.text_splitter import SemanticChunker from langchain.embeddings import OpenAIEmbeddings def semantic_split(text, chunk_size=1000): """语义分割""" embeddings = OpenAIEmbeddings() splitter = SemanticChunker( embeddings, chunk_size=chunk_size, ) chunks = splitter.split_text(text) return chunks # 使用示例 text = """ 第一章 机器学习基础 1.1 机器学习概述 机器学习是人工智能的一个重要分支,它使计算机能够从数据中学习模式和规律。 1.2 主要类型 监督学习:使用标记数据训练模型 无监督学习:从无标记数据中发现模式 强化学习:通过试错学习最优策略 1.3 应用领域 图像识别 自然语言处理 推荐系统 第二章 深度学习 2.1 神经网络基础 深度学习是机器学习的一个子领域,基于人工神经网络。 2.2 主要架构 卷积神经网络(CNN) 循环神经网络(RNN) 变换器(Transformer) """ chunks = semantic_split(text)
优点:
缺点:
特点:基于文档的结构特征进行分割
from langchain.text_splitter import CharacterTextSplitter import re def structured_split(text, pattern=None): """结构化分割""" if pattern is None: # 默认按段落分割 pattern = r'\n\s*\n' # 使用正则表达式分割 sections = re.split(pattern, text) # 过滤空段落 sections = [section.strip() for section in sections if section.strip()] return sections def markdown_structure_split(text): """Markdown文档结构分割""" sections = [] current_section = [] current_level = 0 lines = text.split('\n') for line in lines: line = line.strip() if not line: continue # 检测标题级别 if line.startswith('#'): level = len(line) - len(line.lstrip('#')) # 如果开始新的章节,保存当前章节 if current_section and level <= current_level: sections.append('\n'.join(current_section)) current_section = [] current_level = level current_section.append(line) else: current_section.append(line) # 添加最后一个章节 if current_section: sections.append('\n'.join(current_section)) return sections # 使用示例 text = """ # 第一章 机器学习基础 ## 1.1 机器学习概述 机器学习是人工智能的一个重要分支,它使计算机能够从数据中学习模式和规律。 ## 1.2 主要类型 ### 监督学习 使用标记数据训练模型。 ### 无监督学习 从无标记数据中发现模式。 ### 强化学习 通过试错学习最优策略。 ## 1.3 应用领域 图像识别、自然语言处理、推荐系统等。 # 第二章 深度学习 ## 2.1 神经网络基础 深度学习是机器学习的一个子领域,基于人工神经网络。 """ sections = markdown_structure_split(text) print(f"分割成 {len(sections)} 个章节")
优点:
缺点:
特点:多层级分割,先按大结构分割,再细分
class HierarchicalTextSplitter: """层级文本分割器""" def __init__(self, strategies): self.strategies = strategies def split_text(self, text): """多层分割""" results = [] current_level = [text] for strategy in self.strategies: next_level = [] for chunk in current_level: sub_chunks = strategy(chunk) next_level.extend(sub_chunks) current_level = next_level results = current_level return results # 使用示例 def chapter_level_split(text): """章节级分割""" # 按章节标题分割 chapter_pattern = r'# [^#\n]+\n' chapters = re.split(chapter_pattern, text) # 重新添加标题 title_matches = re.findall(r'# [^#\n]+\n', text) sections = [] for i, chapter in enumerate(chapters): if i < len(title_matches): sections.append(title_matches[i] + chapter) elif chapter.strip(): sections.append(chapter) return sections def section_level_split(text): """小节级分割""" # 按小节标题分割 section_pattern = r'## [^#\n]+\n' sections = re.split(section_pattern, text) title_matches = re.findall(r'## [^#\n]+\n', text) results = [] for i, section in enumerate(sections): if i < len(title_matches): results.append(title_matches[i] + section) elif section.strip(): results.append(section) return results # 创建层级分割器 strategies = [chapter_level_split, section_level_split] splitter = HierarchicalTextSplitter(strategies) # 使用示例 text = """ # 第一章 机器学习基础 ## 1.1 机器学习概述 机器学习是人工智能的一个重要分支,它使计算机能够从数据中学习模式和规律。 ## 1.2 主要类型 ### 监督学习 使用标记数据训练模型。 ### 无监督学习 从无标记数据中发现模式。 # 第二章 深度学习 ## 2.1 神经网络基础 深度学习是机器学习的一个子领域,基于人工神经网络。 """ hierarchical_chunks = splitter.split_text(text) print(f"层级分割成 {len(hierarchical_chunks)} 个片段")
优点:
缺点:
def optimize_overlap(text, chunk_size, min_overlap=50, max_overlap=400): """优化重叠度""" overlap_ratios = [0.1, 0.15, 0.2, 0.25, 0.3] # 10%-30%的重叠比例 best_overlap = min_overlap best_score = float('inf') for ratio in overlap_ratios: overlap = int(chunk_size * ratio) overlap = max(min_overlap, min(overlap, max_overlap)) # 测试分割效果 splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=overlap, ) chunks = splitter.split_text(text) # 计算评分(考虑完整性、冗余度等) score = calculate_split_quality(chunks) if score < best_score: best_score = score best_overlap = overlap return best_overlap def calculate_split_quality(chunks): """计算分割质量评分""" if not chunks: return float('inf') # 1. 检查片段长度 length_variance = variance([len(chunk) for chunk in chunks]) # 2. 检查语义完整性 semantic_scores = [check_semantic_integrity(chunk) for chunk in chunks] semantic_variance = variance(semantic_scores) # 3. 检查边界信息 boundary_scores = [check_boundary_information(chunk) for chunk in chunks] boundary_variance = variance(boundary_scores) # 综合评分 total_score = length_variance + semantic_variance + boundary_variance return total_score
class AdaptiveTextSplitter: """自适应文本分割器""" def __init__(self, min_chunk_size=200, max_chunk_size=2000): self.min_chunk_size = min_chunk_size self.max_chunk_size = max_chunk_size self.content_detector = ContentDetector() def split_text(self, text): """自适应分割""" chunks = [] current_chunk = "" # 按句子分割 sentences = self._split_sentences(text) for sentence in sentences: # 尝试添加到当前片段 test_chunk = current_chunk + sentence if len(test_chunk) > self.max_chunk_size: # 当前片段已达到最大长度,保存 if current_chunk: chunks.append(current_chunk) current_chunk = sentence else: # 单个句子超过最大长度,强制分割 chunks.extend(self._force_split(sentence)) elif len(test_chunk) >= self.min_chunk_size: # 检查是否应该在当前位置分割 if self._should_split_at_boundary(current_chunk, sentence): chunks.append(current_chunk) current_chunk = sentence else: current_chunk = test_chunk else: # 当前片段太小,继续添加 current_chunk = test_chunk # 添加最后一个片段 if current_chunk: chunks.append(current_chunk) return chunks def _split_sentences(self, text): """句子分割""" # 简单的句子分割逻辑 sentences = re.split(r'[。!?.!?]', text) return [s.strip() for s in sentences if s.strip()] def _should_split_at_boundary(self, current_chunk, next_sentence): """判断是否应该在边界分割""" # 检测段落边界 if current_chunk.endswith('\n\n'): return True # 检测章节标题 if re.match(r'^#+\s', next_sentence): return True # 检测列表项 if re.match(r'^\d+\.\s', next_sentence): return True return False def _force_split(self, text): """强制分割超长文本""" chunks = [] current_pos = 0 while current_pos < len(text): end_pos = min(current_pos + self.max_chunk_size, len(text)) chunk = text[current_pos:end_pos] chunks.append(chunk) current_pos = end_pos return chunks
class SplitQualityEvaluator: """分割质量评估器""" def __init__(self): self.metrics = { 'length_distribution': self._evaluate_length_distribution, 'semantic_integrity': self._evaluate_semantic_integrity, 'boundary_quality': self._evaluate_boundary_quality, 'overlap_effectiveness': self._evaluate_overlap_effectiveness, 'structural_preservation': self._evaluate_structural_preservation } def evaluate(self, chunks, original_text): """评估分割质量""" results = {} for metric_name, metric_func in self.metrics.items(): score = metric_func(chunks, original_text) results[metric_name] = score # 计算综合评分 overall_score = sum(results.values()) / len(results) results['overall_score'] = overall_score return results def _evaluate_length_distribution(self, chunks, original_text): """评估长度分布""" if not chunks: return 0.0 lengths = [len(chunk) for chunk in chunks] avg_length = sum(lengths) / len(lengths) variance = sum((l - avg_length) ** 2 for l in lengths) / len(lengths) # 方差越小,分布越均匀 uniformity_score = 1.0 / (1.0 + variance / 1000) return uniformity_score def _evaluate_semantic_integrity(self, chunks, original_text): """评估语义完整性""" if not chunks: return 0.0 total_score = 0.0 for chunk in chunks: # 检查句子完整性 sentence_completeness = self._check_sentence_completeness(chunk) # 检查段落完整性 paragraph_completeness = self._check_paragraph_completeness(chunk) # 检查主题连贯性 coherence_score = self._check_thematic_coherence(chunk) chunk_score = (sentence_completeness + paragraph_completeness + coherence_score) / 3 total_score += chunk_score return total_score / len(chunks) def _evaluate_boundary_quality(self, chunks, original_text): """评估边界质量""" if len(chunks) <= 1: return 1.0 total_score = 0.0 for i in range(len(chunks) - 1): # 检查边界处的语义连贯性 boundary_coherence = self._check_boundary_coherence(chunks[i], chunks[i + 1]) # 检查信息重复度 overlap_score = self._check_boundary_overlap(chunks[i], chunks[i + 1]) boundary_quality = (boundary_coherence + overlap_score) / 2 total_score += boundary_quality return total_score / (len(chunks) - 1)
def evaluate_split_quality(): """评估分割质量""" evaluator = SplitQualityEvaluator() # 测试不同分割策略 test_text = """ # 第一章 机器学习基础 机器学习是人工智能的一个重要分支,它使计算机能够从数据中学习模式和规律。机器学习的核心是让计算机通过经验改进性能,而不仅仅是按照明确编写的指令执行任务。 ## 1.1 机器学习概述 机器学习的基本概念包括:学习算法、训练数据、模型评估等。学习算法是机器学习的核心,它决定了如何从数据中提取模式和规律。 监督学习是最常见的机器学习方法,它使用标记数据来训练模型。无监督学习则从无标记数据中发现隐藏的模式。强化学习通过与环境交互来学习最优策略。 ## 1.2 主要类型 ### 监督学习 监督学习使用标记数据训练模型。标记数据包含输入和期望的输出。常见的监督学习算法包括:线性回归、逻辑回归、决策树、随机森林、支持向量机等。 监督学习的应用场景包括:分类问题、回归问题、异常检测等。分类问题的例子包括垃圾邮件检测、图像分类、情感分析等。 ### 无监督学习 无监督学习从无标记数据中发现隐藏的模式。常见的无监督学习算法包括:K均值聚类、层次聚类、主成分分析、自编码器等。 无监督学习的应用场景包括:客户细分、异常检测、特征提取、降维等。客户细分可以帮助企业更好地了解不同客户群体的需求。 ### 强化学习 强化学习通过试错来学习最优策略。智能体在环境中采取动作,环境给予奖励或惩罚,智能体根据这些反馈调整策略。 强化学习的应用场景包括:游戏AI、机器人控制、自动驾驶、推荐系统等。AlphaGo就是强化学习的成功应用案例。 ## 1.3 应用领域 机器学习在众多领域都有广泛应用。在图像处理领域,机器学习被用于图像分类、目标检测、图像生成等任务。 在自然语言处理领域,机器学习被用于机器翻译、情感分析、文本摘要、问答系统等任务。在推荐系统领域,机器学习被用于个性化推荐、商品推荐、内容推荐等。 医疗健康、金融分析、自动驾驶等领域也广泛应用机器学习技术。这些应用极大地提高了工作效率和准确性。 """ # 测试固定长度分割 fixed_chunks = fixed_length_split(test_text, chunk_size=500, chunk_overlap=100) fixed_quality = evaluator.evaluate(fixed_chunks, test_text) # 测试结构化分割 struct_chunks = markdown_structure_split(test_text) struct_quality = evaluator.evaluate(struct_chunks, test_text) # 比较结果 print("固定长度分割质量:") for metric, score in fixed_quality.items(): print(f" {metric}: {score:.3f}") print("\n结构化分割质量:") for metric, score in struct_quality.items(): print(f" {metric}: {score:.3f}") # 分析差异 print("\n质量对比:") for metric in fixed_quality: diff = struct_quality[metric] - fixed_quality[metric] print(f" {metric}: {diff:+.3f}")
# 分割策略选择指南 分割策略: 固定长度分割: 适用场景: 标准化文档、简单文本、快速原型 优点: 实现简单、处理速度快 缺点: 语义完整性差 推荐: 初期验证、简单应用 语义分割: 适用场景: 复杂文档、知识密集型应用 优点: 语义完整性高、质量好 缺点: 计算成本高 推荐: 生产环境、高精度要求 结构化分割: 适用场景: 格式化文档、Markdown、技术文档 优点: 保持结构层次、易于理解 缺点: 格式依赖性强 推荐: 技术文档、标准化文档 层级分割: 适用场景: 大型文档、多级结构文档 优点: 灵活性高、适应性强 缺点: 实现复杂 推荐: 大型知识库、复杂文档系统
def get_optimal_config(doc_type, doc_length, quality_requirement): """获取最优分割配置""" configs = { 'simple_document': { 'strategy': 'fixed_length', 'chunk_size': 800, 'chunk_overlap': 150, 'quality_requirement': 'medium' }, 'technical_document': { 'strategy': 'structured', 'chunk_size': 1200, 'chunk_overlap': 200, 'quality_requirement': 'high' }, 'knowledge_base': { 'strategy': 'semantic', 'chunk_size': 1000, 'chunk_overlap': 100, 'quality_requirement': 'high' }, 'large_document': { 'strategy': 'hierarchical', 'chunk_size': 1500, 'chunk_overlap': 300, 'quality_requirement': 'high' } } base_config = configs.get(doc_type, configs['simple_document']) # 根据文档长度调整 if doc_length > 10000: base_config['chunk_size'] = int(base_config['chunk_size'] * 1.2) # 根据质量要求调整 if quality_requirement == 'high': base_config['chunk_overlap'] = int(base_config['chunk_overlap'] * 1.5) elif quality_requirement == 'low': base_config['chunk_overlap'] = int(base_config['chunk_overlap'] * 0.5) return base_config # 使用示例 config = get_optimal_config('technical_document', 15000, 'high') print("推荐配置:", config)
def optimized_split_pipeline(text, config=None): """优化的分割流程""" if config is None: config = get_optimal_config('simple_document', len(text), 'medium') # 1. 文本预处理 preprocessed_text = preprocess_text_for_splitting(text) # 2. 选择分割策略 if config['strategy'] == 'fixed_length': splitter = RecursiveCharacterTextSplitter( chunk_size=config['chunk_size'], chunk_overlap=config['chunk_overlap'] ) elif config['strategy'] == 'structured': splitter = MarkdownStructureSplitter() elif config['strategy'] == 'semantic': splitter = SemanticChunker( embedding_model, chunk_size=config['chunk_size'] ) elif config['strategy'] == 'hierarchical': splitter = HierarchicalTextSplitter([ chapter_level_split, section_level_split ]) else: splitter = RecursiveCharacterTextSplitter( chunk_size=config['chunk_size'], chunk_overlap=config['chunk_overlap'] ) # 3. 执行分割 chunks = splitter.split_text(preprocessed_text) # 4. 质量评估 evaluator = SplitQualityEvaluator() quality = evaluator.evaluate(chunks, preprocessed_text) # 5. 质量优化(如果需要) if quality['overall_score'] < 0.8: print("分割质量较低,尝试优化...") # 可以尝试调整参数或使用不同的分割策略 return chunks, quality def preprocess_text_for_splitting(text): """分割前的文本预处理""" import re # 1. 清理多余空白 text = re.sub(r'\n\s*\n', '\n\n', text) text = re.sub(r'[ \t]+', ' ', text) # 2. 标准化标点符号 text = re.sub(r'([。!?])', r'\1\n', text) # 3. 处理特殊格式 # 这里可以添加更多预处理步骤 return text.strip()
文档分割策略是RAG系统数据准备的核心环节。通过本章的学习,你应该掌握:
在实际应用中,建议采用渐进式优化的方法:先使用简单的固定长度分割进行原型验证,然后根据评估结果逐步优化为更复杂的策略。记住,没有"最佳"的分割策略,只有最适合当前需求的策略。
关键是要在语义完整性、处理效率和检索准确性之间找到平衡点,并通过持续的质量监控和优化来提升整体性能。
关键词:RAG知识库实战, 文档分割, 固定长度分割, 语义分割, 结构化分割, 最佳实践
难度:进阶
预计阅读:30分钟