ARTICLE DETAIL

资讯详情

深耕商务建站与企业官网运营的一线实战洞察。

Python极简RAG知识库问答系统:几百行代码实现本地检索增强生成

Python极简RAG知识库问答系统:几百行代码实现本地检索增强生成 简介这是一套面向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框架依赖原生PythonLangChain / 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_size500chunk_overlap50这里的单位是字符不是token。中文场景下一个字符大概对应0.6到1个token500个字符大约就是300到500个token这个长度对大多数模型来说都是安全的。重叠的50个字符用来衔接前后文的语义避免一句话被硬生生切断后后半句丢失了前半句的主语。不过这个策略只是兜底。我在项目里留了一个优化点切块时优先在句号、换行符、问号这些自然边界处切断。具体做法是先暴力切成500字符的块然后检查这个块的结尾是不是在句子中间如果是就往前退到最近的句号处。这样切出来的块语义完整性会好很多检索的精度也会明显提升。2.2 向量化与向量存储文本切好之后下一步就是把每块文本变成一串浮点数也就是Embedding向量。这里有个概念需要澄清很多人问“Embedding模型和普通NLP模型有什么区别”其实简单说Embedding模型的任务是“把意思相近的文本映射到空间中相近的位置”所以它输出的向量天然适合做相似度计算。我在项目里用sentence-transformers库加载BGE模型一次把全部分块编码成向量。这里有个小细节编码的时候一定要设置normalize_embeddingsTrue也就是对向量做L2归一化。原因后面讲检索的时候再说。向量存储我直接用了FAISS的IndexFlatIP这是最基础的内积索引。构建方式很简单把归一化后的向量矩阵传给index.add()就行然后把索引文件保存到本地。这里要注意FAISS的索引文件和分块文本元数据是分开存的索引文件里只有向量数据没有原文。所以我在旁边还存了一个chunks.json里面按顺序放着每一块文本的原文内容这样检索出向量ID之后可以去JSON里把对应的文本找出来。2.3 检索与相似度计算检索的原理其实就是K近邻搜索。用户输入一个问题先把问题用同一个Embedding模型转成向量然后在向量索引里找出和这个向量最相似的K个向量返回对应的文本块。这里解释一下为什么编码时要normalize_embeddingsTrue。因为FAISS的IndexFlatIP计算的是内积内积的大小受向量长度影响。如果两个向量的模长差很多即使方向完全一致内积也会被模长带偏。归一化之后所有向量模长都是1内积就等于余弦相似度检索结果就纯粹由“方向一致性”决定也就是真正的语义相似度。默认top_k5也就是召回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, encodingutf-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_sizeCHUNK_SIZE, overlapCHUNK_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_okTrue) all_chunks [] infos [] for doc_path in glob.glob(os.path.join(DOC_DIR, **/*.*), recursiveTrue): if not os.path.isfile(doc_path): continue print(fprocessing: {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(ftotal chunks: {len(all_chunks)}) model SentenceTransformer(MODEL_NAME) embeddings model.encode(all_chunks, normalize_embeddingsTrue, show_progress_barTrue) embeddings np.asarray(embeddings, dtypefloat32) 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, encodingutf-8) as f: json.dump(infos, f, ensure_asciiFalse, indent2) print(build done.) if __name__ __main__: main()这段代码里值得注意的地方有几个。第一glob.glob用了递归模式docs/下的子目录也能扫描到。第二切块时的if last_period start chunk_size // 2这个条件是为了避免在一个块的太靠前位置切那样会导致块特别短浪费上下文。第三show_progress_barTrue会在编码时打印进度条文档多的时候不至于让你以为程序卡死了。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, encodingutf-8) as f: infos json.load(f) return index, infos def retrieve(model, index, infos, query, top_kTOP_K): q_vec model.encode([query], normalize_embeddingsTrue) q_vec np.asarray(q_vec, dtypefloat32) 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, start1): 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, start1): 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_urlLLM_BASE_URL, api_keyLLM_API_KEY) resp client.chat.completions.create( modelLLM_MODEL, messages[{role: user, content: prompt}], temperature0.1, ) print( answer ) print(resp.choices[0].message.content) if __name__ __main__: main()这个脚本有几个设计细节。第一temperature0.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模型回答答非所问像在自由发挥检索到的内容本身不是答案模型被误导先检查检索结果优化切块和EmbeddingPDF内容提取出来全是乱的PDF是扫描件没有文本层需要接入OCR比如PaddleOCR、Tesseract文档一多构建索引很慢CPU编码太慢或切块太多换GPU跑Embedding或减少重叠字符回答没有引用来源像在编Prompt里没有强制要求基于资料回答在Prompt中加入“资料里没有请说不知道”4.2 我踩过的几个典型坑第一个坑是中文切块时把语义切断了。早期版本我用了一个特别简单的按固定字符数切块的逻辑结果一个完整的句子被从中间劈开比如“公司价值观是”被切到一个块的末尾下一块开头变成了“客户第一”。检索的时候这两个块虽然都有“公司价值观”的表述但语义完整度都不够模型看半天也拼不出完整答案。后来我加了句子边界检测效果立刻好了很多。如果你也自己写切块逻辑一定要记住这个教训。第二个坑是向量归一化问题。有一版我构建索引时没有normalize_embeddingsTrue用的是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包之后我建议你先往里丢几份自己最熟悉的文档把检索结果逐条看一遍熟悉一下什么文本会被什么样的检索词召回来这个手感建立起来之后后续所有的优化你就知道自己该往哪个方向使劲了。本文还有配套的精品资源点击获取
返回列表
PREV
查看更多资讯
NEXT
返回资讯列表