AI 应用中 RAG 检索增强生成的中级实战：从向量库选型到召回效果优化

很多团队做 RAG，第一阶段都很顺：文档切块、做 embedding、塞进向量库、把 top-k 结果丢给大模型，Demo 很快就能跑起来。

但一到真实业务场景，问题就开始出现：

• 用户问得稍微绕一点，召回就偏了
• 明明知识库里有答案，却没被检索出来
• 向量库换了一个，速度上去了，效果却不稳定
• chunk 切得越细越“聪明”，实际上上下文反而碎了
• top-k 拉太多，模型看花眼；拉太少，又漏关键信息

这篇文章不讲“RAG 是什么”的入门概念，而是从中级实战视角，带你把一套 RAG 系统从“能用”推进到“更准、更稳、更可调”。重点放在两件事上：

1. 向量库怎么选
2. 召回效果怎么优化

我会尽量按工程实践来讲，也会穿插一些我自己踩过的坑。

前置知识与环境准备

如果你已经做过基础版 RAG，这部分会很轻松。建议具备：

• 知道 embedding / chunk / top-k 的基本概念
• 会用 Python
• 理解“向量召回不等于最终答案”

本文示例使用：

• Python 3.10+
• sentence-transformers
• faiss-cpu
• rank-bm25
• numpy

安装依赖：

pip install sentence-transformers faiss-cpu rank-bm25 numpy

背景与问题

RAG 的核心价值，是让大模型回答时能“看见”你的私有知识，而不是只靠预训练记忆。

但很多项目里，真正的瓶颈并不在生成，而在召回阶段。因为：

• 召回错了，生成再强也没用
• 大模型擅长“组织语言”，不擅长“凭空找事实”
• 检索质量往往直接决定最终可用率

一个典型链路大概是这样的：

flowchart LR
    A[用户问题] --> B[Query 改写/清洗]
    B --> C[Embedding 编码]
    C --> D[向量检索]
    B --> E[关键词检索]
    D --> F[召回结果合并]
    E --> F
    F --> G[重排 Rerank]
    G --> H[构造 Prompt]
    H --> I[LLM 生成答案]

这里最容易被低估的是中间这几步：

• Query 是否需要改写
• 只做向量检索是否足够
• 是否需要混合检索
• top-k 应该是多少
• 是否需要 rerank
• chunk 应该怎么切

很多“模型答非所问”的根因，其实不是模型，而是召回链路设计太粗糙。

核心原理

这一部分，我们不追求“百科全书式”定义，而是聚焦会影响实战效果的几个关键点。

1. RAG 的本质：先找证据，再让模型作答

RAG 不是“让模型变聪明”，而是“让模型在作答前先查资料”。

可简化成两个阶段：

1. Retrieval：从知识库中召回候选内容
2. Generation：把召回结果作为上下文交给 LLM 生成答案

sequenceDiagram
    participant U as 用户
    participant R as 检索层
    participant V as 向量库/索引
    participant L as 大模型

    U->>R: 提问
    R->>V: 检索相关片段
    V-->>R: 返回候选 chunk
    R->>R: 过滤/重排/拼接上下文
    R->>L: 问题 + 上下文
    L-->>U: 基于证据生成答案

如果检索层召回的是“看起来相近但事实不对”的内容，LLM 往往会一本正经地答错。这也是为什么 RAG 评估里，召回率、MRR、nDCG 这类指标很关键。

2. 向量库到底解决什么问题

向量库本质上解决的是：在大量高维向量里，快速找到与 query 最相近的那些。

你可以把每个 chunk 和用户 query 都编码成向量，然后按距离找近邻。常见度量：

• 余弦相似度
• 内积
• 欧式距离

工程上更重要的是以下维度：

• 数据规模：几万、几百万、几亿条
• 写入频率：离线批量导入还是实时更新
• 延迟要求：几十毫秒还是秒级
• 过滤能力：是否需要 metadata filter
• 部署方式：本地嵌入、云托管、混合架构

3. 向量库选型：不要只看“谁最火”

