AI 应用中的 RAG 实战：从向量检索、重排序到效果评估的完整落地指南

RAG（Retrieval-Augmented Generation，检索增强生成）这两年已经从“看起来很美”的概念，变成了很多 AI 应用真正落地的底层能力：企业知识库问答、客服 Copilot、代码助手、内部文档搜索、智能体工具调用前的上下文补全……几乎都绕不开它。

但我见过很多项目，一开始都把 RAG 理解得过于简单：把文档切块、做向量化、查 TopK、把结果丢给大模型。Demo 往往能跑，但一上线就暴露出问题：

• 明明知识库里有答案，模型却说“我不知道”
• 检索到了一堆“语义相近但不回答问题”的片段
• Chunk 太短，信息不完整；太长，又把噪声带进去
• 一换 Embedding 模型，效果波动很大
• 开发时觉得“看起来还行”，但没有系统评估方法，优化全靠感觉

这篇文章我想带你从工程角度走一遍完整流程：文档切分 → 向量检索 → 重排序 → Prompt 拼装 → 效果评估 → 常见坑排查。不会只讲概念，而是尽量给到可以直接跑起来的代码和验证方法。

背景与问题

先说结论：RAG 的效果，往往不是由某一个模型单点决定的，而是整个检索链路共同决定的。

一个典型 RAG 链路里，至少有 4 个关键环节：

1. 索引构建：文档怎么切、怎么存、元数据怎么设计
2. 初检索（Recall）：向量检索/关键词检索能不能把“可能有用”的内容召回
3. 重排序（Precision）：在候选结果中，把最相关的排到前面
4. 生成（Generation）：Prompt 如何组织，模型如何约束输出

很多团队的问题在于，只盯着第 4 步，也就是“大模型回答得好不好”，却忽略了前面几步。

举个很真实的例子：

用户问：“退款申请提交后多久到账？”

文档库里有三段相关内容：

• A：退款流程总述
• B：退款到账时间说明
• C：财务结算周期说明

向量检索可能把 A、C 排在前面，因为它们和“退款”“到账”都语义相关；但真正能直接回答问题的是 B。
这时候如果没有重排序，模型拿到的上下文就可能“相关但不精确”。

所以，真正可用的 RAG，不是“能搜到点东西”，而是：

• 召回率足够高：别漏掉真正答案
• 排序精度足够高：别把噪声排在前面
• 上下文足够干净：别让模型被错误内容带偏
• 评估方法足够稳定：能持续优化，而不是拍脑袋调参

前置知识与环境准备

这篇文章默认你已经了解这些基础概念：

• Python 基础开发
• 向量 Embedding 是什么
• 大模型 Prompt 基本用法
• API 调用与 .env 配置

本文示例尽量用轻量可运行方案，方便你本地快速验证。我们会用：

• sentence-transformers：做文本向量化
• faiss-cpu：做向量索引
• FlagEmbedding：做重排序（可选，但非常推荐）
• numpy / pandas：辅助处理和评估

安装依赖：

pip install sentence-transformers faiss-cpu pandas numpy scikit-learn
pip install FlagEmbedding

如果你在 macOS 或某些 Linux 环境安装 faiss-cpu 有问题，优先检查 Python 版本和架构兼容性。我当时第一次配环境时，问题不在代码，而在 wheel 包不匹配，这类坑很常见。

核心原理

1. RAG 不是“检索 + 生成”这么简单

从结构上看，一个更完整的 RAG 流程是这样的：

flowchart LR
    A[原始文档] --> B[清洗与切块]
    B --> C[向量化]
    C --> D[向量索引]
    Q[用户问题] --> E[Query 向量化]
    D --> F[召回 TopK]
    E --> F
    F --> G[重排序 Rerank]
    G --> H[拼装上下文 Prompt]
    H --> I[LLM 生成答案]
    I --> J[返回答案与引用]

其中：

