简介:这是一套面向Python初学者与课程设计者的极简RAG知识库系统实现,适用于期末大作业、毕业设计或人工智能方向课程实践,旨在帮助学习者快速掌握检索增强生成(RAG)的核心流程与工程落地方法。资源包共48个文件,含31个Python源码(覆盖文档加载、文本切分、嵌入生成、重排序、Elasticsearch集成及Web服务等核心模块)、3个YAML配置文件(定义系统参数与服务配置)、1个Dockerfile及配套容器化脚本(支持一键部署),另有README.md、Makefile、pyproject.toml等工程化配置文件,完整体现现代Python项目开发规范。压缩包仅152KB,轻量易读,结构清晰——app/、service/、domain/、utils/等目录划分明确,便于理解RAG各环节职责解耦。目前已有69人学习下载,读者可直接运行调试、修改配置适配自有数据源,并基于现有框架拓展向量数据库或大模型接口,是入门RAG工程实践的高性价比参考方案。 最近把手上一个RAG项目彻底清了一遍,剪掉所有花哨的依赖和配置,最后沉淀出一个Python极简RAG知识库系统。这个zip包里的代码量不大,但核心流程一条没少:文档加载、文本切块、向量化、检索、拼接上下文、生成回答,全链路跑通大概只要几百行Python代码。今天这篇就把这套系统的设计思路、关键代码和踩坑记录完整拆出来,想自己做本地知识库问答的同学可以参考着抄作业,尤其是刚接触RAG、不想一上来就上LangChain或LlamaIndex这种重框架的人,这个项目应该能帮你把核心概念落到能跑的代码上。
先说我为什么要做这样一个极简版本。RAG这个词最近被包装得有点吓人,动辄就是知识图谱、Agent、多路召回、重排序,新人一看就劝退。但本质上,RAG可以理解成“给大模型开卷考试”:先从一个知识库里把和问题相关的内容检索出来,再和大模型的问题拼在一起,让模型基于这些内容回答。理解了这层逻辑,剩下的就是怎么把每一步做得更精细的问题。下面我从整体设计开始,一步步拆开这个项目。
1. 项目定位与整体设计思路
1.1 RAG到底解决什么问题
大模型有一个很要命的缺点,就是它学完知识之后,知识就固定在那了。你问它今年刚发生的事情,它大概率会一本正经地编一个答案,这就是所谓的“幻觉”。而知识库的问题在于,企业内部的大量文档、规范、技术资料,是模型训练时根本没见过的东西,直接问模型等于白问。
RAG的解决思路特别直白。你要回答一个问题之前,先不去问大模型,而是去你自建的知识库里做一次检索,把和问题最相关的几段文字捞出来,然后把这些文字和问题一起交给大模型,让它照着这些材料念答案。这样一来,大模型不需要“记得”你的知识库内容,只需要“读懂”你临时塞给它的材料,幻觉问题就大幅缓解了。
这个项目和热词里反复出现的“rag是什么”“rag基础原理”正好对上了。它解决的问题就是:在没有企业级基础设施的情况下,如何用最少的代码搭出一个“能回答问题、能引用原文、能扩展”的最小闭环。
1.2 极简方案的边界在哪里
很多看到“极简”两个字的人会问,那是不是性能很差、功能残缺?其实不是。极简指的是技术栈和代码量上的极简,而不是流程上的缺失。我在设计这个项目时给自己定了几条硬约束:
- 不引入LangChain这类重框架,核心逻辑全部用原生Python和轻量库实现。
- 向量数据库不用Milvus、Weaviate这类分布式服务,用FAISS本地索引文件就够了。
- 大模型部分不做微调,通过OpenAI兼容接口对接本地或云端模型。
- 整体代码量控制在几百行,让一个能看懂Python的人花半天时间就能读完。
这个方案的使用场景很明确:个人知识库、团队内部文档问答、原型验证、教学演示。数据量在几万到几十万字符级别,并发量不高,一台开发机就能跑。如果你的数据量到了百万级文档、需要多人高并发访问,那确实需要换Milvus、上分布式,但那已经不是“极简”该管的范围了。
为了让大家更直观地判断,我列一个对比表:
| 维度 | 极简方案(本项目) | 企业级方案 |
|---|---|---|
| 向量存储 | FAISS本地索引 | Milvus / Qdrant / ES |
| 框架依赖 | 原生Python | LangChain / LlamaIndex |
| 文档量级 | 适合中小规模 | 百万级及以上 |
| 部署方式 | 单机脚本 | 微服务 / K8s |
| 学习成本 | 低,半天能懂 | 高,需要理解一堆抽象概念 |
| 适用场景 | 个人知识库、原型 | 生产环境、高并发业务 |
1.3 技术选型的逻辑
这个项目里有几个核心组件选择,我单独说一下理由,因为这些取舍直接决定了项目为什么能保持“极简”:
向量存储用FAISS而不是Chroma。Chroma也是一个很好的轻量向量库,而且自带增删改查的接口,用起来更省事。但我最后还是用了FAISS,原因是FAISS更贴近底层,你能看到索引是怎么构建的、检索是怎么算相似度的。对于想学习RAG原理的人来说,这种“裸”一点的方式反而更友好。而且FAISS是Meta出品的,性能和稳定性都有保障,单机跑几十万向量完全没问题。
Embedding模型用BGE系列而不是OpenAI的text-embedding-ada-002。中文场景下,本地Embedding模型的效果并不比API差,而且不依赖网络、不产生费用、没有数据隐私问题。我选的是BAAI/bge-small-zh-v1.5,维度512,体积小,普通CPU就能跑,效果在中文语义搜索里属于第一梯队。
生成端用OpenAI兼容接口。这样设计的好处是,无论是调云端模型,还是用Ollama、vLLM、Xinference启动的本地模型,统一走一个接口,代码完全不用改。项目里默认演示的是对接本地Ollama服务,因为这样整个链路可以完全不依赖外网,真正实现“本地知识库”。
2. 核心模块拆解:一个RAG系统最不能省的几块
2.1 文档加载与文本切块策略
文档加载这一步没什么技术含量,但非常容易踩坑。PDF、Word、Markdown、TXT,不同格式有不同的解析方式。我这里用pypdf来解析PDF,用python-docx解析Word,纯文本直接按编码读。这里特别提醒一句,PDF看着简单,实际解析起来是最麻烦的,很多PDF的文本层是坏掉的,或者排版是分栏的,直接提取出来全是乱序。如果你的PDF是扫描件,那必须接OCR,这不是极简项目该干的事,所以我默认跳过了。
真正有技术含量的是切块(Chunking)。这一步决定了检索的效果,也决定了最终回答的质量。切块的核心矛盾在于:块太小,语义不完整,检索出来上下文碎片化;块太大,向量表示的语义会被稀释,而且超过模型上下文窗口后会被截断。
我默认用的是chunk_size=500,chunk_overlap=50,这里的单位是字符,不是token。中文场景下,一个字符大概对应0.6到1个token,500个字符大约就是300到500个token,这个长度对大多数模型来说都是安全的。重叠的50个字符用来衔接前后文的语义,避免一句话被硬生生切断后,后半句丢失了前半句的主语。
不过这个策略只是兜底。我在项目里留了一个优化点:切块时优先在句号、换行符、问号这些自然边界处切断。具体做法是先暴力切成500字符的块,然后检查这个块的结尾是不是在句子中间,如果是,就往前退到最近的句号处。这样切出来的块,语义完整性会好很多,检索的精度也会明显提升。
2.2 向量化与向量存储
文本切好之后,下一步就是把每块文本变成一串浮点数,也就是Embedding向量。这里有个概念需要澄清,很多人问“Embedding模型和普通NLP模型有什么区别”,其实简单说,Embedding模型的任务是“把意思相近的文本映射到空间中相近的位置”,所以它输出的向量天然适合做相似度计算。
我在项目里用sentence-transformers库加载BGE模型,一次把全部分块编码成向量。这里有个小细节,编码的时候一定要设置normalize_embeddings=True,也就是对向量做L2归一化。原因后面讲检索的时候再说。
向量存储我直接用了FAISS的IndexFlatIP,这是最基础的内积索引。构建方式很简单,把归一化后的向量矩阵传给index.add()就行,然后把索引文件保存到本地。这里要注意,FAISS的索引文件和分块文本元数据是分开存的,索引文件里只有向量数据,没有原文。所以我在旁边还存了一个chunks.json,里面按顺序放着每一块文本的原文内容,这样检索出向量ID之后,可以去JSON里把对应的文本找出来。
2.3 检索与相似度计算
检索的原理其实就是K近邻搜索。用户输入一个问题,先把问题用同一个Embedding模型转成向量,然后在向量索引里找出和这个向量最相似的K个向量,返回对应的文本块。
这里解释一下为什么编码时要normalize_embeddings=True。因为FAISS的IndexFlatIP计算的是内积,内积的大小受向量长度影响。如果两个向量的模长差很多,即使方向完全一致,内积也会被模长带偏。归一化之后,所有向量模长都是1,内积就等于余弦相似度,检索结果就纯粹由“方向一致性”决定,也就是真正的语义相似度。
默认top_k=5,也就是召回5个文本块。这个数量不是拍脑袋定的。太少了,可能漏掉关键信息;太多了,拼接起来的上下文会塞满无关内容,反而干扰模型的判断。5个块通常覆盖500到2500个字符,足够回答大多数事实性问题。
如果检索结果的相似度普遍低于0.3,基本可以断定检索失败了。可能出现的原因是:Embedding模型和知识库内容领域不匹配、切块大小不合理、查询词和文档用词差异太大。这些在后面的问题排查章节里细说。
2.4 生成:Prompt拼接与引用溯源
检索到相关内容之后,最后一步就是拼Prompt交给大模型生成回答。这个环节看起来简单,但有一个极其重要的设计原则,我在项目里反复强调:一定要告诉模型“资料里没有就回答不知道”。否则模型还是会仗着“自己懂很多”开始自由发挥,RAG的防幻觉优势就全没了。
项目里的Prompt模板是这样的:
基于以下资料回答问题。如果资料里没有相关信息,请明确说不知道。 资料: {context} 问题:{query} 回答:这个模板里没有花哨的系统和角色设定,原因很简单,极简项目不需要。真正的高手不会指望靠Prompt魔法让模型变聪明,而是靠检索质量让模型“不得不”从有限的上下文里找答案。
关于引用溯源,热词里提到的“rag的引用溯源与groundedness”是个进阶话题。要让回答有据可查,一个很朴素的做法是:检索时把每个文本块编号,在资料里用[1]、[2]这样的标签标出来,然后在Prompt里要求模型在引用到某个资料的信息时,在句子后面标注对应的编号。输出之后,你再根据编号把原文附在回答末尾。这个功能我在极简版本里没有做得很复杂,但保留了编号的接口,后续扩展很方便。
3. 实操复现:从零跑通这个极简系统
3.1 环境准备
先准备一台有Python 3.9以上版本的机器,建议直接用虚拟环境,避免污染全局环境。我习惯在项目目录里创建虚拟环境:
python -m venv .venv source .venv/bin/activate # Windows下是 .venv\Scripts\activate然后是安装依赖。这个项目的依赖非常克制,就这六个库:
pip install faiss-cpu sentence-transformers pypdf python-docx openai这里说明一下几个容易混淆的点。faiss-cpu是CPU版,够用;如果你有NVIDIA显卡,可以装faiss-gpu,但日常使用差距不大,因为FAISS检索本身就是毫秒级的,瓶颈主要在Embedding编码上。openai库是用来调用OpenAI兼容接口的,不只是OpenAI官方服务才能用,Ollama和大部分本地推理框架都兼容。python-docx在只处理纯文本的场景下可以不用装,但考虑到Word文档太常见,我还是加上了。
首先,项目目录结构是一个比较清晰的分离式结构:
rag_demo/ ├── requirements.txt # 依赖清单 ├── build_kb.py # 构建知识库:加载→切块→向量化→存索引 ├── query.py # 查询:检索→拼Prompt→调用LLM→输出回答 ├── docs/ # 放原始文档(txt/md/pdf/docx) └── index/ # 运行时自动生成,存放faiss索引和chunks.json3.2 构建知识库的完整代码
build_kb.py是第一步要运行的脚本,它负责把docs/目录下的所有文档读进来、切成小块、编码成向量、构建索引。
import os import glob import json import numpy as np import faiss from pypdf import PdfReader from docx import Document from sentence_transformers import SentenceTransformer MODEL_NAME = "BAAI/bge-small-zh-v1.5" CHUNK_SIZE = 500 CHUNK_OVERLAP = 50 DOC_DIR = "docs" INDEX_DIR = "index" def read_document(path): if path.endswith((".txt", ".md")): with open(path, "r", encoding="utf-8") as f: return f.read() elif path.endswith(".pdf"): reader = PdfReader(path) return "\n".join(page.extract_text() or "" for page in reader.pages) elif path.endswith(".docx"): doc = Document(path) return "\n".join(p.text for p in doc.paragraphs) return "" def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP): chunks = [] start = 0 while start < len(text): end = min(start + chunk_size, len(text)) if end < len(text): last_period = max(text.rfind("。", start, end), text.rfind("\n", start, end), text.rfind("!", start, end), text.rfind("?", start, end)) if last_period > start + chunk_size // 2: end = last_period + 1 chunks.append(text[start:end].strip()) if end >= len(text): break start = end - overlap return [c for c in chunks if c] def main(): os.makedirs(INDEX_DIR, exist_ok=True) all_chunks = [] infos = [] for doc_path in glob.glob(os.path.join(DOC_DIR, "**/*.*"), recursive=True): if not os.path.isfile(doc_path): continue print(f"processing: {doc_path}") text = read_document(doc_path) if not text.strip(): print(f" warning: empty content, skip {doc_path}") continue chunks = chunk_text(text) for idx, chunk in enumerate(chunks): all_chunks.append(chunk) infos.append({ "id": len(infos), "source": doc_path, "chunk_index": idx, "text": chunk, }) print(f"total chunks: {len(all_chunks)}") model = SentenceTransformer(MODEL_NAME) embeddings = model.encode(all_chunks, normalize_embeddings=True, show_progress_bar=True) embeddings = np.asarray(embeddings, dtype="float32") dimension = embeddings.shape[1] index = faiss.IndexFlatIP(dimension) index.add(embeddings) faiss.write_index(index, os.path.join(INDEX_DIR, "kb.index")) with open(os.path.join(INDEX_DIR, "chunks.json"), "w", encoding="utf-8") as f: json.dump(infos, f, ensure_ascii=False, indent=2) print("build done.") if __name__ == "__main__": main()这段代码里值得注意的地方有几个。第一,glob.glob用了递归模式,docs/下的子目录也能扫描到。第二,切块时的if last_period > start + chunk_size // 2这个条件,是为了避免在一个块的太靠前位置切,那样会导致块特别短,浪费上下文。第三,show_progress_bar=True会在编码时打印进度条,文档多的时候不至于让你以为程序卡死了。
3.3 查询脚本与LLM对接
query.py是第二个脚本,负责接收用户问题、检索、调用大模型生成回答。
import json import sys import numpy as np import faiss from openai import OpenAI from sentence_transformers import SentenceTransformer MODEL_NAME = "BAAI/bge-small-zh-v1.5" INDEX_DIR = "index" TOP_K = 5 LLM_BASE_URL = "http://localhost:11434/v1" # Ollama 默认地址 LLM_MODEL = "qwen2.5:7b" # 换成你实际拉取的模型名 LLM_API_KEY = "not-needed" # 本地服务一般不校验 key def load_model(): model = SentenceTransformer(MODEL_NAME) return model def load_index(): index = faiss.read_index(f"{INDEX_DIR}/kb.index") with open(f"{INDEX_DIR}/chunks.json", "r", encoding="utf-8") as f: infos = json.load(f) return index, infos def retrieve(model, index, infos, query, top_k=TOP_K): q_vec = model.encode([query], normalize_embeddings=True) q_vec = np.asarray(q_vec, dtype="float32") scores, ids = index.search(q_vec, top_k) results = [] for score, idx in zip(scores[0], ids[0]): if idx >= 0 and idx < len(infos): results.append({ "score": float(score), "source": infos[idx]["source"], "text": infos[idx]["text"], }) return results def build_prompt(query, retrieved): context_parts = [] for i, r in enumerate(retrieved, start=1): context_parts.append(f"[{i}] {r['text']}") context = "\n\n---\n\n".join(context_parts) prompt = f"""基于以下资料回答问题。如果资料里没有相关信息,请明确说不知道。 资料: {context} 问题:{query} 回答:""" return prompt def main(): if len(sys.argv) < 2: print("usage: python query.py \"你的问题\"") return query = sys.argv[1] model = load_model() index, infos = load_index() retrieved = retrieve(model, index, infos, query) print("\n===== retrieved chunks =====") for i, r in enumerate(retrieved, start=1): print(f"[{i}] score={r['score']:.4f} source={r['source']}") print(r["text"][:100].replace("\n", " ")) print() prompt = build_prompt(query, retrieved) client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY) resp = client.chat.completions.create( model=LLM_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.1, ) print("===== answer =====") print(resp.choices[0].message.content) if __name__ == "__main__": main()这个脚本有几个设计细节。第一,temperature=0.1,知识库问答这种场景要的是事实准确性,不是创造性,所以温度要低。第二,我把检索到的文本块在拼接前打上了编号,这样做有两个好处,你可以直观看到模型回答时到底参考了哪些资料,后续加引用溯源也方便。第三,retrieve函数里对idx做了边界判断,防止FAISS返回-1这类空结果导致程序崩溃。
3.4 运行验证
先往docs/目录里放几份测试文档。建议第一轮先用你自己的技术方案、简历、公司内部制度这类内容做测试,因为这些内容的准确性你自己心里有数,能直观判断回答对不对。
然后依次执行:
python build_kb.py python query.py "什么是RAG"正常情况下,你会看到脚本先打印检索到的文本块和对应的相似度分数,再打印模型生成的回答。
如果query.py报错提示连接不上localhost:11434,说明本地LLM服务没启动。我用的是Ollama,先执行ollama pull qwen2.5:7b拉模型,再执行ollama serve启动服务。如果你有别的推理服务,直接在LLM_BASE_URL里改地址即可。
4. 常见问题与排查技巧实录
4.1 高频问题速查表
我在做RAG项目的过程中,被问得最多的问题基本集中在下面几类,我整理成了一张表,大家可以直接对照排查:
| 问题现象 | 可能原因 | 解决办法 |
|---|---|---|
| 检索结果和问题完全不相关 | Embedding模型不适合你的领域 | 换text2vec-base-chinese、bge-large-zh,或使用领域微调的Embedding模型 |
| 相似度分数普遍很低(低于0.3) | 切块太小/太大,或查询词和文档表述差异大 | 调整chunk_size,在切块时做句子边界检测 |
| FAISS报维度不匹配的错误 | 构建索引和查询时用了不同模型 | 保证MODEL_NAME完全一致 |
| 模型回答“不知道”,但资料里明明有 | 切块把关键信息切碎了,检索时没召回 | 增大chunk_overlap,或增加TOP_K |
| 模型回答答非所问,像在自由发挥 | 检索到的内容本身不是答案,模型被误导 | 先检查检索结果,优化切块和Embedding |
| PDF内容提取出来全是乱的 | PDF是扫描件,没有文本层 | 需要接入OCR,比如PaddleOCR、Tesseract |
| 文档一多,构建索引很慢 | CPU编码太慢,或切块太多 | 换GPU跑Embedding,或减少重叠字符 |
| 回答没有引用来源,像在编 | Prompt里没有强制要求基于资料回答 | 在Prompt中加入“资料里没有请说不知道” |
4.2 我踩过的几个典型坑
第一个坑是中文切块时把语义切断了。早期版本我用了一个特别简单的按固定字符数切块的逻辑,结果一个完整的句子被从中间劈开,比如“公司价值观是”被切到一个块的末尾,下一块开头变成了“客户第一”。检索的时候,这两个块虽然都有“公司价值观”的表述,但语义完整度都不够,模型看半天也拼不出完整答案。后来我加了句子边界检测,效果立刻好了很多。如果你也自己写切块逻辑,一定要记住这个教训。
第二个坑是向量归一化问题。有一版我构建索引时没有normalize_embeddings=True,用的是IndexFlatL2,查询时手动算余弦相似度,结果分数怎么调都不对。后来我统一改成IndexFlatIP加归一化,逻辑就通顺了。这里想强调一个理念:做检索,你只需要关心“方向一致性”,不需要关心向量本身的长度。归一化之后,内积就是余弦相似度,代码更简单,结果也更好解释。
第三个坑是本地模型的能力上限。我用Qwen2.5-7B做测试,发现它有时候会无视Prompt里的“资料里没有请说不知道”,强行编一个答案。后来我把temperature调到0.1,同时在Prompt里把这句话换成了更生硬的版本:“基于以下资料回答问题,如果资料中没有相关答案,请直接回复:资料中未找到相关信息。”效果好了很多。这说明Prompt的设计确实能影响模型的“服从度”。
第四个坑是增量更新。最开始我想的是每次加一个新文档就把索引全部重建,文档少的时候没问题,文档多了之后构建一次要等好久。后来我才意识到,这种极简架构本来就不适合频繁增量更新,干脆改成“批量重建”策略:一次性把文档都丢进docs/,然后跑一次build_kb.py。如果非要增量,就要用FAISS的IndexIDMap维护文档级别的ID映射,但这会让项目复杂度上一个台阶,就需要权衡取舍了。
5. 从“能跑”到“好用”:几个低成本优化方向
极简版跑通之后,你会发现它“能用”,但距离“好用”还有一段距离。这里我分享几个性价比特别高的优化方向,它们不会破坏项目的极简性,但能明显提升体验。
先说说重排序(Rerank)。现在第一步召回5个文本块,但这里面可能有三块是不相关的,因为它只靠向量相似度。向量相似度擅长捕捉语义相关性,但对“这个块是否真的包含关键答案”这种精确匹配不敏感。重排的做法是,先做一次宽松召回,比如召回20个块,然后用一个专门的Rerank模型(比如bge-reranker-base)对这20个块重新打分,取前5个。这样能极大压缩无关内容,模型看到的结果更干净。
再说说引用溯源。我在前面的Prompt里已经给文本块打了编号,但模型可能不会自动用编号。要真正实现“回答完能告诉你依据在哪”,可以在Prompt里追加一句:“当引用到某份资料的内容时,在句子末尾标注对应的编号,例如[1]。”然后在代码里,把编号映射回原始的文档路径和文本块,附加到回答末尾。这算是RAG落地时客户和领导最喜欢问的东西,因为他们要知道答案可不可信。
然后是元数据过滤。如果知识库里同时有产品文档、技术文档、管理制度,查询“有哪些产品功能”时可能会把技术文档里的内容也捞出来。一个简单的做法是在infos里保存每个块所属的文档分类,在检索时先按分类过滤,再算相似度。这在FAISS里可以用IndexIDMap配合元数据过滤来实现,比继续用裸的IndexFlatIP稍微复杂一点,但很有必要。
最后是Agentic RAG这个方向。热词里出现了很多次“agentic rag”,在极简项目的基础上,它并不神秘。比如,当第一次检索结果相似度都低于0.5时,就说明知识库里可能没有直接答案,这时候可以让模型重新组织一个更宽泛的查询词,做第二轮检索;再比如,当用户问的是“对比A和B”这种复合问题时,可以先拆成两个子问题分别检索,再把结果合并。这些逻辑本质上就是在RAG的外面包了一圈决策能力,但核心的检索和生成模块完全不用改。
根据我个人的实操经验,极简版系统最值得投入精力的不是换更贵的模型,也不是上更复杂的架构,而是先把切块和检索调好。这两个环节做好了,哪怕生成端只是一个7B的本地模型,效果也远远好过“检索稀烂、硬上大模型”的方案。拿到这个zip包之后,我建议你先往里丢几份自己最熟悉的文档,把检索结果逐条看一遍,熟悉一下什么文本会被什么样的检索词召回来,这个手感建立起来之后,后续所有的优化你就知道自己该往哪个方向使劲了。
本文还有配套的精品资源,点击获取