☰
RAG从零到一:知识库问答完整链路与代码拆解
2026/10/20 6:22:58 网站建设 项目流程

1. 大模型“知识偏差”:RAG究竟在解决什么问题

先讲一个我亲身踩过的场景。某次给一个内部设备维修知识库做问答系统,直接把大模型接进来问“某型号打印机的卡纸清除步骤”,模型流利地输出了一整套操作流程——步骤听起来很专业,但和那台设备的维修手册完全对不上。后来查了原始手册,发现模型引用的零配件名称根本不存在。

这个问题的本质,是大模型训练时看到的语料不能保证覆盖你手头的私有资料,也无法更新到最新的事实。它的回答更像一种“统计性补全”,而不是“按证据作答”。RAG(Retrieval-Augmented Generation,检索增强生成)的思路,就是在大模型开口之前,先从一个可控的知识库里检索出相关的资料片段,把这些片段连同问题一起塞给模型,让它“看着材料回答”。

所以RAG的完整价值可以拆成三块来理解:

  • 缓解幻觉:生成内容有外部证据兜底,而不是模型凭空编造。
  • 知识保鲜:把新知识放到可更新的知识库中,模型无需重新训练就能掌握新内容。
  • 私有数据接入:公司内部文档、个人笔记、产品手册这类无法公开训练的数据,可以通过RAG进入问答链路。

这篇文章不是概念复述,我按完整链路一步步带你从零走到一:文档加载、文本切分、向量化、向量库检索、重排序、提示词组装、生成回答,全程用代码逐行拆解,配合我在实际项目里的调参记录和踩坑经历。适合正在做知识库问答、想理解RAG内部机制、以及被检索效果折腾而无从下手的开发者。

2. 完整链路的数据流动:从文档到答案要经过哪些站

在贴代码之前,先建立一张全局地图。很多人学RAG容易一头扎进向量数据库的细节里,却忽略了整个流水线的顺序关系。实际上,RAG的链路可以压缩成一条短语:取、嵌、存、检、排、拼、生。

环节输入输出核心问题
文档加载原始文件(PDF/Word/HTML/文本)结构化文本块如何把不同格式变成纯文本?
文本切分长文本若干chunk(文本块)多长合适?边界切在哪?
向量化chunk集合高维向量用哪个嵌入模型?维度多高?
向量存储向量+原文向量索引用什么库?相似度算法怎么选?
检索用户问题若干候选chunk召回多少个?结果准不准?
重排序多路检索结果打分排序后的chunk怎么融合多路结果?
提示词组装检索资料+用户问题prompt文本上下文放多少?指令怎么写?
生成prompt最终回答模型是否忠实于资料?

站在数据流视角看,整个系统只有两个关键接口:

  • 离线索引流程:把静态文档处理成“可检索的结构”,这一步可以在文档更新时批量执行。
  • 在线问答流程:拿到用户问题,实时检索,生成回答。

离线流程质量决定了“能查到什么”,在线流程质量决定了“能不能查好”。很多人只优化在线部分的提示词,却忘了根子在离线切分和向量化上,结果怎么调都收效有限。

我自己的经验是,做RAG项目第一步不是选模型,而是先真实跑一遍上面这张表格,记录每一步的数据形态变化。只有搞清楚每一站输入输出,后续调参才有坐标。

3. 链路的第一站:文档加载与文本切分代码逐行拆解

3.1 文档加载:统一成“纯文本”才是硬道理

不同格式的文档处理方式完全不一样:TXT直接读,PDF要解析版面,Word要考虑段落结构,HTML要剥离标签。我习惯在项目里做一个统一的加载入口,屏蔽格式差异:

# doc_loader.py from typing import List class DocumentLoader: """统一的文档加载入口,支持常见格式转换""" def __init__(self, file_path: str) -> None: self.file_path = file_path def load(self) -> List[str]: suffix = self.file_path.rsplit(".", 1)[-1].lower() loaders = { "txt": self._load_txt, "md": self._load_markdown, "pdf": self._load_pdf, } if suffix not in loaders: raise ValueError(f"暂不支持的文件类型: {suffix}") return loaders[suffix]() def _load_txt(self) -> List[str]: with open(self.file_path, "r", encoding="utf-8") as fp: return [fp.read()] def _load_markdown(self) -> List[str]: with open(self.file_path, "r", encoding="utf-8") as fp: content = fp.read() # 去除代码围栏,避免把代码marker作为正文检索 cleaned = re.sub(r"```[a-zA-Z]*\n?.*?```", "", content, flags=re.S) return [cleaned] def _load_pdf(self) -> List[str]: # 实际项目中这里会调用一个pdf解析引擎,按页提取文本 # 注意:扫描版PDF必须先走OCR,否则提取出来是空字符串 texts = extract_text_by_page(self.file_path) return texts