• 向量检索负责“广撒网”式召回
• 重排序负责“精筛选”
• 评估负责告诉你：问题到底出在哪一步

2. 为什么单靠向量检索不够

向量检索擅长处理语义相似，但它通常有两个工程问题：

问题一：相似不等于可回答

用户问的是具体事实，但向量模型找的是整体语义接近。
所以它经常会找到“主题相关”的段落，而不是“答案命中”的段落。

问题二：向量模型对细粒度约束不敏感

比如：

• 时间范围：7 天 vs 15 天
• 条件限定：仅企业版可用
• 否定关系：不支持 / 仅支持
• 数值约束：超过 1000 条触发限流

这些信息在语义空间里有时差距没那么大，但业务里差很多。

3. 重排序为什么有效

重排序模型（Reranker）通常是 Cross-Encoder 结构。它不是先把 query 和文档分别编码，而是把 query + chunk 一起输入模型，直接判断它们的匹配程度。

你可以简单理解成：

• 向量检索：快，适合大范围召回
• 重排序：准，适合小范围精排

所以经典做法是：

1. 向量检索先取 Top 20 ~ Top 100
2. 用重排序模型再精排
3. 最终送给 LLM 的可能只保留 Top 3 ~ Top 5

这个组合在实际项目里非常常见，也通常比“只调 Embedding 模型”更划算。

4. 一个实用的两阶段检索架构

sequenceDiagram
    participant U as 用户
    participant R as 检索服务
    participant V as 向量库
    participant RR as 重排序模型
    participant L as 大模型

    U->>R: 提问
    R->>V: 向量召回 TopK
    V-->>R: 候选片段
    R->>RR: query + 候选片段
    RR-->>R: 相关性得分
    R->>L: 拼接高分片段 + Prompt
    L-->>R: 最终答案
    R-->>U: 答案 + 引用来源

实战代码（可运行）

下面我们做一个最小但完整的 Demo：

• 使用本地文档列表模拟知识库
• 用 sentence-transformers 做向量化
• 用 faiss 做向量检索
• 用 FlagEmbedding 做重排序
• 输出最终上下文与排序结果

说明：示例代码是“教学版”，重点是帮助你把链路跑通。生产环境还需要加缓存、异常处理、索引持久化、监控等。

第一步：准备示例文档

# rag_demo.py
documents = [
    {
        "id": "doc_1",
        "title": "退款规则",
        "text": "用户提交退款申请后，系统会在1个工作日内审核。审核通过后，原路退款通常在3到7个工作日内到账。"
    },
    {
        "id": "doc_2",
        "title": "发票说明",
        "text": "电子发票会在支付成功后自动开具，用户可以在订单详情页下载PDF格式发票。"
    },
    {
        "id": "doc_3",
        "title": "会员权益",
        "text": "企业版会员支持团队协作、权限控制和批量数据导出，个人版不支持团队空间。"
    },
    {
        "id": "doc_4",
        "title": "财务结算",
        "text": "平台财务结算周期为T+7，部分银行在节假日期间可能延迟处理打款。"
    },
    {
        "id": "doc_5",
        "title": "退款异常",
        "text": "如果退款超过7个工作日仍未到账，建议先检查原支付账户状态，再联系平台客服处理。"
    }
]

第二步：文档切块

真实场景中，你不会总是拿到这么短的段落。更常见的是长文档，所以我们要先切块。

这里先给一个简单的字符级切块函数：

def chunk_text(text, chunk_size=80, overlap=20):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        if end >= len(text):
            break
        start += chunk_size - overlap
    return chunks

构建 chunk 数据：

chunked_docs = []

for doc in documents:
    chunks = chunk_text(doc["text"], chunk_size=80, overlap=20)
    for idx, chunk in enumerate(chunks):
        chunked_docs.append({
            "chunk_id": f'{doc["id"]}_chunk_{idx}',
            "doc_id": doc["id"],
            "title": doc["title"],
            "text": chunk
        })