很多人一上来就问：“Milvus、Qdrant、Weaviate、PGVector、FAISS 选哪个？”

我的经验是，先看约束，再看功能。下面给一个偏实战的判断框架。

FAISS

适合：

• 本地实验
• 离线批处理
• 单机高性能检索
• 你想完全掌控索引结构

优点：

• 快
• 成熟
• 可选多种 ANN 索引结构

不足：

• 不像完整数据库那样自带丰富的元数据管理和服务能力
• 分布式、在线更新、权限隔离等要自己补

PGVector

适合：

• 已经深度使用 PostgreSQL
• 数据规模中小
• 需要强事务与 SQL 生态

优点：

• 学习成本低
• 元数据过滤方便
• 运维体系统一

不足：

• 极大规模和极致性能场景不如专业向量库

Milvus / Qdrant / Weaviate

适合：

• 中大型线上应用
• 需要更完整的向量数据库能力
• 有 metadata filter、混合检索、集群化需求

优点：

• 检索能力完整
• 工程配套好
• 更适合生产环境

不足：

• 运维复杂度高于本地 FAISS
• 选型不当容易“功能用不上、成本先上来”

一个简单选择建议

• PoC / 本地验证：FAISS
• 已有 Postgres，数据量不大：PGVector
• 正式线上，数据规模中大，有复杂过滤和扩展需求：Qdrant / Milvus / Weaviate

边界条件也要说清楚：
如果你的知识库只有几千条 FAQ，却上来就搞分布式向量集群，多半是过度设计。

4. 召回效果差，通常不是单一问题

召回效果的影响因素，通常至少有这几层：

classDiagram
    class Query{
      用户表达方式
      歧义词/简称
      多轮上下文
    }
    class Chunk{
      切块大小
      重叠策略
      标题保留
    }
    class Embedding{
      模型领域适配
      维度
      多语言能力
    }
    class Retrieval{
      top-k
      相似度阈值
      metadata过滤
      混合检索
    }
    class Rerank{
      交叉编码器
      规则重排
    }

    Query --> Retrieval
    Chunk --> Embedding
    Embedding --> Retrieval
    Retrieval --> Rerank

也就是说，别一看召回不准就先怪向量库。很多时候问题在：

• chunk 切得不合理
• embedding 模型不匹配领域
• query 预处理太弱
• 没做 hybrid search
• top-k 与阈值配置不合理
• 缺少 rerank

从 0 到 1 搭一套可运行的中级 RAG 检索层

下面做一个可以本地运行的小型示例。重点演示：

• 文档切块
• 向量索引
• BM25 检索
• 混合召回
• 简单重排

这不是完整生产系统，但很适合作为调优基线。

实战代码（可运行）

1. 准备示例数据与切块

from dataclasses import dataclass
from typing import List, Dict
import re

@dataclass
class Chunk:
    id: str
    text: str
    metadata: Dict

documents = [
    {
        "doc_id": "doc1",
        "title": "RAG 系统中的 Chunk 切分",
        "content": """
        在 RAG 系统中，切块大小会显著影响召回效果。
        如果 chunk 过小，语义上下文可能丢失；如果 chunk 过大，则会引入噪声。
        实践中常见做法是按段落切分，并设置适度 overlap。
        标题信息建议保留到 chunk 中，这对检索很有帮助。
        """
    },
    {
        "doc_id": "doc2",
        "title": "向量数据库选型建议",
        "content": """
        FAISS 适合本地原型验证和单机高性能检索。
        Qdrant 和 Milvus 更适合线上生产环境，支持更丰富的过滤和服务能力。
        如果团队已经广泛使用 PostgreSQL，PGVector 也是可行方案。
        选型时要关注数据规模、写入频率、延迟目标和过滤需求。
        """
    },
    {
        "doc_id": "doc3",
        "title": "召回优化方法",
        "content": """
        召回优化通常包括查询改写、混合检索、重排和阈值调优。
        单纯依赖向量检索在关键词强约束场景下效果未必最好。
        将 BM25 与向量检索结合，往往能提升首条命中率。
        对于排序靠前但不够精确的候选结果，可引入 reranker 做二次排序。
        """
    },
]

