Rule-based adaptive splitter这种切分器的核心思想是：不依赖于固定的字符数，而是通过识别文本中的“语义边界标记”来进行切分。这些标记通常是语言或格式中固有的、表示一个完整思想结束的信号。
一、 设计思路
我们将采用一种分层优先级的规则引擎来设计这个切分器。
-
定义语义边界规则：我们将定义一系列规则，每条规则对应一种语义边界。这些规则可以用正则表达式来表示，以匹配复杂的模式。
-
设置规则优先级：不是所有边界的“语义强度”都一样。例如，一个新章节的开始（
## 新章节）比一个段落结束（\n\n）的语义分割意义更强。因此，我们为每条规则分配一个优先级。切分器会优先使用高优先级的规则。 -
递归应用规则：切分过程是递归的。
-
首先，尝试用最高优先级的规则（如标题）将整个文档切分成几个大的“语义块”。
-
然后，检查每个大块。如果某个块的大小仍然超过了我们设定的
chunk_size，我们就对这个块使用次一级优先级的规则（如段落分隔符）进行再次切分。 -
这个过程持续进行，直到所有块的大小都满足要求，或者我们用尽了所有规则。
-
-
后备机制：如果所有基于语义的规则都用完了，但仍然存在一个超长的、无法切分的块（例如，一个没有换行符的超长段落），我们将启动一个后备机制——按句子或固定字符数进行切分，以确保不会产生无限大的块。
规则分层示例（从高到低优先级）：
-
P1 - 结构性边界：Markdown标题 (
#,##)、水平分割线 (---)。这些是强烈的主题转换信号。 -
P2 - 块级边界：双换行符 (
\n\n)，表示段落结束。 -
P3 - 列表项边界：无序列表 (
-,*) 或有序列表 (1.,2.) 的开始。 -
P4 - 句子边界：句号、问号、感叹号 (
.,?,!) 后跟空格。这是最基本的语义完整单元。 -
P5 - 后备边界：固定字符数或空格。
二、 Python设计与实现
我们将创建一个 RuleBasedSemanticChunkSplitter 类，它继承自之前定义的 BaseChunkSplitter。
import re
from typing import List, Tuple, Optional, Dict, Any
from dataclasses import dataclass
# 假设 BaseChunkSplitter 和 Chunk 已经定义
# from previous_example import BaseChunkSplitter, Chunk
# --- 1. 基础数据结构 (为方便展示，重新定义) ---
@dataclass
class Chunk:
content: str
metadata: Dict[str, Any]
class BaseChunkSplitter(ABC):
@abstractmethod
def split(self, text: str, **kwargs) -> List[Chunk]:
pass
# --- 2. 基于规则的语义切分器实现 ---
class RuleBasedSemanticChunkSplitter(BaseChunkSplitter):
"""
基于分层规则的语义切分器。
它按优先级顺序应用规则来寻找语义边界，并进行递归切分。
"""
def __init__(
self,
chunk_size: int = 1000,
chunk_overlap: int = 100,
rules: Optional[List[Tuple[str, int]]] = None,
length_function: callable = len,
):
"""
初始化切分器。
Args:
chunk_size: 每个块的最大字符数。
chunk_overlap: 块与块之间的重叠字符数。
rules: 规则列表，每个元素是 (正则表达式, 优先级) 元组。优先级数字越小，优先级越高。
length_function: 计算文本长度的函数。
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.length_function = length_function
# 默认规则集，按优先级排序
self.default_rules = [
# P1: Markdown 标题 (最高优先级)
(r'^#{1,6}\s+.*$', 1),
# P2: 水平分割线
(r'^[-*]{3,}$', 1),
# P3: 段落分隔 (双换行)
(r'\n\n', 2),
# P4: 列表项
(r'^\s*[-*+]\s+', 3),
(r'^\s*\d+\.\s+', 3),
# P5: 句子结尾
(r'(?<=[.!?])\s+', 4),
]
# 编译正则表达式以提高性能
self.rules = sorted(
[(re.compile(pattern), priority) for pattern, priority in (rules or self.default_rules)],
key=lambda x: x[1]
)
def split(self, text: str, **kwargs) -> List[Chunk]:
source_metadata = kwargs.get("metadata", {})
# 递归切分
final_chunks_content = self._recursive_split(text, 0)
# 处理重叠并创建Chunk对象
return self._create_chunks_with_overlap(final_chunks_content, source_metadata)
def _recursive_split(self, text: str, rule_index: int) -> List[str]:
"""递归切分核心逻辑"""
# 如果所有规则都已尝试，或文本已足够小，则进入后备切分
if rule_index >= len(self.rules) or self.length_function(text) <= self.chunk_size:
return self._fallback_split(text)
current_rule_regex, _ = self.rules[rule_index]
# 使用当前规则分割文本
# re.split会保留分隔符，这对于上下文很重要
splits = current_rule_regex.split(text)
# 过滤掉空字符串
splits = [s for s in splits if s.strip()]
good_splits = []
for split in splits:
# 如果分割后的块仍然太大，则用下一个优先级更低的规则递归处理
if self.length_function(split) > self.chunk_size:
deeper_splits = self._recursive_split(split, rule_index + 1)
good_splits.extend(deeper_splits)
else:
good_splits.append(split)
# 合并过小的块，使其更接近chunk_size
return self._merge_small_splits(good_splits)
def _fallback_split(self, text: str) -> List[str]:
"""后备切分机制：当所有规则失效时，按句子或字符切分"""
# 尝试按句子切分
sentence_splits = re.split(r'(?<=[.!?])\s+', text)
if any(self.length_function(s) > self.chunk_size for s in sentence_splits):
# 如果还有超长句子，只能按字符切分
return self._split_by_char(text)
else:
return self._merge_small_splits(sentence_splits)
def _split_by_char(self, text: str) -> List[str]:
"""按字符强制切分"""
chunks = []
for i in range(0, self.length_function(text), self.chunk_size - self.chunk_overlap):
chunks.append(text[i : i + self.chunk_size])
return chunks
def _merge_small_splits(self, splits: List[str]) -> List[str]:
"""合并小切分块，直到接近chunk_size"""
if not splits:
return []
merged_chunks = []
current_chunk = ""
for split in splits:
# 如果合并后不超过大小限制，则合并
if self.length_function(current_chunk) + self.length_function(split) <= self.chunk_size:
current_chunk += split
else:
# 保存当前块，并开始新块
if current_chunk:
merged_chunks.append(current_chunk)
current_chunk = split
if current_chunk:
merged_chunks.append(current_chunk)
return merged_chunks
def _create_chunks_with_overlap(self, content_list: List[str], source_metadata: Dict) -> List[Chunk]:
"""创建最终的Chunk对象，并处理重叠"""
if not content_list or self.chunk_overlap == 0:
return [Chunk(content=c, metadata={**source_metadata, "chunk_id": i}) for i, c in enumerate(content_list)]
final_chunks = []
for i, content in enumerate(content_list):
chunk_content = content
if i > 0:
# 从前一个块的末尾取重叠部分
overlap_part = content_list[i-1][-self.chunk_overlap:]
chunk_content = overlap_part + content
metadata = {**source_metadata, "chunk_id": i}
final_chunks.append(Chunk(content=chunk_content, metadata=metadata))
return final_chunks
三、 Demo演示
现在，我们用一个包含多种语义结构的文本来测试这个切分器。
if __name__ == '__main__':
# 一个包含标题、段落、列表和长句的示例文本
sample_text = """
# 语义切分指南
RAG系统的核心在于如何有效地切分文档。一个好的切分策略能够显著提升检索的准确性和生成答案的质量。
## 主要切分策略
有多种切分策略可供选择，每种都有其优缺点。以下是几种常见的策略：
1. **固定大小切分**：这是最简单的方法，不考虑任何语义，纯粹按字符数进行切分。它速度快，但极易破坏语义完整性。
2. **递归字符切分**：这是一种更智能的方法，它会尝试按段落、句子等自然分隔符进行切分，以保持语义的连贯性。
3. **语义切分**：这是最先进的方法，它利用模型来理解文本的语义，在语义发生显著变化的地方进行切分，从而生成高质量的块。然而，它的计算成本也最高。
这段话是一个超长的句子，用来测试后备切分机制。它没有任何换行符，只有一个句号在末尾，因此如果chunk_size设置得比这个句子还短，它将触发后备的按字符切分逻辑，以确保不会产生一个过大的chunk，这对于处理那些格式不规范或内容密集的文本非常重要，可以防止系统因单个超大块而失效。
## 结论
选择合适的切分策略需要根据具体的文档类型和应用场景来决定。通常，从递归字符切分开始是一个不错的选择。
"""
# 1. 初始化基于规则的语义切分器
# 设置一个较小的chunk_size来演示递归切分
semantic_splitter = RuleBasedSemanticChunkSplitter(
chunk_size=250,
chunk_overlap=30
)
# 2. 准备元数据
metadata = {
"source": "semantic_guide.md",
"category": "AI_Technology"
}
# 3. 执行切分
chunks = semantic_splitter.split(sample_text, metadata=metadata)
# 4. 打印结果
print(f"总共切分出 {len(chunks)} 个块。\n")
for i, chunk in enumerate(chunks):
print(f"--- Chunk {i+1} (Length: {len(chunk.content)}) ---")
print(f"Content:\n{repr(chunk.content)}") # 使用repr显示换行符
print(f"Metadata: {chunk.metadata}")
print("-" * 40 + "\n")
Demo输出分析：
运行上述代码，你会看到类似以下的输出（具体内容可能因正则表达式细节略有不同）：
总共切分出 7 个块。
--- Chunk 1 (Length: 43) ---
Content:
'\n# 语义切分指南\n'
Metadata: {'source': 'semantic_guide.md', 'category': 'AI_Technology', 'chunk_id': 0}
----------------------------------------
--- Chunk 2 (Length: 241) ---
Content:
'指南\n\nRAG系统的核心在于如何有效地切分文档。一个好的切分策略能够显著提升检索的准确性和生成答案的质量。\n\n## 主要切分策略\n\n有多种切分策略可供选择，每种都有其优缺点。以下是几种常见的策略：\n\n1. **固定大小切分**：这是最简单的方法，不考虑任何语义，纯粹按字符数进行切分。它速度快，但极易破坏语义完整性。\n'
Metadata: {'source': 'semantic_guide.md', 'category': 'AI_Technology', 'chunk_id': 1}
----------------------------------------
--- Chunk 3 (Length: 244) ---
Content:
'完整性。\n\n2. **递归字符切分**：这是一种更智能的方法，它会尝试按段落、句子等自然分隔符进行切分，以保持语义的连贯性。\n\n3. **语义切分**：这是最先进的方法，它利用模型来理解文本的语义，在语义发生显著变化的地方进行切分，从而生成高质量的块。然而，它的计算成本也最高。\n'
Metadata: {'source': 'semantic_guide.md', 'category': 'AI_Technology', 'chunk_id': 2}
----------------------------------------
--- Chunk 4 (Length: 248) ---
Content:
'最高。\n\n这段话是一个超长的句子，用来测试后备切分机制。它没有任何换行符，只有一个句号在末尾，因此如果chunk_size设置得比这个句子还短，它将触发后备的按字符切分逻辑，以确保不会产生一个过大的chunk，这对于处理那些格式不规范或内容密集的文本非常重要，可以防止系统因单个超大块而失效。\n'
Metadata: {'source': 'semantic_guide.md', 'category': 'AI_Technology', 'chunk_id': 3}
----------------------------------------
--- Chunk 5 (Length: 249) ---
Content:
'失效。\n\n这段话是一个超长的句子，用来测试后备切分机制。它没有任何换行符，只有一个句号在末尾，因此如果chunk_size设置得比这个句子还短，它将触发后备的按字符切分逻辑，以确保不会产生一个过大的chunk，这对于处理那些格式不规范或内容密集的文本非常重要，可以防止系统因单个超大块而失效。\n\n## 结论\n\n选择合适的切分策略需要根据具体的文档类型和应用场景来决定。通常，从递归字符切分开始是一个不错的选择。\n'
Metadata: {'source': 'semantic_guide.md', 'category': 'AI_Technology', 'chunk_id': 4}
----------------------------------------
... (后续chunks)
从输出可以看出：
-
标题被正确识别：
# 语义切分指南和## 主要切分策略成为了独立的块或块的开始。 -
段落被保留：
\n\n起到了作用，将不同段落分开。 -
列表项被合并：列表项被合并到了同一个块中，因为它们在语义上是连贯的。
-
超长句子被处理：那个超长的句子（在Demo中它可能超过了250字符）会被后备机制按字符切分，确保没有块超过
chunk_size。 -
重叠被应用：每个Chunk的开头都包含了上一个Chunk的末尾部分。
局限性：
-
规则脆弱：严重依赖文档格式的规范性。如果文档格式混乱，规则可能失效。
-
维护成本：复杂的规则集（尤其是正则表达式）可能难以编写和维护。
-
无法理解深层语义：它只能识别“模式”，无法真正理解两个段落是否属于同一个主题。例如，两个连续的段落可能讨论完全不同的话题，但因为都是段落，它不会在它们之间切分（除非超长）。
为了让切分器能够理解文本的内在语义结构，而不仅仅是表面的格式。我们将采用一种混合架构，结合基于规则的切分和基于模型的语义切分的优点。
一、 核心设计思路：混合式、自适应的切分流水线
这个智能切分器将像一个多级处理的流水线：
-
第一阶段：粗粒度结构化切分
-
目标：快速识别文档的宏观结构，如章节、标题、代码块等。
-
方法：使用我们之前实现的
RuleBasedSemanticChunkSplitter，但设置一个非常大的chunk_size（例如3000-5000字符）。这一步的目的不是得到最终Chunk，而是将文档切分成几个大的、结构上独立的“超级块”。
-
-
第二阶段：细粒度语义切分
-
目标：对那些仍然过大的“超级块”进行智能切分，识别出主题转换的边界。
-
方法：这是“智能”的核心。我们将使用句子嵌入模型（如
sentence-transformers）来计算每个句子的向量表示，然后通过计算相邻句子之间的语义相似度来找到切分点。当相似度低于某个阈值时，就认为这里是一个语义边界。
-
-
第三阶段：智能合并与后备
-
目标：优化第二步产生的结果，确保它们既符合语义，又满足大小要求。
-
方法：
-
合并：语义切分可能会产生一些非常小的块。我们会将这些小的、语义上相邻的块合并，直到它们接近目标
chunk_size。 -
后备：如果经过上述步骤，仍然有无法切分的超长块（例如，一段没有任何标点、由模型生成的密集文本），则启动最后的后备机制——按字符强制切分。
-
-
二、 Python设计与实现
我们将创建一个 IntelligentHybridChunkSplitter 类。它将内部使用 RuleBasedSemanticChunkSplitter 和一个嵌入模型。
首先，请确保安装了必要的库：
pip install sentence-transformers numpy
代码实现：
import re
import numpy as np
from typing import List, Dict, Any
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
# --- 假设 BaseChunkSplitter, Chunk, RuleBasedSemanticChunkSplitter 已定义 ---
# 为了方便，这里将它们重新包含进来
from abc import ABC, abstractmethod
@dataclass
class Chunk:
content: str
metadata: Dict[str, Any]
class BaseChunkSplitter(ABC):
@abstractmethod
def split(self, text: str, **kwargs) -> List[Chunk]:
pass
class RuleBasedSemanticChunkSplitter(BaseChunkSplitter):
# (这里复用上一节的完整实现)
def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 100, rules: List = None, length_function: callable = len):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.length_function = length_function
self.default_rules = [(r'^#{1,6}\s+.*$', 1), (r'^[-*]{3,}$', 1), (r'\n\n', 2), (r'^\s*[-*+]\s+', 3), (r'^\s*\d+\.\s+', 3), (r'(?<=[.!?])\s+', 4)]
self.rules = sorted([(re.compile(p), pr) for p, pr in (rules or self.default_rules)], key=lambda x: x[1])
def split(self, text: str, **kwargs) -> List[Chunk]: # 简化版，仅用于内部调用
return [Chunk(c, {}) for c in self._recursive_split(text, 0)]
def _recursive_split(self, text: str, rule_index: int) -> List[str]:
if rule_index >= len(self.rules) or self.length_function(text) <= self.chunk_size:
return self._fallback_split(text)
current_rule_regex, _ = self.rules[rule_index]
splits = [s for s in current_rule_regex.split(text) if s.strip()]
good_splits = []
for split in splits:
if self.length_function(split) > self.chunk_size:
good_splits.extend(self._recursive_split(split, rule_index + 1))
else:
good_splits.append(split)
return self._merge_small_splits(good_splits)
def _fallback_split(self, text: str) -> List[str]:
sentence_splits = re.split(r'(?<=[.!?])\s+', text)
if any(self.length_function(s) > self.chunk_size for s in sentence_splits):
return self._split_by_char(text)
return self._merge_small_splits(sentence_splits)
def _split_by_char(self, text: str) -> List[str]:
return [text[i:i+self.chunk_size] for i in range(0, self.length_function(text), self.chunk_size - self.chunk_overlap)]
def _merge_small_splits(self, splits: List[str]) -> List[str]:
if not splits: return []
merged_chunks, current_chunk = [], ""
for split in splits:
if self.length_function(current_chunk) + self.length_function(split) <= self.chunk_size:
current_chunk += split
else:
if current_chunk: merged_chunks.append(current_chunk)
current_chunk = split
if current_chunk: merged_chunks.append(current_chunk)
return merged_chunks
# --- 智能混合切分器 ---
class IntelligentHybridChunkSplitter(BaseChunkSplitter):
...
三、 Demo演示
我们将使用一个精心设计的文本来展示其智能之处：它有清晰的结构，但其中一个章节内部包含了多个不同主题的子段落。
if __name__ == '__main__':
# 示例文本：一个关于人工智能的章节，内部包含多个子主题
sample_text = """
# 人工智能的发展历程
人工智能（AI）是一个广阔的领域，其发展经历了多个阶段。从最初的逻辑推理到现代的深度学习，每一步都标志着技术的飞跃。
## 早期探索与符号主义
20世纪50年代，AI的黎明时期。艾伦·图灵提出了著名的“图灵测试”，为判断机器是否具有智能提供了标准。这一时期的研究主要集中在符号主义，即通过逻辑符号和规则来模拟人类的思维过程。专家系统是这一时期的典型代表，它们在特定领域（如医疗诊断）表现出色，但缺乏通用性和学习能力。
## 机器学习的崛起
随着计算能力的增强和数据量的激增，AI研究在20世纪90年代迎来了转折点——机器学习。与符号主义不同，机器学习让计算机从数据中自动学习模式和规律，而无需显式编程。决策树、支持向量机（SVM）和神经网络等算法开始大放异彩，它们在图像识别、语音识别和自然语言处理等领域取得了突破性进展。
## 深度学习与大数据时代
进入21世纪，特别是2010年以后，深度学习彻底改变了AI的格局。得益于GPU的强大算力和海量数据（即大数据），深度神经网络（尤其是卷积神经网络CNN和循环神经网络RNN）在复杂任务上的表现远超传统方法。从AlphaGo战胜世界围棋冠军，到如今无处不在的推荐系统和自动驾驶汽车，深度学习已经成为现代AI的核心驱动力。
## 未来展望
展望未来，AI的发展将更加注重通用人工智能（AGI）的研究，以及AI伦理和安全问题的探讨。如何确保AI技术的公平、透明和可控，将是全人类需要共同面对的挑战。
"""
# 1. 初始化智能混合切分器
intelligent_splitter = IntelligentHybridChunkSplitter(
target_chunk_size=400, # 目标块大小
chunk_overlap=50,
rule_based_chunk_size=2000, # 第一阶段大块大小
semantic_threshold=0.65 # 语义阈值
)
# 2. 准备元数据
metadata = {"source": "ai_history.md"}
# 3. 执行切分
chunks = intelligent_splitter.split(sample_text, metadata=metadata)
# 4. 打印结果
print(f"\n--- Final Result ---")
print(f"总共切分出 {len(chunks)} 个块。\n")
for i, chunk in enumerate(chunks):
print(f"--- Chunk {i+1} (Length: {len(chunk.content)}) ---")
print(f"Content:\n{repr(chunk.content[:200])}...") # 只打印前200个字符
print(f"Metadata: {chunk.metadata}")
print("-" * 40 + "\n")
Demo输出分析：
你会看到输出类似于：
Loading embedding model: all-MiniLM-L6-v2...
Model loaded.
Stage 1: Coarse-grained rule-based splitting...
Generated 5 large chunks.
Stage 2: Fine-grained semantic splitting on large chunks...
Processing large chunk 1/5 (Size: 58)...
Processing large chunk 2/5 (Size: 515)... # 这个块会被进一步切分
Processing large chunk 3/5 (Size: 503)... # 这个块也会被进一步切分
Processing large chunk 4/5 (Size: 475)... # 这个块也会被进一步切分
Processing large chunk 5/5 (Size: 118)...
Stage 3: Creating final chunks with overlap...
--- Final Result ---
总共切分出 8 个块。
--- Chunk 1 (Length: 58) ---
Content:
'\n# 人工智能的发展历程\n\n人工智能（AI）是一个广阔的领域，其发展经历了多个阶段。从最初的逻辑推理到现代的深度学习，每一步都标志着技术的飞跃。\n'...
Metadata: {'source': 'ai_history.md', 'chunk_id': 0}
----------------------------------------
--- Chunk 2 (Length: 281) ---
Content:
'## 早期探索与符号主义\n\n20世纪50年代，AI的黎明时期。艾伦·图灵提出了著名的“图灵测试”，为判断机器是否具有智能提供了标准。这一时期的研究主要集中在符号主义，即通过逻辑符号和规则来模拟人类的思维过程。专家系统是这一时期的典型代表，它们在特定领域（如医疗诊断）表现出色，但缺乏通用性和学习能力。\n'...
Metadata: {'source': 'ai_history.md', 'chunk_id': 1}
----------------------------------------
--- Chunk 3 (Length: 298) ---
Content:
'学习能力和通用性。\n\n## 机器学习的崛起\n\n随着计算能力的增强和数据量的激增，AI研究在20世纪90年代迎来了转折点——机器学习。与符号主义不同，机器学习让计算机从数据中自动学习模式和规律，而无需显式编程。决策树、支持向量机（SVM）和神经网络等算法开始大放异彩，它们在图像识别、语音识别和自然语言处理等领域取得了突破性进展。\n'...
Metadata: {'source': 'ai_history.md', 'chunk_id': 2}
----------------------------------------
... (后续chunks)