聊 RAG 的人很多,但能从零到一亲手把整条链路跑通、过程中每一行代码都能解释清楚的,其实没那么多。最近我集中折腾了一个知识库问答项目,核心就是 RAG——检索增强生成,目标是让大模型回答问题时基于内部文档而不是凭空瞎编。整个链路从文档解析、分块、向量化、索引存储,到检索、重排、提示词构造、最终生成,我完整跑了好几遍,踩了不少坑,也整理了一套最小可复现的实现。这篇文章就把这条链路逐段拆开讲,配套代码一个函数一个函数地讲,最后把高频问题排查清单也列出来。如果你是第一次接触 RAG,想搞懂它和微调、长上下文有什么区别,或者你已经跑过 demo 但效果不理想、不知道从哪里调起,这篇应该都适合你。
1. 先搞清楚 RAG 是干嘛的:从上下文长尾问题说起
1.1 为什么大模型需要“外挂知识库”
大模型在训练阶段学到的知识是“截止型”的。一个最近发生的行业事件,只要训练语料里没有,模型只能靠内部关联瞎猜,猜错就成了幻觉。同理,企业内部的产品手册、技术方案、合同条款,这些内容从来不在公开语料里,模型不可能答好。这时候有两类思路:一类是继续训练或微调,把知识塞进参数;另一类就是 RAG,在问答时临时检索相关材料给模型看。
我一开始也纠结过是不是要微调。后来项目里的说明书每周都在改,如果用微调,每次更新都要重训一遍,成本完全扛不住。RAG 的好处恰恰在于知识不进参数,只进索引。文档变了,重新切分、向量化一遍就行,不用动模型本身。这个“可随时更新”的特点,在现实业务里几乎直接决定了技术选型。
打个比方:微调像是让一个学生考前把所有知识点背进脑子,RAG 则是允许他开卷考试,考场上翻目录找答案。开卷的优势是教材更新了不用重新背,劣势是翻书要时间、也要翻得准。RAG 的工程难点,基本都集中在“怎么翻得准、翻得快”。
1.2 RAG、微调、长上下文怎么选
很多人在规划知识问答系统时,会在 RAG、微调、长上下文三条路线里来回摇摆。我一般用下面这张表来判断:
| 维度 | RAG | 微调 | 长上下文 |
|---|---|---|---|
| 知识更新成本 | 只需重建索引 | 需要重新训练 | 每次请求重传全文 |
| 可解释性 | 可溯源到原文片段 | 黑盒 | 可溯源,但成本高 |
| 私有知识适配 | 强 | 中 | 强 |
| 计算成本 | 低到中 | 高 | 中到高 |
| 典型场景 | 企业问答、客服 | 风格迁移、能力增强 | 单次需要整体理解超长文档 |
选型的逻辑不复杂:如果答案“能从某份文档里找到”,优先 RAG;如果想让模型整体学会某种表达风格、某个任务套路,微调更合适;如果单个样本内容特别长且必须全局理解,长上下文可以作为补充,但成本高,而且长文本中间段的注意力衰减问题也得考虑。实际项目里常见的是混合方案:RAG 为主,对反复出错的高频 case 做小规模微调兜底。
1.3 一条链路看全局:从文档到回答
RAG 完整的链路可以分为两大段:
索引构建段:文档加载 → 清洗 → 分块 → 向量化 → 构建索引 问答推理段:用户问题 → 向量化 → 召回候选 → 重排序 → 拼提示词 → 模型生成 → 返回答案
这两段合在一起,才是“完整链路”。只看生成端或者只看检索端,都容易出偏。建索引就像给一本书做目录和书签,检索环节就是你拿着问题去查目录,找到相关页码,再把几页内容抄给模型看。模型不需要背下全书,只需要知道去哪一页找答案,然后基于这几页内容组织回答。
这条链路还有一个容易被忽视的属性:每个环节的输出质量,决定了下一个环节的上限。很多 demo 效果不好,不是模型不行,而是文档没切对,或者检索回来的内容本身就不相关。带着这个“链路思维”去看后面的细节,很多坑你就能提前预判。
2. 文档进、向量出:索引构建的四个关键环节
2.1 文档清洗:垃圾进垃圾出
不管是 PDF 还是 Word,原始文本通常带大量噪声:页眉页脚、页码、超链接、表格里拆碎的字段。这些噪声如果直接进分块,会产生两种坏结果:一是向量被干扰,检索时把页眉这种无关内容当成强相关片段召回;二是分块会被噪声切断,明明同一句话,被截得上下文不连贯。
我在项目里整理了一套清洗流程:先去掉所有页眉页脚和页码,再按章节标题切分,最后用正则去掉连续空白和多余换行。清洗完的文档我会人工抽查十几个段落,确认没有乱码和错位,再进入下一步。很多人忽略这一步,把脏数据直接喂给后面的流程,结果反复调参都没好转,其实问题在源头。清洗看起来不性感,但它可能是性价比最高的环节。
PDF 类文档还要额外留意表格。表格内容一旦被线性读取,字段之间的关系就丢了。我的处理方法是把表格转成“字段名:值”的文本结构,比如“住宿上限:450元/天”,这样既能保留语义,又方便后续检索。否则模型很容易把表头和数值拆到两个块里。
2.2 分块策略:chunk 大小为什么重要
分块是 RAG 链路里影响最大、也最容易被低估的环节。chunk 太小,一个完整业务逻辑散成好几片,检索只能召回其中一片,模型缺乏上下文;chunk 太大,向量被平均化,一个几百字的片段里混了多个主题,检索精度直线下降。我的经验是,中文场景里 chunk_size 落在 256 到 1024 字之间,overlap 取 chunk 的 10% 到 20%,然后根据实际效果微调。
overlap 的作用很微妙。它能避免一句话在边界处被切开后语义丢失,让跨块的信息在每个块里都有完整表达。但 overlap 也会带来冗余,重复片段被多次喂给模型,既浪费 token 又可能造成干扰。要不要加、加多少,取决于数据是不是有强上下文依赖。有条件的团队会把结构化文档按标题层级切,而不是盲目固定长度,这对质量提升非常明显。
我在另一个项目里对比过固定长度分块和按标题分块。同样的问答集,固定长度分块的召回准确率大概在七成,按标题层级分块能到八成五以上。原因很直接:业务文档天然围绕章节组织,按标题切出来的块,每个块的主题都相对纯净,向量表达也更聚焦。
2.3 向量化:Embedding 选型
分块之后的文本不是直接进模型的,它要先变成向量。embedding 模型的质量直接决定“语义相近”能不能被数学上捕捉到。选型时我一般看三点:第一,对中文支持好不好,很多英文模型跑中文检索分数虚高;第二,向量维度与检索库的配合,维度越高,单条占用的内存越大,超大语料下索引成本成倍涨;第三,接口稳定性,线上调用一定要有批量和重试机制。
维度不是越高越好,不要把模型能力简单等同于维度大小。一个针对垂直领域调优过的 768 维模型,往往比一个 1536 维的通用模型在企业资料上更准。向量化之后一定要做归一化,这样才能统一用内积或余弦相似度比较,数值也稳定。否则不同文档向量模长不一致,相似度计算会被模长干扰。
这里还要提一个工程习惯:把向量持久化缓存起来。同一段文本,embedding 一次后就存下来,下次直接读取。个人项目里很容易忽略这个点,数据量一大,每次重建索引都要重新调用 embedding,时间和成本都很浪费。
2.4 向量存储与索引
小项目直接用内存里存列表、暴力比对就行,响应几百毫秒,完全能顶住。数据量大了就必须上近似最近邻索引,原理是用聚类或图结构,把搜索范围从全量缩小到局部。具体实现上,大多数向量数据库默认支持图索引和倒排聚类两类方案。图索引的召回率更高,但建索引慢、内存占用大;聚类方案速度快,但调参讲究。
工程里容易被忽略的是元数据过滤。跟“先查目录再翻页”一个道理,如果能先用时间、来源、文档类型把候选集缩小,再在范围里做相似度检索,查得快也更准。我遇到过不少项目,几百万条文档全量暴力检索,加一个分类过滤后,速度提升了十几倍,准确率也上去了,因为排除了很多主题相近但归属错误的分块。
还有一个细节:索引要支持增量更新。文档不可能一次性全量灌完,新文档进来、旧文档删除,索引都要能灵活变动。做知识库维护的时候,这个能力比性能更重要,否则每次更新都要全量重建,越到后面越痛苦。
3. 提问进、答案出:检索与增强生成的核心机制
3.1 Query 处理:别让用户口语直接撞文档书面语
用户大多习惯用口语提问,比如“报销单要怎么填”,而文档里写的是“报销单填写规范”。这种表述差距会让向量检索召回率下降。处理办法有很多:最简单的是把问题拆成关键词,做一趟关键词召回再和向量召回合并;进阶做法是 HyDE,先让模型根据问题写一个假想的标准答案,然后用这个答案去检索。因为“答案形态”和文档形态更接近,检索往往更准,代价是多一次大模型调用。
还有一种思路是 query 改写:用大模型把口语问题重写成适合检索的、带关键实体和限定条件的查询语句。这个需要在延迟和效果之间做取舍。实际项目里,我在冷启动阶段先用关键词加向量双路召回,积累一批 badcase 后再决定要不要上 HyDE 和改写,避免动不动就堆大模型调用,让链路变得又慢又贵。
3.2 相似度怎么算
向量建好后,相似度计算是下个问题。常用三种:余弦相似度、内积、欧氏距离。只要向量做过归一化,余弦相似度和内积结果是等价的,直接用内积最快。欧氏距离对向量模长敏感,适合那些未归一化、模长本身包含信息的情形,但一般 RAG 场景下,语义相似用余弦更稳。
相似度分数还有一个使用禁忌:不要设死阈值。不同 embedding 模型的分数分布差异很大,同样的“0.7”,在模型 A 里可能已经是强相关,在模型 B 里只是勉强沾边。所以我不会拍脑袋定“相似度必须大于 0.8”,而是先把一批真实问题跑离线评测,观察召回的分数分布,再决定 TopK 取多少。这个习惯帮我少走了很多弯路。
3.3 Rerank:为什么不能直接喂 TopK
检索返回的 TopK 只是“向量空间里距离最近”,不代表语义上最优。向量检索阶段为了速度,通常用的是双塔编码,问题和文档各自编码,交互信息少;而重排阶段用交叉编码器,把问题和一个候选片段拼起来同时送入模型,做精细化打分,相关性判断能力明显更强。一个典型的 RAG 系统里,为了控制延迟,向量检索取 Top50,重排后只留 Top3 到 Top5 进入生成环节。
增加重排是有成本的:交叉编码比向量编码慢几十倍。但在知识密度高、容易混淆的资料库场景里,重排带来的准确率提升非常明显。如果对延迟敏感,可以做一个开关:资料库规模小、主题单一,可以省掉重排;主题杂、易混,重排基本是刚需。我现在的默认策略是:只要有重排能力,就一定开,因为它把“最后那一点不准”堵住了。
3.4 提示词构造:给模型一份“开卷考试守则”
检索质量再高,最后模型也是从提示词里拿到材料的。提示词写得不好,等于把开卷材料拍在学生脸上,但没说清楚“答题只能依据材料”。至少要在系统提示词里写明:你只能基于下面的资料回答;如果资料里没有,直接说不清楚,不要编造;回答要分点,需要时引用资料编号。
上下文拼装也有讲究。我会把重排后的片段按相关性从高到低排列,用明确的编号分隔开,并在最后提醒模型“以上资料可能部分不相关,请忽略无关部分”。这个小动作看起来不起眼,实测能明显减少模型把不相关内容硬编进答案的现象。再进一步,要求模型输出答案时附上引用的片段编号,后续人工溯源会轻松很多。
这里还要注意 token 预算。检索回来的内容拼进提示词,会占据上下文窗口。如果 TopK 取太多,提示词太长,不仅延迟增加,模型还可能被大量低相关文本干扰。我的经验是:先给模型 3 到 5 条高质量片段,宁缺毋滥。开源大模型和商业 API 都遵循这个规律,不是给得越多越好。
4. 实战:手写一个最小 RAG 系统,逐行拆解
4.1 环境准备与实验数据
为了让你在无外部依赖的情况下跑通,我用标准库加 numpy 自己写一个内存向量数据库,embedding 和大模型都用占位函数演示。实际部署时,把占位函数替换成任意 embedding 服务和模型 API 就可以。实验数据用一段虚构的差旅报销制度:
某公司差旅报销制度 一、适用对象:本公司全体正式员工。 二、报销范围:交通费、住宿费、市内交通费。 三、报销标准:一线城市每日住宿上限450元,其他城市上限350元。 四、申请流程:员工出差前需在系统提交出差申请单,出差结束后5个工作日内提交报销单,附发票和行程单。 五、审批规则:金额500元以内由部门负责人审批,500至2000元由财务经理审批,超过2000元需总经理审批。 六、特殊情况:项目紧急出差可由部门主管事后补单,补单需要在3日内完成。这段文本短但结构完整,足够演示从加载到回答的全流程。真实项目里换成长文档,逻辑不变。
4.2 第一步:文档加载、清洗与分块
先看加载和清洗:
import re def load_document(path: str) -> str: """从本地文件读取文本内容。""" with open(path, "r", encoding="utf-8") as f: return f.read()load_document 的逻辑很简单,打开文件、按 UTF-8 读入、返回字符串。真实场景里这里可能是 PDF 解析结果或者从接口拉回来的文本,但接口形态可以保持统一:不管来源,最后都返回一个干净的长字符串。
def clean_text(text: str) -> str: text = re.sub(r"\s+", " ", text) # 把连续空白替换为单个空格 text = re.sub(r"[ \t]+", " ", text) # 去掉多余空格和制表符 return text.strip() # 去掉首尾空白clean_text 用两步正则把文本里的换行、制表符、连续空格统一压成单个空格。为什么要这样?因为分块时如果保留大量换行,同一句话很容易被行边界切断;而压成连续文本后,块切分只按长度走,块内语义完整性更好。代价是丢失了段落结构,所以真实项目里如果文档有明确层级,我会在清洗阶段保留“标题标记”,这里为了演示简洁,先全部压平。
再看分块:
def chunk_text(text: str, chunk_size: int = 200, overlap: int = 20) -> list[str]: chars = list(text) chunks = [] step = chunk_size - overlap for i in range(0, len(chars), step): piece = "".join(chars[i:i + chunk_size]) if len(piece) >= chunk_size // 2: chunks.append(piece) if i + chunk_size >= len(chars): break return chunks这段有几个关键点。step 是 chunk_size 减 overlap,也就是每次前进的步长。比如 chunk_size 200、overlap 20,step 就是 180,意味着每块和上一块有 20 个字符的重叠。为什么需要 overlap?因为切分点落在哪一截很难控制,如果不重叠,一条完整句子可能被拦腰切断,两边各剩一半,检索时哪一半都答不完整。重叠的字符越多,句子的完整性越有保障,但冗余也越大。
循环里先取chars[i:i + chunk_size],然后判断len(piece) >= chunk_size // 2。这个判断是为了过滤尾部不足半个 chunk 的碎片。文本最后一段往往很短,如果不足 chunk 的一半,直接丢弃,避免产生一堆检索价值低的噪声碎片。最后一个if用来在切完最后一块后退出循环,防止因为步长小于块长导致的重复切分。
4.3 第二步:向量化与内存索引
下面实现向量化。为了在无外部服务时能跑通,我用一个确定性的哈希伪随机向量来模拟 embedding 输出:
import hashlib import numpy as np def embed_text(text: str) -> np.ndarray: """将文本转为向量。真实场景请替换为实际 embedding 服务。""" seed = int(hashlib.md5(text.encode("utf-8")).hexdigest()[:8], 16) rng = np.random.default_rng(seed) vec = rng.normal(size=16) return vec / np.linalg.norm(vec)这里解释一下为什么用哈希。embedding 的输入是文本,正常情况下不同文本应该映射到不同向量,而且语义接近的文本向量也应该接近。伪随机向量只能保证“不同文本不同向量”,无法表达语义关系,所以它只适合演示链路形态,不适合真实使用。真实替换时,这个函数内部调用你的 embedding 服务,输入文本、返回几百维的向量数组即可。最后一行做了 L2 归一化,把向量模长变成 1,这样后面可以用内积直接当余弦相似度用。
索引部分用一个极简的内存字典:
class MemoryVectorStore: def __init__(self): self.vectors = [] self.chunks = [] def add(self, vector: np.ndarray, chunk: str): self.vectors.append(vector) self.chunks.append(chunk) def search(self, query_vector: np.ndarray, top_k: int = 5): scores = [float(np.dot(query_vector, v)) for v in self.vectors] top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k] return [(self.chunks[i], scores[i]) for i in top_indices]MemoryVectorStore 只有三个方法。add 把向量和对应文本追加到列表里,search 先算 query 和所有候选向量的内积,再用sorted按分数降序取前 top_k 个索引,最后返回文本和分数。这个实现是 O(n) 的暴力检索,适合几百几千条数据;数据量大了以后,把这里换成真正的近似最近邻索引就可以,接口可以保持不变。
4.4 第三步:检索后的简单重排
向量检索返回的 TopK 可以直接用,但我加了一个轻量重排函数,用来演示“重排阶段”的思路:
def rerank(query: str, candidates: list[tuple[str, float]]) -> list[str]: """轻量重排:按查询关键词在片段中的命中数量排序。""" def _score(candidate: str) -> int: return sum(1 for token in query if token in candidate) return [c for c, _ in sorted(candidates, key=lambda x: _score(x[0]), reverse=True)]这个实现很朴素:统计查询串里的每个字符在候选片段里出现多少次,按命中数排序。真实场景中重排会用交叉编码模型,但思路是一样的——从向量检索的粗排结果里,用更精细的信号挑出真正相关的片段。为什么粗排之后还要挑一轮?因为向量空间里的“近邻”不总等于语义上的“正确”,尤其在文档主题接近的时候,粗排 TopK 里可能混着好几篇没那么相关的文本,轻量重排能先把明显不相关的压下去。
4.5 第四步:提示词构造与生成
生成环节用一个占位函数代表大模型调用:
def generate(prompt: str, sources: list[str]) -> str: """真实场景请替换为大模型接口调用。""" return "(实际部署时返回模型输出)\n参考片段:" + ";".join(sources[:2])然后是主流程:
def answer_question(question: str, store: MemoryVectorStore, top_k: int = 5, keep_n: int = 3) -> str: qv = embed_text(question) # 查询向量化 candidates = store.search(qv, top_k=top_k) # 召回候选 top_segments = rerank(question, candidates)[:keep_n] # 重排后保留前 N 条 context = "\n\n".join(f"[{i+1}] {seg}" for i, seg in enumerate(top_segments)) prompt = ( "请只依据下面的资料回答:\n" + context + "\n\n问题:" + question + "\n\n如果资料中没有答案,请直接回答‘资料中未找到相关信息’。" ) return generate(prompt, sources=top_segments)answer_question 是整条链路的收口。qv 得到查询向量,store.search 做粗排召回,rerank 重排,然后[:keep_n]只取前三。这里 keep_n 是关键,为什么只取 3 条?因为喂给模型的片段越多,噪声越多,上下文越长,生成质量和速度都受影响。拼 context 时给每个片段加编号,方便模型引用。prompt 里那句“如果资料中没有答案,请直接回答”,是控制幻觉的关键一笔,它给了模型一个优雅退出的出口,比逼着它瞎编强得多。
4.6 串联运行与结果观察
这里把流程走一遍。假设已经读入文档、清洗、分块、向量化并 add 进 store:
text = load_document("sales_policy.txt") text = clean_text(text) chunks = chunk_text(text, chunk_size=100, overlap=15) store = MemoryVectorStore() for c in chunks: store.add(embed_text(c), c) question = "公司对差旅报销的审批规则是怎么样的?" answer = answer_question(question, store) print(answer)输出会分两部分:先打印分块列表,看看文本被切成了几块,每块内容是否完整;再打印答案。我建议第一次跑通时,把 store 里的 chuncks 全部打印出来检查一遍。如果某块明显语义割裂,就调整 chunk_size 和 overlap;如果某块还带着页眉页脚,就回到清洗环节。调试的顺序一定是“先看分块,再看检索,最后看生成”,这个顺序能帮你快速定位问题到底出在哪一环。
5. 实战中高频踩坑与排查手册
5.1 检索结果不对,先查这里
| 现象 | 可能原因 | 处理办法 |
|---|---|---|
| 检索返回空 | 清洗过度、文本被清空 | 打印清洗前后字符数,确认内容还在 |
| 向量维度不匹配 | 不同来源的 embedding 模型混用 | 统一模型,统一维度,重建索引 |
| 检索结果全是噪声 | chunk 太大,主题被平均 | 缩小 chunk_size,按标题层级切分 |
| 同一问题结果不稳定 | 向量未归一化 | 统一归一化后再建索引 |
| TopK 相关但生成还是答错 | 提示词缺少约束 | 加上“只依据资料回答,没有则拒绝回答” |
我踩过最典型的一个坑是:某个文档用文本加载器读进来,里面全是不可见字符,清洗正则把正文也误杀了,结果整个索引文件只有几条记录,检索当然什么都找不到。从那以后,我每次清洗完都会打印抽样文本,确认“人眼看得懂、结构没乱”才继续。
5.2 幻觉严重?别急着换模型
遇到模型胡说八道,大部分人第一反应是换更强的模型。但我在项目里观察到的规律是:幻觉和检索质量强相关。如果喂给模型的上下文里根本没有正确答案,再强的模型也会强行编一段。所以排查顺序应该是:先把这次回答对应的检索片段导出来,看片段里有没有能支撑答案的内容。如果没有,说明链路在检索阶段就断了,换模型解决不了。
其次是提示词约束。模型是生成式模型,天生倾向“流畅地输出”。你在提示词里给了它一段资料,但它不一定会严格限制自己的回答范围。写一句“如果资料中没有答案,请直接回答‘资料中未找到相关信息’”,比任何模型选型都更直接地降低幻觉。再进一步,要求答案带引用编号,用户能点开原文核对,这既是体验设计,也是事后审计的手段。
5.3 延迟上来了怎么办
RAG 在线链路里,耗时大头通常不在模型生成,而在检索和重排。优化思路有几个层次。第一,离线把向量索引建好,在线只做查询;第二,向量检索用近似最近邻索引;第三,重排只对 Top50 执行,不要对全量候选执行;第四,加缓存,同一个问题或近似问题直接命中缓存返回;第五,把 embedding 调用和检索并发执行。这些做完,大部分项目的 P95 延迟都能砍掉一大截。
还有一个容易忽略的点:把用户问题里的高频停用词去掉,比如“请问”“一下”“怎么”,可以减少无效检索。不要小看这个操作,它在低延迟场景下能显著降低候选数量,而且往往提高准确率,因为噪音更少了。
5.4 从零到一过程中我建议的调试顺序
一定要按顺序排查,不然你会陷入“改改这里、试试那里”的泥潭。我现在的固定顺序是:
- 打印分块结果,确认块长、块数和内容完整度。
- 打印检索结果,人工判断 Top5 里有没有正确答案。
- 不经过检索,直接把正确答案拼进提示词,单独测试生成效果。
- 最后再联调全链路,观察端到端表现。
这个顺序的核心思路是“单点验证”。检索不好,就先解决检索;生成不行,就先单独验证生成。把复杂链路拆成独立的单点,问题定位会快很多。很多项目跑不起来,不是模型不够强,而是中间某个环节“黑盒”了,不敢去动它。
我做这个项目最大的体会是:RAG 的难点不在某个单一环节,而在于链路思维。每个阶段的输出你都应该能看见、能检查、能单独评价。最后一个再分享一个小技巧:上线前请把被测问题的检索片段全部导出,人工扫一遍。如果检索本身质量不行,后面怎么调生成都是白费力气。这个动作帮我把线上幻觉投诉直接降了一大半,也是整个项目里我认为最值得推荐的一个习惯。