RAG 实战工程

Chunking：固定 512 是初学者陷阱，结构化才是生产配置

背景与原理

固定大小 chunk 的根本问题：语义边界与字符边界不重合。一个 512 token 的切口可能把「为什么」放在 chunk A、「因为...」放在 chunk B——embedding 各自看不到对方，retrieval 永远召不全。这不是 chunk size 调参能解决的，是 chunking 策略问题。

四代 chunking 演进路径：（1）固定大小——naive，已知失败；（2）Recursive character splitter（LangChain 经典）——按段落/句子/词层级回退，是入门到能用的最低门槛；（3）Structure-aware——按 markdown header / 代码 AST / PDF 章节切，保留语义单元；（4）Parent-child / hierarchical——小 chunk 用来 retrieve、大 chunk（父段落或整章）用来给 LLM，避免「召回精确但上下文残缺」。

2024 年 Anthropic 提出的 Contextual Retrieval 是第五代：每个 chunk 在 embed 前先让 Claude 加一段「这 chunk 在原文中的角色」的 contextual prefix。同样的 RAG pipeline，单加这一步 failure rate 降 35%。原理：embedding 模型对孤立 chunk 编码会丢失「这是关于什么的」信号——人工加上 50-100 token 的上下文标签，embedding 立即对齐到正确语义子空间。配合 prompt caching，给 1M token 的 corpus 加 contextual prefix 的成本只有 $1.02——这是 2024 年 RAG 工程最高 ROI 的单一改动。

实战示例

Structure-aware + parent-child + contextual prefix 三件套（最小可工作版）：

from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
import anthropic

# —— 1. Structure-aware: 先按 markdown header 切大块（=父 chunk） ——
header_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
)
parent_chunks = header_splitter.split_text(doc)  # ~1500 token / 块

# —— 2. Parent-child: 再把父 chunk 切成 retrieval 用的小 chunk ——
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400, chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " "]
)

# —— 3. Contextual prefix: 用 Claude 给每个 child 加上下文标签 ——
# 整篇文档放 cache_control=ephemeral，1M token 只付 1 次输入费
client = anthropic.Anthropic()

def contextualize(full_doc, chunk):
    msg = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=120,
        messages=[{"role":"user", "content":[
          {"type":"text",
           "text": f"<document>{full_doc}</document>",
           "cache_control":{"type":"ephemeral"}},  # ← 关键
          {"type":"text",
           "text": f"<chunk>{chunk}</chunk>\nGive a 1-2 sentence "
                   "context placing this chunk in the doc. Just the context."}
        ]}
    )
    return msg.content[0].text

# 索引时：embed(contextual_prefix + child_chunk)
# 生成时：retrieve child → 取回对应 parent 给 LLM 做更全上下文
indexed = []
for parent in parent_chunks:
    for child in child_splitter.split_text(parent.page_content):
        prefix = contextualize(doc, child)
        indexed.append({
          "embed_text": prefix + "\n" + child,  # 用来 embed/BM25
          "parent_text": parent.page_content,    # 给 LLM 用
          "metadata": parent.metadata
        })

Hybrid Search：BM25 不是「老古董」，是 dense 永远追不上的另一半

背景与原理

2020 年以后大家都说「dense embedding 颠覆了 BM25」——这是 academic benchmark 的错觉。真实场景里：用户问 "GPT-4o vs Claude 4.7"，dense embedding 会召回所有「大模型对比」的文章；BM25 直接召回带这两个 token 的文章。"connection refused on port 5432"、"erro_user_42"、"RFC 9457"——这些都是 BM25 完胜 dense 的场景。原因：embedding 把 token 投影到稠密语义空间，稀有词的精确匹配信号被有损压缩；BM25 保留了 token-level 精确度。

Anthropic Contextual Retrieval 报告的 ablation 数据极其干净：在他们的 corpus 上单独 dense retrieval 失败率 5.7%，加 BM25（Contextual BM25 + Contextual Embeddings）降到 2.9%，近乎减半。Microsoft 的 Azure AI Search、Pinecone、Weaviate、Elasticsearch 8.x 全部内置 hybrid 不是巧合——这是 2024 年生产 RAG 的事实标准。

