chunking-strategies

RAG 的分块策略

最佳分块策略是 RAG 表现优异的关键。

为什么分块很重要

分块不当会导致：

• 上下文碎片化（答案被拆分到多个分块中）
• 检索到无关内容（分块中噪声过多）
• 关系丢失（父子内容被分开）
• 浪费 Token（分块过大或过小）

分块方法对比

方法	最适用场景	分块质量	实现难度
固定大小	简单文档、内容统一	中	简单
递归	结构化文档、Markdown	高	中
语义	复杂文档、内容多样	最高	复杂
父子分块	层次化文档	高	中
延迟分块	保留上下文	最高	复杂

模式 1：带重叠的固定大小分块

基础方法——简单而有效：

from langchain.text_splitter import RecursiveCharacterTextSplitter

def create_fixed_chunks(
    text: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50
) -> list[str]:
    """
    Split text into fixed-size chunks with overlap.

    Guidelines:
    - chunk_size: 256-1024 tokens (512 is solid default)
    - overlap: 10-20% of chunk_size
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    return splitter.split_text(text)

模式 2：语义分块

按含义分组，而不是按任意边界分组：

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

def create_semantic_chunks(text: str) -> list[str]:
    """
    Split text based on semantic similarity between sentences.
    Keeps related content together.
    """
    embeddings = OpenAIEmbeddings()

    splitter = SemanticChunker(
        embeddings=embeddings,
        breakpoint_threshold_type="percentile",
        breakpoint_threshold_amount=95  # Higher = fewer, larger chunks
    )

    return splitter.split_text(text)

自定义语义分块

import numpy as np
from sentence_transformers import SentenceTransformer

def semantic_chunk(
    sentences: list[str],
    model_name: str = "all-MiniLM-L6-v2",
    threshold: float = 0.5
) -> list[list[str]]:
    """
    Group sentences by semantic similarity.
    """
    model = SentenceTransformer(model_name)
    embeddings = model.encode(sentences)

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

    for i in range(1, len(sentences)):
        # Cosine similarity between consecutive sentences
        sim = np.dot(embeddings[i-1], embeddings[i]) / (
            np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
        )

        if sim >= threshold:
            current_chunk.append(sentences[i])
        else:
            chunks.append(current_chunk)
            current_chunk = [sentences[i]]

    chunks.append(current_chunk)
    return chunks

模式 3：父子分块

检索小块内容，并连同上下文一起返回：

from llama_index.core.node_parser import (
    HierarchicalNodeParser,
    SentenceSplitter,
    get_leaf_nodes
)
from llama_index.core import StorageContext, VectorStoreIndex
from llama_index.core.retrievers import AutoMergingRetriever

def create_hierarchical_index(documents):
    """
    Create parent-child chunk hierarchy.
    Small chunks for retrieval, auto-merge to parents for context.
    """
    # Define chunk sizes for each level
    node_parser = HierarchicalNodeParser.from_defaults(
        chunk_sizes=[2048, 512, 128]  # Parent → Child → Leaf
    )

    nodes = node_parser.get_nodes_from_documents(documents)
    leaf_nodes = get_leaf_nodes(nodes)

    # Store all nodes
    storage_context = StorageContext.from_defaults()
    storage_context.docstore.add_documents(nodes)

    # Index only leaf nodes
    index = VectorStoreIndex(
        leaf_nodes,
        storage_context=storage_context
    )

    # Retriever auto-merges to parents when siblings retrieved
    retriever = AutoMergingRetriever(
        index.as_retriever(similarity_top_k=12),
        storage_context=storage_context,
        simple_ratio_thresh=0.3  # Merge if 30%+ siblings retrieved
    )

    return retriever

模式 4：延迟分块（2026 年技术）

先嵌入完整文档，再进行分块——保留全局上下文：

def late_chunking(
    document: str,
    model,
    chunk_size: int = 512
) -> list[dict]:
    """
    Late chunking: embed document, then split embeddings.
    Preserves document-level context in chunk embeddings.

    Reference: Jina AI Late Chunking (2024)
    """
    # 1. Get token-level embeddings for full document
    tokens = model.tokenize(document)
    token_embeddings = model.encode_tokens(tokens)

    # 2. Split into chunks
    chunks = []
    for i in range(0, len(tokens), chunk_size):
        chunk_tokens = tokens[i:i + chunk_size]
        chunk_embeddings = token_embeddings[i:i + chunk_size]

        # 3. Pool chunk embeddings (mean pooling)
        chunk_vector = np.mean(chunk_embeddings, axis=0)

        chunks.append({
            "text": model.decode(chunk_tokens),
            "embedding": chunk_vector
        })

    return chunks

模式 5：适配 Markdown/代码的分块

from langchain.text_splitter import (
    MarkdownHeaderTextSplitter,
    Language,
    RecursiveCharacterTextSplitter
)

def chunk_markdown(text: str) -> list[dict]:
    """Split markdown by headers, preserving structure."""
    headers_to_split_on = [
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]

    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on
    )

    return splitter.split_text(text)


def chunk_code(code: str, language: str = "python") -> list[str]:
    """Split code respecting language syntax."""
    lang_map = {
        "python": Language.PYTHON,
        "javascript": Language.JS,
        "typescript": Language.TS,
    }

    splitter = RecursiveCharacterTextSplitter.from_language(
        language=lang_map.get(language, Language.PYTHON),
        chunk_size=1000,
        chunk_overlap=100
    )

    return splitter.split_text(code)

分块大小指南

内容类型	建议大小	重叠量
问答 / FAQ	256-512	25-50
技术文档	512-1024	50-100
法律文档	1024-2048	100-200
代码	500-1000	50-100
对话	256-512	50-100

评估：如何判断分块是否合理

def evaluate_chunking(chunks: list[str], test_queries: list[dict]):
    """
    Evaluate chunk quality with test queries.

    test_queries format:
    [{"query": "What is X?", "expected_chunk_contains": "X is..."}]
    """
    results = {
        "avg_chunk_size": np.mean([len(c) for c in chunks]),
        "chunk_size_std": np.std([len(c) for c in chunks]),
        "total_chunks": len(chunks),
        "retrieval_hits": 0
    }

    for tq in test_queries:
        # Check if expected content is in a single chunk
        for chunk in chunks:
            if tq["expected_chunk_contains"] in chunk:
                results["retrieval_hits"] += 1
                break

    results["hit_rate"] = results["retrieval_hits"] / len(test_queries)
    return results

最佳实践

1. 使分块大小与查询长度匹配 - 分块大小应与预期查询的长度相近
2. 保留语义边界 - 切勿在句子或段落中间拆分
3. 包含元数据 - 为每个分块添加来源、页码、章节信息
4. 使用真实查询进行测试 - 根据实际使用场景进行评估
5. 考虑检索模型 - 有些嵌入模型偏好特定的分块大小

快速决策树

What type of content?
├─ Structured (headers, sections)
│   └─ Use: Markdown/recursive splitter + hierarchy
├─ Unstructured (prose, articles)
│   └─ Use: Semantic chunking
├─ Code
│   └─ Use: Language-aware splitter
└─ Mixed
    └─ Use: Parent-child with semantic leaves