RAG 进阶：Chunking 策略与检索优化的实战取舍

你搭好了 RAG 管线：文档切一切，向量存一存，用户问一句，top-k 检索出来丢给 LLM。结果回答质量不稳定——有时候精准，有时候答非所问。你调了 embedding 模型、换了更大的 LLM，效果提升有限。

问题大概率不在生成端，而在检索端。而检索质量的上限，在你切分文档的那一刻就已经决定了。

为什么检索是瓶颈

RAG 的端到端质量取决于一条最短路径：chunking -> embedding -> retrieval -> reranking -> generation。这条链上任何一环拉垮，后面的环节再怎么优化都是补救。但实际经验中，90% 的检索质量问题可以追溯到 chunking 策略和检索方式的选择。

一个直观的例子：假设你的知识库有一份 API 文档，其中一段描述了某个函数的参数和返回值。如果你用固定 512 token 切分，参数说明和返回值说明可能被切成两个 chunk。用户问"这个函数返回什么"，embedding 检索可能只命中参数说明的 chunk，返回值信息丢失了。

这不是 embedding 模型的问题，不是向量数据库的问题——是 chunking 把语义完整性破坏了。

五种 Chunking 策略详解

1. 固定大小切分（Fixed-Size with Overlap）

最简单的方案：按字符数或 token 数切分，相邻 chunk 之间有重叠。

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
)

chunks = splitter.split_text(your_document)
print(f"共 {len(chunks)} 个 chunk，平均长度 {sum(len(c) for c in chunks) / len(chunks):.0f}")

适用场景：纯文本语料、日志文件、无明显结构的文档。作为 baseline 快速验证管线是否跑通。

不适用：结构化文档（API 文档、法律合同、技术规范）、包含表格或代码块的文档。固定切分会无情地切断表格行、代码块、段落间的逻辑关系。

2. 递归字符切分（Recursive Character Splitting）

LlamaIndex 和 LangChain 的默认方案。按分隔符优先级递归切分：先尝试按双换行切，切不动再按单换行切，再切不动按句号切，以此类推。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    chunk_size=1000,
    chunk_overlap=100,
)

chunks = splitter.split_text(your_document)

为什么比固定切分好：它尊重自然语言的自然边界（段落 > 句子 > 词），大多数情况下不会把一个句子从中间切断。

仍然不足：它不理解文档结构。一个 Markdown 文档里，## API Reference 下面的内容和 ## Getting Started 下面的内容可能在同一个 chunk 里——对检索来说这是噪音。

3. 语义切分（Semantic Chunking）

不按固定大小切，而是按语义相似度切：计算相邻句子的 embedding 相似度，当相似度低于阈值时切分。

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=75,  # 相似度差距排在第 75 百分位时切分
)

chunks = splitter.split_text(your_document)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {len(chunk)} 字符")

优势：每个 chunk 内部语义连贯，检索时命中的 chunk 更可能完整回答用户问题。

代价：需要对每个句子做 embedding，处理速度慢 5-10 倍。适合离线预处理，不适合实时切分。阈值选择对效果影响大——太低则 chunk 太碎，太高则 chunk 太长。

实战中，语义切分更适合长文本文献（论文、报告）和问答对数据集。对于结构化文档，有更好的选择。

4. 文档结构感知切分（Structure-Aware Splitting）

利用文档本身的结构标记（Markdown 标题、HTML 标签、代码块边界、表格行）作为切分点。这是当前生产环境中最推荐的做法。

from langchain.text_splitter import MarkdownHeaderTextSplitter
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 第一步：按 Markdown 标题层级切分
headers_to_split_on = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_chunks = md_splitter.split_text(markdown_document)

# 第二步：对超过 chunk_size 的部分再递归切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
)
final_chunks = text_splitter.split_documents(md_chunks)

# 每个 chunk 自动携带标题元数据
for chunk in final_chunks[:3]:
    print(f"Metadata: {chunk.metadata}")
    print(f"Content: {chunk.page_content[:100]}...")
    print("---")

