RAG 分块策略深度对比：语义分块 vs 文档感知 vs 层级分块

def fixed_size_chunk(text: str, chunk_size: int = 512, overlap: int = 50):
    tokens = tokenizer.encode(text)
    chunks = []
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk = tokens[i:i + chunk_size]
        chunks.append(tokenizer.decode(chunk))
        if i + chunk_size >= len(tokens):
            break
    return chunks

import numpy as np
from sentence_transformers import SentenceTransformer

def semantic_chunk(text: str, model, threshold: float = 0.5):
    # 1. 先按句子切分
    sentences = split_sentences(text)
    
    # 2. 逐句计算 Embedding
    embeddings = model.encode(sentences)
    
    # 3. 计算相邻句子的余弦相似度
    similarities = []
    for i in range(len(embeddings) - 1):
        sim = cosine_similarity(embeddings[i], embeddings[i+1])
        similarities.append(sim)
    
    # 4. 在相似度低于阈值处切分
    chunks = []
    current_chunk = [sentences[0]]
    for i, sim in enumerate(similarities):
        if sim < threshold:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i+1]]
        else:
            current_chunk.append(sentences[i+1])
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

from unstructured import partition_pdf

def document_aware_chunk(file_path: str):
    elements = partition_pdf(file_path)
    
    chunks = []
    current_section = {"title": "", "level": 0, "content": []}
    
    for elem in elements:
        if elem.category == "Title":
            # 遇到新标题，保存当前 section
            if current_section["content"]:
                chunks.append(format_chunk(current_section))
            
            level = elem.metadata.category_depth or 0
            current_section = {
                "title": elem.text,
                "level": level,
                "content": []
            }
        else:
            current_section["content"].append(elem.text)
    
    if current_section["content"]:
        chunks.append(format_chunk(current_section))
    
    return chunks

def format_chunk(section):
    title_prefix = "#" * (section["level"] + 1)
    return f"{title_prefix} {section['title']}\n\n" + "\n".join(section["content"])

class HierarchicalChunker:
    def __init__(self):
        self.levels = [
            {"size": 128, "name": "small"},    # 精确匹配
            {"size": 512, "name": "medium"},   # 标准检索
            {"size": 2048, "name": "large"},   # 上下文扩展
        ]
    
    def chunk(self, text: str):
        results = {}
        for level in self.levels:
            chunks = fixed_size_chunk(text, chunk_size=level["size"], overlap=level["size"]//4)
            results[level["name"]] = chunks
        
        # 建立层级映射
        mappings = self._build_mappings(results)
        return results, mappings
    
    def _build_mappings(self, results):
        """建立 small → medium → large 的映射关系"""
        mappings = []
        for i, small in enumerate(results["small"]):
            # 找到包含这个 small chunk 的 medium 和 large
            medium_idx = self._find_parent(i, len(results["small"]), len(results["medium"]))
            large_idx = self._find_parent(i, len(results["small"]), len(results["large"]))
            mappings.append({
                "small_idx": i,
                "medium_idx": medium_idx,
                "large_idx": large_idx
            })
        return mappings

def recursive_chunk(text: str, chunk_size: int = 512, separators=None):
    if separators is None:
        separators = ["\n\n\n", "\n\n", "\n", ". ", " ", ""]
    
    final_chunks = []
    separator = separators[-1]
    
    for sep in separators:
        if sep == "":
            separator = sep
            break
        if sep in text:
            separator = sep
            break
    
    splits = text.split(separator) if separator else list(text)
    
    good_splits = []
    for split in splits:
        if len(tokenizer.encode(split)) < chunk_size:
            good_splits.append(split)
        else:
            if good_splits:
                merged = merge_splits(good_splits, separator, chunk_size)
                final_chunks.extend(merged)
                good_splits = []
            # 递归处理过长的 split
            if len(separators) > 1:
                final_chunks.extend(
                    recursive_chunk(split, chunk_size, separators[:-1])
                )
            else:
                final_chunks.append(split)
    
    if good_splits:
        merged = merge_splits(good_splits, separator, chunk_size)
        final_chunks.extend(merged)
    
    return final_chunks

class HybridChunker:
    """2026 年推荐分块方案：文档感知 + 语义修正 + 层级扩展"""
    
    def chunk(self, document):
        # Layer 1: 文档结构感知切分
        sections = document_aware_split(document)
        
        # Layer 2: 对过长的 section 做语义二次切分
        refined = []
        for section in sections:
            if token_count(section) > 1024:
                refined.extend(semantic_chunk(section))
            else:
                refined.append(section)
        
        # Layer 3: 为每个 chunk 生成上下文摘要
        for i, chunk in enumerate(refined):
            chunk.context_summary = llm.summarize(
                context=get_context(refined, i, window=2)
            )
        
        return refined