合并算法首选 RRF（Cormack et al. SIGIR 2009）：不需要 normalize 两边的分数，只用 rank。score(d) = Σ 1/(k + rank_i(d))，k 通常取 60。比加权和稳健得多——加权和需要你不停调权重，RRF 对两边分数尺度免疫。Elastic 默认 k=60，Anthropic 报告用的也是 60。

实战示例

10 行 RRF 合并 dense + BM25（用 bm25s 库 + 任意 embedding store）：

import bm25s
from collections import defaultdict

# —— 索引阶段 ——
corpus_texts = [item["embed_text"] for item in indexed]
bm25 = bm25s.BM25()
bm25.index(bm25s.tokenize(corpus_texts, stopwords="en"))
# dense_index = your_vector_store.upsert(...)

# —— 查询阶段 ——
def hybrid_retrieve(query, top_k=50, k=60):
    # 两边各取 top_k
    bm25_results = bm25.retrieve(bm25s.tokenize([query]), k=top_k)
    dense_results = dense_index.query(embed(query), top_k=top_k)

    # —— RRF 合并 ——
    scores = defaultdict(float)
    for rank, doc_id in enumerate(bm25_results.ids[0]):
        scores[doc_id] += 1.0 / (k + rank + 1)
    for rank, hit in enumerate(dense_results.matches):
        scores[hit.id] += 1.0 / (k + rank + 1)

    fused = sorted(scores.items(), key=lambda x: -x[1])
    return [doc_id for doc_id, _ in fused[:top_k]]

# 关键：BM25 和 dense 各召回 50 条，RRF 合并保留 top 50
# 不要在这里截到 top 5——把过滤工作留给下一步的 reranker

Reranker：用 cross-encoder 把 top-50 重排 top-5，是 RAG 性价比最高的一层

背景与原理

Bi-encoder vs cross-encoder 是两类不同模型的本质差异。Bi-encoder（OpenAI text-embedding-3、Cohere embed、BGE）：query 和 doc 各自变成 vector，离线索引、毫秒级查询，但模型在编码时不知道对方存在，错过细微相关性。Cross-encoder（Cohere Rerank、BGE-reranker-v2-m3、ms-marco-MiniLM）：query 和 doc 拼成 [CLS] query [SEP] doc，整条过一次 BERT-like 模型直接出相关性分——精度逼近 LLM-as-judge 但只用 80M-300M 参数。

Cross-encoder 的缺点是计算量与 candidate 数成线性——把 BM25/dense 召回的 top 50-100 重排是可承受的（50ms-300ms），但如果用它做 first-stage retrieval（10万文档），那是上百秒级别。所以正确架构永远是 two-stage：bi-encoder/BM25 召回大候选集（top 50-200）→ cross-encoder reranker 精排到 top 5-10。

真实 ROI 数据。BEIR benchmark：dense baseline NDCG@10 = 0.42，加 ms-marco-MiniLM-L-6-v2 reranker = 0.54（+28%）。Anthropic Contextual Retrieval 报告：在已有 hybrid + contextual 之上再加 Cohere Rerank，failure rate 从 2.9% 进一步降到 1.9%（再砍 35%）。Cohere 自家 BEIR 复现里 rerank-v3 加到 hybrid 上 NDCG +12-18 个点。代价：每查询 +50-200ms、每千次查询约 $1（Cohere 价位）或自托管 BGE 0 现金成本。这是 RAG 里 ROI 最高的单一组件，没有之一。

实战示例

用 BGE-reranker（开源、自托管）+ fallback 到 Cohere（managed）的稳定生产配置：

from FlagEmbedding import FlagReranker
import cohere

# —— 默认：自托管 BGE-reranker-v2-m3（多语种、3 亿参数、可在 8GB 卡上跑）——
#   开源、零 cash cost、p95 ~80ms（A10）
reranker_local = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)

# —— Fallback：Cohere Rerank-v3（managed、稳定、不需要 GPU）——
co = cohere.Client(API_KEY)