注意load()返回的是字符串列表,而不是单个拼接的大字符串。保持分页/分节返回,是为下一步切分保留“天然边界”——比如PDF每一页本身就是一个逻辑单位,切分时可以优先在此处断开,避免一句话被拦腰切断。

实际项目里文档类型常混在一起,我还会在加载阶段顺手记录来源信息,比如文件名、页码,这些元数据后续会一起存进向量库,用来做“回答时标注来源”的功能。没有这一步,最后的回答引用就成了无源之水。

3.2 文本切分:固定窗口、重叠、语义断点是三件套

切分是RAG链路里最容易被低估的环节。切太大,检索到的chunk里噪声多,喂给大模型后它不知道该重点看哪里;切太小,语义被截断,向量化出来的表达很单薄,召回也容易失灵。

我常用的一种切分策略是“递归字符切分”,核心思路是:按优先级顺序尝试不同的分隔符——段落换行、句号、逗号、空格——直到切出的每个块都小于设定的最大长度。

# splitter.py class RecursiveTextSplitter: def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def split(self, text: str) -> List[str]: # 分隔符优先级:从大到小 separators = ["\n\n", "\n", "。", "!", "?", ";", ",", " "] chunks = [] current = "" for para in self._split_by_seps(text, separators): # 累计当前块,超长则落盘为新块 if len(current) + len(para) > self.chunk_size: if current: chunks.append(current) current = para else: current += para if current: chunks.append(current) return self._add_overlap(chunks) def _add_overlap(self, chunks: List[str]) -> List[str]: # 重叠的核心目的是让上下文衔接, # 防止检索时恰好落在两块的交界处,丢了关键前文。 overlapped = [] for i, chunk in enumerate(chunks): if i == 0: overlapped.append(chunk) continue tail = chunks[i-1][-self.chunk_overlap:] overlapped.append(tail + chunk) return overlapped

这里有两个参数决定检索效果的上限:

  • chunk_size:我一般从400到800之间试。太短答案信息量不足,太长则会给后续重排序增加噪声。具体要看内容密度,技术文档、合同条款这类信息密度高的,建议偏小。
  • chunk_overlap:一般取chunk_size的10%到20%。它解决的是“上下文断裂”问题,比如某句话的主语在前一块、动作在后一块,没有重叠这块内容单独向量化后语义会非常不完整。

还要留意一个细节:不要盲目跨标题切分。设备手册里“安全检查”和“故障代码表”是两个不同主题,若强行拼成一个chunk,检索“故障代码”时很可能被“安全检查”的干扰碎片拖累。更好的做法是先用文档自身的标题结构做一次预切分,把预切分结果作为“语义块”,再在块内做长度控制。

这里我补一句实测体会:有一次我把一份跨部门规章制度文档切成1000字的块,检索“报销额度过高如何处理”,结果召回了“加班补贴计算”的内容,因为两块文本在小窗口向量空间里距离很近,语义边界被模糊了。后来把块降到400字并按条款边界切分,同一问题立刻命中正确条款。切分策略对结果的影响,往往比换一个更大更强的模型还要明显。

4. 第二站:向量化与向量库检索的实现细节

4.1 向量化:选模型看的是语义粒度而不是名气

文本切完之后,下一步是把每个chunk映射成一个稠密向量。这一步的选择逻辑是:向量空间里距离近的文本,语义要真的相近,而不是字面上相近。

# embedder.py class EmbeddingClient: def __init__(self, model_name: str, dimension: int = 1024): # 项目里我统一用一套在线嵌入服务, # 选型时重点比较它在垂直领域的相似度表现。 self.model = load_embedding_model(model_name) self.dimension = dimension def embed_documents(self, texts: List[str]) -> List["Embedding"]: # normalize=True的作用是把向量归一化到单位长度, # 这样后续直接算内积等价于计算余弦相似度。 vectors = self.model.encode(texts, normalize=True, batch_size=32) return [Embedding(vec) for vec in vectors] def embed_query(self, query: str) -> "Embedding": vector = self.model.encode([query], normalize=True)[0] return Embedding(vector)