对于代码文档，可以用语言感知的切分器：

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1500,
    chunk_overlap=200,
)
code_chunks = python_splitter.split_text(python_source_code)

核心优势：chunk 自带结构元数据（属于哪个章节、哪个函数），检索时可以把元数据用于过滤和加权。这是其他切分方式做不到的。

Haystack 框架在这方面做得很好——它的 PreProcessor 组件原生支持按段落、标题、句子级别切分，并且自动保留元数据。

5. Late Chunking（延迟切分）

2024 年新兴的模式：先对整个文档做 embedding，然后再切分。核心思想是：长上下文 embedding 模型（如 Jina Embeddings v3）可以处理 8192 token 的输入，先得到整文档级别的语义表示，再在 embedding 空间里做切分。

from transformers import AutoTokenizer, AutoModel
import torch
import numpy as np

tokenizer = AutoTokenizer.from_pretrained("jinaai/jina-embeddings-v3")
model = AutoModel.from_pretrained("jinaai/jina-embeddings-v3", trust_remote_code=True)

def late_chunking(document: str, chunk_size: int = 500) -> list[dict]:
    """先 embedding 整篇文档，再在 token embedding 空间中切分"""
    inputs = tokenizer(document, return_tensors="pt", truncation=True, max_length=8192)
    with torch.no_grad():
        outputs = model(**inputs)

    # 获取每个 token 的 embedding
    token_embeddings = outputs.last_hidden_state.squeeze(0)  # [seq_len, hidden_dim]

    # 在 token embedding 空间中按 chunk_size 分组并平均池化
    tokens = inputs["input_ids"].squeeze(0)
    chunk_embeddings = []
    for i in range(0, len(tokens), chunk_size):
        chunk_emb = token_embeddings[i : i + chunk_size].mean(dim=0)
        chunk_text = tokenizer.decode(tokens[i : i + chunk_size], skip_special_tokens=True)
        chunk_embeddings.append({
            "text": chunk_text,
            "embedding": chunk_emb.numpy(),
        })
    return chunk_embeddings

chunks = late_chunking(your_document, chunk_size=500)
print(f"生成 {len(chunks)} 个 late-chunked embeddings")

理论优势：chunk embedding 保留了全文档上下文信息，解决了"切分后丢失上下文"的根本问题。

实际限制：需要支持长上下文的 embedding 模型，对 GPU 内存要求更高，推理延迟也更高。目前更适合小规模、高精度场景。EmbedAnything 项目提供了高效的 embedding 管线，可以作为这类方案的底层基础设施。

检索优化：混合搜索与 Reranking

选对 chunking 策略解决了"检什么"的问题，但"怎么检"同样重要。纯向量相似度搜索在某些场景下会败给简单的关键词匹配。

混合搜索：Dense + Sparse

混合搜索结合语义检索（dense retrieval）和关键词检索（sparse retrieval / BM25），用倒数排名融合（Reciprocal Rank Fusion, RRF）合并结果。

from rank_bm25 import BM25Okapi
import numpy as np

def hybrid_search(
    query: str,
    chunks: list[str],
    dense_embeddings: np.ndarray,  # 预计算的 chunk embeddings
    bm25: BM25Okapi,
    embedding_model,
    top_k: int = 10,
    alpha: float = 0.5,  # dense 权重，1-alpha 给 sparse
) -> list[tuple[int, float]]:
    """混合搜索：语义 + 关键词，用 RRF 融合"""
    # Dense retrieval
    query_emb = np.array(embedding_model.embed_query(query)).reshape(1, -1)
    dense_scores = np.dot(dense_embeddings, query_emb.T).flatten()
    dense_ranks = np.argsort(-dense_scores)

    # Sparse retrieval (BM25)
    tokenized_query = query.lower().split()
    sparse_scores = bm25.get_scores(tokenized_query)
    sparse_ranks = np.argsort(-sparse_scores)

    # Reciprocal Rank Fusion
    rrf_scores = {}
    for rank, idx in enumerate(dense_ranks):
        rrf_scores[idx] = rrf_scores.get(idx, 0) + alpha / (1 + rank)
    for rank, idx in enumerate(sparse_ranks):
        rrf_scores[idx] = rrf_scores.get(idx, 0) + (1 - alpha) / (1 + rank)

    # 按 RRF 分数排序
    sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results[:top_k]

