RAG-2-文本切块
文本切块是将长篇文档分解成更小、更易于管理的片段的过程。这在RAG系统中至关重要,因为它直接影响到检索到的信息的质量和相关性,从而最终影响大模型生成答案的准确性。
文本切块是将长篇文档分解成更小、更易于管理的片段的过程。这在RAG系统中至关重要,因为它直接影响到检索到的信息的质量和相关性,从而最终影响大模型生成答案的准确性。
为什么文本切块如此重要?
在RAG中,用户查询会与向量数据库中的文本块进行相似性匹配。切块策略的目标是创造出既能包含完整语义信息,又大小合适的“信息单元”。
-
切块过大:
-
缺点:可能包含多个不相关的主题,导致检索时引入噪声。当检索到这样一个大块时,LLM可能难以从大量文本中精准提取出与查询最相关的部分,生成答案的准确性和针对性会下降。
-
优点:能更好地保留上下文信息。
-
-
切块过小:
-
缺点:可能失去关键信息的上下文,导致信息碎片化。例如,一个关键事实可能被分割在两个块中,使得每个块单独看都含义不完整。
-
优点:检索结果可能非常精准,与查询高度匹配。
-
因此,选择一个平衡的切块策略是构建高效RAG系统的关键。
常见的文本切块方法
没有一种“万能”的最佳方法,最佳策略取决于你的文档类型和预期问答任务。
固定大小切块(Fixed-Size Chunking)
最简单、最常用的方法。它使用一个固定的字符数(如512个字符)来分割文本,并可以设置一个重叠量(Overlap)来避免在句子中间切断上下文。
-
如何工作:设置
chunk_size
(块大小)和chunk_overlap
(重叠大小)。-
例如:
chunk_size=500, chunk_overlap=50
-
-
优点:简单、高效、可预测。
-
缺点:可能会在任意位置(如句子或段落中间)切断文本,破坏语义完整性。
-
适用场景:通用性文档,结构相对统一。
基于分隔符的切块(Recursive Chunking / Separator-Based)
一种更“智能”的方法。它尝试利用文本中天然的分隔符(如\n\n
, \n
, .
, !
, ?
, ,
, 空格
)来递归地分割文本,直到达到所需的大小。
-
如何工作:定义一个分隔符优先级列表(例如,先按
\n\n
分,如果块还太大,再按\n
分,再按.
分...),直到每个块都小于预设的大小。 -
优点:能更好地保留句子的完整性,比固定大小切块更符合语言结构。
-
缺点:配置稍复杂。
-
适用场景:大多数文本类型,是固定大小切块的优秀升级替代方案。
语义切块(Semantic Chunking)
这是更前沿、更复杂的方法。它利用机器学习模型来理解文本的语义边界,在“想法”或“主题”发生自然转变的地方进行切分。
-
如何工作:使用嵌入模型或NLP库计算句子或段落之间的相似度。当相似度低于某个阈值时,表明主题发生了变化,就在此处切块。
-
优点:能产生语义上高度连贯和自包含的块,质量最高。
-
缺点:计算成本高,实现最复杂。
-
适用场景:对检索质量要求极高的场景,如学术论文、法律文档等。
基于代理的切块(Agentic Chunking)
一种实验性方法,使用LLM本身来决定如何最佳地切分文档。你可以向LLM提供文档并提示它“将此文档分成几个信息完整的块”。
-
优点:理论上非常灵活和智能。
-
缺点:成本极高、速度慢,不适合大规模应用,目前更多用于研究。
专门针对代码的切块(Code Splitting)
代码具有非常独特的结构(函数、类、导入语句等),需要特殊的切块策略。
-
如何工作:使用编程语言特定的分隔符,例如:
-
Python:
def
,class
,import
,# %%
(Jupyter单元格) -
JavaScript:
function
,class
,import
,}
-
-
优点:保持代码功能的完整性,便于检索整个函数或模块。
-
缺点:需要语言特定的解析器。
-
适用场景:处理源代码库。
代码案例:
1.固定大小切块
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
loader = TextLoader("概述.txt")
documents = loader.load()
# 设置分块器,指定块的大小为50个字符,无重叠
text_splitter = CharacterTextSplitter(
chunk_size=100, # 每个文本块的大小为50个字符
chunk_overlap=10, # 文本块之间没有重叠部分
)
chunks = text_splitter.split_documents(documents)
print("\n=== 文档分块结果 ===")
for i, chunk in enumerate(chunks, 1):
print(f"\n--- 第 {i} 个文档块 ---")
print(f"内容: {chunk.page_content}")
print(f"元数据: {chunk.metadata}")
print("-" * 50)
2.固定字符切块
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
loader = TextLoader("概述.txt")
documents = loader.load()
# 定义分割符列表,按优先级依次使用
separators = ["\n\n", "。", "?", "!", ";", " "]
# 创建递归分块器,并传入分割符列表
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=10,
separators=separators
)
chunks = text_splitter.split_documents(documents)
print("\n=== 文档分块结果 ===")
for i, chunk in enumerate(chunks, 1):
print(f"\n--- 第 {i} 个文档块 ---")
print(f"内容: {chunk.page_content}")
print(f"元数据: {chunk.metadata}")
print("-" * 50)
3.语义切块
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import SimpleDirectoryReader
from llama_index.core.node_parser import SemanticSplitterNodeParser
documents = SimpleDirectoryReader(input_files=["wiki.txt"]).load_data()
openAIEmbedding = OpenAIEmbedding(
api_key="",
api_base="https://api.34ku.com/v1/",
model="text-embedding-3-small"
)
# 创建语义分块器
splitter = SemanticSplitterNodeParser(
buffer_size=3, # 缓冲区大小
breakpoint_percentile_threshold=90, # 断点百分位阈值
embed_model=openAIEmbedding # 使用的嵌入模型
)
# 使用语义分块器对文档进行分块
semantic_nodes = splitter.get_nodes_from_documents(documents)
print("\n=== 语义分块结果 ===")
print(f"语义分块器生成的块数:{len(semantic_nodes)}")
for i, node in enumerate(semantic_nodes, 1):
print(f"\n--- 第 {i} 个语义块 ---")
print(f"内容:\n{node.text}")
print("-" * 50)
4.代理分块
from langchain_core.prompts import ChatPromptTemplate
from model.deepseek import deepseek_llm
prompt = """
你是一名专业的文本分析专家,专门负责将复杂文档进行智能分块处理。
你的任务是将用户提供的文本内容分割成逻辑完整、语义清晰的段落块。请遵循以下原则:
1. 保持语义完整性:每个文本块应该包含完整的信息单元,避免在语义中间切断
2. 合理控制长度:每个文本块建议控制在合适的长度范围内
3. 保持逻辑连贯:同一文本块内的内容应具有高度的相关性和连贯性
4. 识别自然边界:在段落、章节或话题转换处进行分割
5. 输出格式:请以列表形式返回分割后的文本块,每个块作为列表中的一个元素
请分析以下文本并进行专业分块处理:
{input}
"""
prompt_template = ChatPromptTemplate.from_template(prompt)
chain = prompt_template | deepseek_llm
print(chain.invoke({"input": """
80年前,中国人民经过艰苦卓绝的抗日战争,彻底打败日本军国主义侵略者,宣告世界反法西斯战争的完全胜利。中国人民抗日战争的胜利是世界反法西斯战争胜利的重要组成部分,这是中国人民的胜利,也是世界人民的胜利。历史川流不息,精神代代相传。传承和弘扬伟大抗战精神,强化中华儿女团结奋斗的精神纽带,激发出战胜一切困难的决心和勇气,必将为以中国式现代化全面推进强国建设、民族复兴伟业注入强劲信心与不竭动力。
"""}).content)
5.代码分块
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_text_splitters import Language
from langchain_text_splitters import (
Language,
RecursiveCharacterTextSplitter,
)
GAME_CODE = """
class CombatSystem:
def __init__(self):
self.health = 100
self.stamina = 100
self.state = "IDLE"
self.attack_patterns = {
"NORMAL": 10,
"SPECIAL": 30,
"ULTIMATE": 50
}
def update(self, delta_time):
self._update_stats(delta_time)
self._handle_combat()
def _update_stats(self, delta_time):
self.stamina = min(100, self.stamina + 5 * delta_time)
def _handle_combat(self):
if self.state == "ATTACKING":
self._execute_attack()
def _execute_attack(self):
if self.stamina >= self.attack_patterns["SPECIAL"]:
damage = 50
self.stamina -= self.attack_patterns["SPECIAL"]
return damage
return self.attack_patterns["NORMAL"]
class InventorySystem:
def __init__(self):
self.items = {}
self.capacity = 20
self.gold = 0
def add_item(self, item_id, quantity):
if len(self.items) < self.capacity:
if item_id in self.items:
self.items[item_id] += quantity
else:
self.items[item_id] = quantity
def remove_item(self, item_id, quantity):
if item_id in self.items:
self.items[item_id] -= quantity
if self.items[item_id] <= 0:
del self.items[item_id]
def get_item_count(self, item_id):
return self.items.get(item_id, 0)
class QuestSystem:
def __init__(self):
self.active_quests = {}
self.completed_quests = set()
self.quest_log = []
def add_quest(self, quest_id, quest_data):
if quest_id not in self.active_quests:
self.active_quests[quest_id] = quest_data
self.quest_log.append(f"Started quest: {quest_data['name']}")
def complete_quest(self, quest_id):
if quest_id in self.active_quests:
self.completed_quests.add(quest_id)
del self.active_quests[quest_id]
def get_active_quests(self):
return list(self.active_quests.keys())
"""
python_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON, # 指定编程语言为Python
chunk_size=1000,
chunk_overlap=0
)
python_docs = python_splitter.create_documents([GAME_CODE])
print("\n=== 代码分块结果 ===")
for i, chunk in enumerate(python_docs, 1):
print(f"\n--- 第 {i} 个代码块 ---")
print(f"内容:\n{chunk.page_content}")
print(f"元数据: {chunk.metadata}")
print("-" * 50)
更多推荐
所有评论(0)