我见过不少项目在这个环节踩坑:换用某个通用嵌入模型后,检索“内存不足怎么排查”时命中了一堆“硬盘空间不够”的内容——字面有重叠,语义却不在一个频道上。原理是嵌入模型本身是在大规模通用语料上训练的,它对行业内特定术语的表达能力可能不足。

所以选型时,别只看公开榜单的分数,建议自己构造一小套“领域查询-期望答案”对子,跑一次召回测试。我项目里常用的验证方法是:准备15条真实查询,手工标注出每条的正确答案,然后看模型是否能把这些答案排进前10。这个粗略的“命中率”指标,比看抽象分数直观得多。

向量维度也是一个需要关心的参数。维度越高,表达能力通常越强,但存储开销和检索耗时也会同步增长。在项目早期,建议先用线上通用的嵌入服务,后续数据量大到检索延迟不可接受时,再用量化、降维或更换轻量本地模型来优化。

4.2 向量库检索:召回数量的“甜点位”

向量存进库之后,查询时的流程就是:把用户的自然语言问句用同一个嵌入模型编码,得到一个查询向量,然后在库里找最相似的k个向量,取回对应的原始文本。

# retriever.py class Retriever: def __init__(self, vector_store, embedder, top_k: int = 5): self.store = vector_store self.embedder = embedder self.top_k = top_k def retrieve(self, query: str) -> List[Chunk]: q_vec = self.embedder.embed_query(query) # top_k决定了上下文宽度: # 太小容易漏答案,太大容易淹没关键信息。 hits = self.store.search(query_vector=q_vec, top_k=self.top_k) return hits

我最开始做QA系统时top_k直接拍脑袋设成3,效果很不稳定。后来才想明白:给大模型的上下文不是越精准越好,模型需要看到“围绕问题的多视角材料”,才能自己判断哪个片段更可信。常规调法是从5开始,评估命中率后再上下浮动。

向量库选型方面,我的经验是:

  • 数据量在百万级以内,用轻量级本地索引即可,省去部署维护成本。
  • 数据量大且需要高并发,再考虑独立的向量数据库服务。此时要关注索引类型(如HNSW)、内存占用和召回一致性。
  • 开启全文检索的混合检索,在向量召回之外再做一层关键词召回,两个结果合并,能明显提升含专有名词、编号类查询的召回率。

举个例子,检索“PT-2000传感器校准方法”,如果只用向量检索,可能把PT-2000和PT-2020的向量搞混,而关键词检索能精确锁定编号。把两路结果用下面的重排序方法合并,效果非常明显。

5. 第三站:重排序与提示词组装的坑与解法

5.1 重排序:为什么召回top5还不够

向量检索是按向量距离排序的,但这个排序不一定等于“信息相关性排序”。特别是多个来源、多路检索合并之后,需要用一个统一规则决定最终送进大模型的chunk顺序。

最常上手的方案是RRF(Reciprocal Rank Fusion,倒数排名融合)。它不依赖额外模型,只靠“每路检索结果的排名位置”来给chunk打分:

# reranker.py def reciprocal_rank_fusion(result_lists: List[List[Chunk]], k: int = 60) -> List[Chunk]: # 核心公式:score = sum(1 / (k + rank)) # 排名越靠前,贡献分数越高;k是平滑系数,控制排名差异的影响程度。 scores: dict[str, float] = {} for results in result_lists: for rank, chunk in enumerate(results, start=1): scores[chunk.id] = scores.get(chunk.id, 0.0) + 1.0 / (k + rank) ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [chunk for chunk_id, _ in ranked]

RRF的好处是轻量、透明、不依赖额外模型调用,适合作为第一版基线。缺点是它看不出来chunk内容本身到底重不重要,只依赖“被排前面”这个事实。所以第二个阶段,我一般会引入一个“重排序模型”——本质是一个二分类模型,输入是query和chunk的拼接,输出一个相关性分数。

我的建议:先上RRF跑通链路,再评估是否值得上重排序模型。如果RRF版已经能覆盖90%的问题,重排序模型的收益主要集中在长尾查询上,需要结合延迟成本和模型部署成本权衡。

另外,重排序阶段一定要限制送入的chunk数量。我见过有人试图把20个chunk全塞给大模型“让它自己挑”,prompt长度大、噪声多、生成速度慢,回答质量反而下降。我的实践是重排序后保留3到5个chunk,保证信息量又控制prompt长度。

