CharacterTextSplitter - 按固定字符数分块

实现方式：通过CharacterTextSplitter指定chunk_size（如1000字符）和chunk_overlap（重叠字符数）
特点：
- 简单粗暴，直接按字符数切割
- 默认以段落（\n\n）为分隔符，也可指定逗号、句号等
- 若段落超过设定值（如2000字符），会保持原段落不分块
局限性：不够智能，可能破坏语义完整性

示例代码

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter

loader = TextLoader("../data/山西文旅/云冈石窟.txt")
documents = loader.load()

# 设置分块器，指定块的大小为50个字符，无重叠
text_splitter = CharacterTextSplitter(
    chunk_size=100,  # 每个文本块的大小为50个字符
    chunk_overlap=0,  # 文本块之间没有重叠部分
)
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)

执行结果，输出内容

=== 文档分块结果 ===

--- 第 1 个文档块 ---
内容: 云冈石窟
云冈石窟位于中国北部山西省大同市西郊17公里处的武周山南麓，石窟依山开凿，东西绵延1公里。存有主要洞窟45个，大小窟龛252个，石雕造像51000余躯，为中国规模最大的古代石窟群之一，与敦煌莫高窟、洛阳龙门石窟和天水麦积山石窟并称为中国四大石窟艺术宝库。 1961年被国务院公布为全国首批重点文物保护单位，2001年12月14日被联合国教科文组织列入世界遗产名录，2007年5月8日被国家旅游局评为首批国家5A级旅游景区。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 2 个文档块 ---
内容: 云冈五华洞
位于云冈石窟中部的第 9——13窟。这五窟因清代施泥彩绘云冈石窟景观而得名。五华洞雕饰绮丽，丰富多彩，是研究北魏历史、艺术、音乐、舞蹈、书法和建筑的珍贵资料，为云冈石窟群的重要组成部分。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 3 个文档块 ---
内容: 塔洞
云冈东部窟群，指云冈石窟东端1——4，均为塔洞。第1、2窟为同期开的一组，凿于孝文帝迁洛前，窟内中央雕造方形塔柱，四面开龛造像。第一窟主像是弥勒，塔南面下层雕释迦多宝像，上层雕释迦像。浮雕五层小塔，是研究北魏建筑的形象资料。第二窟是释迦像，塔南面下层雕释迦多宝像，上层雕三世佛。两窟南壁窟门两侧都雕有维摩、文殊。第三窟为云冈石窟中规模最大的洞窟，前立壁高约25米，传为昙曜译经楼。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 4 个文档块 ---
内容: 武州山
武周山，亦名武州山，在大同城西山中。宋《太平寰宇记》引《冀州图》云：“武周山在郡西北，东西数百里，南北五十里。山之南面，千仞壁立。”云冈石窟即因武周山南缘斩山开凿。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
...
--- 第 11 个文档块 ---
内容: 1、咨询电话：0352-3206805，景区电瓶车10元/人。 2、优惠群体： （1）60周岁（含60周岁）以上老人，持本人身份证或老年证免票； （2）6周岁（含6周岁）以下或 1.2米以下儿童免票； （3）6周岁（不含6周岁）——18周岁（含18周岁）的未成年人，全日制大学本科及以下学历的学生，凭本人身份证、学生证享受半价票优惠。 （4）全日制大学本科及以学历学生，凭本人身份证、学生证享受半价票优惠。 （5）残疾人持本人身份证和第二代《中华人民共和国残疾证》免票；伤残军人持伤残军人证及本人身份证免票。 （6）现役军人凭军人保障卡免票进入。 （7）导游人员免票，需持本人导游证（不含导游资格证、经理资格证）、派团单、行程计划单。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}

RecursiveCharacterTextSplitter – 递归分块

实现原理：
- 定义优先级分隔符列表（如[“\n\n”, “.”, “,”, “ “]）
- 先尝试第一个分隔符，若块过大则尝试下一个
优势：
- 切割更精细（可细化到句子、短语级别）
- 通过多级分隔符保证块大小接近设定值
注意事项：
- 分隔符不宜过细（如空格），否则会破坏语义
- 英文中”.”需注意区分小数点和句号
对比固定分块：
- 相同设定下，递归分块数更多（如100字符设定产生22块 vs 固定分块11块）
- 但过度分割可能导致关键信息断裂（如价格”1.5元”被拆分为”1”和”.5元”）

