【实操实录】深入探讨了RAG系统中的文本分块策略,固定长度、递归字符分块、用文档结构的Markdown/HTML分块、基于语义和主题的高级分块、小-大分块等混合策略、chunk_size设置
本文摘要:文章深入探讨了RAG系统中的文本分块策略,指出分块质量直接影响系统性能。介绍了几种主要分块方法:1)固定长度和递归字符分块等基础策略;2)利用文档结构的Markdown/HTML分块;3)基于语义和主题的高级分块;4)小-大分块等混合策略。强调应根据文档特性选择策略:从递归分块开始,检查结构化特征,在精度不足时尝试语义分块,对复杂文档采用混合策略。文章提供了Python代码示例,并指出合
🫕𓂂𓏸 ‧₊˚不恰当的分块,就像是给模型提供了一堆被打乱顺序、信息残缺的“坏数据”。模型能力再强,也无法从支离破碎的知识中推理出正确、完整的答案。分块的质量,直接决定了RAG系统性能的下限⁺˚🚂॰‧₊˚
一. 分块的本质:为何与如何
分块的必要性源于两个核心限制🫕𓂂𓏸 ‧₊˚:
⁺˚🚂॰‧₊˚模型上下文窗口:大语言模型(LLM)无法一次性处理无限长度的文本,分块是将长文档切分为模型可以处理的、大小适中的片段。
🎠检索信噪比:在检索时,如果一个文本块包含过多无关信息(噪声),就会稀释核心信号,导致检索器难以精确匹配用户意图。
⁺˚🚂॰‧₊˚理想的分块是在上下文完整性与信息密度之间找到最佳平衡。chunk_size
和 chunk_overlap
是调控这一平衡的基础参数。chunk_overlap
通过在相邻块之间保留部分重复文本,确保了跨越块边界的语义连续性。
保存文档🫕𓂂𓏸 ‧₊˚:
# 将文档保存为临时文件
with open("sample_doc.txt", "w", encoding="utf-8") as f:
f.write(sample_document)
# 加载文档
loader = TextLoader("sample_doc.txt", encoding="utf-8")
documents = loader.load()
print(f"原始文档数量: {len(documents)}")
print(f"原始文档内容长度: {len(documents[0].page_content)}")
二. 分块策略详解与代码实践
2.1.1 固定长度分块🫕𓂂𓏸 ‧₊˚
⁺˚🚂॰‧₊˚这是最直接的方法,按预设的字符数进行切割。它不考虑文本的任何逻辑结构,实现简单,但容易破坏语义完整性。
def fixed_length_chunking():
"""固定长度分块示例"""
# 初始化字符文本分块器
text_splitter = CharacterTextSplitter(
chunk_size=200, # 每个块的字符数
chunk_overlap=20, # 块之间的重叠字符数
separator=" " # 分隔符
)
# 进行分块
chunks = text_splitter.split_documents(documents)
# 输出分块结果
print("\n=== 固定长度分块结果 ===")
print(f"分块数量: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(f"长度: {len(chunk.page_content)}")
print(chunk.page_content)
return chunks
# 执行固定长度分块
fixed_chunks = fixed_length_chunking()
🎠核心思想:按固定字符数 chunk_size
切分文本。
📮适用场景:结构性弱的纯文本,或对语义要求不高的预处理阶段。
2.1.2 递归字符分块🫕𓂂𓏸 ‧₊˚
⁺˚🚂॰‧₊˚LangChain推荐的通用策略。它按预设的字符列表(如 ["\n\n", "\n", " ", ""]
)进行递归分割,尝试优先保留段落、句子等逻辑单元的完整性。
def recursive_character_chunking():
"""递归字符分块示例"""
# 初始化递归字符分块器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=30,
# 分割符列表,按优先级排序
separators=["\n\n", "\n", "。", "!", "?", ",", " ", ""]
)
# 进行分块
chunks = text_splitter.split_documents(documents)
# 输出分块结果
print("\n=== 递归字符分块结果 ===")
print(f"分块数量: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(f"长度: {len(chunk.page_content)}")
print(chunk.page_content)
return chunks
# 执行递归字符分块
recursive_chunks = recursive_character_chunking()
🎠核心思想:按层次化分隔符列表进行递归切分。
📮适用场景:绝大多数文本类型的首选通用策略
🎠参数调优说明:对于固定长度和递归分块,chunk_size
和 chunk_overlap
的设置至关重要:
⁺˚🚂॰‧₊˚chunk_size
: 决定了每个块的大小。块太小,可能导致上下文信息不足,模型无法充分理解;块太大,则可能引入过多噪声,降低检索的信噪比,并增加API调用成本,通常根据嵌入模型的最佳输入长度和文本特性来选择,例如 256, 512, 1024。
⁺˚🚂॰‧₊˚chunk_overlap
: 决定了相邻块之间的重叠字符数。设置合理的重叠(如 chunk_size
的10%-20%)可以有效防止在块边界处切断完整的语义单元(如一个长句子),是保证语义连续性的关键
2.1.3 基于句子的分块🫕𓂂𓏸 ‧₊˚
⁺˚🚂॰‧₊˚以句子为最小单元进行组合,确保了最基本的语义完整性。
但是会导致切割出来的文本特别细碎,大模型无法联系句与句之间的内容联系,导致只能根据词与词的匹配度进行检索匹配,无法根据语义进行判断分析
def sentence_based_chunking():
"""基于句子的分块示例"""
# 自定义中文句子分割函数
def chinese_sentence_splitter(text):
# 使用正则表达式分割中文句子
sentence_endings = re.compile(r'([。!?;,.!?;])')
sentences = sentence_endings.split(text)
# 重组句子
sentences = [sentences[i] + sentences[i+1]
for i in range(0, len(sentences)-1, 2)]
# 过滤空句子
return [s.strip() for s in sentences if s.strip()]
# 获取文档内容
text = documents[0].page_content
# 分割句子
sentences = chinese_sentence_splitter(text)
print(f"\n=== 中文句子分割结果 ===")
print(f"句子数量: {len(sentences)}")
for i, sent in enumerate(sentences[:5]): # 只显示前5个句子
print(f"句子 {i+1}: {sent}")
# 将句子组合成块
chunk_size = 3 # 每个块包含3个句子
chunks = []
for i in range(0, len(sentences), chunk_size):
chunk_sentences = sentences[i:i+chunk_size]
chunk_text = " ".join(chunk_sentences)
chunks.append(chunk_text)
# 输出分块结果
print("\n=== 基于句子的分块结果 ===")
print(f"分块数量: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(f"句子数量: {len(chunks[i].split('。'))-1}")
print(chunk)
return chunks
🎠核心思想:将文本分割成句子,再将句子聚合成块。
📮适用场景:对句子完整性要求高的场景,如法律文书、新闻报道
🎠注意事项:语言模型的选择许多标准库的默认配置是为英文设计的。例如,nltk.tokenize.sent_tokenize
默认使用基于英文训练的Punkt模型进行分句。如果直接用于处理中文文本,会因无法识别中文标点而导致分句失败。 处理中文时,必须采用适合中文的分割方法,例如:
⁺˚🚂॰‧₊˚基于中文标点符号(如 。!?)的正则表达式进行切分。
使用加载了中文模型的NLP库(如 spaCy, HanLP 等)进行更准确的分句
2.2 结构感知分块
⁺˚🚂॰‧₊˚利用文档固有的结构信息(如标题、列表、对话轮次)作为分块边界,这种方法逻辑性强,能更好地保留上下文。
2.2.1 结构化文本分块🫕𓂂𓏸 ‧₊˚
def structured_text_chunking():
"""结构化文本分块示例"""
# 定义要识别的标题层级
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
# 初始化Markdown分块器
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on,
strip_headers=False, # 保留标题文本在块中
)
# 进行分块
chunks = markdown_splitter.split_text(sample_document)
# 输出分块结果
print("\n=== 结构化文本分块结果 ===")
print(f"分块数量: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(f"元数据: {chunk.metadata}")
print(chunk.page_content)
return chunks
# 执行结构化文本分块
structured_chunks = structured_text_chunking()
🎠核心思想:根据Markdown的标题层级或HTML的标签来定义块的边界。
📮适用场景:格式规范的Markdown、HTML文档
2.2.2 对话式分块🫕𓂂𓏸 ‧₊˚
def conversational_chunking():
"""对话式分块示例"""
# 示例对话内容
conversation = """
用户: 你好,什么是大语言模型?
助手: 大语言模型是一种能够理解和生成人类语言的人工智能模型。
用户: 它有什么应用场景呢?
助手: 大语言模型的应用场景很广泛,包括文本生成、问答系统、代码生成等。
用户: 能详细说说文本生成的应用吗?
助手: 当然可以。文本生成可以用于写作、翻译、摘要等任务。在内容创作领域,
它可以帮助作者生成文章初稿,提高写作效率。
用户: 那RAG技术又是什么呢?
助手: RAG是检索增强生成的缩写,是一种结合信息检索和文本生成的技术,
能够让大语言模型在回答问题时参考外部知识库。
"""
# 按对话轮次分块
# 分割正则表达式,匹配"用户:"或"助手:"
pattern = r'(用户:|助手:)'
parts = re.split(pattern, conversation)
# 组合成对话块
chunks = []
for i in range(1, len(parts), 2):
speaker = parts[i].strip()
content = parts[i+1].strip()
chunks.append(f"{speaker} {content}")
# 输出分块结果
print("\n=== 对话式分块结果 ===")
print(f"分块数量: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(chunk)
return chunks
# 执行对话式分块
conversation_chunks = conversational_chunking()
🎠核心思想:根据对话的发言人或轮次进行分块。
📮适用场景:客服对话、访谈记录、会议纪要
2.3 语义与主题分块
⁺˚🚂॰‧₊˚这类方法超越了文本的物理结构,根据内容的语义含义进行切分。
2.3.1 语义分块🫕𓂂𓏸 ‧₊˚
🎠核心思想:计算相邻句子/段落的向量相似度,在语义发生突变(相似度低)的位置进行切分。
📮适用场景:知识库、研究论文等需要高精度语义内聚的文档
def semantic_chunking():
"""语义分块示例"""
# 使用之前定义的中文句子分割函数
def chinese_sentence_splitter(text):
sentence_endings = re.compile(r'([。!?;,.!?;])')
sentences = sentence_endings.split(text)
sentences = [sentences[i] + sentences[i+1]
for i in range(0, len(sentences)-1, 2)]
return [s.strip() for s in sentences if s.strip()]
# 获取文档内容并分割成句子
text = documents[0].page_content
sentences = chinese_sentence_splitter(text)
print(f"\n=== 语义分块 - 句子数量: {len(sentences)} ===")
# 初始化嵌入模型
try:
# 如果你有OpenAI API密钥,可以使用OpenAI的嵌入模型
# os.environ["OPENAI_API_KEY"] = "your-api-key"
# embeddings = OpenAIEmbeddings()
# 这里使用SentenceTransformers作为替代(开源免费)
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
sentence_embeddings = model.encode(sentences)
# 计算句子间的相似度
similarities = []
for i in range(len(sentence_embeddings) - 1):
sim = cosine_similarity(
[sentence_embeddings[i]],
[sentence_embeddings[i+1]]
)[0][0]
similarities.append(sim)
# 设置阈值,确定分块边界
breakpoint_threshold = 0.5 # 语义相似度阈值
breakpoints = [i for i, sim in enumerate(similarities) if sim < breakpoint_threshold]
# 根据边界进行分块
chunks = []
start = 0
for breakpoint in breakpoints:
end = breakpoint + 1
chunk = " ".join(sentences[start:end])
chunks.append(chunk)
start = end
# 添加最后一个块
chunks.append(" ".join(sentences[start:]))
# 输出分块结果
print("\n=== 语义分块结果 ===")
print(f"分块数量: {len(chunks)}")
print(f"分块边界位置: {breakpoints}")
for i, chunk in enumerate(chunks):
print(f"\n块 {i+1}:")
print(chunk)
return chunks
except Exception as e:
print(f"语义分块出错: {e}")
return []
# 执行语义分块
semantic_chunks = semantic_chunking()
🎠参数调优说明:SemanticChunker
的效果高度依赖 breakpoint_threshold_amount
这个阈值参数。
📮阈值的作用:可以将其理解为一个“语义变化敏感度”的控制器。当相邻句子的语义相似度差异超过这个阈值时,就在此处进行切分。
🎠阈值较低:切分会非常敏感,即使微小的语义变化也可能导致分块。这会产生大量非常小且高度内聚的块。
📮阈值较高:对语义变化的容忍度更大,只有在话题发生显著转变时才会切分,从而产生更少、更大的块。
⁺˚🚂॰‧₊˚这个值没有固定的最佳答案,需要根据您的文档内容和领域进行反复实验和调整,以达到最理想的分块效果。
2.3.2 基于主题的分块🫕𓂂𓏸 ‧₊˚
🎠核心思想:利用主题模型(如LDA)或聚类算法,在文档的宏观主题发生转换时进行切分。
🫕适用场景:长篇、多主题的报告或书籍
📮注意事项:基于主题的分块在实践中需要谨慎使用,因为它存在一些固有挑战:
🎠数据依赖性强:主题模型和聚类算法通常需要足够长的文本和明确的主题区分度才能生效。对于短文本或主题交叉频繁的文档,效果可能不佳。
📮预处理要求高:文本清洗、停用词去除、词形还原等预处理步骤对最终的主题识别质量有巨大影响。
🎠超参数敏感:需要预先设定主题数量(n_topics
)等超参数,而这往往难以准确估计。
⁺˚🚂॰‧₊˚因此,当直接应用此类方法时,可能会发现分出的块在逻辑上不连贯,或者与实际主题边界不符。 此方法更适合作为一种探索性工具,在主题边界清晰的长文档上使用,并需要进行充分的实验验证。
2.4 高级分块策略
2.4.1 小-大分块🫕𓂂𓏸 ‧₊˚
🎠核心思想:使用小块(如句子)进行高精度检索,然后将包含该小块的原始大块(如段落)作为上下文送入LLM。
🎠适用场景:需要高检索精度和丰富生成上下文的复杂问答场景
2.4.2 代理式分块🫕𓂂𓏸 ‧₊˚
🎠核心思想:利用一个LLM Agent来模拟人类的阅读理解过程,动态决定分块边界。•适用场景:实验性项目,或处理高度复杂、非结构化的文本
def parent_child_chunking():
"""小-大分块策略示例"""
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 初始化向量存储
try:
# 使用OpenAI嵌入
# embeddings = OpenAIEmbeddings()
# 这里使用SentenceTransformers作为替代
from langchain.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(model_name='all-MiniLM-L6-v2')
# 定义子分块器(小块,用于检索)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10)
# 定义父分块器(大块,用于生成上下文)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
# 向量存储用于存储小块
vectorstore = Chroma(
collection_name="split_parents",
embedding_function=embeddings
)
# 存储用于保存大块
store = InMemoryStore()
# 初始化ParentDocumentRetriever
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
# 添加文档
retriever.add_documents(documents)
# 测试检索
query = "RAG技术包括哪些部分?"
retrieved_docs = retriever.get_relevant_documents(query)
print("\n=== 小-大分块检索结果 ===")
print(f"检索到的文档数量: {len(retrieved_docs)}")
for i, doc in enumerate(retrieved_docs):
print(f"\n检索到的文档 {i+1}:")
print(f"长度: {len(doc.page_content)}")
print(doc.page_content)
return retriever
except Exception as e:
print(f"小-大分块出错: {e}")
return None
# 执行小-大分块
parent_child_retriever = parent_child_chunking()
3. 混合分块:平衡效率与质量
⁺˚🚂॰‧₊˚在实践中,单一策略往往难以应对所有情况,混合分块结合了多种策略的优点
def hybrid_chunking():
"""混合分块策略示例:结构化 + 递归分块"""
# 第一步:使用Markdown分块器进行结构化分块(粗粒度)
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on,
strip_headers=False,
)
# 进行第一次分块(粗分)
initial_chunks = markdown_splitter.split_text(sample_document)
print("\n=== 混合分块 - 第一次分块结果 ===")
print(f"第一次分块数量: {len(initial_chunks)}")
# 第二步:对过大的块使用递归分块器进行二次分块(细分)
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=20,
separators=["\n\n", "\n", "。", "!", "?", ",", " ", ""]
)
final_chunks = []
for chunk in initial_chunks:
# 如果块太大,进行二次分块
if len(chunk.page_content) > 300:
sub_chunks = recursive_splitter.split_text(chunk.page_content)
# 保留元数据
for sub_chunk in sub_chunks:
final_chunks.append({
"content": sub_chunk,
"metadata": chunk.metadata
})
else:
final_chunks.append({
"content": chunk.page_content,
"metadata": chunk.metadata
})
# 输出最终分块结果
print("\n=== 混合分块 - 最终结果 ===")
print(f"最终分块数量: {len(final_chunks)}")
for i, chunk in enumerate(final_chunks):
print(f"\n块 {i+1}:")
print(f"元数据: {chunk['metadata']}")
print(f"长度: {len(chunk['content'])}")
print(chunk['content'])
return final_chunks
# 执行混合分块
hybrid_chunks = hybrid_chunking()
🎠核心思想:先用一种宏观策略(如结构化分块)进行粗粒度切分,再对过大的块使用更精细的策略(如递归或语义分块)进行二次切分。
🎠适用场景:处理结构复杂且内容密度不均的文档。
代码示例 (结构化 + 递归混合):
4. 如何选择最佳分块策略?
⁺˚🚂॰‧₊˚面对众多策略,合理的选择路径比逐一尝试更重要
第一步:从基准策略开始
🎠默认选项:RecursiveCharacterTextSplitter
无论处理何种文本,这都是最稳妥的起点。它在通用性、简单性和效果之间取得了很好的平衡,使用它建立一个性能基线
第二步:检查结构化特征
🫕优先选项:结构感知分块 在应用基准策略后,检查你的文档是否具有明确的结构,如Markdown标题、HTML标签、代码或对话格式。如果有,可以切换到MarkdownHeaderTextSplitter
等相应的结构化分块方法。这是成本最低、收益较好的优化步骤。
第三步:当精度成为瓶颈时
🎠进阶选项:语义分块 或 小-大分块 如果基础策略和结构化策略的检索效果仍不理想,无法满足业务需求,需要更高维度的语义信息。
🫕SemanticChunker
:适用于需要块内语义高度一致的场景。
ParentDocumentRetriever
(小-大分块):适用于既要保证检索精准度,又需要为LLM提供完整上下文的复杂问答场景。
第四步:应对极端复杂的文档
🎠高级实践:混合分块 对于那些结构复杂、内容密度不均、混合多种格式的复杂文档,单一策略难以应对,可以先用MarkdownHeaderTextSplitter
进行宏观切分,再对过长的块用RecursiveCharacterTextSplitter
进行二次细分
致谢
⊹꙳🫧˖✧谢谢大家的阅读,很多不足支出,欢迎大家在评论区指出,如果我的内容对你有帮助,
可以点赞 , 收藏 ,大家的支持就是我坚持下去的动力!
请赐予我平静,去接受我无法改变的 :赐予我勇气,去改变我能改变的!

为武汉地区的开发者提供学习、交流和合作的平台。社区聚集了众多技术爱好者和专业人士,涵盖了多个领域,包括人工智能、大数据、云计算、区块链等。社区定期举办技术分享、培训和活动,为开发者提供更多的学习和交流机会。
更多推荐
所有评论(0)