# 预处理
tokenized_corpus = [chunk.lower().split() for chunk in chunks]
bm25_index = BM25Okapi(tokenized_corpus)

什么时候混合搜索收益最大：用户查询包含精确术语、产品名称、错误代码时。BM25 在精确匹配上完胜语义搜索，而语义搜索在模糊查询、同义词场景上更强。两者互补。

LlamaIndex 原生支持混合搜索——配置 VectorIndex + BM25 检索器并设置 RRF 融合参数即可。Lantern 作为 PostgreSQL 向量扩展，也支持在 SQL 层面实现 dense + sparse 联合查询。

Reranking 管线

初次检索（无论是纯向量还是混合搜索）拿到 top-k（比如 20 条），再用 cross-encoder 模型精排，选出最相关的 top-n（比如 5 条）。

from sentence_transformers import CrossEncoder

# Cross-encoder 比 bi-encoder 慢但更准确
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, chunks: list[str], top_n: int = 5) -> list[dict]:
    """用 cross-encoder 对检索结果重新排序"""
    pairs = [(query, chunk) for chunk in chunks]
    scores = reranker.predict(pairs)

    ranked = sorted(
        [{"text": chunk, "score": float(score)} for chunk, score in zip(chunks, scores)],
        key=lambda x: x["score"],
        reverse=True,
    )
    return ranked[:top_n]

# 使用示例
initial_results = ["chunk1 内容...", "chunk2 内容...", "chunk3 内容..."]
reranked = rerank("用户的查询", initial_results, top_n=3)
for r in reranked:
    print(f"Score: {r['score']:.4f} | {r['text'][:80]}")

关键点：Reranking 是当前性价比最高的检索优化手段。一个小的 cross-encoder（MiniLM 级别）就能带来 10-20% 的检索精度提升，代价是额外 10-50ms 延迟。

查询扩展（Query Expansion）

用户的查询往往太短、太模糊。用 LLM 生成子查询，从不同角度检索，再合并结果。

from openai import OpenAI

client = OpenAI()