def split_text(text: str, max_len: int = 80, overlap: int = 20) -> List[str]:
    text = re.sub(r"\s+", " ", text).strip()
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + max_len, len(text))
        chunk = text[start:end]
        chunks.append(chunk)
        if end == len(text):
            break
        start = end - overlap
    return chunks

chunks: List[Chunk] = []
for doc in documents:
    parts = split_text(doc["title"] + "。 " + doc["content"], max_len=80, overlap=20)
    for i, part in enumerate(parts):
        chunks.append(
            Chunk(
                id=f'{doc["doc_id"]}_chunk_{i}',
                text=part,
                metadata={"doc_id": doc["doc_id"], "title": doc["title"]}
            )
        )

print(f"chunk 数量: {len(chunks)}")
for c in chunks:
    print(c.id, c.text)

2. 建立向量索引

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")

texts = [c.text for c in chunks]
embeddings = model.encode(texts, normalize_embeddings=True)
embeddings = np.array(embeddings).astype("float32")

dim = embeddings.shape[1]
index = faiss.IndexFlatIP(dim)  # 归一化后可用内积近似余弦相似度
index.add(embeddings)

print("向量索引构建完成，向量数：", index.ntotal)

3. 建立 BM25 索引

from rank_bm25 import BM25Okapi

def tokenize(text: str):
    # 简化版分词：中文场景生产中建议接入更合理的 tokenizer
    return list(text)

tokenized_corpus = [tokenize(c.text) for c in chunks]
bm25 = BM25Okapi(tokenized_corpus)

这里我专门提醒一句：
中文检索不要轻视分词问题。
示例里为了可运行用了“按字切分”，但在线上环境，最好接入更符合业务语料的中文分词或搜索分析器。否则 BM25 效果会被低估。

4. 实现向量检索、BM25 检索与混合召回

def vector_search(query: str, top_k: int = 5):
    q_emb = model.encode([query], normalize_embeddings=True)
    q_emb = np.array(q_emb).astype("float32")
    scores, indices = index.search(q_emb, top_k)

    results = []
    for score, idx in zip(scores[0], indices[0]):
        results.append({
            "chunk": chunks[idx],
            "score": float(score),
            "source": "vector"
        })
    return results

def bm25_search(query: str, top_k: int = 5):
    scores = bm25.get_scores(tokenize(query))
    top_indices = np.argsort(scores)[::-1][:top_k]

    results = []
    for idx in top_indices:
        results.append({
            "chunk": chunks[idx],
            "score": float(scores[idx]),
            "source": "bm25"
        })
    return results

def hybrid_search(query: str, top_k: int = 5, alpha: float = 0.6):
    v_res = vector_search(query, top_k=top_k * 2)
    b_res = bm25_search(query, top_k=top_k * 2)

    merged = {}

    def normalize(scores):
        if not scores:
            return []
        s_min, s_max = min(scores), max(scores)
        if s_max == s_min:
            return [1.0 for _ in scores]
        return [(s - s_min) / (s_max - s_min) for s in scores]

    v_scores = normalize([x["score"] for x in v_res])
    b_scores = normalize([x["score"] for x in b_res])

    for item, ns in zip(v_res, v_scores):
        cid = item["chunk"].id
        merged.setdefault(cid, {"chunk": item["chunk"], "vector_score": 0.0, "bm25_score": 0.0})
        merged[cid]["vector_score"] = ns

    for item, ns in zip(b_res, b_scores):
        cid = item["chunk"].id
        merged.setdefault(cid, {"chunk": item["chunk"], "vector_score": 0.0, "bm25_score": 0.0})
        merged[cid]["bm25_score"] = ns

    final_results = []
    for cid, item in merged.items():
        final_score = alpha * item["vector_score"] + (1 - alpha) * item["bm25_score"]
        final_results.append({
            "chunk": item["chunk"],
            "score": final_score,
            "vector_score": item["vector_score"],
            "bm25_score": item["bm25_score"],
        })

    final_results.sort(key=lambda x: x["score"], reverse=True)
    return final_results[:top_k]

