21 种 RAG 分块技术最全总结！

import re  

defsentence_based_chunking(text, sentences_per_chunk=3):
"""基于句子的分块  

    Args:  
        text: 输入文本  
        sentences_per_chunk: 每个分块包含的句子数  
    """
# 使用正则表达式分割句子  
    sentences = re.split(r'[.!?。！？]+', text)  
    sentences = [s.strip() for s in sentences if s.strip()]  

    chunks = []  
for i in range(0, len(sentences), sentences_per_chunk):  
        chunk_sentences = sentences[i:i + sentences_per_chunk]  
        chunk = '。'.join(chunk_sentences)  
if chunk:  # 确保分块不为空  
            chunks.append(chunk + '。')  

return chunks  

# 使用示例  
text = """机器学习是人工智能的一个重要分支。它通过算法让计算机从数据中学习模式。  
深度学习是机器学习的子集。它使用神经网络来模拟人脑的工作方式。  
目前深度学习在图像识别和自然语言处理领域取得了巨大成功。"""

chunks = sentence_based_chunking(text, sentences_per_chunk=2)  
for i, chunk in enumerate(chunks):  
    print(f"分块 {i+1}: {chunk}")

import PyPDF2  
from io import BytesIO  

defpage_based_chunking_pdf(pdf_path):
"""基于页面的PDF分块  

    Args:  
        pdf_path: PDF文件路径  
    """
    chunks = []  

try:  
with open(pdf_path, 'rb') as file:  
            pdf_reader = PyPDF2.PdfReader(file)  

for page_num, page in enumerate(pdf_reader.pages):  
                text = page.extract_text()  
if text.strip():  # 确保页面有内容  
                    chunk = {  
'page_number': page_num + 1,  
'content': text.strip()  
                    }  
                    chunks.append(chunk)  

except Exception as e:  
        print(f"PDF处理错误: {e}")  

return chunks  

# 简化版本：模拟页面分块  
defsimulate_page_chunking(text, chars_per_page=500):
"""模拟基于页面的分块  

    Args:  
        text: 输入文本  
        chars_per_page: 每页字符数  
    """
    chunks = []  
    page_num = 1

for i in range(0, len(text), chars_per_page):  
        page_content = text[i:i + chars_per_page]  
        chunk = {  
'page_number': page_num,  
'content': page_content  
        }  
        chunks.append(chunk)  
        page_num += 1

return chunks  

# 使用示例  
text = "这是一个长文档的内容。" * 100
chunks = simulate_page_chunking(text, chars_per_page=200)  
print(f"总页数: {len(chunks)}")  
print(f"第一页内容: {chunks[0]['content'][:50]}...")

import json  
import re  
from bs4 import BeautifulSoup  

defjson_structured_chunking(json_data):
"""JSON数据结构化分块"""
    chunks = []  

if isinstance(json_data, list):  
for i, item in enumerate(json_data):  
            chunk = {  
'type': 'json_item',  
'index': i,  
'content': json.dumps(item, ensure_ascii=False, indent=2)  
            }  
            chunks.append(chunk)  
elif isinstance(json_data, dict):  
for key, value in json_data.items():  
            chunk = {  
'type': 'json_field',  
'key': key,  
'content': json.dumps({key: value}, ensure_ascii=False, indent=2)  
            }  
            chunks.append(chunk)  

return chunks  

defhtml_structured_chunking(html_content):
"""HTML内容结构化分块"""
    soup = BeautifulSoup(html_content, 'html.parser')  
    chunks = []  

# 按段落分块  
for i, p in enumerate(soup.find_all('p')):  
if p.get_text().strip():  
            chunk = {  
'type': 'paragraph',  
'index': i,  
'content': p.get_text().strip()  
            }  
            chunks.append(chunk)  

# 按标题分块  
for level in range(1, 7):  # h1-h6  
for i, h in enumerate(soup.find_all(f'h{level}')):  
            chunk = {  
'type': f'heading_{level}',  
'index': i,  
'content': h.get_text().strip()  
            }  
            chunks.append(chunk)  

return chunks  

# 使用示例  
json_data = [  
    {"name": "张三", "age": 25, "department": "技术部"},  
    {"name": "李四", "age": 30, "department": "产品部"}  
]  

chunks = json_structured_chunking(json_data)  
for chunk in chunks:  
    print(f"类型: {chunk['type']}, 内容: {chunk['content'][:50]}...")

from transformers import AutoTokenizer  
import tiktoken  

