RAG（检索增强生成）听起来简单：把文档存起来、查到相关片段、塞进提示词。但实际做出来「能演示」和「能用」之间差得很远——大部分 RAG 效果差，问题都不在模型，而在切分、召回、重排这三步。

这篇给一条能落地的完整链路，每一步都说清「为什么这么做」和「常见错法」。

一、先看清链路：五个环节，三个是坑

离线：文档 → 解析 → 切分 → 向量化 → 入库
在线：问题 → 改写 → 检索 → 重排 → 生成

五个环节里，效果损失最大的三个：
  ① 切分（chunking）——切坏了，后面全废
  ② 召回（retrieval）——只取 top-k 相似，常漏掉关键片段
  ③ 重排（rerank）——不重排，噪声片段会挤掉真正有用的

二、解析：不同格式的坑不一样

第一步就要判断 PDF 是文本型还是扫描型：用 pdfplumber 抽一页文字，如果字符数接近 0，就是扫描件，必须走 OCR。不做这一步，后面全是空向量。

三、切分：这是最影响效果的一步

3.1 不要用「固定字数」切

# 错误做法：每 500 字切一刀
def bad_split(text, size=500):
    return [text[i:i+size] for i in range(0, len(text), size)]
1. 后果：句子被拦腰切断，语义碎片化，检索到也用不上

3.2 按结构切（推荐）

# 按标题层级切，保留层级路径
import re

def split_by_heading(text, max_len=800):
    """按 markdown 标题切分，过长的段再按段落切。"""
    1. 用标题作为切分点
    parts = re.split(r'\n(?=#{1,4} )', text)
    chunks = []
    for p in parts:
        if len(p) <= max_len:
            chunks.append(p.strip())
        else:
            1. 太长就继续按空行切，再贪心合并到 max_len
            paras = [x for x in p.split('\n\n') if x.strip()]
            buf = ''
            for para in paras:
                if len(buf) + len(para) <= max_len:
                    buf += para + '\n\n'
                else:
                    if buf:
                        chunks.append(buf.strip())
                    buf = para + '\n\n'
            if buf:
                chunks.append(buf.strip())
    return [c for c in chunks if c]

3.3 三个关键的切分参数

# 关键技巧：给每个 chunk 加上「面包屑」
def with_breadcrumb(chunk, heading_path):
    1. heading_path = ["第三章", "3.2 配置", "3.2.1 数据库"]
    return f"{' > '.join(heading_path)}\n\n{chunk}"

1. 效果：用户问「怎么配置数据库」，即使正文没提「数据库」，
1.      标题路径里有，也能被召回

四、向量化与入库：模型选择与多路召回

4.1 嵌入模型怎么选

中文场景别直接用 OpenAI 的 embedding：它在中文语义上的表现不如 bge-m3，而且按量计费。自托管 bge-m3 在 4 核 CPU 上单条 20-50ms，够用。

4.2 存哪里：向量库选型

选 pgvector：起步阶段向量数不会太多，而「和业务数据同库」能省掉大量同步逻辑。等上千万条再换 Qdrant。

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
  id          BIGSERIAL PRIMARY KEY,
  doc_id      BIGINT NOT NULL,
  heading     TEXT,             -- 面包屑标题路径
  content     TEXT NOT NULL,
  embedding   vector(1024),     -- bge-m3 的维度
  tsv         tsvector,         -- 关键词检索用
  created_at  TIMESTAMPTZ DEFAULT now()
);

-- 向量索引（HNSW 比 IVFFlat 召回率更高，构建慢）
CREATE INDEX idx_chunks_vec ON chunks
  USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

-- 关键词索引（混合检索要用）
CREATE INDEX idx_chunks_tsv ON chunks USING gin (tsv);

五、检索：单一向量检索会漏，必须混合

只做向量检索的三大盲区：

• 专有名词：型号「X-2000」、错误码「E1042」这类，向量抓不住精确匹配
• 短查询：用户只输入「报错 1040」，向量语义太稀薄
• 否定与条件：「不含某字段的文档」这类，向量难以表达

5.1 混合检索：向量 + 关键词，各取一半