5.2 提示词组装:指令要“逼”模型用材料说话

组装prompt这一步,很多人直接模板拼接,却忽略了两个关键设计:一是要明确告诉模型“只基于资料回答”,二是要设定一个当资料不足时怎么办的兜底行为。

# prompt_builder.py PROMPT_TEMPLATE = """你是内部知识库的问答助手。 请严格基于下面提供的资料回答用户的问题。 判断标准: 1. 如果资料里有明确答案,直接给出,并标注引用自哪个来源。 2. 如果资料不足或内容相互矛盾,明确说“资料中未找到相关依据”,禁止自行推测。 3. 不要使用资料之外的常识来补全答案。 资料内容: {context} 用户问题:{query} 回答:""" def build_prompt(query: str, retrieved_chunks: List[Chunk]) -> str: # 给chunk加编号,方便模型在回答里引用来源。 context = "\n\n".join( f"[来源{i}] {chunk.text}" for i, chunk in enumerate(retrieved_chunks, start=1) ) return PROMPT_TEMPLATE.format(context=context, query=query)

一个小改动对结果改变很大:给chunk加[来源1]这类标记。模型在回答里会自然写出“根据来源2”,这为前端展示引用来源提供了极大的便利,也方便后续做答案溯源和人工校验。

另一个容易忽略的点是“资料内容”和“用户问题”之间的排布顺序。我项目里测试过两种布局:资料在问题之前,和问题在资料之前。后者更容易让模型先理解问题,再带着问题去资料里找答案,实际生成内容的忠实度略高。这个差异可能来自模型在注意力机制上对靠近输入末尾内容的关注更强,所以把问题放在资料后面,等于让模型“带着最新指令”去读前面的资料。

5.3 生成阶段:参数对忠实度的影响比想象中大

很多人生成阶段只用默认参数,效果不稳定时才回头调。两个参数对“忠实于资料”的影响特别大:

  • temperature(温度):值越高,候选词分布越平滑,输出越随机;值越低,模型越倾向高频词,幻觉明显减少。做知识库问答,我一般定在0.1到0.3之间。
  • max_tokens(最大生成长度):给太长,模型容易在资料耗尽之后开始“自由发挥”;先估算资料能支撑的回答长度,再设置一个偏短的输出上限。

还有一个trick:在prompt末尾直接加一句“回答控制在200字以内”,比在API参数里限制max_tokens效果更自然。因为生成阶段的token限制是硬截断,可能出现话没说完就被切断;而提示词引导是软约束,模型会自己压缩表达,产出的答案完整度更高。

6. 把链路串起来:一个文档问答系统的完整实战

介绍完各个环节后,我直接贴一份自己项目里跑通的完整骨架。它是一套“电子设备维修手册问答系统”,文档不复杂,适合展示端到端流程。

# rag_pipeline.py from typing import List class RAGPipeline: def __init__(self, doc_path: str, top_k: int = 5): # 1. 加载 loader = DocumentLoader(doc_path) docs = loader.load() # 2. 切分 splitter = RecursiveTextSplitter(chunk_size=500, chunk_overlap=60) self.chunks = [] for doc in docs: self.chunks.extend(splitter.split(doc)) # 3. 向量化 + 存储 self.embedder = EmbeddingClient(model_name="domain-semantic-v1", dimension=1024) self.store = VectorStore() vectors = self.embedder.embed_documents(self.chunks) self.store.insert(vectors, texts=self.chunks) # 4. 检索器 self.retriever = Retriever(self.store, self.embedder, top_k=top_k) def answer(self, query: str) -> str: # 召回 hits = self.retriever.retrieve(query) # 此处可扩展:关键词检索 + RRF融合 # 组装prompt prompt = build_prompt(query, hits) # 调用大模型生成 return llm_generate(prompt, temperature=0.2)

运行一个真实问题“打印机报错E02如何处理”:

输入查询: 打印机报错E02如何处理 召回chunk: [来源1] 错误代码E02表示纸盒未正确关闭……请重新推入纸盒,确认卡扣弹回。 [来源2] 维护保养前必须先断开电源…… 生成结果: 根据来源1,E02错误代码表示纸盒未正确关闭,建议重新推入纸盒并确认卡扣弹回。

需要注意的是,链路跑通只代表“能工作”,离“好用”还差两步:离线评估和线上监控。