示例代码

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("../data/山西文旅/云冈石窟.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)

执行结果，输出内容


=== 文档分块结果 ===

--- 第 1 个文档块 ---
内容: 云冈石窟
云冈石窟位于中国北部山西省大同市西郊17公里处的武周山南麓，石窟依山开凿，东西绵延1公里。存有主要洞窟45个，大小窟龛252个，石雕造像51000余躯，为中国规模最大的古代石窟群之一
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 2 个文档块 ---
内容: ，与敦煌莫高窟、洛阳龙门石窟和天水麦积山石窟并称为中国四大石窟艺术宝库。 1961年被国务院公布为全国首批重点文物保护单位，2001年12月14日被联合国教科文组织列入世界遗产名录
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 3 个文档块 ---
内容: ，2007年5月8日被国家旅游局评为首批国家5A级旅游景区。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

--- 第 4 个文档块 ---
内容: 云冈五华洞
位于云冈石窟中部的第 9——13窟。这五窟因清代施泥彩绘云冈石窟景观而得名。五华洞雕饰绮丽，丰富多彩，是研究北魏历史、艺术、音乐、舞蹈、书法和建筑的珍贵资料
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------
...
--- 第 22 个文档块 ---
内容: ，需持本人导游证（不含导游资格证、经理资格证）、派团单、行程计划单。
元数据: {'source': '../data/山西文旅/云冈石窟.txt'}
--------------------------------------------------

分块的重要性

示例代码

import os
from dotenv import load_dotenv

load_dotenv()   

from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import VectorStoreIndex
from llama_index.core import Settings
from llama_index.readers.file import PDFReader
from llama_index.core.node_parser import SentenceSplitter


embed_model = OpenAIEmbedding(model="text-embedding-3-small")
llm = OpenAI(
    model=os.getenv("DEEPSEEK_MODEL_NAME"),  # DeepSeek API 支持的模型名称
    api_key=os.getenv("DEEPSEEK_API_KEY"),  # 从环境变量加载API key
    api_base=os.getenv("DEEPSEEK_BASE_URL")
)

Settings.embed_model = embed_model
Settings.llm = llm
Settings.node_parser = SentenceSplitter(chunk_size=250, chunk_overlap=20) # 50, 100, 250将得到不同的结果，为什么？

# Load PDF using standard PDFReader
loader = PDFReader()
documents = loader.load_data(
    file="../data/复杂PDF/uber_10q_march_2022_page26.pdf"
)

# Create index directly from documents
index = VectorStoreIndex.from_documents(documents)

# Create query engine
query_engine = index.as_query_engine(
    similarity_top_k=3,
    verbose=True
)

query = "how much is the Loss from operations for 2022?"

response = query_engine.query(query)
print("\n************LlamaIndex Query Response************")
print(response)

# Display retrieved chunks
print("\n************Retrieved Text Chunks************")
for i, source_node in enumerate(response.source_nodes):
    print(f"\nChunk {i+1}:")
    print("Text content:")
    print(source_node.text)
    print("-" * 50)

调整块的大小

关键发现：
- 100字符分块：返回错误答案$1.938B（无意义数值）
- 150字符分块：返回$1.524B（2021年正确值，但非问题要求的2022年）
- 250字符分块：正确返回$482M（2022年实际值）
原理分析：
- 过小分块割裂表格数据关联性
- 适当分块保留完整表格结构，使模型能正确对应年份与数值
- 过大分块（如2500字符）可能引入干扰信息

基于特定格式（如代码）分块

普通分块问题：
- 函数定义被强行拆分
- 类成员变量与方法分离
语言专用分块：
- 按class和def自然分割
- 保留完整类结构（如CombatSystem类完整包含所有方法）
- 支持多种语言（Python/Java/SQL等）的语法感知分割

示例代码

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_text_splitters import Language

separators = RecursiveCharacterTextSplitter.get_separators_for_language(Language.JS)
print(separators)

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)

执行结果，输出内容：

['\nfunction ', '\nconst ', '\nlet ', '\nvar ', '\nclass ', '\nif ', '\nfor ', '\nwhile ', '\nswitch ', '\ncase ', '\ndefault ', '\n\n', '\n', ' ', '']

=== 代码分块结果 ===

--- 第 1 个代码块 ---
内容:
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"]
元数据: {}
--------------------------------------------------

--- 第 2 个代码块 ---
内容:
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)
元数据: {}
--------------------------------------------------

--- 第 3 个代码块 ---
内容:
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())
元数据: {}
--------------------------------------------------
……

语义分块

示例代码

