文本切块是将长篇文档分解成更小、更易于管理的片段的过程。这在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: defclassimport# %% (Jupyter单元格)

    • JavaScript: functionclassimport}

  • 优点:保持代码功能的完整性,便于检索整个函数或模块。

  • 缺点:需要语言特定的解析器。

  • 适用场景:处理源代码库。

代码案例:

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)

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