deftoken_based_chunking_transformers(text, model_name="bert-base-chinese", max_tokens=512):
"""使用Transformers分词器进行基于Token的分块"""
    tokenizer = AutoTokenizer.from_pretrained(model_name)  

# 对整个文本进行分词  
    tokens = tokenizer.tokenize(text)  
    chunks = []  

for i in range(0, len(tokens), max_tokens):  
        chunk_tokens = tokens[i:i + max_tokens]  
# 将tokens转换回文本  
        chunk_text = tokenizer.convert_tokens_to_string(chunk_tokens)  
        chunks.append(chunk_text)  

return chunks  

deftoken_based_chunking_tiktoken(text, model="gpt-3.5-turbo", max_tokens=1000):
"""使用tiktoken进行基于Token的分块（适用于OpenAI模型）"""
try:  
        encoding = tiktoken.encoding_for_model(model)  
except KeyError:  
        encoding = tiktoken.get_encoding("cl100k_base")  

# 编码文本  
    tokens = encoding.encode(text)  
    chunks = []  

for i in range(0, len(tokens), max_tokens):  
        chunk_tokens = tokens[i:i + max_tokens]  
# 解码回文本  
        chunk_text = encoding.decode(chunk_tokens)  
        chunks.append(chunk_text)  

return chunks  

# 简化版本：字符近似token分块  
defsimple_token_chunking(text, max_tokens=1000, chars_per_token=4):
"""简化的基于Token的分块（字符数近似）"""
    max_chars = max_tokens * chars_per_token  
    chunks = []  

for i in range(0, len(text), max_chars):  
        chunk = text[i:i + max_chars]  
        chunks.append(chunk)  

return chunks  

# 使用示例  
text = "人工智能技术的发展正在改变我们的生活方式。" * 50
chunks = simple_token_chunking(text, max_tokens=100, chars_per_token=2)  
print(f"分块数量: {len(chunks)}")  
print(f"第一个分块长度: {len(chunks[0])} 字符")

from sentence_transformers import SentenceTransformer  
import numpy as np  
from sklearn.metrics.pairwise import cosine_similarity  

defsemantic_chunking(text, similarity_threshold=0.7, min_chunk_size=50):
"""基于语义相似性的分块  

    Args:  
        text: 输入文本  
        similarity_threshold: 相似性阈值  
        min_chunk_size: 最小分块大小  
    """
# 初始化句子编码器  
    model = SentenceTransformer('all-MiniLM-L6-v2')  

# 按句子分割  
    sentences = text.split('。')  
    sentences = [s.strip() + '。'for s in sentences if s.strip()]  

if len(sentences) < 2:  
return [text]  

# 计算句子嵌入  
    embeddings = model.encode(sentences)  

    chunks = []  
    current_chunk = [sentences[0]]  

for i in range(1, len(sentences)):  
# 计算当前句子与当前分块的相似性  
        current_embedding = embeddings[i:i+1]  
        chunk_embeddings = embeddings[len(chunks) * len(current_chunk):len(chunks) * len(current_chunk) + len(current_chunk)]  

# 计算平均相似性  
        similarities = cosine_similarity(current_embedding, chunk_embeddings)  
        avg_similarity = np.mean(similarities)  

if avg_similarity >= similarity_threshold:  
            current_chunk.append(sentences[i])  
else:  
# 如果分块足够大，保存当前分块  
            chunk_text = ''.join(current_chunk)  
if len(chunk_text) >= min_chunk_size:  
                chunks.append(chunk_text)  
            current_chunk = [sentences[i]]  

# 添加最后一个分块  
if current_chunk:  
        chunk_text = ''.join(current_chunk)  
if len(chunk_text) >= min_chunk_size:  
            chunks.append(chunk_text)  
elif chunks:  # 如果最后一个分块太小，合并到前一个分块  
            chunks[-1] += chunk_text  
else:  
            chunks.append(chunk_text)  

return chunks  

# 简化版本：基于关键词相似性的语义分块  
defsimple_semantic_chunking(text, keywords_per_chunk=3):
"""简化的语义分块（基于关键词）"""
    sentences = text.split('。')  
    sentences = [s.strip() for s in sentences if s.strip()]  

    chunks = []  
    current_chunk = []  
    current_keywords = set()  

for sentence in sentences:  
# 简单提取关键词（这里用长度>2的词）  
        words = [word for word in sentence if len(word) > 2]  
        sentence_keywords = set(words[:keywords_per_chunk])  

# 计算关键词重叠  
        overlap = len(current_keywords.intersection(sentence_keywords))  

