简介：这份源码面向希望深入掌握大模型检索增强生成（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_size=500, overlap=80): # 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_size=400~600、overlap=50~100；如果是法律、医疗这类长句密集的文本，块可以放大到 800，overlap 提到 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_embeddings=True) dim = emb.shape[1] # 用内积索引，配合归一化向量做余弦检索 index = faiss.IndexFlatIP(dim) index.add(np.array(emb, dtype="float32")) # 检索示例 query_vec = model.encode(["报销标准是多少"], normalize_embeddings=True) scores, ids = index.search(np.array(query_vec, dtype="float32"), top_k=3)逻辑说明：normalize_embeddings=True把向量归一化，这样 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_k=5, score_threshold=0.35): q_vec = model.encode([query], normalize_embeddings=True) scores, ids = index.search(np.array(q_vec, dtype="float32"), 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"[{i+1}] {item['text']}" for i, item in enumerate(retrieved) ) return PROMPT_TEMPLATE.format(context=context, question=question)逻辑说明：模板里“只根据材料回答”和“无法回答”这两句是防幻觉的关键，缺了模型就会自由发挥。材料编号方便你在答案里看到引用，也方便排查是哪段材料导致的错误。参数上，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负责 embedding，faiss-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 提取文本用pypdf，Markdown 直接读。提取完统一走上一章的切块函数。
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", encoding="utf-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_path="index.faiss", meta_path="meta.pkl"): emb = model.encode(chunks, normalize_embeddings=True) index = faiss.IndexFlatIP(emb.shape[1]) index.add(np.array(emb, dtype="float32")) faiss.write_index(index, index_path) # 块文本和索引分开存，靠顺序对应 with open(meta_path, "wb") as f: pickle.dump(chunks, f) def load_index(index_path="index.faiss", meta_path="meta.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_size=500, overlap=80) 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_k=5): # 多个查询的召回结果合并，按块文本去重，保留最高分 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(), key=lambda x: x["score"], reverse=True) return ranked[:top_k]逻辑说明：用块文本做去重键，同一块被多个查询召回时保留最高分，最后统一排序截断。参数上，top_k是最终进上下文的条数，别设太大。这套流程的验证方法是：准备 30~50 个真实问题，标注正确答案所在块，分别测基础版和进阶版的命中率，用数据决定值不值得上重排。
我自己做 RAG 最大的教训是：别一上来就堆框架和花哨功能，先把切块、阈值、提示词这三样调扎实，命中率和可信度就赢过一大半项目。每次改参数都留个记录，不然调着调着就忘了哪版最好。希望帮到你。
本文还有配套的精品资源，点击获取
