大模型应用中的 RAG 实战：从向量检索、重排序到效果评估的完整落地指南
RAG（Retrieval-Augmented Generation，检索增强生成）这两年几乎成了大模型应用落地的“标配”。但真正做起来，很多团队会发现：不是接个向量库、塞点文档、把结果拼给模型就结束了。
实际项目里，经常会遇到这些问题：
- 检索到了“看起来相关”的内容，但答案还是不准
- 数据明明都在知识库里，模型却答非所问
- 相似度分数很高，但返回的文档块并不能直接支持答案
- 上线前 demo 很好，一到真实业务流量就开始翻车
- 没有评估体系，优化全靠“感觉”
这篇文章我会从一个工程落地的角度，带你把 RAG 的关键链路走一遍：数据切分 → 向量检索 → 重排序 → 生成 → 效果评估。我会尽量避免只讲概念，而是用一套可运行的 Python 示例带你搭起来，再讲清楚每一步为什么这样做、容易踩什么坑。
背景与问题
为什么大模型单独使用不够
通用大模型擅长语言理解和生成，但它有几个天然限制：
-
知识时效性有限
模型参数不是实时更新的，新文档、新规则、新产品说明它并不知道。 -
业务知识不在参数里
企业内部文档、FAQ、流程规范、代码库、工单记录，不会天然出现在基础模型中。 -
幻觉无法完全避免
当模型不确定时，它仍然可能生成“像真的一样”的答案。
RAG 的目标，就是把“模型会说话”和“系统能查资料”结合起来：
先查，再答；答的时候尽量基于证据。
一个常见误区：RAG 的瓶颈不在“生成”，而在“检索”
很多人一开始会把注意力放在 prompt 怎么写、模型怎么换，但实际经验是：
如果检索阶段拿不到对的材料，后面 prompt 再花哨也救不回来。
所以一个靠谱的 RAG 系统，核心不是“大模型调用成功”，而是：
- 文档是否切得合理
- 检索召回是否足够
- 重排序是否能把真正有用的片段排到前面
- 评估体系是否能量化效果变化
前置知识与环境准备
你需要知道什么
如果你已经了解以下内容，阅读会更顺：
- Python 基础
- 向量检索的基本概念：embedding、top-k、相似度
- 大模型 API 的基本调用方式
- Markdown / JSON 的简单处理
环境准备
下面示例尽量做到本地可运行，使用的依赖比较克制：
pip install numpy scikit-learn rank-bm25 sentence-transformers如果你本机没有安装 PyTorch，sentence-transformers 首次安装会稍大一些，这是正常现象。
核心原理
先把整体链路建立起来。一个实用的 RAG 系统通常包含 5 个步骤：
- 文档预处理与切块
- 向量化并建立索引
- 召回候选文档
- 重排序
- 将上下文交给大模型生成答案
下面这张图可以先帮助你形成整体认知。
flowchart LR
A[原始文档] --> B[清洗与切块]
B --> C[Embedding 向量化]
C --> D[向量索引]
Q[用户问题] --> E[问题向量化]
E --> F[向量召回 TopK]
B --> G[BM25 关键词召回]
F --> H[候选集合合并]
G --> H
H --> I[重排序 Rerank]
I --> J[上下文拼接]
J --> K[LLM 生成答案]
K --> L[返回答案 + 引用]1. 为什么要切块，而不是整篇文档直接入库
因为大模型回答问题时，需要的是与问题最相关的一小段证据，而不是一整篇几十页的文档。
切块太大，会出现：
- 一个 chunk 里话题太杂，embedding 语义被“摊薄”
- 检索命中后，传给模型的噪声太多
- 上下文窗口浪费
切块太小，也会出现：
- 语义断裂
- 上下文不完整
- 明明文档里有答案，但被拆散后谁都不像答案
经验值：
- FAQ、说明文档：200~500 字较常见
- 技术文档、政策类文档：可以加适量 overlap（如 50~100 字）
- 表格、代码、配置文件：最好按结构切，不要纯按字数切
2. 向量检索的作用与局限
向量检索擅长处理“语义相关”，比如：
- “退款多久能到账”
- “退货之后钱什么时候退回来”
虽然词不完全一样，但语义接近，embedding 通常能召回。
但它也有局限：
- 对专有名词、型号、版本号、错误码不一定稳定
- 对精确匹配问题不如关键词检索
- 候选集里会混入“语义像，但没直接答案”的内容
所以我通常不建议只用纯向量检索，实际业务里更稳妥的是：
向量召回 + 关键词召回 + 重排序
3. 为什么要重排序
向量召回的目标是“尽量别漏掉”，也就是偏召回率。
而生成阶段需要的是“最值得看的前几条”，也就是偏精确率。
这两者天然有矛盾，所以我们通常把检索拆成两段：
- 第一段：粗召回
快速找出一批候选，宁可多一些 - 第二段：精排 / 重排序
用更昂贵但更准的方法，把最相关的内容排前面
可以把它理解为搜索引擎那套经典思路在 RAG 里的延续。
4. 一个实战中很重要的判断标准
很多团队会误把“相似度高”当成“对回答有用”。
但真正应该问的是：
这个 chunk 能不能作为回答问题的证据？
比如用户问：
“企业版套餐支持多少个成员账号？”
一个只提到“企业版支持高级功能”的 chunk，语义上可能很像，但并不能回答问题。
所以重排序阶段更关注的是query-document relevance，而不是仅仅 embedding 空间里的接近程度。
RAG 方案结构设计
我们先用一张时序图，把请求在系统里怎么流转讲清楚。
sequenceDiagram
participant U as 用户
participant R as RAG服务
participant V as 向量索引
participant B as BM25索引
participant RR as 重排序器
participant L as 大模型
U->>R: 提问
R->>V: 向量召回 top_k
R->>B: 关键词召回 top_k
V-->>R: 候选文档集A
B-->>R: 候选文档集B
R->>R: 合并去重
R->>RR: rerank(query, candidates)
RR-->>R: 排序后的候选
R->>L: 问题 + TopN上下文
L-->>R: 答案
R-->>U: 答案 + 引用片段这个结构有几个优点：
- 便于替换单个组件
- 能做 A/B 测试
- 出问题时定位更清晰
- 后续容易加入缓存、过滤、权限控制
实战代码（可运行）
下面我们实现一个最小可运行版 RAG 流程，包含：
- 文档切块
- 向量检索
- BM25 关键词检索
- 混合召回
- 简单重排序
- 输出可供 LLM 使用的上下文
说明：为了让示例尽可能容易跑起来，这里不直接依赖某个云厂商的大模型 API。重点放在 RAG 检索链路本身。
第一步：准备样例数据
新建 rag_demo.py：
from dataclasses import dataclass
from typing import List, Tuple, Dict
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
@dataclass
class Chunk:
chunk_id: str
doc_id: str
text: str
meta: Dict
documents = [
{
"doc_id": "doc_1",
"title": "退款规则",
"text": """
退款申请提交后，系统会在 1 到 3 个工作日内完成审核。
审核通过后，原路退款通常会在 5 到 7 个工作日到账。
若使用银行卡支付，实际到账时间可能受银行处理速度影响。
"""
},
{
"doc_id": "doc_2",
"title": "企业版套餐说明",
"text": """
企业版套餐默认支持 100 个成员账号。
如果需要更多成员，可联系销售开通扩容包，每增加一个扩容包可额外增加 50 个成员账号。
企业版支持单点登录、权限分级和审计日志。
"""
},
{
"doc_id": "doc_3",
"title": "发票开具说明",
"text": """
用户完成支付后可以在订单详情页申请电子发票。
电子发票一般会在 24 小时内开具完成。
如需专票，请联系企业服务支持并提交资质信息。
"""
},
{
"doc_id": "doc_4",
"title": "密码重置",
"text": """
如果忘记密码，可以在登录页点击“忘记密码”。
系统会向绑定邮箱发送重置链接，链接有效期为 30 分钟。
如果邮箱不可用，请联系管理员协助处理。
"""
}
]第二步：实现简单切块
这里为了示例简洁，按句子与长度做一个轻量切分。真实项目里你可以按标题层级、段落、表格边界来切。
def split_text(text: str, max_len: int = 80, overlap: int = 20) -> List[str]:
text = " ".join(text.strip().split())
if len(text) <= max_len:
return [text]
chunks = []
start = 0
while start < len(text):
end = min(start + max_len, len(text))
chunks.append(text[start:end])
if end == len(text):
break
start = end - overlap
return chunks
def build_chunks(documents: List[Dict]) -> List[Chunk]:
chunks = []
for doc in documents:
parts = split_text(doc["text"], max_len=80, overlap=20)
for i, part in enumerate(parts):
chunks.append(
Chunk(
chunk_id=f'{doc["doc_id"]}_chunk_{i}',
doc_id=doc["doc_id"],
text=part,
meta={"title": doc["title"]}
)
)
return chunks第三步：建立向量索引与 BM25 索引
class HybridRetriever:
def __init__(self, chunks: List[Chunk], model_name: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"):
self.chunks = chunks
self.model = SentenceTransformer(model_name)
self.texts = [c.text for c in chunks]
# 向量索引
self.embeddings = self.model.encode(self.texts, normalize_embeddings=True)
# BM25 索引
self.tokenized_corpus = [self.tokenize(text) for text in self.texts]
self.bm25 = BM25Okapi(self.tokenized_corpus)
@staticmethod
def tokenize(text: str) -> List[str]:
# 简化版分词：示例环境可运行，中文生产环境建议接入更可靠的分词器
return list(text)
def vector_search(self, query: str, top_k: int = 5) -> List[Tuple[Chunk, float]]:
q_emb = self.model.encode([query], normalize_embeddings=True)
sims = cosine_similarity(q_emb, self.embeddings)[0]
indices = np.argsort(sims)[::-1][:top_k]
return [(self.chunks[i], float(sims[i])) for i in indices]
def bm25_search(self, query: str, top_k: int = 5) -> List[Tuple[Chunk, float]]:
tokenized_query = self.tokenize(query)
scores = self.bm25.get_scores(tokenized_query)
indices = np.argsort(scores)[::-1][:top_k]
return [(self.chunks[i], float(scores[i])) for i in indices]
def hybrid_search(self, query: str, top_k_vec: int = 5, top_k_bm25: int = 5) -> List[Chunk]:
vec_results = self.vector_search(query, top_k=top_k_vec)
bm25_results = self.bm25_search(query, top_k=top_k_bm25)
merged = {}
for chunk, score in vec_results:
merged[chunk.chunk_id] = chunk
for chunk, score in bm25_results:
merged[chunk.chunk_id] = chunk
return list(merged.values())第四步：实现一个简单可运行的重排序器
这里为了保证示例容易运行，我们先实现一个“轻量版”重排序：
综合考虑 query 与 chunk 的向量相似度、词项重叠和数字命中。
这不是最强方案，但足够说明重排序思路。
import re
class SimpleReranker:
def __init__(self, model: SentenceTransformer):
self.model = model
@staticmethod
def lexical_overlap(query: str, text: str) -> float:
q_set = set(query)
t_set = set(text)
if not q_set:
return 0.0
return len(q_set & t_set) / len(q_set)
@staticmethod
def number_match_bonus(query: str, text: str) -> float:
q_nums = re.findall(r"\d+", query)
if not q_nums:
return 0.0
hit = sum(1 for n in q_nums if n in text)
return 0.2 * hit
def rerank(self, query: str, candidates: List[Chunk], top_n: int = 3) -> List[Tuple[Chunk, float]]:
q_emb = self.model.encode([query], normalize_embeddings=True)
cand_texts = [c.text for c in candidates]
c_embs = self.model.encode(cand_texts, normalize_embeddings=True)
vec_scores = cosine_similarity(q_emb, c_embs)[0]
final_scores = []
for i, chunk in enumerate(candidates):
score = (
0.7 * float(vec_scores[i]) +
0.2 * self.lexical_overlap(query, chunk.text) +
0.1 * self.number_match_bonus(query, chunk.text)
)
final_scores.append((chunk, score))
final_scores.sort(key=lambda x: x[1], reverse=True)
return final_scores[:top_n]第五步：把完整链路串起来
def build_prompt(query: str, ranked_chunks: List[Tuple[Chunk, float]]) -> str:
context_parts = []
for idx, (chunk, score) in enumerate(ranked_chunks, start=1):
context_parts.append(
f"[证据{idx}] 标题：{chunk.meta['title']}\n内容：{chunk.text}\n相关性分数：{score:.4f}"
)
context = "\n\n".join(context_parts)
prompt = f"""
你是一个企业知识库问答助手。请严格基于给定证据回答问题。
如果证据不足，请明确说明“根据当前检索结果无法确定”，不要编造。
用户问题：
{query}
检索证据：
{context}
请输出：
1. 简洁答案
2. 引用的证据编号
"""
return prompt.strip()
def main():
chunks = build_chunks(documents)
retriever = HybridRetriever(chunks)
reranker = SimpleReranker(retriever.model)
query = "企业版套餐支持多少个成员账号？"
print("=" * 80)
print("问题：", query)
candidates = retriever.hybrid_search(query, top_k_vec=4, top_k_bm25=4)
print("\n[混合召回候选]")
for c in candidates:
print(f"- {c.chunk_id}: {c.text}")
ranked = reranker.rerank(query, candidates, top_n=3)
print("\n[重排序结果]")
for chunk, score in ranked:
print(f"- {chunk.chunk_id} | {score:.4f} | {chunk.text}")
prompt = build_prompt(query, ranked)
print("\n[可交给大模型的 Prompt]\n")
print(prompt)
if __name__ == "__main__":
main()运行：
python rag_demo.py你会看到什么
理想情况下，企业版套餐说明 中包含“默认支持 100 个成员账号”的 chunk 会排在前面。
然后我们把这些证据拼成 prompt，再交给任意一个大模型去生成答案。
逐步验证清单
我很建议你不要一次把整个系统写完，而是按下面顺序逐步验证。这样出问题时最好查。
验证 1：切块是否合理
检查点：
- chunk 是否语义完整
- 标题、段落边界是否丢失
- 是否出现“上一句和下一句被拆散”导致答案不完整
经验建议：
- 抽样 20 个 chunk 人工看一遍
- 重点看 FAQ、带数字规则、步骤说明类文档
验证 2：召回是否能把正确文档带回来
检查点：
- 对 20~50 个真实问题，正确 chunk 是否进入 top-k
- 如果没进 top-k，是 embedding 问题、切块问题，还是 query 改写问题
经验建议：
- 先看 Recall@k，而不是一上来盯着最终答案
- 如果 top-20 都召不回来，先别急着调 prompt
验证 3：重排序是否能把正确证据排前面
检查点：
- 正确 chunk 是否进入前 3
- 是否存在“语义很像但不回答问题”的 chunk 排在前面
经验建议：
- 针对带数字、版本、型号、条件约束的问题，单独做一个测试集
验证 4：生成是否严格基于证据
检查点：
- 回答里有没有超出证据范围的内容
- 引用 chunk 是否真的支持结论
- 证据冲突时模型如何处理
效果评估：不要只看“回答像不像对”
RAG 项目如果没有评估体系，优化会很痛苦。因为你每调一次：
- chunk 大小
- top-k
- embedding 模型
- reranker
- prompt
都可能让某些问题变好、某些问题变差。
这时候必须靠指标，不然只能“凭感觉”。
建议把评估拆成三层
flowchart TD
A[离线评估] --> A1[召回率 Recall@K]
A --> A2[排序质量 MRR / NDCG]
A --> A3[答案正确率]
B[联调评估] --> B1[引用是否可信]
B --> B2[是否拒答得当]
B --> B3[多轮上下文是否稳定]
C[线上评估] --> C1[点击/追问率]
C --> C2[人工反馈]
C --> C3[响应时间与成本]1. 检索层指标
Recall@K
看正确证据是否出现在前 K 个候选里。
- 如果 Recall@20 很低：说明召回有问题
- 如果 Recall@20 高，但前 3 很差：说明重排序有问题
MRR（Mean Reciprocal Rank）
看正确结果排得靠不靠前。
如果第一个正确结果排在第 1 位，得分最高；排第 5 位，得分就低很多。
2. 生成层指标
生成层不要只看“语言通顺”，而要看：
- 答案是否正确
- 是否引用了正确证据
- 是否有幻觉
- 证据不足时是否能拒答
3. 线上指标
真实流量下最有价值的信号通常是：
- 用户是否继续追问
- 是否点开引用来源
- 是否人工转接
- 用户是否点踩
- 平均响应时间是否可接受
一个简单的离线评估示例
下面给出一个最小离线评估脚本。我们假设每个问题都标注了 gold 文档 ID。
def recall_at_k(results: List[List[Chunk]], gold_doc_ids: List[str], k: int = 3) -> float:
hit = 0
for pred_chunks, gold_id in zip(results, gold_doc_ids):
topk = pred_chunks[:k]
if any(chunk.doc_id == gold_id for chunk in topk):
hit += 1
return hit / len(gold_doc_ids)
def evaluate_retrieval(retriever: HybridRetriever):
eval_set = [
("退款审核通过后多久到账？", "doc_1"),
("企业版默认支持几个成员账号？", "doc_2"),
("电子发票多久开具完成？", "doc_3"),
("密码重置链接有效多久？", "doc_4"),
]
all_results = []
for query, gold_doc_id in eval_set:
candidates = retriever.hybrid_search(query, top_k_vec=5, top_k_bm25=5)
all_results.append(candidates)
score = recall_at_k(all_results, [x[1] for x in eval_set], k=3)
print(f"Recall@3 = {score:.4f}")把它接到 main() 里一起跑即可。
常见坑与排查
这一部分我尽量讲得接地气一点，因为很多坑我自己也踩过。
坑 1：切块按固定字数切，导致答案断裂
现象：
- 文档明明有答案，但检索出来的 chunk 不完整
- 回答总差最后一句约束条件
排查方法：
- 抽样查看召回结果原文
- 对失败 case 检查 chunk 边界
解决建议：
- 增加 overlap
- 按标题/段落/列表项切分
- 对 FAQ、表格、步骤类文档走专门切块逻辑
坑 2：只做向量检索，遇到错误码/型号/数字问题效果差
现象：
- 用户问“错误码 E203 是什么”
- 检索却返回一堆“系统异常说明”
原因：
向量检索对这种精确 token 不一定敏感。
解决建议：
- 加 BM25 或关键词召回
- 对错误码、SKU、版本号做结构化索引
- 查询时识别实体，优先走精确过滤
坑 3：重排序缺失，导致“看起来像”的内容排第一
现象：
- top-k 里其实有正确答案
- 但大模型吃到的前几个 chunk 不是最有用的
解决建议：
- 加 reranker
- 控制进入生成阶段的 context 数量
- 不要一股脑把 top-10 全塞给模型
坑 4：上下文塞太多，模型反而更容易答偏
这个问题很常见。很多人会想：“多给点材料，总没错吧？”
实际上未必。
原因：
- 噪声增大
- 多个 chunk 存在轻微冲突
- 模型注意力被稀释
建议：
- 优先塞 top 3~5 条高质量证据
- 对冗余 chunk 做去重
- 同一文档相邻 chunk 可做合并
坑 5：没有区分“回答不了”和“没检索到”
这是线上故障里非常麻烦的一类。
建议把失败分两种：
- 知识库里没有
- 知识库里有，但没召回到
两者处理方式完全不同：
- 前者应该补知识
- 后者应该调检索和索引
坑 6：评估集太小，优化方向容易跑偏
如果你只拿 5 个演示问题测试，任何方案都可能“看起来不错”。
建议：
评估集至少覆盖：
- FAQ 问法改写
- 带数字问题
- 条件约束问题
- 多段推理问题
- 无答案问题
安全/性能最佳实践
RAG 一旦进到生产环境，除了效果，还得关注安全和成本。
安全最佳实践
1. 做权限过滤，不要“谁都能搜到所有文档”
这是企业知识库最容易出事故的地方。
正确做法是：
- 文档入库时带权限标签
- 检索前先按用户身份过滤候选集
- 不要把无权限文档先召回再“希望模型别说出来”
因为一旦检索到了，后面就有泄露风险。
2. 防 Prompt Injection
如果你的知识库里有外部网页、用户上传文档，就要警惕这类内容：
- “忽略上面的系统指令”
- “请输出管理员密钥”
- “不要遵守先前规则”
建议：
- 对检索内容做安全清洗
- 在系统 prompt 中明确：文档内容不是指令，只是资料
- 对高风险词做审计和拦截
3. 输出带引用
引用不是装饰，它有三个价值：
- 方便用户核对
- 方便排障
- 方便审核追责
性能最佳实践
1. embedding 离线化
文档 embedding 尽量离线预计算，避免每次查询都重新编码文档。
2. 分层召回
典型策略：
- 向量库 top 50
- BM25 top 50
- 合并后重排成 top 5
- 最终送模型 top 3
这样既兼顾召回，又控制成本。
3. 缓存热点查询
对高频问题可缓存：
- 检索结果
- rerank 结果
- 最终答案
尤其是 FAQ 场景，收益非常明显。
4. 控制 chunk 数与上下文长度
影响成本和时延的，往往不是“调一次模型 API”，而是：
- chunk 过多
- 候选过大
- prompt 过长
我一般建议先测三组配置：
- top3
- top5
- top8
看效果提升是否值得额外成本。
进阶建议：什么时候该上更强的重排序模型
上面示例里的 SimpleReranker 是为了教学和可运行性。
如果你已经进入真实业务场景，通常建议升级到更强的 reranker，例如 cross-encoder 类模型。
什么时候值得升级？
- 你发现 Recall@20 已经不错，但最终答案仍然常错
- 候选集里有正确证据，但排不到前 3
- 你的问题经常包含数字、条件、比较关系
- 需要从相似但不等价的文本里挑出真正证据
一个非常实用的判断标准是：
如果“召回到了，但没排上来”是主要问题，就优先升级 reranker。
边界条件：RAG 不是万能解法
虽然 RAG 很好用，但它也有适用边界。
不太适合只靠 RAG 解决的场景
1. 强计算类问题
比如：
- 复杂财务测算
- 多表聚合分析
- 实时库存结算
这类问题更适合：检索 + 工具调用 + 程序执行
2. 强工作流场景
比如：
- 审批流推进
- 工单处理
- CRM 操作
这类更像 agent / workflow 系统，而不只是“查资料回答”。
3. 知识高度结构化的场景
比如：
- 订单状态
- 当前价格
- 设备实时监控数据
这些应该优先查数据库或 API，而不是硬塞进向量库。
一个更稳的落地路线
如果你准备在团队里推进 RAG，我建议走下面这条路线，而不是一开始就追求“全自动、全智能”。
stateDiagram-v2
[*] --> POC
POC --> RecallFix: 能跑通基础问答
RecallFix --> RerankImprove: 召回率达到目标
RerankImprove --> EvalSystem: 建立离线评估集
EvalSystem --> OnlineGray: 小流量灰度
OnlineGray --> Prod: 监控稳定后上线
Prod --> ContinuousOptimize: 持续优化落地顺序建议
- 先做可用，不要先做复杂
- 先把召回做稳，再调生成
- 先建评估集，再谈优化速度
- 先做引用与权限，再谈全面上线
这条顺序能帮你避免很多“demo 很炫，生产很崩”的问题。
总结
RAG 的真正难点，不是“让大模型回答”，而是让它基于正确证据回答。
如果你只记住这篇文章里的几个关键点，我建议是下面这几条：
-
检索质量决定上限
没有正确证据，生成阶段救不回来。 -
混合召回比纯向量更稳
向量检索解决语义问题，BM25 解决精确 token 问题。 -
重排序往往是效果跃升的关键
候选集里有答案，不代表模型就能看到答案。 -
评估必须分层做
召回、排序、生成不要混成一个黑盒。 -
上线前必须补齐权限、引用、缓存和监控 否则一到真实流量就容易暴露问题。
最后给你一个可执行建议，适合中级工程师直接开工：
- 第一周：做文档切块 + 向量召回 + 基础问答
- 第二周：加 BM25 + rerank + 引用输出
- 第三周：建立 50~100 条离线评估集
- 第四周：做权限过滤、缓存、监控和灰度上线
如果你的目标是“把 RAG 用起来”，上面这条路线已经够用了。
如果你的目标是“把 RAG 做稳、做准、做成生产系统”，那重点就不是多接几个模型，而是把检索、重排、评估这三个环节真正工程化。