企业官网建设流程全解析

简介：这是一套面向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 元数据，再解析切块、向量化写入向量库，两边用同一个文档 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 到 600，overlap取 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(host="localhost", user="root", password="your_pwd", database="rag_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(host="localhost", user="root", password="your_pwd", database="rag_system") cursor = conn.cursor(dictionary=True) 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(host="localhost", user="root", password="your_pwd", database="rag_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_k=top_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_k=5) 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): can't connect to local mysql server through socket '/tmp/mysql.sock'。

原因：MySQL 服务没启动，或者连接配置里用了 socket 方式但路径不对。

解决：先确认 MySQL 服务在跑，systemctl status mysql看一眼；连接参数里显式指定host="127.0.0.1"和port=3306，强制走 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_k=10 ) keyword_results = keyword_index.search(query, doc_ids=accessible_ids, top_k=10) # 用文档 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(), key=lambda x: x[1], reverse=True) return [doc_id for doc_id, _ in ranked[:5]]

逻辑说明：alpha是融合权重，取 0.7 表示更信任向量检索，关键词做补充。这里用排名倒数做简易打分，工程上够用；追求更精细可以换成 RRF 或归一化分数。accessible_ids同时传给两路检索，保证权限过滤不丢。

验证检索质量有个笨但有效的办法：准备 20 到 30 个真实问题，人工标注每个问题的正确文档，然后跑一遍看命中率。命中率低于 70% 就别急着调模型，先回头查切块和 embedding。

从那以后我每次搭 RAG 系统，都强制先把「权限过滤 + 检索命中率」这两件事跑通再接前端，不然界面做得再漂亮，答非所问一样留不住人。希望帮到你。

本文还有配套的精品资源，点击获取