5. 加一个轻量重排

生产里常用 cross-encoder reranker，这里为了示例可运行，我们先用一个“规则增强版重排”：

• 如果 chunk 中直接包含 query 的关键词，额外加分
• 标题命中也额外加分

def simple_rerank(query: str, candidates: List[Dict], top_k: int = 3):
    keywords = [w for w in re.findall(r"[\u4e00-\u9fa5A-Za-z0-9]+", query) if len(w) >= 2]
    reranked = []

    for item in candidates:
        text = item["chunk"].text
        title = item["chunk"].metadata.get("title", "")
        bonus = 0.0

        for kw in keywords:
            if kw in text:
                bonus += 0.15
            if kw in title:
                bonus += 0.1

        reranked.append({
            **item,
            "final_score": item["score"] + bonus
        })

    reranked.sort(key=lambda x: x["final_score"], reverse=True)
    return reranked[:top_k]

6. 跑一个完整检索流程

def retrieve(query: str):
    print(f"\n=== Query: {query} ===")

    hybrid = hybrid_search(query, top_k=5, alpha=0.6)
    reranked = simple_rerank(query, hybrid, top_k=3)

    for i, item in enumerate(reranked, 1):
        chunk = item["chunk"]
        print(f"\n[{i}] {chunk.id}")
        print(f"title: {chunk.metadata['title']}")
        print(f"score={item['score']:.4f}, final_score={item['final_score']:.4f}")
        print(chunk.text)

retrieve("RAG 里 chunk 太小会有什么问题？")
retrieve("向量数据库怎么选？")
retrieve("为什么要混合检索和重排？")

如果一切正常，你会看到结果大致集中到对应主题上，而且相较于单一向量检索，混合召回在关键词强约束问题上更稳。

逐步验证清单

很多人调 RAG 时容易“一次改三件事”，最后根本不知道哪一步起了作用。更建议按下面顺序验证：

第一步：只看向量检索

检查：

• top-5 是否至少有 2~3 条明显相关
• 是否存在“语义相近但主题不对”的误召回

第二步：只看 BM25

检查：

• 精确关键词问题是否优于向量检索
• 同义表达是否明显变差

第三步：做混合召回

检查：

• 首条命中率是否提升
• 是否减少“有答案但没召回”的情况

第四步：加 rerank

检查：

• top-1 是否比 top-5 更可靠
• 无关结果是否被压下去

第五步：调 chunk 策略

检查：

• 更大 chunk 是否提高上下文完整性
• 更小 chunk 是否提高局部匹配精度
• overlap 是否减少边界截断问题

召回优化的几个高收益动作

这一节是实战里最值钱的部分。

1. Chunk 不是越小越好，也不是越大越好

一个常见误区是：“切得越细，匹配越精准。”
实际上过细的 chunk 会造成：

• 标题与正文分离
• 关键定义被截断
• 上下文丢失
• rerank 时缺少完整证据

我的建议：

• FAQ/短知识：chunk 可小一些
• 规章制度/技术文档：按标题+段落切
• 长篇说明文：保留 10%~20% overlap

如果文档天然有层级结构，优先保留：

• 文档标题
• 小节标题
• 段落内容
• 来源 URL / 时间 / 版本号

2. Embedding 模型要匹配语言与领域

通用 embedding 模型不一定适合你的业务。

比如：

• 医疗、法律、金融术语多的场景
• 中英混合文档
• 缩写和产品名很多的企业知识库

这时要重点测试：

• 多语言能力
• 专业术语区分能力
• 短 query 与长段落的对齐效果

我实际做项目时，碰到过一个问题：
通用模型对“审批流”“工作流”“流程编排”这几个词拉得太近，导致检索混淆。换了更适配领域语料的 embedding 后，误召回明显下降。

3. 混合检索通常比单路检索更稳

向量检索擅长：

