ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python极简RAG知识库系统实战:从零搭建本地问答MVP

Python极简RAG知识库系统实战:从零搭建本地问答MVP 简介这份资源是Python实现的极简RAG知识库系统源码包面向人工智能方向毕业设计、课程设计及对检索增强生成感兴趣的学习者帮助理解如何将检索技术与生成模型结合构建问答系统。包内共48个文件以31个Python脚本为主体涵盖加载器、转换器、嵌入器、检索器等核心功能模块另有YAML配置、Dockerfile、Makefile等部署与工程化文件整个压缩包仅152KB轻量易用。代码目录结构清晰包含app、utils、web、service、domain等分层便于按功能阅读与二次开发。已有61人学习下载适合用于快速搭建RAG原型或作为毕设项目基础。通过阅读源码和配置文件可以掌握文档预处理、索引构建、检索排序、生成回答的完整流程同时学习到利用Docker进行环境封装和项目管理的方法对提升工程实践能力有直接帮助。1. 解压即用的Python极简RAG知识库系统它到底解决什么问题当你拿到一个名为“Python极简RAG知识库系统.zip”的源码包时第一反应可能是这里面到底是个什么黑匣子拆开来看它就是用Python把“检索增强生成Retrieval-Augmented GenerationRAG”这一套东西做成最小可用的知识库问答系统。简单的说你扔给它一堆本地文档TXT、Markdown、PDF它能基于这些文档内容回答你的问题而不是靠大模型“瞎编”。这套方案在本地跑通只需要几个核心依赖前后端代码加起来不到一千行却能把“喂给大模型私有知识”这件事落到实处。为什么我推荐用Python来搭极简RAG因为Python拥有最成熟的自然语言处理生态向量模型有sentence-transformers向量检索有FAISS接口服务有FastAPI这三件套在Windows、Linux、macOS上都能跑不需要碰Java或C那种重型工程。这个标题里的“zip”也点明了一个现实——大多数人拿到的不是Git仓库而是一个压缩包。解压、装依赖、跑通、改参数这就是接下来半小时你要做的事。适合谁明确一点这不是给人闲逛科普的而是给要动手做本地问答、做RAG实战或做私有知识库MVP的开发者。你可能会担心“极简”是不是意味着功能残缺。恰恰相反真正踩过RAG坑的人都知道项目体量与难度不成正比。极简方案绕开了庞杂的分布式组件放弃了实时同步却保住了最核心的链路文档切分、向量化、检索、拼接Prompt、交给大模型生成答案。下面我掏出自己常用的那套工程方案带你把这条链路从头捋一遍。2. 先把检索链路立住再谈代码RAG的四个核心环节与选型理由2.1 极简RAG的全链路嵌入、索引、查询、生成一个都不能少RAG的知识库系统在工程上可以拆成四个连续的环节嵌入Embedding、索引Index、查询Query、生成Generation。嵌入环节负责把文本变成向量索引环节把向量存进一个可快速检索的结构查询环节把用户问题变成同样的向量去索引里拿相似度最高的若干片段生成环节把“问题片段”拼成Prompt交给大模型输出最终答案。我见过不少翻车案例有人只做了“向量化相似度搜索”就以为完事了结果用户问“发票报销流程是什么”系统返回的是完全无关的闲聊内容。原因就在于生成环节没有做或者检索到的片段根本没被拼接进Prompt。极简RAG不是砍掉某个环节而是每个环节都用最轻量的方式实现。你那份zip包解压后如果按我的习惯组织核心会有五个Python文件embedding.py负责加载向量模型index.py负责建索引与检索splitter.py负责文件切分llm.py负责调用大模型接口main.py负责命令行与API入口。文件不多但每一块都缺一不可。另外有句大实话RAG的瓶颈往往不在模型而在检索效果。很多人以为换个大模型就好了实际上检索回来的片段如果是错的大模型再强也会一本正经地给出错误答案。所以这套系统里我会把更多的参数调试精力放在切分大小、TopK、相似度阈值上而不是纠结是用GPT还是Claude。2.2 技术选型为什么是FAISS 本地Embedding而不是LangChain全家桶做RAG入门很多人第一反应是直接上LangChain。我的建议是别。LangChain封装层次多出了问题你很难判断是哪个环节挂了而且它自带的概念太多——Document Loader、Text Splitter、Vector Store、Retriever——对“极简”系统来说是个负担。你需要的只是一个能跑的检索链路而不是一整套框架。所以这里采用FAISS作为向量索引库用faiss-cpu这个包就够了不需要GPU。FAISS是Meta开源的单机跑还在百万量级以下没有压力索引文件还能落盘保存下次启动直接load这比每次重新向量化整个文档省时间得多。Embedding模型我习惯用sentence-transformers加载本地模型比如BAAI/bge-small-zh-v1.5中文或shibing624/text2vec-base-chinese。选本地模型的原因只有一个向量化文档通常在离线环境做你总不希望每次启动都要访问HuggingFace下载权重。把这些模型下载一次缓存在本地之后即使断网也能用。至于大模型生成环节为了极简和避免被某个云厂商绑定我采用的是一种“兼容OpenAI接口”的方式——本地起一个Ollama服务或者连接任何提供OpenAI兼容API的在线服务只改环境变量里的base_url和api_key就行。这里要给一个选型表格方便你按场景决定组件组件极简推荐替代方案选择理由Embedding模型BAAI/bge-small-zh-v1.5text2vec, OpenAI text-embedding-3-small本地运行、免费、维度低向量索引FAISS (IndexFlatIP)Chroma, Milvus单文件落盘无服务依赖文本切分自写按段落长度切分LangChain RecursiveCharacterTextSplitter逻辑透明可控性强生成模型Ollama qwen2.5OpenAI / DeepSeek API私有部署成本低接口一致Web APIFastAPIFlask自带交互文档异步支持好2.3 依赖安装与目录结构拿到zip后第一件事不是读代码而是跑通环境解压zip后第一件事不是急着打开源码看而是确认目录结构和安装依赖。一个常规的Python安装步骤在Windows上注意勾选“Add Python to PATH”在macOS/Linux上检查python3 --version。有的读者习惯用conda那也完全可以。我一般会给项目准备一个requirements.txt内容极简到只剩六个包fastapi、uvicorn、sentence-transformers、faiss-cpu、numpy、openai。其中openai库不是用来访问OpenAI官网的而是因为它对兼容接口支持得最好社区生态也统一。如果你拿到的zip里没有requirements.txt那就手动执行下面这段命令pip install fastapi uvicorn sentence-transformers faiss-cpu numpy openai这里解释一下为什么没有包含torch。sentence-transformers会自动拉取PyTorch所以不需要你显式安装但代价是首次安装体积很大约2GB。如果你在安装时遇到网络慢常见的处理是使用国内镜像源pip install -i https://pypi.tuna.tsinghua.edu.cn/simple fastapi uvicorn sentence-transformers faiss-cpu numpy openai安装结束后建议先做一个自带冒烟测试在命令行输入python -c from sentence_transformers import SentenceTransformer; print(SentenceTransformer(BAAI/bge-small-zh-v1.5))。如果能正常打印模型信息说明torch和模型加载这条路是通的。这里有个血泪经验很多人把整个环境装完卡在这一步发现HuggingFace连不上才想起来没有提前下载模型。所以我强烈建议你第一次运行时单独执行一个下载缓存模型的脚本把模型权重先拽到本地。后面所有向量化操作都不会再走外网。3. 用Python实现核心模块从文档切分到能回答问题的检索3.1 文档读取与切分一个自写的按段落长度切分器大部分知识库文档是Markdown或纯文本。PDF也能转但极简系统里我先把注意力放在纯文本和Markdown上。切分的核心思路是先按段落分再按字符长度截断同时保留相邻片段之间的重叠避免一句话被截成两半导致语义断裂。这里给出一个最小可用的切分器代码# splitter.py import re from typing import List def split_document(text: str, chunk_size: int 500, chunk_overlap: int 50) - List[str]: 将纯文本按段落和最大长度切分为chunk列表 # 先按空行切出段落 paragraphs re.split(r\n\s*\n, text.strip()) chunks [] current for para in paragraphs: para para.strip() if not para: continue # 如果当前chunk加上新段落会超长先把当前chunk收掉 if len(current) len(para) 1 chunk_size: if current: chunks.append(current) current para else: current current \n para if current else para # 如果段落本身超过chunk_size按硬长度切分 while len(current) chunk_size: chunks.append(current[:chunk_size]) current current[chunk_size - chunk_overlap:] # 保留重叠 if current: chunks.append(current) # 进一步截断防止单chunk仍然过长 final_chunks [] for chunk in chunks: if len(chunk) chunk_size: start 0 while start len(chunk): final_chunks.append(chunk[start:start chunk_size]) start chunk_size - chunk_overlap else: final_chunks.append(chunk) return final_chunks这段代码的逻辑并不复杂先用正则按两个以上换行切出自然段落保证同一个主题尽量不被拆散然后以chunk_size500为上限把段落合并成块。如果某个段落本身就超过500字就按chunk_size - chunk_overlap的步长强制截断同时保留50字的尾部与下一块重叠减少关键信息正好落在边界的情况。关于chunk_size和chunk_overlap这两个参数我说下经验值。中文场景下500字大约对应一个能自圆其说的知识点段落如果你的文档是技术规范句子密、逻辑强建议调到300如果是访谈类、口语化内容可以放宽到800。重叠值一般取chunk_size的10%到20%。如果检索老是漏掉关键信息优先检查是不是chunk_overlap设成了0。这个参数看着不起眼但在RAG实战里它直接影响召回率。3.2 向量化与入库用FAISS建立索引并落盘文档切好之后进入嵌入环节。这里使用sentence-transformers库加载本地中文模型把每个份chunk变成向量。首先我们要把切好的chunk全部向量化再写入FAISS索引最后把索引文件和chunk原文一起保存下来这样下次查询就不用重新跑一遍文档。下面这个代码块解决的是“索引建立与保存”# index.py import json import numpy as np import faiss from sentence_transformers import SentenceTransformer def build_index(chunks: list, model_name: str BAAI/bge-small-zh-v1.5, index_path: str faiss.index, meta_path: str chunks.json): model SentenceTransformer(model_name) # 批量编码normalize_embeddingsTrue让向量变成单位向量便于用内积计算相似度 embeddings model.encode(chunks, normalize_embeddingsTrue, batch_size32) dim embeddings.shape[1] # IndexFlatIP是内积索引配合归一化向量等价于余弦相似度 index faiss.IndexFlatIP(dim) index.add(embeddings.astype(float32)) # 保存索引和原始文本 faiss.write_index(index, index_path) with open(meta_path, w, encodingutf-8) as f: json.dump(chunks, f, ensure_asciiFalse, indent2) print(f建索引完成共{len(chunks)}个chunk向量维度{dim})这里有几个值得展开讲的点。normalize_embeddingsTrue是一个常见的细节它把所有向量都变成模长为1然后IndexFlatIP内积的结果就是余弦相似度数值范围在-1到1之间便于比较和设阈值。如果你不归一化内积分数受向量长度影响结果会看起来很大或很小容易被误解。batch_size32是控制显存或内存占用如果你的机器内存只有8GB建议把batch_size调到16。FAISS索引建好后原文件可以删掉了下次查询只需要加载faiss.index和chunks.json。这一点对需要频繁更新知识库的人来说很重要你不需要重新对旧文档做全文切分和向量化只要对新文档执行同样的流程再把它追加到原索引里就行。追加操作见后面的“增量更新”小节。3.3 检索与生成拼接Prompt拿回一个有理有据的回答检索是整个知识库系统最直接暴露效果的一环。思路很直接把用户问题喂给同一个embedding模型得到查询向量再用FAISS去索引库中查TopK个最相似的chunk最后把这些chunk塞进Prompt。我会把检索和生成写成两个函数便于分开调试# rag_core.py import numpy as np import faiss from sentence_transformers import SentenceTransformer def retrieve(query: str, model, index, chunks: list, top_k: int 3, score_threshold: float 0.45): 检索相关chunk返回[(chunk, score)] q_vec model.encode([query], normalize_embeddingsTrue).astype(float32) scores, indices index.search(q_vec, top_k) results [] for score, idx in zip(scores[0], indices[0]): if score score_threshold: continue # 过滤掉相似度过低的片段 results.append((chunks[idx], float(score))) return results def generate_answer(query: str, contexts: list, client, model_name: str qwen2.5:7b) - str: 基于检索到的contexts生成最终回答 context_block \n\n---\n\n.join([f[知识片段 {i1}]\n{c} for i, c in enumerate(contexts)]) prompt f你是企业内部知识库助手。请严格根据下面的知识片段回答问题。如果知识片段中没有答案请直接说“知识库中没有找到相关信息”不要编造。 知识片段 {context_block} 用户问题{query} response client.chat.completions.create( modelmodel_name, messages[{role: user, content: prompt}], temperature0.2, max_tokens512 ) return response.choices[0].message.contentscore_threshold0.45是我在中文知识库上一个比较中庸的起点。这个值受embedding模型影响很大bge系列的相似度分数普遍偏高text2vec则偏低。建议你在自己的数据上跑几条样本打印出score和对应chunk肉眼判断一下阈值该设多少。如果不设阈值返回的全部结果都会被塞进Prompt轻则浪费token重则让模型被无关片段干扰。检索环节调得像玄学时可以先关闭生成直接打印检索结果确认返回片段是否合理再去调生成。记住检索错误是根因生成错误只是结果的放大器。4. 把整个系统跑起来命令行交互与FastAPI接口4.1 让系统先以命令行模式转动最小可用的main.py不要让API和前端干扰你验证核心链路。我会先写一个命令行入口用一段脚本把“加载索引 → 接收问题 → 检索 → 生成”完整跑通。这里直接给代码# main_cli.py import json import faiss from openai import OpenAI from sentence_transformers import SentenceTransformer def load_model(): model SentenceTransformer(BAAI/bge-small-zh-v1.5) index faiss.read_index(faiss.index) with open(chunks.json, r, encodingutf-8) as f: chunks json.load(f) return model, index, chunks def cli_loop(model, index, chunks): client OpenAI( base_urlhttp://localhost:11434/v1, # 这个地址指向Ollama的兼容接口 api_keyollama # Ollama本地服务不校验key占位即可 ) print(知识库已加载输入问题开始问答输入exit退出) while True: query input(\n问题).strip() if query.lower() in (exit, quit): break contexts retrieve(query, model, index, chunks, top_k3) if not contexts: print(未找到足够相关的知识片段请换个问法或检查索引。) continue answer generate_answer(query, [c for c, _ in contexts], client) print(回答, answer) if __name__ __main__: model, index, chunks load_model() cli_loop(model, index, chunks)此处base_url指向了Ollama的默认地址。如果你使用在线大模型API只需要把base_url改成对应服务商提供的兼容地址api_key改成你的keymodel_name改成对应模型名。代码里没有硬编码model_name传入generate_answer而是使用了默认参数这样切换模型只需要改一处字符串。首次跑通后我需要你做三个验证动作。第一问一个答案明确在文档里的问题比如“报销流程分几步”确认回答内容与原文一致。第二问一个文档里没有的问题确认系统会回答“没有找到相关信息”而不是强行编造。第三故意模糊提问比如把“报销”写成“爆消”看看检索是否还能命中。如果第三步效果很差说明你的embedding模型对错别字的容错不高这属于正常现象后面可以考虑在前处理时加入纠错。4.2 用FastAPI暴露一个交接给前端的HTTP接口命令行跑通之后前端或小程序才能接进来。FastAPI是我见过最省事的API方法它自带Swagger交互文档你写完接口后打开/docs就能手动测试。下面这个接口接收JSON格式的{question: ...}返回答案和命中的知识片段# api.py from fastapi import FastAPI from pydantic import BaseModel import uvicorn from main_cli import load_model, retrieve, generate_answer app FastAPI() model, index, chunks load_model() class Query(BaseModel): question: str app.post(/ask) def ask(q: Query): contexts retrieve(q.question, model, index, chunks, top_k3) if not contexts: return {answer: 知识库中没有找到相关信息, contexts: []} answer generate_answer(q.question, [c for c, _ in contexts], client) return {answer: answer, contexts: [c for c, _ in contexts]} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)这里有一个值得注意的问题client对象在api.py里被我直接引用了但上面代码没有定义它。实际上我建议把它放在load_model()里一起创建或者单独封装一个create_client()。避免在全局写死某个模型的依赖影响后续切换。改法很简单在load_model里加一行client OpenAI(base_url..., api_key...)并随函数返回。这里为了演示接口结构刻意省略但实际项目中不要这样写否则你会遇到NameError这是新手最容易踩的坑之一。启动API服务的命令是uvicorn api:app --host 0.0.0.0 --port 8000。注意端口不能被防火墙拦截局域网其他机器想要访问时host必须设成0.0.0.0而不是127.0.0.1。如果你是在macOS上搭建这套系统直接用这个方式也是一样不需要额外处理macOS专属依赖。4.3 知识库首次启动的三步自检黑匣子不可信眼见为实很多人在系统跑起来后就急着接入大模型结果回答质量不理想却找不到问题在哪。我建议你启动后先做三步黑盒测试尤其是本地生成的环节。第一步检查索引加载时间如果加载时间超过几十秒说明SentenceTransformer在读模型时走了网络下载这没做到位正确情况下应该是几百毫秒加载本地缓存。第二步检索打印调试在retrieve函数里临时把检索分数打印出来确认第一次查询时分数普遍低于0.4那说明知识库向量与问题向量完全没有对齐第三步生成答案时把拼接好的Prompt打印出来观察到底有没有把知识片段正确嵌入。这三大步就像设备的自检灯能帮你确认故障出在检索、生成还是模型加载。这节看起来不是写代码但对于实际部署非常重要。如果你跳过了这里后面任何错误都会被黑匣子吞掉到时候你只能抓瞎。我自己在做RAG实战时往往要花一半时间在这个自检环节而不是修改Prompt。5. 避坑清单RAG知识库系统最容易翻车的五个细节5.1 换Embedding模型后向量维度对不上导致索引崩溃现象你之前用text2vec建了索引今天换了个新模型再加载旧索引时直接报错提示维度不一致。原因不同embedding模型的输出维度不同text2vec输出768维bge-small输出512维faiss索引一旦建立维度就已经固定在索引文件里了。解决在重建索引前检查新模型的输出维度并与现有索引维度比较。我一般习惯把模型名和维度写进一个config.json索引文件和它绑定在一起。凡是更换模型必须重新执行build_index全量更新不能混用。这里还有一个隐藏坑哪怕两个模型维度相同它们生成的向量空间也是不同的使用旧的chunk向量与新的查询向量做检索结果等于随机。所以不要因为维度一样就直接沿用旧索引这是RAG实战里最隐蔽的翻车原因之一。5.2 chunk切分过小或过大检索结果总是不对劲现象回答中经常出现“知识库中没有找到相关信息”但你明明把答案写进了文档。原因chunk_size设置得太小时一个完整知识点被拆成两半检索Term匹配的分值被稀释设置得太大时单个chunk包含太多无关内容又会被大模型当成噪音。解决用小样本测试集逐个验证chunk_size。我建议在你的真实文档里挑5个具有代表性的问题分别用300、500、800三个chunk_size跑一遍看哪个参数下命中率最高。这个参数是纯测试驱动的不要靠查表属于“你的数据你做主”的范畴。另外chunk_overlap设成0会让问题更严重。试想知识边界刚好落在一句话中间前后两个chunk都只有半句话检索出来自然是残缺信息。保证重叠区域在50字以上基本能避免这类惨案。5.3 大模型接口调用时token超限回答被截断现象知识片段过长的请求直接报错“maximum context length exceeded”或者回答到一半戛然而止。原因你没有统计prompt的总token数把多个chunk一股脑拼进去超出了模型上下文窗口。解决在generate_answer里加一个字符长度校验比如总长度超过context_limit_chars就减少chunk列表。更合理的做法是让retrieve返回片段时就按score从高到低累加长度直到达到某个上限。比如设置max_context_chars1500每往里加一个chunk前判断是否超限超了就不再加入。这比单纯靠TopK更可控因为不同chunk长度差别很大。5.4 多线程场景下重复加载模型内存直接爆掉现象你给FastAPI接上了多worker部署比如启动四个uvicorn worker每个worker都会加载一份embedding模型和FAISS索引8GB内存的机器直接OutOfMemory。原因embedding模型文件几百MB到1GB不定多进程复制导致内存翻三倍。解决把模型加载挪到应用启动时只执行一次并用全局变量持有如果你必须多worker就限制worker数量不超过机器内存能承受的范围或者把embedding服务单独拆成一个预加载模式。极简系统建议单worker就够了毕竟本地知识库的QPS要求通常不高。这个坑在知乎“rag知识库能存储图片嘛”这种新手问答里也常被忽视大家只想着把功能堆上去忘了资源边界。5.5 中文文档编码报错decode后满屏乱码现象读取本地TXT或Markdown文件时报UnicodeDecodeError或者内容里中文乱码。原因文件实际编码可能是GBK或GB18030但open()默认用UTF-8去解。解决极简方案里读取文件时用errorsignore不是好办法它会静默丢字。正确做法是先探测编码优先UTF-8失败后再尝试GB18030。一段通用代码def read_text_file(path: str) - str: for enc in (utf-8, gb18030, latin-1): try: with open(path, r, encodingenc) as f: return f.read() except UnicodeDecodeError: continue raise ValueError(无法识别文件编码)这段代码按顺序尝试三种编码用户正常中文文档基本能落地。这里有个额外提醒如果你是从Word复制到文本里的内容经常夹带特殊不可见字符建议在切分前用正则过滤掉\u00a0不间断空格和零宽字符否则你会看到chunk里莫名其妙多出古怪空位。6. 让极简RAG真正可用三个低成本进阶技巧当你的极简系统跑通后还有三个不换架构就能明显提升体验的技巧重排序、增量更新、定期评估。先说重排序很多人检索回来的TopK结果看着分数差不多但里面混着完全无关的内容。常见做法是再用一个cross-encoder模型对TopK结果和问题进行打分取分数最高的两三条。sentence-transformers也支持cross-encoder比如cross-encoder/ms-marco-MiniLM-L-6-v2这个模型体积小但重排序效果却很明显。你只在检索后多花几十毫秒就能把召回的死角洗掉一层。第二个技巧是增量更新。前面说FAISS索引支持add但很多入门者不知道。新文档来了之后只需要对新文档做切分和向量化然后index.add(new_embeddings)同时把新chunk追加到chunks.json里。不要重建整个索引。不过这里有一个必须注意的前提FAISS的IndexFlatIP不允许删除指定向量如果你删除了旧文档必须全量重建。所以极简思路里我建议把更新策略设计成“定期重建索引”比如每天凌晨重建一次而不是实时删改。这样既简单又稳定避免在“删除向量”这个功能上踩进FAISS的限制里。第三个技巧是验证兜底。RAG系统最大的幻觉不是大模型编造而是你不知道自己系统的检索质量在哪个水平。我一般会准备20个“问题-期望答案关键词”对每次调整chunk或模型后跑一遍统计命中率。这部分脚本很短就是一个遍历测试集、记录命中关键词的循环。用它来测试比肉眼抽查要可靠得多。如果命中率低于60%那说明知识切分或阈值设置有问题先不要急着去微调大模型Prompt。Prompt调得再多检索跑偏也是白搭。最后说说我的一个习惯我会把每个调过的参数组合写在experiment_log.md里比如“chunk_size500, top_k3, threshold0.45命中率70%”。因为RAG参数之间的影响不是线性叠加的今天调大的阈值明天调小的chunk效果可能互相抵消。没有日志你连是自己调参调差了还是数据变了都分不清。这也是我从多次翻车中学到的教训。希望这篇笔记能帮你绕开那些我走过的弯路让你手里这份“Python极简RAG知识库系统.zip”真正跑起来成为你有据可查的私有问答工具。本文还有配套的精品资源点击获取
返回列表