if overlap > 0or len(current_chunk) == 0:  
            current_chunk.append(sentence + '。')  
            current_keywords.update(sentence_keywords)  
else:  
            chunks.append(''.join(current_chunk))  
            current_chunk = [sentence + '。']  
            current_keywords = sentence_keywords  

if current_chunk:  
        chunks.append(''.join(current_chunk))  

return chunks  

# 使用示例  
text = """人工智能技术发展迅速。机器学习是AI的核心技术。深度学习推动了AI革命。  
自然语言处理技术不断进步。聊天机器人越来越智能。语言模型能力显著提升。  
计算机视觉应用广泛。图像识别准确率不断提高。自动驾驶技术日趋成熟。"""

chunks = simple_semantic_chunking(text)  
for i, chunk in enumerate(chunks):  
    print(f"语义分块 {i+1}: {chunk}")

import re  

defrecursive_chunking(text, max_chunk_size=500, separators=None):
"""递归分块方法  

    Args:  
        text: 输入文本  
        max_chunk_size: 最大分块大小  
        separators: 分隔符列表，按优先级排序  
    """
if separators isNone:  
        separators = ['\n\n', '\n', '。', '！', '？', '，', ' ']  

defsplit_text(text, max_size, sep_index=0):
"""递归分割文本"""
if len(text) <= max_size or sep_index >= len(separators):  
return [text] if text.strip() else []  

        separator = separators[sep_index]  
        chunks = []  

# 按当前分隔符分割  
        parts = text.split(separator)  

        current_chunk = ""
for part in parts:  
# 如果添加这部分不会超过大小限制  
if len(current_chunk + separator + part) <= max_size:  
if current_chunk:  
                    current_chunk += separator + part  
else:  
                    current_chunk = part  
else:  
# 保存当前分块  
if current_chunk:  
if len(current_chunk) > max_size:  
# 如果当前分块仍然太大，递归处理  
                        chunks.extend(split_text(current_chunk, max_size, sep_index + 1))  
else:  
                        chunks.append(current_chunk)  

# 开始新分块  
if len(part) > max_size:  
# 如果这部分太大，递归处理  
                    chunks.extend(split_text(part, max_size, sep_index + 1))  
                    current_chunk = ""
else:  
                    current_chunk = part  

# 添加最后一个分块  
if current_chunk:  
if len(current_chunk) > max_size:  
                chunks.extend(split_text(current_chunk, max_size, sep_index + 1))  
else:  
                chunks.append(current_chunk)  

return chunks  

return split_text(text, max_chunk_size)  

# 使用示例  
text = """这是第一段内容，包含了关于人工智能技术发展的详细介绍。  

这是第二段内容，讨论了机器学习在各个领域的应用。这段内容比较长，可能需要进一步分割。  

这是第三段的短内容。  

这是第四段非常长的内容，包含了大量的技术细节和实现方案，需要详细说明深度学习、自然语言处理、计算机视觉等多个技术领域的最新进展。"""

chunks = recursive_chunking(text, max_chunk_size=100)  
for i, chunk in enumerate(chunks):  
    print(f"递归分块 {i+1} (长度: {len(chunk)}): {chunk[:50]}...")

import re  
from typing import List, Dict, Any  

classHybridChunker:
"""混合分块器，结合多种分块策略"""

def__init__(self, max_chunk_size=1000, overlap_size=100):
        self.max_chunk_size = max_chunk_size  
        self.overlap_size = overlap_size  

defdetect_content_type(self, text: str) -> str:
"""检测内容类型"""
# 检测是否包含代码  
if re.search(r'```|def |class |import ', text):  
return'code'

# 检测是否包含表格标记  
if'|'in text and text.count('|') > 5:  
return'table'

# 检测是否包含列表  
if re.search(r'^\s*[-*+]\s', text, re.MULTILINE):  
return'list'

# 检测是否包含标题结构  
if re.search(r'^#+\s', text, re.MULTILINE):  
return'structured'

return'plain_text'

defchunk_by_content_type(self, text: str, content_type: str) -> List[str]:
"""根据内容类型选择分块策略"""

if content_type == 'code':  
return self._chunk_code(text)  
elif content_type == 'table':  
return self._chunk_table(text)  
elif content_type == 'list':  
return self._chunk_list(text)  
elif content_type == 'structured':  
return self._chunk_structured(text)  
else:  
return self._chunk_plain_text(text)  