• 语义匹配
• 同义表达
• 自然语言问法

BM25 擅长：

• 精确关键词
• 术语、版本号、报错码
• 缩写、命令、函数名

所以在很多企业知识库里，hybrid search 几乎是默认选项。尤其当用户会问：

• “报错 E203 怎么处理”
• “v2.3 API 限流规则”
• “pgvector 和 faiss 区别”

这种问题里，关键词是很强的信号，不能只靠向量相似度。

4. top-k 不要盲目调大

很多人觉得“召回不准，那我就多拿点”。
这招有时有效，但副作用也明显：

• 噪声变多
• Prompt 变长
• 模型注意力分散
• 成本上升，延迟变高

比较实用的做法是：

• 检索阶段：先取较大的候选集，比如 top-20
• 重排阶段：压缩到 top-5 或 top-8
• 最终喂给 LLM：保留 top-3 到 top-5

这通常比“直接 top-10 全塞给模型”效果更稳定。

5. Query 改写往往是低成本高回报

用户提问可能很口语：

• “这个库到底咋选”
• “为啥总查不准”
• “多文档拼起来会乱吗”

你可以在检索前做轻量改写，例如：

• 补全主语
• 提取关键词
• 展开缩写
• 把口语改写成检索友好表达

例如：

• 原 query：这个库到底咋选
• 改写后：RAG 向量数据库选型标准，包括 FAISS、Qdrant、Milvus、PGVector 的适用场景

不要小看这一步，它常常能明显改善召回稳定性。

常见坑与排查

下面这些问题，我基本都见过。

坑 1：明明知识库里有答案，却检索不到

排查顺序建议：

1. 看 chunk 是否切碎了
2. 看 embedding 是否正常生成
3. 看 query 是否太口语或太短
4. 看 top-k 是否过小
5. 看 metadata filter 是否误过滤
6. 看相似度阈值是否设太高

一个很实用的方法：
把“正确答案所在 chunk”拿出来，直接和 query 算相似度。如果相似度本身就不高，多半不是向量库问题，而是 embedding / chunk / query 表达的问题。

坑 2：召回结果都“差不多相关”，但 top-1 总错

这通常是排序问题，不是召回问题。

解决思路：

• 加 reranker
• 标题命中加权
• 文档新鲜度加权
• 对 FAQ 类知识做规则置顶
• 把文档级与 chunk 级打分结合起来

很多业务里，top-5 里已经有答案，但 top-1 不稳定。此时最值得投资源的往往是重排，而不是继续换向量库。

坑 3：离线评估很好，线上用户还是说不好用

常见原因：

• 离线测试问题集太干净
• 用户真实 query 更口语、更省略
• 多轮上下文没拼进去
• 检索结果虽然相关，但不够“可回答”

建议你至少准备三类评估集：

• 标准问法
• 口语问法
• 模糊/歧义问法

如果只在“理想题库”上测，线上体验往往会打折。

坑 4：索引更新后效果波动很大

要检查：

• 是否混用了不同 embedding 模型
• 旧向量是否未清理干净
• chunk 规则是否变了
• 文档版本是否重复入库
• 归一化方式是否一致

这是个非常典型的线上事故来源。
同一个索引里的向量，最好来自同一套 embedding 模型与同一套预处理流程。

坑 5：中文检索效果不稳定

重点看：

• 分词质量
• 标点与全半角处理
• 简繁体统一
• 英文术语和中文混排
• 数字、版本号、错误码保留策略

中文 RAG 很少是“直接上英文教程里的默认设置就完事”。特别是 BM25、关键词检索、日志报错检索这类场景，文本规范化非常关键。

安全/性能最佳实践

RAG 项目一旦上线，除了效果，还要考虑安全和性能。

安全方面

1. 做好知识源权限隔离

如果知识库里有不同角色可见的数据，检索阶段就必须带权限过滤。
不要先召回，再指望生成阶段“别答出来”。那太晚了。

2. 防止 Prompt 注入污染

知识文档本身可能包含恶意内容，比如：

