RAG系统文本分块优化指南:9种实用策略让检索精度翻倍

B站影视 韩国电影 2025-06-04 10:32 2

摘要:检索增强生成(Retrieval-Augmented Generation, RAG)技术通过将外部知识检索与大语言模型生成能力相结合,实现了基于检索文本块(chunk)上下文的高质量内容生成。RAG系统的性能很大程度上依赖于文本分块策略的选择和实施。

检索增强生成(Retrieval-Augmented Generation, RAG)技术通过将外部知识检索与大语言模型生成能力相结合,实现了基于检索文本块(chunk)上下文的高质量内容生成。RAG系统的性能很大程度上依赖于文本分块策略的选择和实施。

文本分块是RAG系统中的关键预处理环节,文本块定义为从大型文档或数据集中按照特定规则和策略分割而成的文本片段,这些片段将被嵌入并索引到RAG知识库中以支持检索操作。例如,简单的文档分割可以产生两个独立的文本块,如下图所示。

文本块作为可管理的嵌入、索引和检索单元发挥着核心作用。检索系统通过这些单元为查询找到相关上下文,相较于将完整文档传递给生成模型,传递最相关的文本段落在计算效率和响应质量方面都具有明显优势。

不同的分块策略各有其适用场景、优势和局限性。本文将深入分析九种主要的文本分块策略及其具体实现方法。下图概括了我们将要讨论的内容。

文本分块是将大型文档分解为可管理处理单元的结构化过程。该过程的目标是将文档重新组织为多个部分,使RAG系统能够根据输入查询高效检索最相关的内容片段。

RAG系统的工作机制是在推理阶段对查询进行嵌入编码,然后从向量数据库中检索前k个最相关的文本块。这些文本块作为上下文信息提供给生成模型,以产生准确且相关的响应。

文本块的粒度可以从句子级别扩展到段落或章节级别,具体取决于应用场景和所采用的分割规则。这些规则构成了我们所说的分块策略。目前存在多种分块策略,每种策略都有其特定的优势和限制条件。

固定大小分块是一种将文本按照统一长度标准进行分割的策略,分割标准可以是词数、标记数或字符数。例如,可以将文档分割为每块包含200个词的片段。

该方法基于直接的文本分割操作,实现简单且计算效率高。当需要确保输入数据具有一致尺寸时(如为具有固定输入维度的机器学习模型提供数据),通常采用这种策略。

固定大小分块的主要优势包括实现简单快速,需要最少的计算资源或语言分析,可以快速部署;同时具有良好的可预测性,生成的统一大小文本块简化了数据库中的存储、索引和检索操作。

然而,该策略也存在明显的局限性。首先是上下文碎片化问题,由于分割位置的随意性,经常会切断句子或段落,从而破坏语义完整性。其次是缺乏灵活性,无法适应文本的自然结构和流程,可能将相关概念分离到不同的文本块中。

以下代码演示了固定大小分块策略的实现:

def fixed_size_chunk(text, chunk_size=500):
words = text.split
return [" ".join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
chunks = fixed_size_chunk(pdf_text, chunk_size=500)

基于句子的分块策略根据句子边界对文本进行分割。实现方式可以采用基于句点符号的简单规则,或者使用spaCy、NLTK等自然语言处理库来准确识别句子边界。

该策略尊重语法和上下文单元的完整性,特别适用于句子完整性至关重要的任务,如机器翻译或情感分析等应用场景。

基于句子分块的主要优势在于语义保持能力,通过保持完整句子结构,显著降低了语义信息丢失的风险。该策略遵循自然的语言边界,与人类阅读习惯保持一致,提高了结果的可解释性。

该策略的局限性主要体现在文本块大小的可变性上。短句子会产生过小的文本块,而长句子可能超出预期的大小限制,导致处理的不一致性。此外该方法依赖自然语言处理工具的质量,句子分词器在处理复杂标点符号或非标准文本时可能出现错误。

使用spaCy库实现基于句子的分块策略的代码如下:

import spacy
nlp = spacy.load("en_core_web_sm")
def sentence_chunk(text):
doc = nlp(text)
return [sent.text.strip for sent in doc.sents]
chunks = sentence_chunk(pdf_text)

基于语义的分块策略通过分析文本的语义内容来进行分组。该方法通常使用嵌入技术计算句子或段落之间的语义相似性,根据设定的语义阈值(如余弦相似度)形成聚类,确保文本块内容在语义上具有高度相关性。

该策略的主要优势在于能够生成具有强语义关联的文本块,这种上下文相关性使其特别适合检索增强生成应用。同时,该方法具有良好的适应性,能够动态适应文档内容,避免了严格大小限制带来的问题。

基于语义分块的局限性主要体现在计算成本方面,嵌入生成和相似性计算是资源密集型操作,在大规模数据集上可能面临性能瓶颈。该方法对阈值参数较为敏感,需要仔细调整以平衡文本块大小和相关性,存在过度碎片化或过度聚合的风险。

以下代码实现了基于语义的分块策略:

import spacy
from sentence_transformers import SentenceTransformer, util
nlp = spacy.load("en_core_web_sm")
model = SentenceTransformer("all-MiniLM-L6-v2")
def semantic_embedding_chunk(text, threshold=0.75):
"""
使用句子嵌入将文本分割成语义块。
使用 spaCy 进行句子分割,使用 SentenceTransformer 生成嵌入。
:param text: 要分块的完整文本。
:param threshold: 用于将句子添加到当前块的余弦相似度阈值。
:return: 语义块列表(每个块为字符串)。
"""
# 句子分割
doc = nlp(text)
sentences = [sent.text.strip for sent in doc.sents if sent.text.strip]
chunks =
current_chunk_sentences =
current_chunk_embedding = None
for sentence in sentences:
# 为当前句子生成嵌入
sentence_embedding = model.encode(sentence, convert_to_tensor=True)
# 如果开始一个新的块,则用当前句子初始化它
if current_chunk_embedding is None:
current_chunk_sentences = [sentence]
current_chunk_embedding = sentence_embedding
else:
# 计算当前句子与块嵌入之间的余弦相似度
sim_score = util.cos_sim(sentence_embedding, current_chunk_embedding)
if sim_score.item >= threshold:
# 将句子添加到当前块,并更新块的平均嵌入
current_chunk_sentences.append(sentence)
num_sents = len(current_chunk_sentences)
current_chunk_embedding = ((current_chunk_embedding * (num_sents - 1)) + sentence_embedding) / num_sents
else:
# 完成当前块并开始一个新块
chunks.append(" ".join(current_chunk_sentences))

# 如果存在,则附加最后一个块
if current_chunk_sentences:

return chunks
semantic_chunks = semantic_embedding_chunk(pdf_text, threshold=0.75)
for i, chunk in enumerate(semantic_chunks):
print(f"Chunk {i+1}:\n{chunk}\n{'-'*60}")

通过调整阈值参数,可以控制文本块的语义聚合程度,实现不同粒度的分块效果。

递归分块是一种迭代式文本分割策略,按照层次化的分割规则(如段落、句子、词汇)逐步分割文本,直到文本块满足预设的大小约束。例如,当一个长段落超出大小限制时,系统会进一步将其分割为句子和更小的标记组合。

递归分块的主要优势在于能够严格遵守大小限制,确保所有生成的文本块都符合预设的尺寸要求。同时,该策略具有良好的灵活性,能够优雅地处理各种长度的文本内容,适应不同类型的文档结构。

该策略的局限性主要体现在碎片化风险和处理效率方面。过度分割可能破坏文本的连贯性和语义完整性,而多层递归处理会显著增加计算时间,在处理大规模文档时可能影响系统性能。

递归分块的实现代码如下:

import PyPDF2
def iterative_chunk(text, max_length=500):
chunks =
while len(text) > max_length:
separators = ["\n\n", "\n", " ", ""]
found = False
for sep in separators:
if sep == "":
# 如果没有找到分隔符,则直接在 max_length 处截断文本。
chunk = text[:max_length]
chunks.append(chunk.strip)
text = text[max_length:]
found = True
break
idx = text.rfind(sep, 0, max_length)
if idx != -1 and idx != 0:
chunk = text[:idx]

text = text[idx:]
found = True
break
if not found:
# 如果没有找到合适的分隔符,则直接在 max_length 处截断。

if text.strip:
chunks.append(text.strip)
return chunks
chunks = iterative_chunk(pdf_text, max_length=500)

开发者可以根据具体需求调整递归逻辑,实现最适合特定应用场景的分块模式。

滑动窗口分块策略通过在文本上移动固定大小的窗口来创建具有重叠内容的文本块。例如,使用200个标记的窗口大小,每次移动150个标记(重叠50个标记),这种重叠机制确保上下文信息能够跨越相邻文本块传递,有效减轻边界信息丢失的问题。

该策略的主要优势在于上下文连续性的保持,重叠区域确保了跨文本块的概念关系得以保留,这对于需要理解文档整体脉络的应用特别重要。同时,窗口大小和重叠程度的可调性使其能够适应不同的应用需求和文档特征。

滑动窗口分块的局限性主要表现为存储冗余和计算负载增加。重复内容会增加存储成本并使去重处理变得复杂,而重叠区域的存在会增加下游任务的处理步骤和计算资源消耗。

Python实现代码如下:

import PyPDF2
def sliding_window_chunk(text, window_size=100, overlap=20):
words = text.split
chunks =
step = window_size - overlap
for i in range(0, len(words), step):
chunk = " ".join(words[i:i+window_size])
chunks.append(chunk)
return chunks
chunks = sliding_window_chunk(pdf_text, window_size=100, overlap=20)

层次化分块策略利用文档的内在结构(如标题、章节、小节)来创建具有嵌套关系的文本块。该策略反映了文档的逻辑组织结构,例如章节作为父级文本块包含多个小节子块,而小节又包含段落级别的内容。这种结构类似于XML或HTML的树形组织框架。

层次化分块的主要优势在于结构保真性,能够保持文档的逻辑流程和多层次上下文关系。该策略支持多粒度访问模式,用户可以从广泛的主题概念深入到具体的细节内容,实现灵活的信息检索。

该策略的局限性主要体现在对文档结构的依赖性上,需要处理格式良好的结构化文档,对于非结构化文本的处理能力有限。此外,实现复杂性较高,需要构建稳健的算法来检测标题、列表或标记标签等结构元素。

层次化分块可以采用手动和自动两种实现方式。手动方式根据预定义的标记进行分割:

# 手动层次化分块
import PyPDF2
def hierarchical_chunk_manual(text, markers=["INTRODUCTION", "CONCLUSION"]):
"""
根据手动提供的标记将文本分割成块。
"""
lines = text.splitlines
chunks =
current_chunk =
for line in lines:
# 如果在行中找到任何手动标记并且已经有累积的内容
if any(marker in line for marker in markers) and current_chunk:
chunks.append("\n".join(current_chunk).strip)
current_chunk = [line]
else:
current_chunk.append(line)
if current_chunk:

return chunks
manual_chunks = hierarchical_chunk_manual(pdf_text, markers=["INTRODUCTION", "CONCLUSION"])

自动层次化分块通过算法逻辑检测文档结构,甚至可以结合大语言模型来实现:

# 自动层次化分块
def detect_markers(text):
"""
自动检测潜在的标题标记。
启发式方法:任何短行(<= 10 个词)且全部大写或以冒号结尾的行都被视作标记。
"""
lines = text.splitlines
markers =
for line in lines:
words = line.split
if words and len(words) <= 10:
if line.isupper or line.endswith(":"):
markers.append(line.strip)
return list(set(markers))
def hierarchical_chunk_auto(text):
"""
使用自动检测到的标记将文本分割成块。
返回块和检测到的标记。
"""
auto_markers = detect_markers(text)
# 按出现顺序对标记进行排序

detected =
for line in lines:
for marker in auto_markers:
if marker in line and marker not in detected:
detected.append(marker)
chunks =
current_chunk =
for line in lines:
# 如果找到任何检测到的标记并且有累积的文本,则开始一个新的块
if any(marker in line for marker in detected) and current_chunk:
chunks.append("\n".join(current_chunk).strip)
current_chunk = [line]
else:
current_chunk.append(line)
if current_chunk:

return chunks, detected
auto_chunks, auto_markers = hierarchical_chunk_auto(pdf_text)

基于主题的分块策略应用机器学习聚类算法(如潜在狄利克雷分配LDA、k-means聚类)按照主题相关性对文本进行分组。该方法通过词频分析或嵌入向量模式识别,生成代表连贯主题的文本块,例如"气候变化"或"市场趋势"等专题内容。

该策略的主要优势在于主题一致性,能够显著提高主题驱动应用程序中的检索准确性。同时具有良好的动态分组能力,无需预定义规则即可自适应文档内容,实现智能化的内容组织。

基于主题分块的局限性主要体现在主题重叠问题上,语义模糊的文本可能同时属于多个聚类,需要额外的消歧处理。此外,主题建模算法的计算开销较大,在大规模数据集上的扩展性存在挑战。

以下代码实现了基于主题的分块策略:

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.decomposition import LatentDirichletAllocation
def topic_based_chunk(text, n_topics=2):
# 将文本分割成句子(使用句点后跟空格作为简单的分割符)
sentences = text.split('. ')
vectorizer = CountVectorizer(stop_words='english')
X = vectorizer.fit_transform(sentences)
lda = LatentDirichletAllocation(n_components=n_topics, random_state=42)
lda.fit(X)
topic_distribution = lda.transform(X)
topics = {}
for i, sent in enumerate(sentences):
topic = topic_distribution[i].argmax
topics.setdefault(topic, ).append(sent)
return topics
topics = topic_based_chunk(pdf_text, n_topics=2)

特定模态分块策略专门处理包含多种内容类型的混合文档,将文本、图像、表格等不同模态的内容分离为相应的处理片段。该方法为每种模态采用专门的处理工具,例如将表格提取为CSV格式,图像保存为PNG格式,文本内容组织为段落结构,并使用OCR等技术处理图像中的文本信息。

该策略的主要优势在于能够保持不同模态内容的完整性,维护非文本元素的结构和语义信息。同时,针对每种内容类型的定制化处理能够实现最优的信息提取和处理效果。

特定模态分块的局限性主要表现为系统复杂性的显著增加,需要为每种模态建立专门的处理管道,并设计逻辑将不同模态的输出重新整合。此外,该方法需要多种专门工具的支持,如表格提取器、图像分类器等,增加了资源开销和维护成本。

以下代码演示了特定模态分块策略的基础实现:

import PyPDF2
import pdfplumber
def extract_pdf_text(pdf_path):
"""
使用 PyPDF2 从 PDF 的每一页提取文本,并将其作为一个字符串返回。
"""
reader = PyPDF2.PdfReader(pdf_path)
full_text = ""
for page in reader.pages:
page_text = page.extract_text
if page_text:
full_text += page_text + "\n\n" # 使用双换行符分隔页面
return full_text
def extract_tables(pdf_path):
"""
使用 pdfplumber 从 PDF 中提取表格。
返回一个表格列表(每个表格是一个列表的列表)。
"""
tables =
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
page_tables = page.extract_tables
for table in page_tables:
tables.append(table)
return tables
def extract_images(pdf_path):
"""
使用 PyPDF2 从 PDF 中提取图像。
注意:PyPDF2 的图像提取功能有限。
返回一个包含图像数据和元数据的字典列表。
"""

images =
for page_num, page in enumerate(reader.pages):
resources = page.get("/Resources")
if resources and "/XObject" in resources:
xObject = resources["/XObject"].get_object
for obj in xObject:
if xObject[obj].get("/Subtype") == "/Image":
data = xObject[obj].get_data
# 从过滤器确定图像类型
if "/Filter" in xObject[obj]:
if xObject[obj]["/Filter"] == "/DCTDecode":
ext = "jpg"
elif xObject[obj]["/Filter"] == "/FlateDecode":
ext = "png"
else:
ext = "bin"
else:
ext = "bin"
images.append({"page": page_num, "data": data, "ext": ext})
return images
def modality_chunk(pdf_path):
"""
从 PDF 文件中提取文本、表格和图像。
- 文本被分割成段落(使用双换行符)。
- 表格使用 pdfplumber 提取。
- 图像使用 PyPDF2 提取。
返回一个字典,键为:"text_chunks"、"tables" 和 "images"。
"""
text = extract_pdf_text(pdf_path)
text_chunks = [p.strip for p in text.split("\n\n") if p.strip]
tables = extract_tables(pdf_path)
images = extract_images(pdf_path)
return {"text_chunks": text_chunks, "tables": tables, "images": images}
result = modality_chunk(pdf_file)

智能代理分块策略利用大语言模型和智能代理技术动态识别最优的文本分割点或选择最适合的分块技术。该方法依赖智能代理执行类似人类专家在分析和分割文档时的决策过程,实现高度智能化的文本处理。

该策略的主要优势在于具备上下文感知能力,能够生成类似人类专家水平的精细化文本块。同时具有出色的适应性,能够根据文档类型、写作风格和内容复杂性进行动态调整,提供个性化的分块解决方案。

智能代理分块的局限性主要体现在成本和延迟方面,多次大语言模型API调用会显著增加费用和处理时间。此外,该方法对提示工程的质量高度敏感,输出质量很大程度上依赖于精心设计的提示模板。

以下代码基于Ranjith的智能代理分块方法,通过大语言模型生成的命题执行分块,并使用智能代理创建文本块组:

import uuid
import google.generativeai as genAI
import time
from typing import List, Dict
import os
class AgenticChunker:
def __init__(self):
self.chunks = {} # 块信息
self.agent = genAI.GenerativeModel("gemini-2.0-flash") # 更新的模型名称
self.chunk_id_length = 5 # 用于截断块 ID
self.configure = genAI.configure(api_key="YOUR-API-KEY")
def add_propositions(self, propositions: List[str]):
"""带速率限制地添加多个命题"""
for idx, proposition in enumerate(propositions):
print(f"Processing proposition {idx+1}/{len(propositions)}")
self.add_proposition(proposition)
time.sleep(4) # 在命题之间等待 4 秒以遵守 15 rpm 的限制
def add_proposition(self, proposition: str):
"""将单个命题添加到适当的块中"""
print(f"Evaluating: {proposition[:50]}...") # 显示截断的命题
if not self.chunks:
print("No existing chunks - creating first chunk")
self.create_new_chunk(proposition)
return
relevant_chunk_id = self.find_relevant_chunk(proposition)
if relevant_chunk_id:
print(f"Adding to existing chunk: {self.chunks[relevant_chunk_id]['title']}")
self.add_proposition_to_chunk(relevant_chunk_id, proposition)
else:
print("Creating new chunk for proposition")

def add_proposition_to_chunk(self, chunk_id: str, proposition: str):
"""将命题添加到现有块并更新元数据"""
self.chunks[chunk_id]["propositions"].append(proposition)
# 批量更新以减少 API 调用:当达到 3 个命题的倍数时更新
if len(self.chunks[chunk_id]["propositions"]) % 3 == 0:
self.chunks[chunk_id]["summary"] = self.update_chunk_summary(self.chunks[chunk_id])
self.chunks[chunk_id]["title"] = self.update_chunk_title(self.chunks[chunk_id])
def _generate_content(self, prompt: str) -> str:
"""Gemini API 调用的包装器,具有错误处理和速率限制功能"""
try:
response = self.agent.generate_content(prompt)
# 每次 API 调用后等待 5 秒,以避免超出速率限制
time.sleep(5)
return response.text.strip
except Exception as e:
print(f"API Error: {str(e)}")
return "" # 返回空字符串以防止管道故障
def update_chunk_summary(self, chunk: Dict) -> str:
"""生成更新的块摘要"""
prompt = f"""
您是一组代表关于相似主题的句子的块的管理者。
刚刚添加了一个新的命题。生成一个非常简短的 1 句话摘要,告知查看者该块的内容。
仅回复新的摘要,不要包含其他内容。
块的命题:
{chr(10).join(chunk['propositions'][-3:])}
当前摘要: {chunk['summary'] if 'summary' in chunk else ''}
"""
return self._generate_content(prompt)
def update_chunk_title(self, chunk: Dict) -> str:
"""生成更新的块标题"""
prompt = f"""
您是一组代表相关命题的块的管理者。
刚刚添加了一个新的命题。生成一个非常简短的更新块标题(2-4 个词),总结该块的主题。
仅回复新的标题,不要包含其他内容。
块的命题:

块摘要: {chunk['summary'] if 'summary' in chunk else ''}
当前块标题: {chunk['title'] if 'title' in chunk else ''}
"""

def get_new_chunk_summary(self, proposition: str) -> str:
"""为新块生成初始摘要"""
prompt = f"""
您是一组代表讨论相似主题的句子组的块的管理者。
根据以下命题,生成一个非常简短的 1 到 2 句话的摘要,描述新块的主题。
仅回复新的块摘要,不要包含其他内容。
命题:
{proposition}
"""

def get_new_chunk_title(self, summary: str) -> str:
"""根据其摘要为新块生成初始标题"""
prompt = f"""
您是一组块的管理者。根据以下摘要,生成一个简洁的标题(2-4 个词),抓住块的精髓。
仅回复新的块标题,不要包含其他内容。
块摘要:
{summary}
"""

def create_new_chunk(self, proposition: str):
"""使用初始命题创建新块"""
new_chunk_id = str(uuid.uuid4)[:self.chunk_id_length] # 唯一的块 ID
new_chunk_summary = self.get_new_chunk_summary(proposition)
new_chunk_title = self.get_new_chunk_title(new_chunk_summary)
self.chunks[new_chunk_id] = {
'chunk_id': new_chunk_id,
'propositions': [proposition],
'title': new_chunk_title,
'summary': new_chunk_summary,
'chunk_index': len(self.chunks)
}
print(f"Created new chunk {new_chunk_id}: {new_chunk_title}")
def find_relevant_chunk(self, proposition: str) -> str:
"""为命题找到匹配的块"""
prompt = f"""
确定以下命题是否应属于现有块之一。
如果应该,则返回块 ID。如果不应该,则返回 'NO_MATCH'。
现有块 (ID: 标题 - 摘要):
{self._format_chunk_outline}
命题:
{proposition}
仅回复匹配的块 ID 或 'NO_MATCH'。
"""
response = self._generate_content(prompt)
resp = response.strip
if resp == "NO_MATCH" or resp not in self.chunks:
return None
return resp
def _format_chunk_outline(self) -> str:
"""格式化块信息以供 LLM 输入"""
return "\n".join(
f"{c['chunk_id']}: {c['title']} - {c['summary']}"
for c in self.chunks.values
)
def pretty_print_chunks(self):
"""显示所有块及其元数据"""
print("\n----- Chunks Created -----\n")
for _, chunk in self.chunks.items:
print(f"Chunk ID : {chunk['chunk_id']}")
print(f"Title : {chunk['title'].strip}")
print(f"Summary : {chunk['summary'].strip}")
print("Propositions:")
for prop in chunk['propositions']:
print(f" - {prop}")
print("\n")

def print_chunks(self):
self.pretty_print_chunks
chunker = AgenticChunker
chunker.add_propositions(pdf_text)
chunker.print_chunks

本文详细分析了九种主要的RAG系统文本分块策略,每种策略都有其特定的适用场景和技术特点。在实际应用中,选择合适的分块策略需要综合考虑文档类型、应用需求、计算资源和性能要求等因素。

固定大小分块适用于需要一致性输入的场景,基于句子和语义的分块策略更适合注重语义完整性的应用。递归和滑动窗口分块提供了灵活的大小控制,而层次化和主题分块策略则适合处理结构化内容。特定模态分块专门处理多媒体文档,智能代理分块则代表了未来发展的方向。

开发者可以根据具体需求选择单一策略或组合多种策略,以实现最优的RAG系统性能。通过深入理解这些分块策略的原理和实现方法,我们能够构建更加高效和准确的RAG系统,为用户提供更好的信息检索和生成体验。

作者:Cornellius Yudha Wijaya

来源:deephub

相关推荐