def expand_query(original_query: str, n_sub_queries: int = 3) -> list[str]:
    """用 LLM 生成子查询，覆盖不同检索角度"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "你是一个查询扩展助手。根据用户的原始查询，生成 "
                    f"{n_sub_queries} 个不同角度的子查询，用于提高检索覆盖率。"
                    "每个子查询一行，不要编号，不要解释。"
                ),
            },
            {"role": "user", "content": original_query},
        ],
        temperature=0.3,
        max_tokens=200,
    )
    sub_queries = response.choices[0].message.content.strip().split("\n")
    return [original_query] + [q.strip() for q in sub_queries if q.strip()]

expanded = expand_query("如何优化 RAG 检索效果")
for q in expanded:
    print(f"  -> {q}")

查询扩展在用户习惯简短提问（比如中文场景下"怎么配置"这种模糊查询）时效果最好。但要注意：每个子查询都会触发一次检索，检索延迟和成本是原来的 N 倍。生产中通常限制在 3-5 个子查询。

生产级管线：把所有东西组合起来

下面是一个完整的、可运行的 chunking + 检索管线，综合了结构感知切分、混合搜索和 reranking。

"""
生产级 RAG 检索管线
依赖：pip install langchain langchain-openai rank-bm25 sentence-transformers
"""
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder
from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

# --- 配置 ---
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100
INITIAL_RETRIEVAL_K = 20
FINAL_TOP_N = 5
RERANK_MODEL = "cross-encoder/ms-marco-MiniLM-L-6-v2"

# --- Step 1: 结构感知切分 ---
def chunk_documents(documents: list[str]) -> list[dict]:
    """按 Markdown 结构切分，保留元数据"""
    header_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
    )
    sub_splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP
    )

    all_chunks = []
    for doc in documents:
        md_chunks = header_splitter.split_text(doc)
        for chunk in md_chunks:
            if len(chunk.page_content) > CHUNK_SIZE:
                sub_chunks = sub_splitter.split_documents([chunk])
                all_chunks.extend(sub_chunks)
            else:
                all_chunks.append(chunk)

    return [
        {"text": c.page_content, "metadata": c.metadata}
        for c in all_chunks
    ]

# --- Step 2: 构建 Embedding + BM25 索引 ---
def build_index(chunks: list[dict]):
    """同时构建 dense 和 sparse 索引"""
    texts = [c["text"] for c in chunks]
    embedder = OpenAIEmbeddings(model="text-embedding-3-small")
    dense_embeddings = np.array(embedder.embed_documents(texts))

    tokenized = [text.lower().split() for text in texts]
    bm25 = BM25Okapi(tokenized)

    return dense_embeddings, bm25, embedder

# --- Step 3: 混合搜索 + Reranking ---
def retrieve(
    query: str,
    chunks: list[dict],
    dense_embeddings: np.ndarray,
    bm25: BM25Okapi,
    embedder: OpenAIEmbeddings,
) -> list[dict]:
    """完整的检索管线：混合搜索 -> Reranking -> 返回 top-n"""
    texts = [c["text"] for c in chunks]

    # Dense retrieval
    query_emb = np.array(embedder.embed_query(query)).reshape(1, -1)
    dense_scores = np.dot(dense_embeddings, query_emb.T).flatten()
    dense_ranks = np.argsort(-dense_scores)

    # Sparse retrieval
    sparse_scores = bm25.get_scores(query.lower().split())
    sparse_ranks = np.argsort(-sparse_scores)

    # RRF 融合
    rrf_scores = {}
    for rank, idx in enumerate(dense_ranks[:INITIAL_RETRIEVAL_K]):
        rrf_scores[idx] = rrf_scores.get(idx, 0) + 0.5 / (1 + rank)
    for rank, idx in enumerate(sparse_ranks[:INITIAL_RETRIEVAL_K]):
        rrf_scores[idx] = rrf_scores.get(idx, 0) + 0.5 / (1 + rank)

    # 取融合后 top-k
    candidates = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    candidate_texts = [texts[idx] for idx, _ in candidates[:INITIAL_RETRIEVAL_K]]

    # Cross-encoder reranking
    reranker = CrossEncoder(RERANK_MODEL)
    pairs = [(query, text) for text in candidate_texts]
    rerank_scores = reranker.predict(pairs)

    results = []
    for (idx, _), score in zip(candidates[:INITIAL_RETRIEVAL_K], rerank_scores):
        results.append((idx, float(score)))
    results.sort(key=lambda x: x[1], reverse=True)

    return [
        {**chunks[idx], "relevance_score": score}
        for idx, score in results[:FINAL_TOP_N]
    ]

# --- 运行 ---
if __name__ == "__main__":
    sample_docs = [
        "# RAG 优化指南\n\n## 概述\n\nRAG（检索增强生成）是一种结合信息检索和文本生成的技术。\n\n## Chunking 策略\n\n选择合适的 chunking 策略是 RAG 质量的关键。\n\n### 固定大小切分\n\n最简单的方案，按固定 token 数切分。\n\n### 语义切分\n\n按语义相似度切分，保证每个 chunk 语义连贯。",
        "# 向量数据库对比\n\n## 概述\n\n本文对比主流向量数据库的性能和特性。\n\n## Pinecone\n\n完全托管的向量数据库服务。\n\n## Milvus\n\n开源高性能向量数据库。",
    ]

    chunks = chunk_documents(sample_docs)
    print(f"共切分 {len(chunks)} 个 chunk")
    dense_emb, bm25_idx, embedder = build_index(chunks)
    print("索引构建完成")

    results = retrieve("RAG 如何切分文档", chunks, dense_emb, bm25_idx, embedder)
    for r in results:
        print(f"\n[Score: {r['relevance_score']:.4f}] Metadata: {r['metadata']}")
        print(f"  {r['text'][:120]}...")

决策框架：什么文档用什么切分策略

文档类型	推荐切分策略	原因	检索方式
API 文档 / 技术手册	结构感知（标题 + 代码块边界）	自然按函数/类/模块划分，元数据可过滤	混合搜索 + Reranking
法律合同 / 条款文档	递归字符切分 + 条款编号正则	条款间逻辑独立但长度不一	纯向量搜索 + Reranking
论文 / 研究报告	语义切分	论述逻辑连贯，语义边界比结构边界更准	混合搜索
FAQ / 问答对	按问答对切分，不拆分	每个 QA 对天然是一个检索单元	BM25 即可
日志 / 无结构文本	固定大小 + overlap	没有结构可用，固定切分是唯一选择	BM25 + 关键词过滤
表格数据	按行切分，保留表头作为元数据	表格行是天然切分点，表头提供语义	向量搜索 + 元数据过滤
混合格式文档	结构感知 + 针对表格/代码特殊处理	不同部分需要不同策略	混合搜索 + Reranking

通用建议：如果你不确定文档类型，从递归字符切分 + 混合搜索开始，这是最稳妥的 baseline。然后针对检索质量差的 case 分析原因，逐个优化。

AutoRAG 项目提供了自动化的 RAG 参数搜索功能，可以帮你自动测试不同的 chunking 参数和检索策略组合，省去手动调参的时间。

三个常见的生产坑

坑 1：忽略了 chunk 里的元数据

切分后每个 chunk 自带的元数据（来源文档、章节标题、页码）是最容易被忽略的检索增强手段。

# 错误做法：只存 text + embedding
vector_store.add(text=chunk, embedding=emb)

# 正确做法：存储并利用元数据
vector_store.add(
    text=chunk,
    embedding=emb,
    metadata={
        "source": "api-docs",
        "section": "authentication",
        "doc_type": "api_reference",
    }
)

# 检索时用元数据过滤
results = vector_store.search(
    query="如何认证",
    filter={"doc_type": "api_reference"},  # 只搜 API 文档
    top_k=10,
)

元数据过滤可以在向量搜索之前大幅缩小检索范围，不仅提高精度，还降低延迟。Pathway 的 llm-app 框架在这一点上设计得很好——它的数据管线原生支持在索引时附加丰富的元数据。

坑 2：Reranker 成为延迟瓶颈

Cross-encoder reranking 是串行操作，20 条候选 x 10ms/条 = 200ms 额外延迟。在要求 P95 < 500ms 的系统中，这占了近一半。

解决方案：

• 减少 initial retrieval 的 top-k（从 30 降到 15）
• 用更小的 reranker 模型（MiniLM 而不是 large）
• 对 reranker 做 batch 推理
• 考虑用 LLM 做轻量级 reranking（只输出 rank 不生成文本）

坑 3：Embedding 模型与查询分布不匹配

你用多语言 embedding 模型索引了中文文档，但用户查询是中英混合的（比如 "RAG chunking 策略"）。很多 embedding 模型在混合语言输入上表现会下降。

解决方案：

• 索引时同时生成中英文 chunk（用 LLM 翻译后分别 embedding）
• 查询扩展时生成中英文子查询
• 选择明确支持跨语言的 embedding 模型（如 multilingual-e5、BGE-M3）

总结

• Chunking 决定了检索质量的上限。花在 chunking 策略选择上的时间，比花在调 embedding 模型上的时间回报更高。从结构感知切分开始，按需调整。
• 混合搜索是当前生产环境的标配。纯向量搜索在精确匹配场景下败给 BM25，纯 BM25 在语义理解场景下败给向量搜索。两者融合是最稳妥的方案。
• Reranking 是性价比最高的单点优化。加一个 cross-encoder reranker，检索精度通常提升 10-20%，代价是 10-50ms 延迟。
• 元数据不是可选项。在索引阶段就把来源、章节、类型等元数据挂上，检索时用元数据过滤，效果提升立竿见影。
• 没有万能的 chunking 策略。API 文档、论文、FAQ、法律合同需要不同的切分方式。根据你的文档类型选策略，不要照搬教程。