• “忽略上文”
• “输出系统提示词”
• “泄露内部规则”

对外部来源文档要做清洗，对 Prompt 构造要有明确边界，例如：

• 把检索内容当作“参考资料”而不是“指令”
• 系统提示中明确要求模型不能执行资料中的命令性文本

3. 敏感信息脱敏

入库前尽量处理：

• 手机号
• 身份证号
• 邮箱
• 密钥、Token、数据库连接串

不要让“检索做得太好”反而放大数据泄露风险。

性能方面

1. 建立分层索引策略

常见做法：

• 热门知识走缓存
• 高频 FAQ 走规则直出
• 长尾问题走向量召回
• 大规模数据按租户/业务线分片

这能显著降低延迟和成本。

2. 先粗召回，再精排

标准套路通常是：

1. ANN 快速召回 top-50
2. 混合召回合并候选
3. rerank 压到 top-5
4. 再进入生成

这比“所有候选都精排”更现实。

3. 控制上下文长度

不是所有召回结果都值得进 Prompt。
建议对每个 chunk 做：

• 去重
• 截断
• 按来源合并
• 保留标题与出处

否则上下文过长，成本和幻觉都会上升。

4. 做检索日志与可观测性

至少记录：

• 原始 query
• 改写后的 query
• 召回结果 ID
• 各路打分
• 最终 answer
• 用户反馈

没有这些日志，后面优化基本靠猜。

一个更接近生产的调优思路

如果你现在手上已经有一个“能跑”的 RAG，我建议按下面顺序优化，不容易走弯路：

stateDiagram-v2
    [*] --> 跑通基础RAG
    跑通基础RAG --> 建立评测集
    建立评测集 --> 优化Chunk策略
    优化Chunk策略 --> 测试Embedding模型
    测试Embedding模型 --> 接入混合检索
    接入混合检索 --> 增加Rerank
    增加Rerank --> 加入Query改写
    加入Query改写 --> 线上A/B验证
    线上A/B验证 --> [*]

这个顺序的好处是：

• 先抓大头，再抠细节
• 每一步都能被验证
• 不会把“效果问题”和“系统复杂度”混在一起

向量库选型的实战建议清单

最后把选型这件事收敛成一份可执行清单。

如果你在做原型验证

优先考虑：

• FAISS
• 本地 embedding
• 小规模数据集
• 简单 metadata 管理

目标不是“一步到位”，而是尽快验证：

• chunk 策略
• embedding 模型
• 混合检索是否有收益

如果你准备上线

重点看：

• 支持 metadata filter 吗
• 支持多租户隔离吗
• 更新延迟可接受吗
• 索引重建成本高吗
• 是否有成熟监控与备份方案

如果你已经在线上但效果不稳定

不要急着换库，先检查：

• chunk 设计
• embedding 模型
• hybrid search
• rerank
• query 改写
• 日志与评估集是否完善

很多时候，真正提升效果的不是“换最强库”，而是把召回链路补完整。

总结

RAG 做到中级阶段，关注点就不再是“能不能检索”，而是：

• 能不能稳定召回正确证据
• 能不能在速度、成本、效果之间取得平衡
• 能不能在业务变化时持续调优

如果你只记住三件事，我建议是：

1. 向量库选型要看场景，不要只看流行度
   PoC 用 FAISS 很合适，生产环境再考虑更完整的向量数据库能力。

2. 召回优化是系统工程，不是单点调参
   chunk、embedding、hybrid search、rerank、query 改写，往往要配合起来看。

3. 先建立评估与日志，再谈优化
   没有可观测性，你很难知道问题出在检索、排序还是生成。

如果你现在的 RAG 已经“能跑”，下一步最值得投入的通常不是把链路搞得更复杂，而是先把下面这套最小闭环做好：

• 一套靠谱的 chunk 策略
• 一个适配业务的 embedding 模型
• 混合检索
• 简单 rerank
• 检索日志和离线评测集

做到这一步，你的 RAG 往往就会从“偶尔答对”进入“多数情况下答得准而且稳定”的阶段。