在离线评估阶段,我通常会准备一组真实的用户问题,把“检索命中率”和“生成答案人工评分”作为两个维度。检索命中率可以自动化统计,生成答案质量则需要人工抽样打分。如果命中率已经很高,但答案还是很差,问题通常在prompt或生成参数;如果命中率就很低,就要回到切分、向量化、召回策略去优化,这一步别在prompt上调来调去,浪费时间。

线上监控阶段,除了常见的接口耗时和失败率,我会额外记录两类指标:无引用回答的比例和“资料不足”兜底话术出现的比例。这两个数字能直接反映检索质量是否恶化,比随机抽几条日志靠谱得多。

7. 上线前必须做的评估:我踩过的三个典型坑

7.1 只测“标准问题”,漏了“模糊问题”

第一次上线时,我准备的测试集全是“打印机卡纸怎么办”这类清晰问题,结果线上真实用户提问五花八门:有人问“机器老卡”,有人问“纸卡住了怎么搞”,还有人问“E02是啥意思”。标准的检索链路对模糊问题表现很不好,因为嵌入模型对口语化表达的语义映射不稳定。

解决方法是准备一个“口语化问题集”,把同一问题的不同问法都放进去,评估时分别看每类的命中率。真实场景里模糊问题的比例远比想象的高,不做这层测试,上线后效果会当场打脸。

7.2 只看top1,忽略候选池

早期我做评估时,只关心检索返回的第一个chunk对不对。后来发现,这会给优化方向带来严重误导:top1不对不代表链路有问题,可能是embedding模型的候选排序不够好,也可能是切分把正确答案切碎了,正确答案在top7但没进top3,最终也没进prompt。

正确的评估口径是:先看top5/top10命中率,再结合重排序后的效果。如果top10命中率很高,说明是排序问题,重点优化rerank;如果top10里根本没有正确答案,那是召回问题,要改切分或向量化。

7.3 幻觉检测全凭肉眼,没有量化

“幻觉”在这类系统里被频繁提起,但很多人检查幻觉的方式是肉眼读几条生成结果。这个做法有一定作用,但没法规模化。我在后期做了一个简单的“忠实度检查”:把生成的回答拆成若干个事实断言,再去检索到的chunk里做包含度匹配,判断每个断言是否有出处。实用性非常高。

实现思路是,利用大模型自身做信息抽取:让模型从回答中提取关键断言列表,然后对这些断言做向量化,与原文chunk计算语义相似度,低于阈值的断言就标记为“疑似无依据”。这个方案不完美,但能快速给出一个“点评分”指标,用来对比不同切分参数和提示词版本,比漫无目的地肉眼读结果高效得多。

另外,线上日志一定要按“query文本”和“召回chunk文本”做结构化存储。这样后续观察“无引用”“兜底话术出现”时,可以直接回放当时的召回情况,快速定位是切分问题、向量化问题、还是prompt问题。

这三个坑理解透了,至少能避免我在第一阶段浪费的两周时间——那两周我一直在怀疑重排序模型不够强,后来才发现根源是切分粒度太粗,正确答案根本进不了候选池,重排序再怎么排也救不回来。

8. 从零到一之后的几条优化路径

链路跑通、评估稳定后,如果你还有余力继续往下做,我的优先级建议是这样的:

  • 第一优先:多路召回。在向量检索基础上叠加关键词检索,用RRF融合,对专有名词、编号、缩写类问题提升最明显。
  • 第二优先:query改写。用户提问太口语化时,先让大模型把问题改写成更规范的检索式,再去做向量检索。成本是一次小模型调用,收益是召回质量稳定提升。
  • 第三优先:上下文压缩。把召回的chunk先做一轮“只保留与query相关句子”的压缩,再拼进prompt,能显著降低长chunk的噪声。
  • 第四优先:评价驱动的半自动优化。把人工评估的结果沉淀成小数据集,每次改切分参数、换嵌入模型时都跑一遍同一套评估集,让优化方向可量化,而不是靠感觉。

在实际操作中,我个人的体会是:RAG系统的性能天花板,很少由模型决定,更多由“文本切分是否尊重内容结构”“召回候选池是否足够宽”“重排序是否有效过滤噪声”这三个工程细节决定。先确保这三个环节扎实,再去追最新的大模型版本才有意义。

最后分享一个减少烦恼的小技巧:在向量库里每一条chunk都存上“来源文档ID”和“原文页码”。线上回答中的引用按钮,直接从这里取数据展示。用户体验的提升非常直接,而且后续做人工审核、日志排查时,这个字段能帮你节省大量时间。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询