def_chunk_code(self, text: str) -> List[str]:
"""代码分块策略"""
# 按代码块分割  
        code_blocks = re.split(r'```[\s\S]*?```', text)  
        chunks = []  
for block in code_blocks:  
if len(block) > self.max_chunk_size:  
# 按函数或类分割  
                parts = re.split(r'\n(?=def |class )', block)  
                chunks.extend(parts)  
else:  
                chunks.append(block)  
return [chunk.strip() for chunk in chunks if chunk.strip()]  

def_chunk_table(self, text: str) -> List[str]:
"""表格分块策略"""
        lines = text.split('\n')  
        chunks = []  
        current_chunk = []  

for line in lines:  
if'|'in line:  
                current_chunk.append(line)  
else:  
if current_chunk:  
                    chunks.append('\n'.join(current_chunk))  
                    current_chunk = []  
if line.strip():  
                    chunks.append(line)  

if current_chunk:  
            chunks.append('\n'.join(current_chunk))  

return chunks  

def_chunk_list(self, text: str) -> List[str]:
"""列表分块策略"""
# 按列表项分组  
        items = re.split(r'\n(?=\s*[-*+]\s)', text)  
        chunks = []  
        current_chunk = ""

for item in items:  
if len(current_chunk + item) > self.max_chunk_size:  
if current_chunk:  
                    chunks.append(current_chunk.strip())  
                current_chunk = item  
else:  
                current_chunk += '\n' + item if current_chunk else item  

if current_chunk:  
            chunks.append(current_chunk.strip())  

return chunks  

def_chunk_structured(self, text: str) -> List[str]:
"""结构化文档分块策略"""
# 按标题分割  
        sections = re.split(r'\n(?=#+\s)', text)  
        chunks = []  

for section in sections:  
if len(section) > self.max_chunk_size:  
# 进一步按段落分割  
                paragraphs = section.split('\n\n')  
                current_chunk = ""

for para in paragraphs:  
if len(current_chunk + para) > self.max_chunk_size:  
if current_chunk:  
                            chunks.append(current_chunk.strip())  
                        current_chunk = para  
else:  
                        current_chunk += '\n\n' + para if current_chunk else para  

if current_chunk:  
                    chunks.append(current_chunk.strip())  
else:  
                chunks.append(section.strip())  

return [chunk for chunk in chunks if chunk]  

def_chunk_plain_text(self, text: str) -> List[str]:
"""纯文本分块策略"""
# 结合段落和句子分块  
        paragraphs = text.split('\n\n')  
        chunks = []  
        current_chunk = ""

for para in paragraphs:  
if len(current_chunk + para) > self.max_chunk_size:  
if current_chunk:  
                    chunks.append(current_chunk.strip())  

# 如果单个段落过长，按句子分割  
if len(para) > self.max_chunk_size:  
                    sentences = re.split(r'[.!?。！？]+', para)  
                    temp_chunk = ""
for sent in sentences:  
if len(temp_chunk + sent) > self.max_chunk_size:  
if temp_chunk:  
                                chunks.append(temp_chunk.strip() + '。')  
                            temp_chunk = sent  
else:  
                            temp_chunk += sent + '。'if temp_chunk else sent  

if temp_chunk:  
                        current_chunk = temp_chunk  
else:  
                        current_chunk = ""
else:  
                    current_chunk = para  
else:  
                current_chunk += '\n\n' + para if current_chunk else para  

if current_chunk:  
            chunks.append(current_chunk.strip())  

return chunks  

defchunk(self, text: str) -> List[Dict[str, Any]]:
"""主分块方法"""
        content_type = self.detect_content_type(text)  
        chunks = self.chunk_by_content_type(text, content_type)  

# 为每个分块添加元数据  
        result = []  
for i, chunk in enumerate(chunks):  
            result.append({  
'id': i,  
'content': chunk,  
'content_type': content_type,  
'length': len(chunk)  
            })  

return result  

# 使用示例  
text = """# 人工智能技术概述  

人工智能技术包括多个领域：  

- 机器学习  
- 深度学习    
- 自然语言处理  
- 计算机视觉  

## 代码示例  

def train_model(data):  
    model = create_model()  
    model.fit(data)  
    return model  


这是一个简单的模型训练函数。"""

chunker = HybridChunker(max_chunk_size=200)  
chunks = chunker.chunk(text)  

for chunk in chunks:  
    print(f"分块 {chunk['id']} ({chunk['content_type']}, {chunk['length']} 字符):")  
    print(f"{chunk['content'][:100]}...")  
    print("-" * 50)