from llama_index.core import SimpleDirectoryReader
from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser,
)
# from llama_index.embeddings.openai import OpenAIEmbedding 
from llama_index.embeddings.huggingface import HuggingFaceEmbedding 
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-zh") # 加载本地嵌入模型

documents = SimpleDirectoryReader(input_files=["../data/黑悟空/黑悟空wiki.txt"]).load_data()

# 创建语义分块器
splitter = SemanticSplitterNodeParser(
    buffer_size=3,  # 缓冲区大小
    breakpoint_percentile_threshold=90, # 断点百分位阈值
    embed_model=embed_model     # 使用的嵌入模型
)
# 创建基础句子分块器（作为对照）
base_splitter = SentenceSplitter(
    # chunk_size=512
)

# 使用语义分块器对文档进行分块
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)

# 使用基础句子分块器对文档进行分块
base_nodes = base_splitter.get_nodes_from_documents(documents)
print("\n=== 基础句子分块结果 ===")
print(f"基础句子分块器生成的块数：{len(base_nodes)}")
for i, node in enumerate(base_nodes, 1):
    print(f"\n--- 第 {i} 个句子块 ---")
    print(f"内容:\n{node.text}")
    print("-" * 50)

buffer_size参数:
- 默认值为1，控制评估语义相似度时组合的句子数量
- 设置为1时每个句子单独考虑
- 设置为3时每3个句子作为一组评估相似度
breakpoint_percentile_threshold参数:
- 默认值为95，控制创建分割点的阈值
- 表示余弦不相似度的百分位数值
- 数值越小生成的节点越多（更容易达到分割值）
- 数值越大生成的节点越少（需要更大不相似度才会分割）

参数组合影响:

buffer_size：默认值为1
- 这个参数控制评估语义相似度时，将多少个句子组合在一起当设置为1时，每个句子会被单独考虑
  当设置大于1时，会将多个句子组合在一起进行评估例如，如果设置为3，就会将每3个句子作为一个组来评估语义相似度
breakpoint_percentile_threshold：默认值为95
- 这个参数控制何时在句子组之间创建分割点,它表示余弦不相似度的百分位数阈值,当句子组之间的不相似度超过这个阈值时，就会创建一个新的节点
  - 数值越小，生成的节点就越多（因为更容易达到分割阈值）
  - 数值越大，生成的节点就越少（因为需要更大的不相似度才会分割）
这两个参数共同影响文本的分割效果：
- buffer_size 决定了评估语义相似度的粒度
- breakpoint_percentile_threshold 决定了分割的严格程度

例如：
- 如果 buffer_size=2 且 breakpoint_percentile_threshold=90：每2个句子会被组合在一起,当组合之间的不相似度超过90%时就会分割,这会产生相对较多的节点
- 如果 buffer_size=3 且 breakpoint_percentile_threshold=98：每3个句子会被组合在一起,需要更大的不相似度才会分割,这会产生相对较少的节点

基础句子分块与语义分块特点

基础句子分块特点:
- 按固定字符数机械划分
- 不考虑内容语义连贯性
- 实现简单但可能切断语义关联
语义分块特点:
- 基于内容语义自动划分
- 识别主题边界形成自然段落
- 适合内容结构复杂的文档

使用Unstructured基于文档结构分块

基于对文档语义结构的识别，而非简单地根据空行或换行符来拆分文本。

Basic策略将文本元素顺序合并到同一个分块内，直到达到最大字符数（max_characters）或软限制（new_after_n_chars）。如果单个元素（如一段特别长的正文或一张很大的表格）本身超过了最大字符数，则会对该元素进行二次拆分。表格元素会被视为独立的分块；如果表格过大，也会被拆分为多个TableChunk。可以通过overlap与overlap_all等参数设置分块重叠。
By Title策略在保留Basic策略的基础行为的同时，会在检测到新的标题（Title元素）后立刻关闭当前分块，并开启一个新的分块。可以通过multipage_sections和combine_text_under_n_chars等参数进一步控制如何合并或拆分跨页片段、短小片段等。

通过API调用Unstructured时，还有以下两种额外的智能分块策略。

lBy Page：确保每页的内容独立分块。
lBy Similarity：利用嵌入模型将主题相似的元素组合成块。

参考学习地址：https://docs.unstructured.io/api-reference/partition/chunking

Jean's Blog

RAG组件--文本分块之方法和实现

CharacterTextSplitter - 按固定字符数分块

RecursiveCharacterTextSplitter – 递归分块

分块的重要性

基于特定格式（如代码）分块

语义分块

基础句子分块与语义分块特点

使用Unstructured基于文档结构分块