print(f"总 chunk 数: {len(chunked_docs)}")
for item in chunked_docs:
    print(item)

实战建议：
中文场景下，切块最好优先按标题、段落、列表项、句号等语义边界切，不要只按固定字符数硬切。硬切虽然简单，但很容易把关键句拆断。

第三步：向量化并建立 FAISS 索引

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer

embedding_model = SentenceTransformer("BAAI/bge-small-zh-v1.5")

texts = [item["text"] for item in chunked_docs]
embeddings = embedding_model.encode(
    texts,
    normalize_embeddings=True
).astype("float32")

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

print("向量索引已建立，向量数:", index.ntotal)

这里选择 IndexFlatIP 的原因是：

• 向量先做归一化
• 用内积检索时，结果接近余弦相似度
• Demo 阶段最简单直接

如果你的数据规模到几十万、几百万，就要考虑 IVF、HNSW、PQ 等更高性能索引结构。

第四步：实现基础检索

def retrieve(query, top_k=5):
    query_vec = embedding_model.encode(
        [query],
        normalize_embeddings=True
    ).astype("float32")

    scores, indices = index.search(query_vec, top_k)
    results = []
    for score, idx in zip(scores[0], indices[0]):
        item = chunked_docs[idx]
        results.append({
            "chunk_id": item["chunk_id"],
            "doc_id": item["doc_id"],
            "title": item["title"],
            "text": item["text"],
            "score": float(score)
        })
    return results

测试一下：

query = "退款申请提交后多久到账？"
results = retrieve(query, top_k=5)

for r in results:
    print("-" * 60)
    print("title:", r["title"])
    print("score:", round(r["score"], 4))
    print("text:", r["text"])

你通常会看到一个现象：
退款规则、退款异常、财务结算 这些都可能被召回，但排序未必完全符合“最适合回答用户问题”的顺序。

这就是重排序该出场的时候了。

第五步：加入重排序

使用 FlagEmbedding 的 reranker：

from FlagEmbedding import FlagReranker

reranker = FlagReranker(
    "BAAI/bge-reranker-base",
    use_fp16=False
)

def rerank(query, retrieved_docs, top_n=3):
    pairs = [[query, doc["text"]] for doc in retrieved_docs]
    rerank_scores = reranker.compute_score(pairs)

    merged = []
    for doc, score in zip(retrieved_docs, rerank_scores):
        item = doc.copy()
        item["rerank_score"] = float(score)
        merged.append(item)

    merged.sort(key=lambda x: x["rerank_score"], reverse=True)
    return merged[:top_n]

组合成完整检索链路：

def search(query, recall_top_k=5, final_top_n=3):
    recalled = retrieve(query, top_k=recall_top_k)
    reranked = rerank(query, recalled, top_n=final_top_n)
    return recalled, reranked

运行：

query = "退款申请提交后多久到账？"
recalled, reranked = search(query)

print("\n=== 初检索结果 ===")
for item in recalled:
    print(f'{item["title"]} | score={item["score"]:.4f} | {item["text"]}')

print("\n=== 重排序结果 ===")
for item in reranked:
    print(f'{item["title"]} | rerank_score={item["rerank_score"]:.4f} | {item["text"]}')

大概率你会看到重排序后，“退款规则”这种直接回答问题的 chunk 更靠前。

第六步：拼装给 LLM 的上下文

哪怕你暂时不接大模型，也建议先把“送入模型的上下文”打印出来，因为这一步直接决定最终回答质量。

def build_context(reranked_docs):
    context_parts = []
    for i, doc in enumerate(reranked_docs, start=1):
        context_parts.append(
            f"[资料{i}] 标题：{doc['title']}\n内容：{doc['text']}"
        )
    return "\n\n".join(context_parts)

query = "退款申请提交后多久到账？"
_, reranked = search(query)
context = build_context(reranked)

