1. 为什么我又把RAG翻出来重做了一遍
RAG这个词,从2023年火到现在,网上的教程一抓一大把,但真正落到自己业务里,你会发现一个很尴尬的现实:跑通Demo只要一个下午,想让它稳定可用,可能要折腾一个月。我前后在三个不同规模的项目里落地过RAG,从最初用LangChain拼一个“能问答就行”的原型,到后来要给几十个内部用户提供知识库检索服务,中间踩的坑足够写一本小册子。
这篇文章不讲虚的,就讲一件事:一个能真正跑起来的RAG系统,从数据准备到上线调优,完整步骤长什么样,每一步为什么这么做,以及我在哪些地方摔过跤。核心关键词就三个:RAG、落地实践、踩坑记录。适合谁看?如果你已经知道RAG大概是什么,但自己动手时发现检索结果总是不对、回答总是胡编、或者不知道该从哪一步开始优化,那这篇内容就是写给你的。零基础也能看懂,因为我会把每个环节的“为什么”讲清楚,而不是甩一堆代码让你自己悟。
先说一个我自己的判断:RAG的瓶颈从来不在模型本身,而在数据管线和检索策略。很多人一上来就纠结用哪个LLM、要不要微调,结果忽略了最基础的问题——你的文档切对了吗?你的检索真的能命中吗?我见过太多项目,模型换了一轮又一轮,效果就是上不去,最后发现是切块策略有问题。所以这篇文章的重点会放在数据侧和检索侧,这两块做好了,哪怕用一个普通的开源模型,效果也能超过大部分“堆配置”的方案。
2. 整体架构设计与技术选型思路
2.1 先想清楚:你的RAG到底要解决什么问题
在动手之前,必须先回答一个问题:你的用户会问什么类型的问题?这个问题决定了后面所有的技术选择。我把它分成三类:
第一类是事实型查询,比如“XX产品的保修期是多久”“XX流程的第三步是什么”。这类问题的答案通常集中在文档的某一段落里,检索精度要求高,但对上下文理解要求低。
第二类是归纳型查询,比如“总结一下这份报告的核心结论”“对比A方案和B方案的优缺点”。这类问题需要跨多个段落甚至多个文档整合信息,对检索的召回率要求高,单靠向量相似度往往不够。
第三类是推理型查询,比如“如果客户要求延期交付,按照合同条款应该怎么处理”。这类问题需要结合多个知识点进行推理,对知识库的结构化程度要求最高。
我自己的项目里,第一类和第二类占了80%以上。所以我的架构设计优先保证这两类的效果,第三类通过引入结构化知识(比如把合同条款做成表格)来辅助。如果你一上来就想做全能型RAG,大概率会陷入“什么都想要,什么都做不好”的困境。
2.2 技术栈选择:为什么我最终选了这套组合
市面上的RAG工具链太多了,LangChain、LlamaIndex、Haystack,还有各种国产框架。我试过其中大部分,最终稳定下来的组合是这样的:
| 环节 | 选型 | 理由 |
|---|---|---|
| 文档解析 | Unstructured + 自写规则 | 通用解析器处理PDF表格容易丢结构,关键文档我加了自定义规则 |
| 文本切分 | 递归切分 + 语义切分混合 | 纯递归切分会把完整语义块切断,纯语义切分又太慢 |
| 向量模型 | BGE-M3 | 中文效果好,支持多语言,本地部署成本低 |
| 向量库 | Milvus | 数据量上到百万级后,Milvus的检索延迟明显优于FAISS |
| 检索策略 | 混合检索(向量+关键词) | 纯向量检索对专有名词和编号不敏感,必须加BM25兜底 |
| 重排序 | BGE-Reranker | 召回阶段放宽,精排阶段收紧,效果提升明显 |
| 生成模型 | Qwen2.5-14B | 中文理解好,支持长上下文,本地部署可控 |
这套组合不是拍脑袋定的。我试过用OpenAI的embedding,效果确实好,但数据出境的合规问题过不了;试过用FAISS,小数据量很快,但数据上到50万条以后,内存占用和检索延迟都成了瓶颈;试过纯向量检索,结果用户搜“第三章第二节”这种带编号的内容,向量模型完全找不到。所以每一个选型背后,都是实际踩坑后的妥协。
提示:如果你刚开始做,数据量在10万条以内,FAISS完全够用,没必要上Milvus。但如果你预期数据会持续增长,建议一开始就用Milvus,迁移成本比你想的高。
2.3 数据流设计:从原始文档到可检索知识库
整个数据流我分成四个阶段:
阶段一:采集与解析。原始文档可能是PDF、Word、Excel、HTML、Markdown,甚至图片。我的做法是先用Unstructured做统一解析,输出带元数据的文本块。对于表格,我单独用Camelot提取,转成Markdown表格再嵌入。对于扫描件,先用OCR处理,但OCR结果一定要人工抽检,我遇到过OCR把“0”识别成“O”导致检索完全失效的情况。
阶段二:清洗与切分。解析出来的文本有很多噪音,比如页眉页脚、页码、重复的标题。我写了一套规则清洗,然后做切分。切分策略后面会详细讲,这里只说一个原则:切分的目标是让每个块在语义上自包含,同时不超过模型的上下文限制。
阶段三:向量化与索引。每个文本块用BGE-M3生成向量,同时保留原始文本和元数据(来源文件、页码、章节标题)。元数据非常重要,后面做过滤和溯源都靠它。
阶段四:检索与生成。用户提问后,先做查询改写,然后混合检索召回Top-K,再用Reranker精排,最后把Top-N个块拼成上下文送给LLM生成回答。
这个流程看起来简单,但每个阶段都有坑。下面我逐个拆解。
3. 核心细节解析与实操要点
3.1 文档切分:RAG效果的第一道生死线
切分是RAG里最容易被低估的环节。我见过太多人直接用LangChain的RecursiveCharacterTextSplitter,设一个chunk_size=1000就完事了。结果就是:检索出来的块要么缺头少尾,要么包含大量无关内容。
我的切分策略是三层切分:
第一层按文档结构切。如果文档有明确的章节标题(比如Markdown的##、Word的标题样式),优先按章节切。这样每个块天然有语义边界。我写了一个解析器,把标题层级提取出来,作为每个块的元数据。
第二层按语义切。对于没有明确结构的文档,我用语义相似度做切分。具体做法是:先把文本按句子切分,然后计算相邻句子的向量相似度,相似度低于阈值的地方就是切分点。这个方法比固定长度切分效果好很多,但计算量大,我只在关键文档上用。
第三层按长度兜底。如果某个块还是太长(超过模型上下文限制),再用递归切分强制切断。但这时候我会加一个重叠窗口,通常是块大小的10%-15%,保证切断处的语义不会完全丢失。
# 语义切分的核心逻辑示意 from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer('BAAI/bge-m3') def semantic_split(text, threshold=0.6): sentences = split_into_sentences(text) embeddings = model.encode(sentences) chunks = [] current_chunk = [sentences[0]] for i in range(1, len(sentences)): sim = np.dot(embeddings[i-1], embeddings[i]) / ( np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i]) ) if sim < threshold: chunks.append(' '.join(current_chunk)) current_chunk = [sentences[i]] else: current_chunk.append(sentences[i]) chunks.append(' '.join(current_chunk)) return chunks注意:语义切分的阈值不要设得太高,否则会把完整段落切碎。我实测下来,0.6-0.7之间比较平衡。另外,语义切分对短文档效果不明显,长文档(超过5000字)才值得用。
3.2 向量化:模型选型与批量处理的坑
向量模型的选择直接决定检索效果。我试过OpenAI的text-embedding-3-large,效果确实好,但有两个问题:一是成本,二是数据合规。后来换成本地部署的BGE-M3,效果差距在可接受范围内,但成本和可控性好了很多。
BGE-M3有几个特点值得注意:它支持多语言,中文效果在开源模型里属于第一梯队;它支持长文本,最大输入长度8192个token;它还能同时输出稠密向量和稀疏向量,这对混合检索非常友好。
但用BGE-M3也有坑。第一个坑是批量大小。如果你一次性把几百个块塞进去编码,显存很容易爆。我的经验是,单卡24G显存,批量大小设在32-64之间比较稳。第二个坑是归一化。BGE-M3输出的向量需要做L2归一化,否则余弦相似度计算会出问题。第三个坑是指令前缀。BGE系列模型对查询和文档有不同的指令前缀,查询要加"为这个句子生成表示以用于检索相关文章:",文档不需要。这个细节如果不注意,检索效果会打折扣。
# BGE-M3 批量编码的正确姿势 from FlagEmbedding import BGEM3FlagModel model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True) # 文档编码,不加指令前缀 doc_embeddings = model.encode( documents, batch_size=32, max_length=8192, return_dense=True, return_sparse=True ) # 查询编码,加指令前缀 query_embedding = model.encode( ["为这个句子生成表示以用于检索相关文章:" + query], batch_size=1, max_length=512, return_dense=True, return_sparse=True )3.3 混合检索:为什么纯向量检索不够用
纯向量检索有一个致命缺陷:它对精确匹配不敏感。比如用户搜“GB/T 19001-2016”,向量模型可能返回一堆关于“质量管理体系”的文档,但就是找不到那个标准号。再比如用户搜“第三章第二节”,向量模型完全无法理解这种结构化引用。
所以我在向量检索之外,加了一路BM25关键词检索。两路检索各自召回Top-K,然后合并去重。合并策略有两种:一种是简单加权,向量检索权重0.7,BM25权重0.3;另一种是RRF(Reciprocal Rank Fusion),按排名融合。我实测下来,RRF更稳定,因为它不依赖分数归一化。
# RRF 融合示例 def rrf_fusion(vector_results, bm25_results, k=60): scores = {} for rank, doc_id in enumerate(vector_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) for rank, doc_id in enumerate(bm25_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(scores.items(), key=lambda x: x[1], reverse=True)提示:BM25的实现我推荐用
rank_bm25这个库,轻量够用。如果你的数据量很大,可以考虑Elasticsearch,但运维成本会上去。
3.4 重排序:召回放宽,精排收紧
混合检索召回的结果可能有几十条,但真正能用的可能只有三五条。这时候就需要重排序。我用的BGE-Reranker是一个交叉编码器,它会把查询和每个候选块拼在一起打分,精度比向量相似度高很多,但速度慢。
所以我的策略是:召回阶段放宽到Top-50,精排阶段收紧到Top-5。这样既保证了召回率,又控制了延迟。Reranker的批量大小也要注意,我一般设在16-32之间,再大延迟就明显了。
重排序还有一个好处:它可以过滤掉那些“看起来相似但实际无关”的块。我遇到过很多次,向量检索返回的块和查询在字面上很像,但内容完全不相关。Reranker能有效识别这种情况。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
我假设你用的是Linux环境,有NVIDIA显卡。如果没有显卡,CPU也能跑,但速度会慢很多。以下是核心依赖:
# 创建虚拟环境 python -m venv rag_env source rag_env/bin/activate # 核心依赖 pip install FlagEmbedding==1.2.10 pip install pymilvus==2.4.0 pip install rank_bm25==0.2.2 pip install unstructured==0.12.0 pip install camelot-py==0.11.0 pip install langchain==0.2.0 pip install langchain-community==0.2.0Milvus我用的是Docker部署,单机版足够:
docker run -d --name milvus-standalone \ -p 19530:19530 \ -p 9091:9091 \ -v ./milvus_data:/var/lib/milvus \ milvusdb/milvus:v2.4.0注意:Milvus的版本要和pymilvus匹配,我遇到过版本不匹配导致连接失败的情况。另外,Milvus的数据目录一定要挂载出来,否则容器重启数据就没了。
4.2 文档解析与清洗的完整流程
文档解析我分成三步:格式识别、内容提取、噪音清洗。
格式识别很简单,看文件后缀就行。但要注意,有些PDF是扫描件,需要用OCR。我用的OCR是PaddleOCR,中文识别效果不错。
内容提取用Unstructured,但它的默认配置对中文支持一般。我改了几个参数:
from unstructured.partition.pdf import partition_pdf elements = partition_pdf( filename="doc.pdf", strategy="hi_res", # 高精度模式,保留布局信息 languages=["chi_sim", "eng"], # 中文简体+英文 infer_table_structure=True, # 提取表格结构 include_page_breaks=True # 保留分页信息 )噪音清洗我写了一套正则规则,主要处理这几类问题:
- 页眉页脚:通常出现在每页的固定位置,我通过统计每页首尾行的重复频率来识别。
- 页码:纯数字行,且出现在页面边缘。
- 重复标题:如果某个标题在多个页面重复出现,只保留第一次。
- 乱码:非中文、非英文、非数字的连续字符。
清洗完之后,我会把文本按章节重组,保留标题层级作为元数据。这一步很关键,因为后面的检索过滤和溯源都依赖元数据。
4.3 向量库的Schema设计与索引配置
Milvus的Schema设计直接影响检索性能。我的Schema是这样的:
from pymilvus import CollectionSchema, FieldSchema, DataType fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=8192), FieldSchema(name="dense_vector", dtype=DataType.FLOAT_VECTOR, dim=1024), FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR), FieldSchema(name="source_file", dtype=DataType.VARCHAR, max_length=512), FieldSchema(name="chapter", dtype=DataType.VARCHAR, max_length=256), FieldSchema(name="page_num", dtype=DataType.INT64) ] schema = CollectionSchema(fields, description="RAG knowledge base")索引配置我用的是HNSW,参数M=16,efConstruction=200。这个配置在召回率和速度之间比较平衡。如果你更看重速度,可以用IVF_FLAT,但召回率会降一些。
index_params = { "metric_type": "IP", # 内积,因为向量已经归一化 "index_type": "HNSW", "params": {"M": 16, "efConstruction": 200} } collection.create_index("dense_vector", index_params)提示:
metric_type选IP还是L2取决于你的向量是否归一化。BGE-M3输出的是归一化向量,用IP等价于余弦相似度。如果你不确定,用COSINE最保险,但性能会略低。
4.4 检索链路的完整实现
检索链路我封装成一个类,核心方法就一个retrieve:
class RAGRetriever: def __init__(self, milvus_collection, bm25_index, reranker): self.collection = milvus_collection self.bm25 = bm25_index self.reranker = reranker def retrieve(self, query, top_k=50, top_n=5): # 查询改写 rewritten_query = self.rewrite_query(query) # 向量检索 query_embedding = model.encode( ["为这个句子生成表示以用于检索相关文章:" + rewritten_query] ) vector_results = self.collection.search( data=query_embedding['dense_vecs'], anns_field="dense_vector", param={"metric_type": "IP", "params": {"ef": 128}}, limit=top_k, output_fields=["chunk_text", "source_file", "chapter"] ) # BM25检索 bm25_results = self.bm25.get_top_n( rewritten_query, self.corpus, n=top_k ) # RRF融合 fused = rrf_fusion(vector_results, bm25_results) # 重排序 candidates = [self.get_chunk_text(doc_id) for doc_id, _ in fused[:top_k]] rerank_scores = self.reranker.compute_score( [[rewritten_query, c] for c in candidates] ) reranked = sorted( zip(candidates, rerank_scores), key=lambda x: x[1], reverse=True ) return reranked[:top_n]查询改写这一步很多人会忽略,但它对效果影响很大。我的做法是用一个小模型(比如Qwen2.5-1.5B)把用户的口语化问题改写成更适合检索的形式。比如用户问“那个啥,就是关于报销的流程是啥来着”,改写成“报销流程 步骤 要求”。
4.5 生成环节的Prompt设计与上下文拼接
生成环节的Prompt我改了十几版,最终稳定下来的模板是这样的:
你是一个知识库助手,请根据以下参考资料回答用户问题。 参考资料: {context} 用户问题:{question} 回答要求: 1. 只根据参考资料回答,不要编造信息。 2. 如果参考资料中没有相关信息,直接说“根据现有资料无法回答”。 3. 回答时注明信息来源(文件名和章节)。 4. 如果参考资料中有矛盾信息,指出矛盾并说明。上下文拼接也有讲究。我按Reranker分数从高到低排列,但会做去重和截断。去重是防止同一个文档的多个块重复出现,截断是保证总长度不超过模型的上下文限制。我一般保留Top-5个块,总长度控制在3000 token以内。
注意:上下文不是越多越好。我试过把Top-10都塞进去,结果模型反而被无关信息干扰,回答质量下降。Top-5是一个比较平衡的值。
5. 常见问题与排查技巧实录
5.1 检索命中率低:从查询侧和索引侧双向排查
检索命中率低是最常见的问题。我的排查思路是:先看查询侧,再看索引侧。
查询侧的问题包括:查询太短、查询太口语化、查询包含错别字。解决办法是查询改写和查询扩展。查询扩展我用的是同义词词典+LLM生成。比如用户搜“年假”,扩展成“年假 年度休假 带薪休假”。
索引侧的问题包括:切分太碎、向量模型不适合领域、索引参数不合理。我遇到过一次,切分太碎导致每个块只有一两句话,向量模型无法捕捉完整语义。后来把块大小从200字调到500字,命中率提升了30%。
还有一个隐蔽的问题:元数据过滤太严。我一开始为了精确,加了很严格的元数据过滤,结果把很多相关块过滤掉了。后来改成软过滤,只在用户明确指定来源时才过滤。
5.2 回答胡编乱造:如何让模型“闭嘴”
模型胡编是RAG的另一个大坑。我的解决办法有三层:
第一层是Prompt约束。上面那个Prompt模板里明确说了“只根据参考资料回答”,但光靠这个不够。
第二层是引用溯源。我要求模型在回答里注明每个信息的来源,这样用户能自己判断。同时我在后处理里检查模型引用的来源是否真的在上下文里,如果不在,就标记为“可能不可靠”。
第三层是置信度阈值。如果Reranker的最高分低于某个阈值(我设的是0.5),就直接返回“根据现有资料无法回答”,不送给LLM生成。这个策略虽然会降低回答率,但能大幅减少胡编。
5.3 性能瓶颈:从检索延迟到生成速度的优化
性能问题我遇到过几次。第一次是检索延迟高,排查发现是Milvus的索引没建好,ef参数设得太小。后来把ef从64调到128,召回率上去了,延迟只增加了5ms。
第二次是生成速度慢。Qwen2.5-14B在单卡A100上生成速度大概是30 token/s,一个200字的回答要7秒左右。我的优化是:用vLLM做推理加速,开启连续批处理,速度提升了3倍。
第三次是内存占用高。BGE-M3和Reranker同时加载,显存占用接近20G。我的解决办法是:把Reranker放到CPU上跑,虽然慢一点,但显存压力小了很多。实测下来,Reranker在CPU上处理50个候选块大概需要200ms,可以接受。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 检索结果不相关 | 切分太碎/向量模型不匹配 | 人工检查Top-10结果 | 调整切分策略/换模型 |
| 专有名词搜不到 | 纯向量检索对精确匹配不敏感 | 测试带编号的查询 | 加BM25混合检索 |
| 回答胡编 | Prompt约束不够/上下文无关 | 检查上下文和回答的对应关系 | 加引用溯源/置信度阈值 |
| 检索延迟高 | 索引参数不合理/数据量大 | 看Milvus的查询耗时 | 调ef参数/加缓存 |
| 生成速度慢 | 模型太大/没有推理加速 | 测token生成速度 | 用vLLM/换小模型 |
| 显存不够 | 模型太多/批量太大 | 看nvidia-smi | 模型分卡/减小批量 |
| 表格内容丢失 | 解析器不支持表格 | 检查解析结果 | 用Camelot单独提取 |
| 多文档冲突 | 没有冲突处理机制 | 看回答是否矛盾 | Prompt里加冲突说明 |
5.5 几个我踩过的坑和独家技巧
坑一:PDF解析的隐藏字符。有些PDF里藏着不可见字符,比如零宽空格、软连字符。这些字符会让向量模型产生奇怪的向量。我的解决办法是在清洗阶段用正则把这些字符全部删掉。
坑二:向量库的删除操作。Milvus的删除是软删除,数据不会立即释放。如果你频繁更新知识库,数据会越积越多。我的做法是定期做compact,或者干脆重建collection。
坑三:Reranker的输入长度限制。BGE-Reranker的最大输入长度是512 token,如果你的块超过这个长度,会被截断。我的做法是在切分阶段就控制块大小,确保不超过512 token。
技巧一:用LLM做查询改写。我试过用规则做查询改写,效果一般。后来换成用Qwen2.5-1.5B做改写,效果好很多。成本也不高,一次改写大概0.1秒。
技巧二:缓存高频查询。我统计了一下,Top-20的查询占了总查询量的40%。所以我在检索前面加了一层Redis缓存,命中缓存的查询直接返回,延迟从200ms降到5ms。
技巧三:定期评估检索质量。我建了一个测试集,包含100个问题和对应的标准答案。每次调整参数后,跑一遍测试集,看命中率和回答准确率的变化。这个习惯让我避免了很多“感觉变好了但实际变差了”的情况。
6. 上线后的持续优化与扩展方向
RAG系统上线不是终点,而是起点。我上线后做了几件事:
第一是日志分析。我记录了每次查询的原始问题、改写后的问题、检索结果、生成回答、用户反馈。通过分析这些日志,我发现了很多之前没注意到的问题。比如有些查询的改写完全跑偏了,导致检索失败。
第二是bad case收集。我让用户可以对回答点赞点踩,点踩的case我会人工分析。分析了几百个bad case后,我发现大部分问题集中在三类:切分不合理、检索没命中、生成胡编。针对这三类,我分别做了优化。
第三是知识库更新。知识库不是一成不变的,我建了一个定时任务,每周扫描一次源文档目录,有变化的文档自动重新解析、切分、向量化。但这里有个坑:更新时要保证原子性,不能出现新旧数据混在一起的情况。我的做法是建两个collection,更新时写新collection,更新完切换别名。
第四是多路召回扩展。除了向量和BM25,我还加了基于知识图谱的召回。把文档里的实体和关系抽出来,建成一个简单的图谱。对于推理型查询,图谱召回能提供向量检索找不到的信息。这块我还在探索,目前效果有限,但方向是对的。
最后分享一个我个人的体会:RAG的优化是一个迭代过程,没有一劳永逸的方案。我一开始总想找一个“最优配置”,后来发现根本不存在。不同的数据、不同的查询分布、不同的用户预期,都需要不同的策略。所以我的建议是:先跑通一个最小可用版本,然后根据实际bad case持续迭代。每次只改一个变量,改完跑测试集,确认有效再继续。这样虽然慢,但每一步都走得扎实。