def rerank(query, candidates, top_n=8):
    # candidates 是 hybrid 召回的 top 50（list of {id, text}）
    try:
        pairs = [(query, c["text"]) for c in candidates]
        scores = reranker_local.compute_score(pairs, normalize=True)
    except Exception:
        result = co.rerank(
            model="rerank-v3.5",
            query=query,
            documents=[c["text"] for c in candidates],
            top_n=top_n
        )
        return [candidates[r.index] for r in result.results]

    ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return [c for c, _ in ranked[:top_n]]

# —— 关键：reranker 之后的 top_n 才是真正给 LLM 的 chunks ——
# 不要给 LLM 50 个 chunk 让它「自己挑」——浪费 token + lost-in-the-middle
top_chunks = rerank(query, hybrid_candidates, top_n=8)

Query Transformation：用户的 query 才是 RAG 最弱的一环

背景与原理

核心矛盾：corpus 里的 chunks 是「答案样态」（陈述句、长上下文），用户 query 是「问题样态」（短、含代词、上下文缺失）。embedding 空间里两者距离天然就远。这就是为什么 demo 里能用「正确措辞」查到答案、用户实际查的话总差一点——不是模型不行，是 query 没在答案的语义邻域里。

三种主流 query transformation：

• HyDE（Hypothetical Document Embeddings, Gao et al. ACL 2023）：让 LLM 先假装回答这个问题，用假答案的 embedding 去 retrieve。直觉：假答案再差，也比 query 更接近真答案的语义空间。BEIR 上 zero-shot 提升 NDCG +6-15 个点。
• Multi-query（LangChain / RAG-Fusion）：让 LLM 把原 query 改写成 3-5 个变体（同义、视角不同、更具体），各自 retrieve，结果 RRF 合并。覆盖 paraphrase 盲区。
• Step-back prompting（Zheng et al. ICLR 2024，Google DeepMind）：让 LLM 先生成一个更抽象/更上位的问题，retrieve 抽象答案，再用抽象答案+原 query 一起回答。对推理类问答提升 7-27%。

三个 transformation 不互斥。Anthropic 2024 工程经验：HyDE + multi-query 在语义模糊查询上叠加效果最好；step-back 对需要先理解原则再代入的问答（数学、法律、医学）增益最大。但成本也叠加——每查询多 1-2 次 LLM 调用，延迟 +500ms-1s。所以策略上 query transformation 应当按查询类型路由：短而具体的 ID 类查询不需要；自然语言长查询全开。

实战示例

HyDE + multi-query 组合，按查询类型路由：

import anthropic, re
client = anthropic.Anthropic()

# —— 1. HyDE: 生成假答案做 embedding ——
def hyde(query):
    msg = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=200,
        messages=[{"role":"user", "content":
          f"Write a concise, factual 3-4 sentence answer to this question. "
          f"If you don't know, write what such an answer would likely contain.\n\n{query}"}]
    )
    return msg.content[0].text  # 假答案，用它的 embedding 而非 query 的

# —— 2. Multi-query: 改写成 3 个变体 ——
def multi_query(query):
    msg = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=300,
        messages=[{"role":"user", "content":
          f"Rewrite the following query as 3 distinct search queries. "
          f"Each on its own line, no numbering, no quotes.\n\nQuery: {query}"}]
    )
    return [q.strip() for q in msg.content[0].text.strip().split("\n") if q.strip()]

# —— 3. 路由：根据 query 形态选 transformation ——
def route_and_retrieve(query):
    # 短且含 ID/version/error code → 不做 transformation，直接 BM25 主导
    if len(query.split()) < 5 or re.search(r'\b[A-Z]{2,}[-_]?\d+\b|v\d+\.\d+', query):
        return hybrid_retrieve(query, top_k=50)

    # 自然语言长查询 → HyDE + multi-query 全开
    queries = [query] + multi_query(query)
    candidates = defaultdict(float)
    for i, q in enumerate(queries):
        # 用 HyDE 假答案做 dense；BM25 还是用原 query（关键词不应该被改写）
        hyp = hyde(q)
        for rank, doc_id in enumerate(hybrid_retrieve_split(
              dense_query=hyp, bm25_query=q, top_k=50)):
            candidates[doc_id] += 1.0 / (60 + rank + 1)  # RRF
    return sorted(candidates, key=candidates.get, reverse=True)[:50]