prompt = f"""你是企业知识库助手。请严格基于提供的资料回答问题。
如果资料不足，请明确说“根据现有资料无法确认”，不要编造。

{context}

用户问题：{query}
请给出简洁、准确的回答，并尽量指出依据来自哪条资料。
"""

print(prompt)

一个好的 Prompt，至少要做到三件事：

• 约束回答来源：基于资料回答
• 约束幻觉：资料不足时明确说不知道
• 保留引用：方便调试和建立信任

逐步验证清单

很多人做 RAG 卡住，不是因为不会，而是因为一次性堆太多模块，出了问题不知道查哪。我的建议是按下面顺序逐步验证：

验证 1：Chunk 是否可读

检查每个 chunk：

• 是否包含完整语义
• 是否把一句话切断
• 是否丢了标题层级信息
• 是否有大量无意义噪声（页眉页脚、版权声明、导航菜单）

验证 2：向量召回是否命中

拿 20 个典型问题，人工看 Top5/Top10 是否包含正确答案片段。

只要 Top10 里经常没有答案，问题就不在重排序，而在更前面：

• 文档切块不合理
• embedding 模型不适合该领域
• query 改写缺失
• 元数据过滤不正确

验证 3：重排序是否提升前排精度

对比：

• 仅向量检索 Top3
• 向量检索 Top20 + Rerank Top3

如果第二种方式明显更准，说明你的链路方向是对的。

验证 4：LLM 是否被上下文带偏

即使检索没问题，LLM 也可能：

• 漏看某条资料
• 过度总结
• 把多个 chunk 混成错误结论

所以必须把“最终上下文”和“模型答案”一起审查。

效果评估：不要只凭感觉

RAG 优化最怕一句话：“我觉得这次好像更准了。”

工程上，至少要有一套小规模但稳定的评估集。
最简单的方式是构造一个问答对数据集：

eval_data = [
    {
        "question": "退款申请提交后多久到账？",
        "gold_doc_id": "doc_1"
    },
    {
        "question": "退款超过7个工作日没到账怎么办？",
        "gold_doc_id": "doc_5"
    },
    {
        "question": "个人版是否支持团队空间？",
        "gold_doc_id": "doc_3"
    }
]

评估指标 1：Recall@K

看正确文档是否出现在前 K 条结果中。

def evaluate_recall_at_k(eval_data, k=5):
    hit = 0
    for item in eval_data:
        results = retrieve(item["question"], top_k=k)
        doc_ids = [r["doc_id"] for r in results]
        if item["gold_doc_id"] in doc_ids:
            hit += 1
    return hit / len(eval_data)

print("Recall@3:", evaluate_recall_at_k(eval_data, k=3))
print("Recall@5:", evaluate_recall_at_k(eval_data, k=5))

评估指标 2：MRR（Mean Reciprocal Rank）

看正确答案排得有多靠前。

def evaluate_mrr(eval_data, k=10):
    total_score = 0.0
    for item in eval_data:
        results = retrieve(item["question"], top_k=k)
        rank = 0
        for idx, r in enumerate(results, start=1):
            if r["doc_id"] == item["gold_doc_id"]:
                rank = idx
                break
        if rank > 0:
            total_score += 1.0 / rank
    return total_score / len(eval_data)

print("MRR@10:", evaluate_mrr(eval_data, k=10))

评估指标 3：重排序后 Top1 命中率

def evaluate_rerank_top1(eval_data, recall_k=10):
    hit = 0
    for item in eval_data:
        recalled = retrieve(item["question"], top_k=recall_k)
        reranked = rerank(item["question"], recalled, top_n=1)
        if reranked and reranked[0]["doc_id"] == item["gold_doc_id"]:
            hit += 1
    return hit / len(eval_data)

print("Rerank Top1 Accuracy:", evaluate_rerank_top1(eval_data, recall_k=5))

如果你想再进一步，可以把评估拆成两层：

