大模型应用中的 RAG 落地实践:从知识库构建到检索增强效果优化
RAG(Retrieval-Augmented Generation,检索增强生成)这两年几乎成了企业做大模型应用的“标准配置”。原因很现实:单靠基础大模型的参数记忆,既难保证时效性,也很难把企业内部知识、流程规则、历史案例稳定地注入进去。
但真正开始落地时,很多团队会发现:“接了向量库”不等于“RAG 做好了”。模型还是答非所问、召回一堆无关文本、上下文塞满了却没提高效果,甚至查询一多成本和时延就一起飙升。
这篇文章我想从工程实践角度,把 RAG 从“能跑”讲到“能用”——包括知识库怎么建、检索链路怎么设计、代码怎么落地,以及优化时最常见的坑。
背景与问题
先说一个很典型的场景:企业想做一个内部知识问答系统,数据来源包括产品文档、FAQ、制度规范、工单记录和操作手册。最初方案通常很朴素:
- 把文档切块;
- 用 embedding 向量化;
- 放进向量数据库;
- 用户提问时,召回 top-k 文本;
- 拼进 prompt 让大模型回答。
这条链路没有问题,但实际效果常常不稳定,核心原因通常出在下面几个环节:
- 知识库质量差:文档重复、过时、格式混乱、结构丢失;
- 切块策略不合理:切得太碎,语义不完整;切得太大,召回噪声多;
- 检索单一:只做向量召回,对关键词、缩写、数字编号不敏感;
- 上下文拼接粗糙:召回内容太多,模型反而抓不到重点;
- 评估体系缺失:没有离线集,也没有线上反馈闭环,只能靠“感觉优化”。
如果把 RAG 看成一个系统,它并不是“一个模型能力问题”,而是一个典型的数据工程 + 检索工程 + Prompt 工程 + 评估工程的组合问题。
核心原理
RAG 的基本工作流
RAG 的目标很简单:先找资料,再生成答案。它试图用外部知识来弥补大模型参数记忆的局限。
flowchart TD
A[原始知识源: PDF/Markdown/FAQ/数据库] --> B[清洗与结构化]
B --> C[切块 Chunking]
C --> D[向量化 Embedding]
D --> E[向量索引/检索系统]
U[用户问题] --> Q[Query 理解与改写]
Q --> R[召回 Retriever]
E --> R
R --> RR[重排 Reranker]
RR --> P[上下文构建 Prompt]
P --> LLM[大模型生成答案]
LLM --> O[最终回复]
这个流程里,真正决定效果的不是某一个点,而是整条链路的配合。
一个实用的理解框架:RAG 的四层
我更习惯把 RAG 拆成四层来看:
- 数据层:原始文档是否可靠、版本是否统一、结构是否保留;
- 索引层:切块、向量化、元数据设计、索引更新;
- 检索层:召回、过滤、重排、多路检索融合;
- 生成层:Prompt 约束、引用片段组织、答案格式控制。
这四层里,很多团队把注意力都放在“换更强模型”上,但实际提升往往来自前两层和第三层。
为什么单纯向量检索不够
向量检索擅长“语义相似”,但在下面这些场景会吃亏:
- 错别字、简称、缩写;
- 产品型号、错误码、单号、合同号;
- 非常短的问题,如“报错 E102”;
- 强依赖关键词精确匹配的制度、条款类查询。
所以工程上更稳妥的方案通常是:混合检索(Hybrid Search)= 关键词检索 + 向量检索,再做重排。
flowchart LR
Q[用户查询] --> A[BM25 关键词召回]
Q --> B[Vector 向量召回]
A --> C[候选集合合并]
B --> C
C --> D[Reranker 重排]
D --> E[Top-N 上下文]
E --> F[LLM 生成]
知识库构建为什么是成败关键
很多效果问题,最后追根溯源都不是模型本身不行,而是知识库有这些问题:
- 同一内容多个版本并存;
- 表格、标题层级、代码块在预处理时丢失;
- 文档切块没有保留段落关系;
- 元数据缺失,没法按部门、时间、来源过滤;
- 更新机制不清晰,索引长期滞后。
如果知识源本身混乱,RAG 只是把混乱更高效地喂给模型。
方案设计:从知识库构建到检索链路
这一部分讲一套比较适合中型项目的落地思路。
1. 知识源治理
建议先对知识源做分层:
- 高可信知识:官方制度、产品文档、标准 SOP;
- 中可信知识:FAQ、内部 wiki;
- 低可信知识:工单记录、聊天摘要、临时笔记。
这样做的意义在于,后续检索时可以:
- 按可信度加权;
- 控制低可信内容只作候选,不直接喂给生成;
- 在回答中优先引用高可信来源。
2. 切块策略
切块没有万能参数,但有几个经验值很实用:
- 纯文本文档:300~800 字符一个 chunk,保留 50~120 字符 overlap;
- 操作手册/教程:按标题层级切,尽量保留步骤完整性;
- FAQ:一问一答为最小单位;
- 表格/代码:不要粗暴按字符截断,尽量保留结构。
我自己踩过的坑之一,就是把长文固定按 500 字切块,结果“问题描述”和“处理步骤”被切到两个 chunk,召回总是只命中一半,模型自然答不完整。
3. 元数据设计
元数据不是附属品,它直接决定你后面能不能做过滤、审计和优化。建议至少保留:
doc_idtitlesourcesectionupdated_atcategorypermission_tagchunk_id
有了这些字段,后面才能做:
- 按部门过滤;
- 按时间优先;
- 按文档类型加权;
- 返回引用来源;
- 定位脏数据。
4. 检索链路设计
一条更稳的查询链路通常是:
- 用户问题预处理;
- 查询改写/补全;
- 关键词召回;
- 向量召回;
- 合并去重;
- 重排;
- 动态截断上下文;
- 生成答案并附引用。
如果用户问题比较短,查询改写尤其重要。比如:
- 原始问题:
退款怎么走? - 改写后:
退款申请流程、审批节点、退款到账时间、异常处理
当然,改写不能瞎扩展,否则会把召回空间越扩越偏。最好限定在业务词表和文档域内。
实战代码(可运行)
下面我用一个可运行的简化 Python 示例演示 RAG 的核心流程。为了方便本地体验,这里不接真实大模型 API,而是重点展示:
- 文档切块
- TF-IDF 检索
- 简单重排
- 上下文拼装
- 生成回答模板
这个示例适合理解流程,也方便你后续替换成真正的 embedding、向量库和 LLM。
安装依赖
pip install scikit-learn numpy
示例代码
from dataclasses import dataclass
from typing import List, Dict, Tuple
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import re
@dataclass
class Chunk:
chunk_id: str
doc_id: str
title: str
content: str
source: str
category: str
class SimpleRAG:
def __init__(self, chunks: List[Chunk]):
self.chunks = chunks
self.vectorizer = TfidfVectorizer()
self.doc_texts = [self._normalize(c.content) for c in chunks]
self.doc_matrix = self.vectorizer.fit_transform(self.doc_texts)
def _normalize(self, text: str) -> str:
text = text.lower().strip()
text = re.sub(r"\s+", " ", text)
return text
def search(self, query: str, top_k: int = 5) -> List[Tuple[Chunk, float]]:
q = self.vectorizer.transform([self._normalize(query)])
sims = cosine_similarity(q, self.doc_matrix)[0]
pairs = list(zip(self.chunks, sims))
pairs.sort(key=lambda x: x[1], reverse=True)
return pairs[:top_k]
def rerank(self, query: str, candidates: List[Tuple[Chunk, float]]) -> List[Tuple[Chunk, float]]:
keywords = set(re.findall(r"[\u4e00-\u9fa5A-Za-z0-9]+", query.lower()))
reranked = []
for chunk, base_score in candidates:
content_tokens = set(re.findall(r"[\u4e00-\u9fa5A-Za-z0-9]+", chunk.content.lower()))
keyword_hit = len(keywords & content_tokens)
score = base_score + 0.05 * keyword_hit
reranked.append((chunk, score))
reranked.sort(key=lambda x: x[1], reverse=True)
return reranked
def build_prompt(self, query: str, top_chunks: List[Tuple[Chunk, float]]) -> str:
context_parts = []
for idx, (chunk, score) in enumerate(top_chunks, 1):
context_parts.append(
f"[资料{idx}] 标题:{chunk.title}\n来源:{chunk.source}\n内容:{chunk.content}\n"
)
context = "\n".join(context_parts)
prompt = f"""你是企业知识助手。请严格根据提供资料回答问题。
如果资料不足,请明确说“根据当前知识库无法确认”。
回答时尽量分点,并附上引用资料编号。
用户问题:
{query}
检索到的资料:
{context}
请开始回答:
"""
return prompt
def answer(self, query: str, top_k: int = 3) -> Dict:
candidates = self.search(query, top_k=top_k * 2)
reranked = self.rerank(query, candidates)[:top_k]
prompt = self.build_prompt(query, reranked)
# 这里用一个模拟生成逻辑,真实项目中可替换成 LLM API
answer_lines = ["根据知识库,相关信息如下:"]
for idx, (chunk, _) in enumerate(reranked, 1):
answer_lines.append(f"{idx}. {chunk.content[:80]}... [资料{idx}]")
answer_lines.append("如需精确执行,请以原始制度或流程文档为准。")
answer = "\n".join(answer_lines)
return {
"query": query,
"prompt": prompt,
"answer": answer,
"references": [
{
"chunk_id": c.chunk_id,
"doc_id": c.doc_id,
"title": c.title,
"source": c.source
}
for c, _ in reranked
]
}
def build_demo_chunks() -> List[Chunk]:
return [
Chunk(
chunk_id="c1",
doc_id="d1",
title="退款流程说明",
content="退款申请需由业务人员在系统提交,金额在1000元以内由主管审批,超过1000元需财务复核。审批通过后,预计3到5个工作日到账。",
source="制度中心",
category="流程制度"
),
Chunk(
chunk_id="c2",
doc_id="d2",
title="退款异常处理",
content="若退款失败,需检查收款账户信息是否正确。如账户异常,业务人员应联系客户更新信息后重新发起退款流程。",
source="客服知识库",
category="异常处理"
),
Chunk(
chunk_id="c3",
doc_id="d3",
title="发票开具规则",
content="发票申请需在付款完成后提交,电子发票将在1个工作日内发送至登记邮箱。",
source="财务FAQ",
category="财务"
),
Chunk(
chunk_id="c4",
doc_id="d4",
title="审批权限矩阵",
content="主管可审批1000元及以下退款申请,部门经理可审批5000元及以下特殊退款,超过额度需财务负责人确认。",
source="审批手册",
category="权限规则"
),
]
if __name__ == "__main__":
rag = SimpleRAG(build_demo_chunks())
result = rag.answer("退款怎么走?超过1000元谁审批?")
print("==== Answer ====")
print(result["answer"])
print("\n==== References ====")
for ref in result["references"]:
print(ref)
运行后你会得到什么
这个示例虽然简单,但已经包含了 RAG 的几个关键动作:
- 先检索;
- 再按关键词命中做轻量重排;
- 最后把结果组织进 Prompt。
真实项目里,你可以这样替换:
TfidfVectorizer→ embedding 模型;- 本地矩阵 → FAISS / Milvus / pgvector / Elasticsearch;
- 模拟回答 → OpenAI / 通义 / 百川 / 私有部署模型;
- 简单 rerank → bge-reranker / cross-encoder。
检索与生成协同优化
很多人优化 RAG,容易只盯着召回分数,但最后用户感知的是“答案好不好用”。所以检索和生成必须一起看。
1. 动态 Top-K,而不是固定值
固定 top-k 常常是个隐形坑:
k太小:漏信息;k太大:噪声增加,Prompt 变长,模型注意力被稀释。
更好的做法是根据分数动态截断,例如:
- 只保留分数高于阈值的 chunk;
- 或者只保留和第一名分差不大的结果;
- 再限制总 token 数。
2. 不要把所有召回结果原样塞给模型
正确姿势通常是:
- 先去重;
- 再按来源可信度排序;
- 合并相邻 chunk;
- 去掉内容高度重合的片段;
- 最后保留最能支撑答案的证据。
3. 给模型明确边界
Prompt 里一定要说清楚:
- 仅根据资料回答;
- 资料不足时要明确说明;
- 回答要附引用;
- 不允许编造制度、时间、额度。
这类约束对企业场景尤其重要,因为用户往往更在意“能否追溯”而不是“回答是否自然”。
常见坑与排查
下面这些问题,我基本都见过,甚至有些是线上出过事故后才补上的。
坑 1:召回很准,但答案还是错
现象:检索结果里其实有正确答案,但模型输出还是偏了。
排查方向:
- Prompt 是否明确要求“基于资料作答”;
- 上下文是否过长,关键信息被淹没;
- 引用片段是否顺序混乱;
- 模型是否倾向使用先验常识覆盖资料内容。
处理建议:
- 缩短上下文;
- 提升引用片段密度;
- 在 prompt 中强调“冲突时以资料为准”;
- 输出结构改成“结论 + 依据”。
坑 2:数字、编号、型号类问题效果差
现象:比如“E102 是什么错误”“A12 合同怎么处理”,向量检索召回不稳定。
原因:这类问题更依赖精确匹配,不是纯语义问题。
处理建议:
- 增加 BM25 / 倒排索引;
- 为编号字段单独建索引;
- 预处理时保留大小写、连字符、特殊编码模式。
坑 3:文档一更新,答案还在引用旧版本
现象:用户明明看到了新制度,机器人还在按旧规则回答。
原因:
- 索引未及时增量更新;
- 老文档未失效;
- 元数据没有版本字段。
处理建议:
- 文档入库做版本控制;
- 检索时优先最新版本;
- 旧版本只保留审计用途,不参与默认召回。
坑 4:切块太碎,回答支离破碎
现象:召回都是“半句话”,模型拼接后像东一块西一块。
处理建议:
- 按标题和语义边界切块;
- 保留 overlap;
- 对相邻 chunk 做 merge;
- FAQ 场景尽量保持问答完整。
坑 5:线上效果和离线测试完全两回事
现象:离线评估看起来不错,用户还是频繁点踩。
原因:
- 测试集太理想化;
- 真实用户问题更短、更口语化、更含糊;
- 没覆盖异常查询和追问场景。
处理建议:
- 从真实日志回流构建评测集;
- 把“模糊问题”“缩写问题”“错别字问题”单独建桶评估;
- 做多轮问答压测,而不是只看单轮命中。
安全/性能最佳实践
RAG 不只是效果问题,还会碰到权限、安全和成本问题。企业场景里这块不能后补,最好从一开始就设计进去。
安全实践
1. 权限过滤前置
检索前或检索后,必须基于用户身份做权限过滤,不能把“本不该看到的 chunk”先召回再交给模型。因为一旦进了 Prompt,就有泄露风险。
sequenceDiagram
participant U as 用户
participant A as 应用层
participant R as 检索服务
participant K as 知识库
participant M as 大模型
U->>A: 提问 + 身份信息
A->>R: 查询 + 权限标签
R->>K: 带权限过滤检索
K-->>R: 可访问文档片段
R-->>A: 重排后的上下文
A->>M: 安全上下文 + Prompt
M-->>A: 回答
A-->>U: 最终结果
2. 敏感信息脱敏
知识库中如果包含手机号、身份证号、合同金额、客户隐私字段,建议在入库时脱敏,或者在召回出库时按权限做显示控制。
3. 防 Prompt 注入
如果知识库内容来自开放来源,文档里可能包含恶意提示,如“忽略之前所有规则”。应在生成前增加内容清洗或模板隔离,明确将文档内容视为“资料”,不是“指令”。
性能实践
1. 分层缓存
可以缓存三类数据:
- embedding 结果缓存;
- 热门查询检索结果缓存;
- 最终答案缓存(需带版本和权限维度)。
2. 索引增量更新
不要每次全量重建索引。对于文档中心类场景,增量更新通常更现实:
- 新增文档:直接向索引追加;
- 更新文档:逻辑删除旧 chunk,写入新 chunk;
- 删除文档:标记失效,不参与召回。
3. 控制上下文成本
上下文越长,成本和时延通常越高。建议做三件事:
- 限制总 token;
- 相似 chunk 去重;
- 先摘要后拼接,特别是超长文档场景。
4. 多路召回的并发执行
向量召回、关键词召回、规则检索可以并发跑,再统一合并,通常比串行更稳。
一个可落地的评估方法
如果你想知道 RAG 有没有真的变好,至少要有两层评估:
离线评估
准备一个问答集,每条包含:
- 用户问题;
- 标准答案或关键事实;
- 应命中的文档片段;
- 文档来源约束。
常见指标可以看:
- Recall@k:正确片段是否进入前 k;
- MRR / NDCG:正确片段排位是否靠前;
- Answer Faithfulness:答案是否忠于资料;
- Citation Accuracy:引用是否对应正确来源。
线上评估
重点关注:
- 首次命中率;
- 用户追问率;
- 点赞/点踩率;
- 人工转接率;
- 平均响应时延;
- 单次调用成本。
如果只看“模型回复流畅度”,很容易误判。真正有价值的是:用户是否更快拿到可信答案。
实战建议:一条更稳的落地路线
如果你准备从 0 到 1 做一个 RAG 系统,我建议按这个顺序推进:
- 先做知识治理:清理过期和重复文档;
- 定义元数据规范:来源、版本、权限先补齐;
- 从单路检索起步:先把最基本召回打通;
- 再引入混合检索和重排:不要一上来堆复杂组件;
- 建立评测集:没有评估,就没有优化闭环;
- 最后做生成优化:包括 Prompt、引用、摘要和缓存。
很多团队容易反过来:先接最强模型,再慢慢补知识库。这种做法短期看起来快,长期几乎都会返工。
总结
RAG 的价值,不在于“让大模型知道更多”,而在于让模型在正确的时刻拿到正确的知识,并基于这些知识稳定作答。
真正落地时,建议记住这几点:
- 知识库质量优先于模型花样;
- 混合检索通常比纯向量检索更稳;
- 切块、元数据、重排是效果分水岭;
- 权限、安全、版本控制必须前置设计;
- 一定要建立离线与线上结合的评估体系。
如果你现在的 RAG 系统已经“能回答”,下一步不要急着换模型,先去看三件事:
- 召回内容是不是对;
- 上下文是不是干净;
- 回答是不是可追溯。
把这三件事做好,RAG 的效果通常会有非常明显的提升。对于大多数企业应用来说,这比单纯追逐更大的模型,更现实,也更有产出。