async def hybrid_search(conn, query: str, k: int = 20):
    """向量召回 + 关键词召回，各自取 k，再合并去重。"""
    emb = await embed(query)          # 调用 bge-m3

    vec_rows = await conn.fetch("""
        SELECT id, content, heading,
               1 - (embedding <=> $1::vector) AS score
        FROM chunks
        ORDER BY embedding <=> $1::vector
        LIMIT $2
    """, emb, k)

    kw_rows = await conn.fetch("""
        SELECT id, content, heading,
               ts_rank(tsv, plainto_tsquery('simple', $1)) AS score
        FROM chunks
        WHERE tsv @@ plainto_tsquery('simple', $1)
        ORDER BY score DESC
        LIMIT $2
    """, query, k)

    1. 合并：用 RRF（Reciprocal Rank Fusion）而不是简单加权
    merged = rrf_merge(vec_rows, kw_rows, k=60)
    return merged

5.2 RRF 合并：为什么比加权分数好

def rrf_merge(list_a, list_b, k: int = 60, top_n: int = 10):
    """RRF：按名次倒数累加，不需要归一化分数。
    score = Σ 1/(k + rank)
    """
    scores = {}
    docs = {}
    for rows in (list_a, list_b):
        for rank, r in enumerate(rows, start=1):
            did = r['id']
            scores[did] = scores.get(did, 0) + 1 / (k + rank)
            docs[did] = r
    ordered = sorted(scores.items(), key=lambda x: -x[1])[:top_n]
    return [{**docs[did], 'rrf': s} for did, s in ordered]

为什么用 RRF 而不是「向量分 × 0.7 + 关键词分 × 0.3」：两类分数的量纲完全不同（余弦相似度 0-1，ts_rank 可能是 0.01-0.5），加权需要归一化，而归一化本身会引入偏差。RRF 只看名次，天然可比。

六、重排：这一步能提升 10-20% 准确率

召回取 20 条，重排后取前 5 条喂给模型。重排用交叉编码器（cross-encoder），比向量相似度准得多：

from sentence_transformers import CrossEncoder

1. bge-reranker 系列，中文效果好
reranker = CrossEncoder('BAAI/bge-reranker-base', max_length=512)

def rerank(query: str, candidates: list[dict], top_n: int = 5):
    pairs = [(query, c['content']) for c in candidates]
    scores = reranker.predict(pairs)
    ordered = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return [{**c, 'rerank': float(s)} for c, s in ordered[:top_n]]

为什么值得花这 100ms：召回基于「独立编码」，问题和文档各自转成向量再算距离，丢失了交互信息。重排把两者拼在一起过一遍模型，能捕捉「这段是否真的回答了这个问题」。

七、生成：提示词决定「不胡说」

PROMPT = """你是资料问答助手。规则：
1. 只根据下面提供的资料回答，不要用你自己的知识补充
2. 如果资料里没有答案，直接说「资料中没有相关信息」，不要编
3. 回答时标注依据的片段编号，如 [1]

资料：
{context}

问题：{question}

回答："""

• 「资料里没有就说没有」必须明确写：不写模型一定会用训练知识补，这是 RAG 幻觉的主因
• 要求标注片段编号：既是可追溯性，也能让用户判断可信度
• 资料要带编号和来源：格式化成「[1] 来自 xxx.md：内容」，而不是裸拼文本

八、评估：没有评估就没有优化

RAG 最容易陷入「感觉好像好了一点」的循环。必须有量化指标：

# 最小可用的评估集（手工做 30 条就够起步）
EVAL = [
  {"q": "如何配置数据库连接", "gold_chunk_id": 42},
  {"q": "报错 1040 怎么解决", "gold_chunk_id": 128},
  1. ...
]

def recall_at_k(search_fn, eval_set, k=20):
    hit = 0
    for e in eval_set:
        results = search_fn(e['q'], k=k)
        if any(r['id'] == e['gold_chunk_id'] for r in results):
            hit += 1
    return hit / len(eval_set)

九、五个最常见的错法

十、总结：效果差先查这三处

• 切分：块是否语义完整？是否带标题路径？overlap 够不够？
• 召回：是否只有向量？有没有关键词兜底？k 取多大？
• 重排：有没有交叉编码器重排？还是直接取 top-k？
• 提示词：有没有明确禁止用外部知识？有没有要求标注来源？
• 评估：有没有量化指标？还是只看几个例子觉得「还行」？