• Retrieval 指标：看能不能找到
• Generation 指标：看回答是否忠于资料、是否正确完整

评估链路图

flowchart TD
    A[评估样本 question + gold] --> B[向量召回]
    B --> C{TopK中是否包含正确文档}
    C -->|是| D[计算 Recall@K]
    C -->|否| E[定位召回问题]
    B --> F[重排序]
    F --> G{Top1是否命中}
    G -->|是| H[计算 Top1 Accuracy]
    G -->|否| I[定位排序问题]
    F --> J[送入 LLM]
    J --> K[人工或自动评估答案质量]

常见坑与排查

这一节我尽量讲得“像现场”，因为这些坑真的很常见。

坑 1：Chunk 太碎，答案被拆散

现象：

• 检索结果看起来相关
• 但每个片段都只包含半句话
• LLM 回答不完整，甚至答错

原因：

• 固定长度切块太短
• overlap 太小
• 没按语义边界切分

排查方法：

• 直接打印命中的 chunk
• 看“正确答案是否完整存在于单个 chunk 中”

建议：

• 中文知识库常用 300~800 字符起步试验
• overlap 可先从 50~150 字符尝试
• 优先保留标题层级、段落边界

坑 2：召回很多“相关废话”

现象：

• TopK 里都是同主题内容
• 但真正回答问题的片段排不靠前

原因：

• 向量检索只能保证语义近，不保证答案最优
• 文档里“背景介绍”比“具体规则”更常见

排查方法：

• 对同一个 query 打印 Top10
• 看是否“主题相关但无法作答”的 chunk 太多

建议：

• 加入 reranker
• 减少模板性、介绍性内容的权重
• 保留更丰富元数据，比如文档类型、章节标题、更新时间

坑 3：文档预处理带入噪声

现象：

• 命中 chunk 包含导航栏、页脚、版权声明、重复标题
• 相似检索总被这些文本污染

原因：

• PDF/HTML 抽取后没清洗
• 重复内容被反复索引

排查方法：

• 随机抽查 50 个 chunk
• 统计重复文本比例

建议：

• 清洗无意义模板文本
• 去重
• 对标题、正文、表格分开处理

坑 4：只测少量 query，误以为效果很好

现象：

• Demo 漂亮
• 一上真实用户问题，命中率明显下降

原因：

• 测试问题太“标准”
• 没覆盖口语化、错别字、缩写、业务别称

排查方法：

评估集要覆盖：

• 标准问法
• 口语问法
• 不完整问法
• 含歧义问法
• 带过滤条件的问题

建议：

• 至少先做 50~100 条代表性问题
• 每次改参数都固定跑评估

坑 5：重排序变准了，但变慢了

现象：

• 质量提升明显
• 但接口延迟变高

原因：

• Cross-Encoder 本来就比向量检索重
• recall_top_k 设太大

排查方法：

记录每一阶段耗时：

• embedding
• 检索
• rerank
• 生成

建议：

• 先用向量召回 Top20
• rerank 后保留 Top3~5
• 高频问题加缓存
• 对低价值问题可降级：只走向量检索

安全/性能最佳实践

RAG 不只是“效果工程”，也是“系统工程”。下面这些建议，我认为在生产环境里非常关键。

1. 给回答加来源引用

不要只返回答案，最好返回：

• 文档标题
• 文档 ID
• chunk 内容摘要
• 命中分数

这样做有三个好处：

• 方便用户确认可信度
• 方便开发排查问题
• 方便做反馈闭环

2. 对“无答案”场景做显式兜底

很多系统最大的问题，不是答错，而是一本正经地编错。

建议设置两层兜底：

• 检索得分过低：直接提示“知识库中未找到足够依据”
• LLM Prompt 中显式要求：资料不足时不得编造

3. 元数据过滤要前置

如果知识库有租户、部门、时间、权限等维度，一定要在检索前或检索时做过滤。

典型例子：

