简介这是一套面向Python全栈开发者与RAG技术学习者的智能文档检索系统完整源码围绕检索增强生成RAG构建解决企业或个人知识库中文档解析、向量化存储与智能问答的落地问题。系统采用Streamlit前端搭配Python后端集成MySQL数据库、向量存储、异步文档处理、流式响应、用户认证、角色权限控制与文件类型限制等模块并配有登录、密码重置、个人资料、管理员面板等页面适合作为课程设计、毕业设计或技术练手项目。资源包共62个文件以24个py源码为核心辅以18个pyc编译文件、5个css与3个js前端资源、4个xml配置及docx说明文档等整体约175KB目录按modules、pages、uploads等分层组织结构清晰。目前已有72人学习下载。读者可获得一套可直接运行的RAG检索问答工程理解文档解析、向量化、增强检索与流式输出的完整链路并参考权限控制与数据库连接池等实现思路快速搭建自己的智能文档检索应用。1. 从一份能跑起来的 RAG 文档检索系统说起很多人第一次接触 RAG是从「把 PDF 丢给大模型问答」开始的结果要么检索召回一堆无关段落要么回答里全是幻觉。这份资源给的是一个完整可落地的智能文档检索系统Python 后端负责文档解析、向量化、检索与流式问答MySQL 存用户、角色、会话和文档元数据Streamlit 做前端交互还带用户认证、角色权限控制和文件类型限制。它解决的不是「RAG 是什么」而是「一套能登录、能分权限、能上传文档、能流式回答的 RAG 系统到底怎么拼起来」。适合已经会点 Python、想拿一个能改能扩的 RAG 项目练手或直接二次开发的人也适合想搞清楚向量存储和 MySQL 各自该管什么的人。2. 系统骨架拆解Python 后端、MySQL 与向量存储各管什么2.1 为什么用 MySQL 而不是全塞进向量库RAG 项目最容易犯的错是把所有东西都往向量库里塞。向量库存的是语义向量擅长相似度检索但它不擅长做用户认证、角色权限、会话归属这类强关系、强事务的查询。这份资源把职责切得很清楚MySQL 管结构化数据向量库管语义检索。具体分工是这样的数据类别存储位置原因用户账号、密码哈希MySQL需要唯一约束、事务、登录校验角色与权限映射MySQL关系型查询权限判断频繁文档元数据文件名、类型、上传者、时间MySQL需要按用户/角色过滤文档切块后的向量向量存储语义相似度检索会话与消息记录MySQL需要按会话 ID 关联查询常见做法是文档上传后先落 MySQL 元数据再解析切块、向量化写入向量库两边用同一个文档 ID 关联。检索时先用 MySQL 按当前用户的角色过滤出「他有权访问的文档 ID 集合」再在向量库里只对这个子集做相似度搜索。这一步很关键否则权限控制就是摆设——你前端藏了按钮后端检索照样把别人的文档召回出来。2.2 文档解析与向量化的落地步骤文档处理是整条链路里最容易翻车的一环。这份资源支持文件类型限制说明它在入口就做了白名单校验。我一般会按下面的顺序搭# document_processor.py import os from typing import List # 允许的文件类型白名单和前端上传组件保持一致 ALLOWED_EXTENSIONS {.pdf, .txt, .md, .docx} def validate_file(filename: str) - bool: 校验文件扩展名是否在白名单内防止上传可执行文件 ext os.path.splitext(filename)[1].lower() return ext in ALLOWED_EXTENSIONS def split_text(text: str, chunk_size: int 500, overlap: int 50) - List[str]: 按固定长度切块保留 overlap 避免语义被切断 chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) start end - overlap # 回退 overlap 个字符保证上下文连续 return chunks逻辑说明validate_file在文件进入解析流程前就拦掉非法类型这是第一道防线。split_text用固定窗口加重叠切块chunk_size控制单块长度太大检索粒度粗太小语义不完整overlap是防止一句话正好被切在边界上导致两边都读不通。参数上中文文档我一般chunk_size取 400 到 600overlap取 50 到 100英文可以适当放大。向量化这一步常见做法是用一个 embedding 模型把每个 chunk 转成向量连同文档 ID、chunk 序号一起写入向量库。注意向量库里的每条记录都要带上doc_id和owner_role这类过滤字段否则后面做权限过滤时你只能全量召回再在内存里筛性能会很难看。2.3 Streamlit 前端与流式响应的接法Streamlit 做 RAG 前端有个天然优势写起来快st.chat_message和st.chat_input直接给你一套聊天界面。但流式响应要接对否则用户会盯着转圈等好几秒。# app.py import streamlit as st from backend.qa_chain import stream_answer st.title(智能文档检索系统) # 会话状态里保存历史消息避免每次重跑丢上下文 if messages not in st.session_state: st.session_state.messages [] for msg in st.session_state.messages: st.chat_message(msg[role]).write(msg[content]) if prompt : st.chat_input(输入你的问题): st.session_state.messages.append({role: user, content: prompt}) st.chat_message(user).write(prompt) with st.chat_message(assistant): # 用 write_stream 逐块渲染实现打字机效果 response st.write_stream(stream_answer(prompt)) st.session_state.messages.append({role: assistant, content: response})逻辑说明st.session_state是 Streamlit 的会话级存储页面重跑时不会丢。st.write_stream接收一个生成器每 yield 一个片段就渲染一次这就是流式响应的前端落点。后端stream_answer需要是一个生成器函数内部先做检索、拼 prompt再调用大模型流式接口逐 token 吐出。参数上要注意如果后端一次性返回整段再切分那不叫流式用户体感没区别必须是从模型接口层就是流式的。3. 用户认证与角色权限控制别让检索绕过权限3.1 认证流程与密码存储用户认证这块很多人图省事直接明文存密码这是血泪经验级别的错误。正确做法是存哈希用bcrypt或passlib都行。# auth.py from passlib.hash import bcrypt import mysql.connector def register_user(username: str, password: str, role: str user): 注册用户密码只存哈希绝不存明文 hashed bcrypt.hash(password) conn mysql.connector.connect(hostlocalhost, userroot, passwordyour_pwd, databaserag_system) cursor conn.cursor() cursor.execute( INSERT INTO users (username, password_hash, role) VALUES (%s, %s, %s), (username, hashed, role) ) conn.commit() cursor.close() conn.close() def verify_user(username: str, password: str) - dict: 校验登录返回用户信息或 None conn mysql.connector.connect(hostlocalhost, userroot, passwordyour_pwd, databaserag_system) cursor conn.cursor(dictionaryTrue) cursor.execute(SELECT * FROM users WHERE username %s, (username,)) user cursor.fetchone() cursor.close() conn.close() if user and bcrypt.verify(password, user[password_hash]): return user return None逻辑说明bcrypt.hash每次生成的盐不同同一个密码两次哈希结果不一样这是正常且必要的。bcrypt.verify会自动从存储的哈希里提取盐来比对。参数上role字段决定后续权限常见取值是admin和user。注意 MySQL 连接这里用了参数化查询%s千万别用字符串拼接否则就是 SQL 注入的活靶子。3.2 角色权限如何作用到检索层权限控制不能只做在界面上。这份资源带角色权限控制正确的落点是检索前先根据当前用户角色从 MySQL 查出可访问的文档 ID 列表再把这个列表作为过滤条件传给向量库。# retriever.py def get_accessible_doc_ids(user_role: str, user_id: int) - list: 根据角色返回可访问的文档 ID 列表 conn mysql.connector.connect(hostlocalhost, userroot, passwordyour_pwd, databaserag_system) cursor conn.cursor() if user_role admin: # 管理员可访问全部文档 cursor.execute(SELECT id FROM documents) else: # 普通用户只能访问自己上传的文档 cursor.execute(SELECT id FROM documents WHERE uploader_id %s, (user_id,)) ids [row[0] for row in cursor.fetchall()] cursor.close() conn.close() return ids def retrieve(query_vector, accessible_ids: list, top_k: int 5): 在可访问文档范围内做向量检索 if not accessible_ids: return [] # 没有任何权限直接返回空避免越权 results vector_store.search( query_vector, filter{doc_id: {$in: accessible_ids}}, # 关键过滤条件 top_ktop_k ) return results逻辑说明get_accessible_doc_ids是权限的唯一事实来源管理员拿全量普通用户拿自己的。retrieve里的filter参数是防止越权的核心向量库必须支持按元数据过滤否则你只能召回后再筛既慢又容易漏。top_k控制返回条数一般 3 到 8 之间太多会稀释相关性太少可能漏掉关键信息。提示如果你的向量库不支持元数据过滤那就得在应用层做二次筛选但一定要在拼 prompt 之前筛完别把无权限内容送进模型。3.3 会话隔离与流式问答的权限校验多用户系统里会话必须隔离。每个会话记录要带user_id查询历史消息时强制带上这个条件。流式问答的入口也要再校验一次权限因为用户可能伪造请求直接打后端接口。# qa_chain.py def stream_answer(prompt: str, user_id: int, user_role: str): 流式问答主流程每一步都带权限校验 accessible_ids get_accessible_doc_ids(user_role, user_id) if not accessible_ids: yield 你当前没有可访问的文档请先上传。 return query_vector embed(prompt) docs retrieve(query_vector, accessible_ids, top_k5) if not docs: yield 没有检索到相关内容换个问法试试。 return context \n.join([d[text] for d in docs]) full_prompt f根据以下资料回答问题\n{context}\n\n问题{prompt} # 调用大模型流式接口逐块 yield for chunk in llm.stream(full_prompt): yield chunk逻辑说明这个生成器把权限校验、检索、拼 prompt、流式输出串成一条线。accessible_ids为空时直接返回提示不浪费一次模型调用。context拼接时要注意长度超过模型上下文窗口就得截断或做重排。llm.stream必须是真正的流式接口否则前端打字机效果出不来。4. 避坑与排查RAG 系统上线前必须过的几道坎4.1 检索召回一堆无关内容现象用户问「合同违约金怎么算」系统召回的是「合同签署日期」相关段落答非所问。原因切块粒度太粗一个 chunk 里混了好几个主题或者 embedding 模型对中文语义区分度不够。解决把chunk_size调小到 300 到 400增加overlap换一个中文表现更好的 embedding 模型检索后加一层重排用交叉编码器对 top 20 重新打分再取 top 5。4.2 MySQL 连接报错 2002现象启动后端时报error 2002 (hy000): cant connect to local mysql server through socket /tmp/mysql.sock。原因MySQL 服务没启动或者连接配置里用了 socket 方式但路径不对。解决先确认 MySQL 服务在跑systemctl status mysql看一眼连接参数里显式指定host127.0.0.1和port3306强制走 TCP 而不是 socket检查用户权限里root是否允许从localhost连接。4.3 流式响应变成一次性输出现象前端等了五秒然后整段答案一次性蹦出来没有打字机效果。原因后端把模型返回的完整结果缓存后才 yield或者用了非流式的模型接口。解决确认llm.stream底层调的是流式 API检查生成器里有没有list()或join()把流提前消费掉Streamlit 侧确认用的是st.write_stream而不是st.write。4.4 权限过滤失效导致越权现象普通用户搜到了管理员上传的文档内容。原因检索时没传filter或者filter字段名和向量库里的元数据字段对不上。解决在retrieve里打印实际传入的 filter 和向量库返回的元数据确认字段名一致写一个测试用例用普通用户身份检索管理员文档断言结果为空。4.5 文件上传后检索不到现象文档上传成功MySQL 里也有记录但问答时检索不到。原因向量化写入失败但没报错或者文档 ID 关联错了。解决在向量化写入后加一条日志打印写入的向量条数和 doc_id检索时先用 doc_id 直接查向量库确认数据在不在检查 MySQL 里的 doc_id 和向量库里的 doc_id 是不是同一个值。5. 进阶玩法把检索质量再往上抬一档系统能跑通只是起点真正决定体验的是检索质量。我一般会在基础 RAG 之上加两个东西查询改写和混合检索。查询改写是指用户的问题先经过一次轻量处理比如把「它多少钱」补全成「XX 产品多少钱」再拿去检索。这一步能明显提升多轮对话里的召回率。混合检索则是把向量相似度和关键词匹配结合起来向量擅长语义关键词擅长精确命中两者加权融合后效果通常比单走向量好。# hybrid_retriever.py def hybrid_search(query: str, accessible_ids: list, alpha: float 0.7): 向量检索与关键词检索加权融合alpha 控制向量权重 query_vector embed(query) vector_results vector_store.search( query_vector, filter{doc_id: {$in: accessible_ids}}, top_k10 ) keyword_results keyword_index.search(query, doc_idsaccessible_ids, top_k10) # 用文档 ID 做归一化融合alpha 越大越偏向语义 scores {} for rank, doc in enumerate(vector_results): scores[doc[id]] scores.get(doc[id], 0) alpha * (1 / (rank 1)) for rank, doc in enumerate(keyword_results): scores[doc[id]] scores.get(doc[id], 0) (1 - alpha) * (1 / (rank 1)) ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return [doc_id for doc_id, _ in ranked[:5]]逻辑说明alpha是融合权重取 0.7 表示更信任向量检索关键词做补充。这里用排名倒数做简易打分工程上够用追求更精细可以换成 RRF 或归一化分数。accessible_ids同时传给两路检索保证权限过滤不丢。验证检索质量有个笨但有效的办法准备 20 到 30 个真实问题人工标注每个问题的正确文档然后跑一遍看命中率。命中率低于 70% 就别急着调模型先回头查切块和 embedding。从那以后我每次搭 RAG 系统都强制先把「权限过滤 检索命中率」这两件事跑通再接前端不然界面做得再漂亮答非所问一样留不住人。希望帮到你。本文还有配套的精品资源点击获取