Optimizing RAG at Scale: Chunking, Retrieval, and the Bayesian Search That Cut Latency 40%

大规模 RAG 优化：分块、检索以及将延迟降低 40% 的贝叶斯搜索

How we moved from “semantic search + hope” to a measured, tunable retrieval pipeline with 95% recall@10. 我们是如何从“语义搜索 + 祈祷”转向一套可衡量、可调优，且 recall@10 达到 95% 的检索流水线的。

The RAG Reality Check

RAG 的现实检验

Everyone ships RAG the same way: chunk by 512 tokens, embed with text-embedding-3-small, top-k=5, stuff into context. It works for demos. Then you hit production: 每个人部署 RAG 的方式都如出一辙：按 512 个 token 分块，使用 text-embedding-3-small 进行嵌入，取 top-k=5，然后塞入上下文。这在演示时没问题，但一旦进入生产环境，问题就来了：

• Legal contracts: 512 tokens splits clauses mid-sentence.
• 法律合同： 512 个 token 会把条款从中间截断。
• API docs: 1000-token chunks drown signal in noise.
• API 文档： 1000 个 token 的分块会让有效信息淹没在噪音中。
• Customer tickets: Conversational context needs overlap, not fixed windows.
• 客户工单： 对话上下文需要重叠，而不是固定的窗口。
• Latency: 500ms embedding + 200ms vector search + 300ms LLM = 1s+ per query.
• 延迟： 500ms 嵌入 + 200ms 向量搜索 + 300ms LLM = 每次查询超过 1 秒。

We rebuilt our retrieval layer from first principles. Here’s what actually moves metrics. 我们从第一性原理出发重建了检索层。以下是真正能提升指标的方法。

---

Chunking: One Size Fits None

分块：没有一种尺寸适合所有人

# rag/chunking.py
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class Chunk:
    text: str
    metadata: dict
    token_count: int
    chunk_id: str

class ChunkingStrategy(ABC):
    @abstractmethod
    def chunk(self, document: str, metadata: dict) -> list[Chunk]: ...

class FixedTokenChunker(ChunkingStrategy):
    """Baseline. Good for homogeneous content."""
    def __init__(self, chunk_size=512, overlap=50):
        self.chunk_size = chunk_size
        self.overlap = overlap

class RecursiveChunker(ChunkingStrategy):
    """Respects structure: markdown headers, code blocks, paragraphs."""
    def __init__(self, separators=["\n## ", "\n### ", "\n\n", "\n", " "], chunk_size=512):
        self.separators = separators
        self.chunk_size = chunk_size

class SemanticChunker(ChunkingStrategy):
    """Uses embedding similarity to find natural boundaries."""
    def __init__(self, model="text-embedding-3-small", threshold=0.7):
        self.model = model
        self.threshold = threshold

class AgenticChunker(ChunkingStrategy):
    """LLM decides boundaries. Expensive but highest quality for complex docs."""
    def __init__(self, model="gpt-4o-mini"):
        self.model = model

Our production config by document type: 我们针对不同文档类型的生产环境配置如下：

Document Type	Strategy	Chunk Size	Overlap	Recall@10
Legal contracts	Recursive (clause-aware)	1024	100	94%
API reference	Recursive (function-aware)	768	50	96%
Support tickets	Semantic + conversation turns	512	75	91%
Internal wiki	Agentic (LLM)	1500	200	97%

文档类型	策略	分块大小	重叠	Recall@10
法律合同	递归（条款感知）	1024	100	94%
API 参考	递归（函数感知）	768	50	96%
支持工单	语义 + 对话轮次	512	75	91%
内部维基	智能体 (LLM)	1500	200	97%

---

Hybrid Retrieval: BM25 + Vector + Rerank

混合检索：BM25 + 向量 + 重排序

Pure vector search misses exact matches (error codes, function names). Pure BM25 misses semantic matches. Hybrid wins. 纯向量搜索会漏掉精确匹配（如错误代码、函数名）。纯 BM25 会漏掉语义匹配。混合检索才是赢家。

# rag/retrieval.py
class HybridRetriever:
    def __init__(self, vector_store, bm25_index, reranker, weights=(0.4, 0.3, 0.3)):
        self.vector = vector_store
        self.bm25 = bm25_index
        self.reranker = reranker
        self.weights = weights # vector, bm25, reranker

    async def retrieve(self, query: str, k=20, final_k=5):
        # Stage 1: Parallel retrieval
        vector_results = await self.vector.search(query, k=k)
        bm25_results = await self.bm25.search(query, k=k)
        
        # Stage 2: Reciprocal Rank Fusion
        fused = self._rrf(vector_results, bm25_results, k=50)
        
        # Stage 3: Cross-encoder rerank (top 50 → top 5)
        reranked = await self.reranker.rerank(query, fused[:50])
        return reranked[:final_k]

Why cross-encoder rerank? Bi-encoder (embedding) similarity ≈ 0.75 correlation with relevance. Cross-encoder ≈ 0.92. The 50→5 funnel costs 50ms but gains 15% recall. 为什么要用交叉编码器（Cross-encoder）重排序？双编码器（嵌入）相似度与相关性的相关系数约为 0.75，而交叉编码器约为 0.92。50 进 5 的漏斗虽然增加了 50ms 延迟，但召回率提升了 15%。

---

Query Transformation: Don’t Search What User Asked

查询转换：不要直接搜索用户的问题

Users ask badly. Transform first. 用户的问题通常问得很烂，先进行转换。

# rag/query_transform.py
class QueryTransformer:
    async def expand(self, query: str, context: dict = None) -> list[str]:
        """Generate multiple search queries from one user question."""
        # ... (LLM generates diverse queries covering intent)

Query expansion results: 查询扩展结果：

• Single query recall@10: 78%

• 3 expanded queries (union): 94%

• 5 expanded queries (union): 96%

• Cost: 3-5x embedding calls, but parallelizable.

• 单次查询 recall@10：78%

• 3 次扩展查询（并集）：94%

• 5 次扩展查询（并集）：96%

• 成本：3-5 倍的嵌入调用，但可以并行处理。

---

Bayesian Optimization: Stop Guessing Hyperparameters

贝叶斯优化：停止猜测超参数

chunk_size=512, top_k=5, similarity_threshold=0.7 — who chose these? We treat retrieval as a black-box function f(chunk_size, overlap, top_k, weights) → recall@10, latency and optimize with Bayesian search. chunk_size=512, top_k=5, similarity_threshold=0.7 —— 这些参数是谁选的？我们将检索视为一个黑盒函数 f(chunk_size, overlap, top_k, weights) → recall@10, latency，并使用贝叶斯搜索进行优化。

# rag/optimization.py
import optuna

def objective(trial: optuna.Trial) -> tuple[float, float]:
    config = RetrievalConfig(
        chunk_size=trial.suggest_categorical("chunk_size", [256, 512, 768, 1024, 1536]),
        overlap=trial.suggest_int("overlap", 0, 200, step=25),
        top_k=trial.suggest_int("top_k", 5, 50, step=5),
        # ...
    )
    # Evaluate on golden set (200 queries)
    recall, latency = evaluate_config(config, golden_set)
    return recall, latency / 1000 # seconds