• A 部门不该搜到 B 部门文档
• 下线制度不能和现行制度混在一起
• 海外区和国内区规则不能串

很多安全问题，不是模型“泄密”，而是检索阶段就把不该给的数据拿出来了。

4. 检索与生成分开监控

建议至少记录这些指标：

• 检索耗时
• rerank 耗时
• LLM 耗时
• Recall@K
• Top1 命中率
• 无答案率
• 用户追问率 / 人工转接率

一旦线上效果下降，你才能快速判断是：

• 知识库变了
• embedding 模型变了
• reranker 出问题了
• LLM Prompt 被改坏了

5. 为性能设计分层策略

对于不同业务场景，可以做分层处理：

6. 混合检索通常比纯向量更稳

如果你的场景里有很多关键词、编号、产品名、API 名称、错误码，建议考虑：

• BM25 / 关键词检索
• 向量检索
• 混合召回后再重排序

比如用户搜：

• ERR_429
• API v2 billing
• SLA 99.9

这类 query 纯向量未必稳定，但关键词检索很有优势。

一个更完整的工程分层视角

classDiagram
    class DocumentProcessor {
        +clean()
        +split()
        +attach_metadata()
    }

    class Retriever {
        +embed_query()
        +vector_search()
        +keyword_search()
    }

    class Reranker {
        +score_pairs()
        +sort_candidates()
    }

    class Generator {
        +build_prompt()
        +generate_answer()
        +add_citations()
    }

    class Evaluator {
        +calc_recall()
        +calc_mrr()
        +judge_answer()
    }

    DocumentProcessor --> Retriever
    Retriever --> Reranker
    Reranker --> Generator
    Generator --> Evaluator

这张图想表达的核心是：
RAG 不该被写成一个“大函数”，而应该拆成可替换、可评估的模块。

这样你后面换 Embedding 模型、换向量库、换 reranker、换 LLM，都不会牵一发动全身。

实战建议：参数怎么起步

如果你现在准备做一个中小规模中文知识库，我会建议先从这组参数起步：

• chunk_size：300~600 字符
• overlap：50~100 字符
• embedding 模型：中文通用模型先用 bge-small-zh-v1.5 或同级别模型
• recall_top_k：10~20
• rerank_top_n：3~5
• 索引类型：数据量小先 Flat，数据量大再上 ANN
• 评估集规模：最少 50 条，最好 100 条以上

边界条件也要明确：

• 如果文档极短、结构规整，切块策略可以简单些
• 如果文档是合同、制度、FAQ 混合体，必须做更细的清洗和元数据设计
• 如果问题要求非常高的事实准确率，必须加引用和拒答机制

总结

如果把 RAG 落地这件事说得再直白一点，我的经验是：

先保证“找得到”，再优化“排得准”，最后才是“答得好”。

一个靠谱的 RAG 系统，建议按这个顺序建设：

1. 把文档切对：语义完整、噪声少、元数据齐全
2. 把召回做稳：先看 Recall@K，别一上来只看最终答案
3. 把重排序加上：它通常是“低成本高回报”的精度增强手段
4. 把 Prompt 收紧：明确要求基于资料回答，资料不足要拒答
5. 把评估建起来：固定评估集，持续跑指标
6. 把监控补齐：知道慢在哪、错在哪、退化在哪

如果你现在已经有一个“能跑”的 RAG Demo，我建议你下一步不要急着换更大的模型，而是先做这三件最有性价比的事：

• 抽样检查 50 个 chunk，看切分是否合理
• 做 30~50 条问题评估集，先算 Recall@K
• 在向量召回后加一层 reranker，再对比 Top1 命中率

很多时候，效果提升并不来自“更贵的模型”，而来自更扎实的检索工程。

希望这篇文章能帮你把 RAG 从“概念会了”推进到“真的能落地”。如果你已经开始做企业知识库或 AI 搜索，这套链路基本就是后续迭代的主干。