Python RAG 源码实战：从零搭建知识库，解决检索不准与答案编造

简介这份源码面向希望深入掌握大模型检索增强生成RAG技术的Python开发者与算法学习者提供一套可运行的最佳实践工程范例帮助理解检索与生成如何协同提升文本处理能力适用于搜索引擎、智能问答、自动文稿撰写等场景。资源包共22个文件、约527KB以7个XML配置文件和5个Python源码文件为核心前者负责环境与参数配置后者承载检索、查询、提示词等算法逻辑另含Markdown与文本说明、Git忽略配置、PNG示意图、IDEA工程文件及开源许可结构完整、便于二次开发。目前已有946人学习下载。读者可从中获得清晰的目录组织、模块划分思路与RAG实现骨架对照源码快速搭建实验环境理解配置与代码的配合方式并借助文档与图示降低上手门槛适合作为进阶学习与项目落地的参考模板。1. 从一份 Python RAG 源码说起为什么你搭的知识库总在“胡说八道”你大概率遇到过这种场景把公司几十份 PDF 丢进一个开源 RAG 项目问它“报销标准是多少”它答得头头是道数字却是编的。翻回原文一查压根没这句话。这不是模型笨而是检索环节把不相关的段落塞进了上下文大模型只能顺着“喂”进来的错误材料往下编。基于 Python 的大模型 RAG 检索增强生成本质就是给大模型外挂一个可查证的知识库先把文档切块、向量化、存进向量库提问时先检索出最相关的几段再连同问题一起交给大模型生成答案。它解决的是大模型“不知道你私有数据”和“爱编造”这两个硬伤适合手里有文档、想快速做出可问答知识库的 Python 开发者。这一章先把 RAG 的骨架立住后面几章带你从零跑通一套能落地的源码结构把检索命中率和答案可信度真正调上来。2. RAG 源码的四个核心模块切块、向量化、检索、生成一套能用的 RAG 源码拆开看就是四件事文档怎么切、切完怎么变成向量、提问时怎么找回最相关的块、找回来怎么喂给大模型。很多人一上来就抄 LangChain 的链式调用跑通了却不知道哪一步在拖后腿。我一般先把这四个模块单独拎出来每个都能独立替换和调试出问题才知道该改哪。2.1 文档切块chunk_size 和 overlap 怎么定切块是 RAG 里最容易被忽视、又最影响效果的一步。切太大一个块里混了好几个主题检索时噪声大切太小一句话被拦腰截断语义不完整。常见做法是按字符数切配合重叠区防止边界信息丢失。下面是一个不依赖重型框架的最小切块实现def split_text(text, chunk_size500, overlap80): # chunk_size: 每块目标字符数中文按字符算 # overlap: 相邻块重叠字符数防止句子被切断后语义丢失 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] chunks.append(chunk) # 下一块起点回退 overlap 个字符形成重叠 start end - overlap if start 0 or end len(text): break return chunks逻辑说明循环每次取chunk_size个字符下一块起点回退overlap保证跨块的句子在两块里都出现。参数上中文技术文档我一般用chunk_size400~600、overlap50~100如果是法律、医疗这类长句密集的文本块可以放大到 800overlap 提到 150。判断切得好不好有个土办法随机抽几个块读一遍如果每块都能独立看懂在讲什么就合格了。2.2 向量化与向量库选型本地小模型还是调 API切完块要转成向量才能做语义检索。这里有个选型分叉调云端 embedding API效果好但按量计费、数据出本地用本地开源 embedding 模型免费、数据不出门但需要一点显存或 CPU 算力。个人知识库和内部文档我倾向本地模型配合 FAISS 这种轻量向量库几十万块也能扛。from sentence_transformers import SentenceTransformer import faiss import numpy as np # 本地 embedding 模型首次运行会自动下载权重 model SentenceTransformer(BAAI/bge-small-zh-v1.5) chunks [第一段文本, 第二段文本] # 实际来自切块结果 # 生成向量并归一化归一化后内积等价于余弦相似度 emb model.encode(chunks, normalize_embeddingsTrue) dim emb.shape[1] # 用内积索引配合归一化向量做余弦检索 index faiss.IndexFlatIP(dim) index.add(np.array(emb, dtypefloat32)) # 检索示例 query_vec model.encode([报销标准是多少], normalize_embeddingsTrue) scores, ids index.search(np.array(query_vec, dtypefloat32), top_k3)逻辑说明normalize_embeddingsTrue把向量归一化这样 FAISS 的内积索引IndexFlatIP算出来就是余弦相似度省去额外转换。top_k是召回条数一般先取 3~5后面再重排。参数上bge-small-zh-v1.5维度 512速度快、显存占用低适合起步追求精度可换bge-large-zh但检索延迟会上去。注意向量库和 embedding 模型必须配套换模型就得重建索引否则向量空间对不上检索结果全是乱的。2.3 检索策略top_k、阈值和重排检索不是把 top_k 结果一股脑塞给大模型就完事。召回太多噪声进上下文模型容易被带偏召回太少可能漏掉关键信息。我的做法是两段式先用向量检索召回 10~20 条再用一个重排模型rerank精排取前 3~5 条。没有重排模型时至少加一个相似度阈值过滤。def retrieve(query, model, index, chunks, top_k5, score_threshold0.35): q_vec model.encode([query], normalize_embeddingsTrue) scores, ids index.search(np.array(q_vec, dtypefloat32), top_k * 4) results [] for score, idx in zip(scores[0], ids[0]): # 低于阈值的直接丢弃避免噪声进上下文 if score score_threshold: continue results.append({text: chunks[idx], score: float(score)}) if len(results) top_k: break return results逻辑说明先多召回top_k * 4再按阈值筛最后截断到top_k。score_threshold是关键参数设太高会漏召回设太低噪声多。经验值bge 系列中文模型0.35~0.45 之间比较稳具体要拿你的问题集测。判断阈值合不合适看两个指标——召回率该找到的有没有找到和精确率找到的是不是都相关两者此消彼长取平衡点。2.4 生成环节提示词模板与上下文拼接检索回来的块怎么拼进提示词直接决定答案质量。核心原则两条一是明确告诉模型“只根据给定材料回答材料里没有就说不知道”二是给材料编号方便模型引用来源。下面是一个能直接用的提示词模板PROMPT_TEMPLATE 你是一个严谨的知识库助手。请只根据下面提供的材料回答问题。 如果材料中没有相关信息直接回答“根据现有资料无法回答”不要编造。 材料 {context} 问题{question} 回答 def build_prompt(question, retrieved): # 给每段材料编号便于追溯来源 context \n\n.join( f[{i1}] {item[text]} for i, item in enumerate(retrieved) ) return PROMPT_TEMPLATE.format(contextcontext, questionquestion)逻辑说明模板里“只根据材料回答”和“无法回答”这两句是防幻觉的关键缺了模型就会自由发挥。材料编号方便你在答案里看到引用也方便排查是哪段材料导致的错误。参数上context总长度要控制在大模型上下文窗口内一般留出 1/3 给问题和回答剩下给材料超长就减少召回条数或压缩块大小。3. 从零跑通一套 RAG 源码环境、依赖与最小可运行流程上一章拆了模块这一章把它们串成一条能跑的流水线。我见过太多人卡在环境上——Python 版本不对、依赖冲突、模型下载失败。这一章按顺序走每一步都给可复制的命令和代码跑完你手里就有一个能问答的最小 RAG。3.1 环境准备Python 版本与依赖清单Python 版本建议 3.10 或 3.11太老的版本部分库不支持太新的版本有些依赖还没跟上。用虚拟环境隔离别往系统 Python 里装。# 创建并激活虚拟环境 python -m venv rag_env source rag_env/bin/activate # Windows 用 rag_env\Scripts\activate # 安装核心依赖 pip install sentence-transformers faiss-cpu numpy # 如需调用大模型 API再装对应 SDK例如 pip install openai逻辑说明sentence-transformers负责 embeddingfaiss-cpu是 CPU 版向量库有 GPU 可换faiss-gpu。依赖装完先跑一句python -c import faiss, sentence_transformers验证没报错再往下。注意faiss-cpu和faiss-gpu不能同时装冲突了先pip uninstall干净再装。3.2 文档加载与切块把 PDF 和 Markdown 变成块真实文档多是 PDF、Word、Markdown 混着来。PDF 提取文本用pypdfMarkdown 直接读。提取完统一走上一章的切块函数。from pypdf import PdfReader def load_pdf(path): reader PdfReader(path) text for page in reader.pages: # extract_text 对扫描版 PDF 无效需先做 OCR text page.extract_text() or return text def load_markdown(path): with open(path, r, encodingutf-8) as f: return f.read() # 统一入口 def load_document(path): if path.endswith(.pdf): return load_pdf(path) elif path.endswith((.md, .txt)): return load_markdown(path) raise ValueError(f不支持的格式: {path})逻辑说明extract_text()对纯文本 PDF 有效扫描件返回空字符串这种情况得先上 OCR否则后面全是空块。加载完接切块函数把长文本切成块列表。参数上PDF 提取常带多余换行和页眉页脚切块前可以用正则清一遍减少噪声。3.3 建索引与持久化一次构建多次查询每次提问都重新算向量太浪费建好索引要存盘。FAISS 支持直接写文件下次启动读回来即可。import faiss import numpy as np import pickle def build_and_save(chunks, model, index_pathindex.faiss, meta_pathmeta.pkl): emb model.encode(chunks, normalize_embeddingsTrue) index faiss.IndexFlatIP(emb.shape[1]) index.add(np.array(emb, dtypefloat32)) faiss.write_index(index, index_path) # 块文本和索引分开存靠顺序对应 with open(meta_path, wb) as f: pickle.dump(chunks, f) def load_index(index_pathindex.faiss, meta_pathmeta.pkl): index faiss.read_index(index_path) with open(meta_path, rb) as f: chunks pickle.load(f) return index, chunks逻辑说明向量存进 FAISS 索引文件原始块文本用 pickle 单独存两者靠添加顺序一一对应所以重建索引时块顺序不能变。参数上IndexFlatIP是精确检索数据量到百万级可以考虑IndexIVFFlat做近似检索换速度但需要额外训练索引。注意索引文件和元数据文件要一起备份丢一个就对不上。3.4 串起问答链路一个可运行的 main 函数把加载、切块、建索引、检索、生成串起来就是一个完整的最小 RAG。def main(): model SentenceTransformer(BAAI/bge-small-zh-v1.5) text load_document(docs/manual.pdf) chunks split_text(text, chunk_size500, overlap80) build_and_save(chunks, model) index, chunks load_index() while True: question input(提问q 退出) if question q: break retrieved retrieve(question, model, index, chunks) if not retrieved: print(未检索到相关内容) continue prompt build_prompt(question, retrieved) # 这里接你的大模型调用把 prompt 发出去拿回答 print(prompt) # 先打印看拼出来的提示词对不对 if __name__ __main__: main()逻辑说明先建索引再进问答循环retrieve返回空说明阈值卡太严或知识库里真没有直接提示用户而不是硬答。调试阶段先把拼好的 prompt 打印出来确认材料拼对了再接大模型能省很多排查时间。参数上chunk_size、overlap、score_threshold三个值建议做成配置项方便不同文档集切换。4. 检索质量调优命中率上不去的四个真实原因RAG 跑通容易答得准难。检索命中率rag hit rate是核心指标——用户问的问题正确答案所在的那块有没有被召回。命中率上不去后面生成再强也白搭。这一章讲四个我踩过的真实原因每个都能对应到具体参数。4.1 切块把答案切碎了跨块信息丢失现象用户问“第三章提到的三个条件是什么”检索回来的块每个都只提到一个条件模型只能答出一个。原因答案本身跨了多个块而检索只按单块相似度排序跨块信息天然吃亏。解决一是加大 overlap让相邻块共享更多内容二是对列表、步骤类内容切块时按结构切而不是按字符数切保证一个逻辑单元在一块里。我一般会在切块前先按标题层级分段再对每段做字符切块效果比纯字符切好不少。4.2 查询和文档用词不一致语义鸿沟现象文档里写“差旅费报销标准”用户问“出差能报多少钱”检索不到。原因字面不重合向量相似度也不够高。解决一是换更强的 embedding 模型中文场景 bge 系列比通用多语言模型好二是加查询改写让大模型先把用户口语化问题改写成几个检索友好的查询再分别检索合并结果。查询改写这一步对命中率提升明显代价是多一次大模型调用。4.3 top_k 和阈值设错召回不足或噪声过多现象要么该找到的没找到要么找回来一堆不相关的。原因top_k太小漏召回score_threshold太高误杀太低放噪声进来。解决先关掉阈值把 top_k 开到 20人工看召回结果里正确答案排第几这个排名就是你的上限再逐步调阈值观察命中率和噪声的平衡点。别拍脑袋定阈值一定要拿真实问题集测。4.4 向量库和模型不匹配换了模型没重建索引现象换了 embedding 模型后检索结果全乱相似度普遍偏低。原因旧索引是用旧模型算的向量新查询用新模型算两个向量空间对不上。解决换 embedding 模型必须重建索引没有例外。我一般把模型名写进索引元数据加载时校验不一致就报错提示重建避免这种玄学问题浪费半天。5. 避坑与排查RAG 上线前必须过的五道坎前面讲的是怎么调好这一章讲怎么不翻车。下面五条都是我在真实项目里踩过的每条按现象、原因、解决写照着排查能省不少时间。5.1 答案编造材料里没有却答得煞有介事现象问一个知识库里根本没有的问题模型照样给出一段像模像样的答案。原因提示词没约束“无法回答”的行为或者检索阈值太低把不相关材料喂了进去。解决提示词里明确写“材料中没有就回答无法回答”同时提高score_threshold检索为空时直接返回固定话术不调大模型。5.2 中文乱码PDF 提取出来全是问号现象PDF 加载后文本是乱码或空白。原因PDF 用了非标准字体编码或本身是扫描件。解决先判断是文本型还是扫描型扫描型必须走 OCR文本型乱码可换pdfplumber等库重试。加载环节加一个校验提取文本长度异常就报警别让空块进索引。5.3 检索延迟高每次提问等好几秒现象问答响应慢用户等不及。原因embedding 模型太大、向量库没建索引、或每次都在重算文档向量。解决文档向量只算一次并持久化查询向量用轻量模型数据量大时把IndexFlatIP换成 IVF 类近似索引。延迟和精度要权衡先测出瓶颈在哪一步再优化。5.4 上下文超长材料太多把窗口撑爆现象报错提示超出模型上下文长度或回答被截断。原因召回条数太多、块太大拼起来超过窗口。解决控制召回条数对材料做去重和压缩必要时只保留与问题最相关的句子。我一般按“窗口的 1/3 给材料”来倒推能放几条超了就减。5.5 更新知识库后答案还是旧的现象文档改了问答还是老答案。原因索引没重建或者缓存没清。解决文档变更后触发重建索引流程把模型名、文档版本写进元数据加载时校验版本。别指望向量库自动感知文档变化它只认你喂进去的向量。6. 进阶用重排和查询改写把命中率再提一档基础版跑通后想再往上提命中率最划算的两招是重排和查询改写。重排是在向量召回之后加一道精排用交叉编码器cross-encoder对“问题-块”逐对打分精度比向量相似度高代价是慢。查询改写是让大模型把用户问题扩写成多个检索查询覆盖不同表述再合并召回结果。这两招叠加命中率通常能比基础版明显提升但都会增加延迟和调用成本要不要上取决于你的场景对准确率的容忍度。一个实用的组合流程是向量召回 20 条 → 重排取前 5 条 → 拼提示词生成。重排模型可以用bge-reranker系列和 embedding 模型配套。查询改写则放在检索前用一次大模型调用生成 2~3 个变体查询分别检索后按相似度合并去重。下面是一个合并去重的小工具def merge_results(result_lists, top_k5): # 多个查询的召回结果合并按块文本去重保留最高分 best {} for results in result_lists: for item in results: key item[text] if key not in best or item[score] best[key][score]: best[key] item ranked sorted(best.values(), keylambda x: x[score], reverseTrue) return ranked[:top_k]逻辑说明用块文本做去重键同一块被多个查询召回时保留最高分最后统一排序截断。参数上top_k是最终进上下文的条数别设太大。这套流程的验证方法是准备 30~50 个真实问题标注正确答案所在块分别测基础版和进阶版的命中率用数据决定值不值得上重排。我自己做 RAG 最大的教训是别一上来就堆框架和花哨功能先把切块、阈值、提示词这三样调扎实命中率和可信度就赢过一大半项目。每次改参数都留个记录不然调着调着就忘了哪版最好。希望帮到你。本文还有配套的精品资源点击获取