企业官网建设流程全解析

简介：这份资源是一套基于RAG检索增强生成技术构建的智能文档检索系统完整源码，面向希望深入理解检索增强生成落地实践的Python开发者与AI应用学习者。系统整合了用户认证、文档解析与向量化、向量存储、智能问答与流式响应、角色权限控制及文件类型限制等核心能力，后端以Python实现，前端采用Streamlit搭建，数据层依托MySQL数据库，覆盖从文档上传到问答输出的完整链路。压缩包共62个文件，约175KB，以24个py源码为主体，辅以pyc编译文件、css与js前端样式脚本、xml配置及md说明文档，模块划分清晰，便于按功能定位代码。目前已有72人学习下载。读者可从中获取可运行的RAG项目骨架、异步文档处理与增强检索实现、向量库封装、权限与认证逻辑以及前端交互界面，适合作为课程设计、毕业项目或二次开发的基础参考。

1. 从一份带权限的 RAG 文档检索系统说起：为什么多数 demo 活不过第二周

我见过太多 RAG 项目死在演示到上线的路上。第一周跑通langchain + faiss + streamlit，老板点头，第二周加个用户登录，第三周发现张三能搜到李四的合同，第四周向量库和 MySQL 里的文档状态对不上，第五周没人敢碰这套代码。问题不在 RAG 本身，而在于「智能文档检索系统」这七个字里藏着一堆 demo 阶段被忽略的工程约束：用户认证、角色权限、文件类型限制、向量与关系数据的双写一致性、流式响应的中断处理。

这篇讲的就是把这些约束一次性补齐的落地路径。技术栈是 Python 后端 + Streamlit 前端 + MySQL 存元数据与权限 + 向量库做语义检索 + RAG 做问答生成，覆盖文档解析、向量化、检索、流式输出、角色权限控制、文件类型白名单这一整条链路。适合已经跑通过最小 RAG demo、准备把它变成内部能用的系统的后端或全栈工程师。如果你还在纠结 RAG 是什么，建议先补基础；如果你已经卡在权限和双写一致性上，直接跳到第 3、4 章。

我一般把这类系统拆成四层：接入层（Streamlit + 认证）、业务层（权限判定 + 文档管理）、检索层（向量化 + 召回 + 重排）、存储层（MySQL + 向量库）。每层都有独立的坑，混在一起调就是玄学。

2. 系统骨架：MySQL 存什么、向量库存什么、Streamlit 怎么接

2.1 为什么元数据和向量必须分两处存

很多人第一反应是「向量库不是也能存 metadata 吗，为什么还要 MySQL」。能存，但不该全存。向量库的 metadata 过滤能力参差不齐，做等值过滤还行，做「用户属于哪个部门、部门有哪些角色、角色能看哪些文档分类」这种多跳关联查询就是灾难。而 MySQL 天生干这个。

我的分工原则很明确：

关键设计：向量库里每个向量只带一个chunk_id作为主键，其余全部回 MySQL 查。这样权限过滤在 MySQL 侧完成，先算出「当前用户能看的 doc_id 列表」，再去向量库做带doc_id in (...)的过滤检索。别指望向量库帮你做权限，那是把安全逻辑交给一个不擅长它的组件。

2.2 建表：四张核心表的最小结构

-- 用户表 CREATE TABLE users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(64) UNIQUE NOT NULL, password_hash VARCHAR(255) NOT NULL, role_id INT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 角色权限表：用 JSON 存可访问的文档分类，简单直接 CREATE TABLE roles ( id INT PRIMARY KEY AUTO_INCREMENT, role_name VARCHAR(32) NOT NULL, allowed_categories JSON NOT NULL, -- 如 ["public","finance"] can_upload TINYINT DEFAULT 0 ); -- 文档元信息表 CREATE TABLE documents ( id INT PRIMARY KEY AUTO_INCREMENT, filename VARCHAR(255) NOT NULL, file_type VARCHAR(16) NOT NULL, category VARCHAR(32) NOT NULL, uploader_id INT NOT NULL, status ENUM('pending','indexed','failed') DEFAULT 'pending', chunk_count INT DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_category (category), INDEX idx_status (status) ); -- 分块映射表：向量库回查用 CREATE TABLE chunks ( id INT PRIMARY KEY AUTO_INCREMENT, doc_id INT NOT NULL, chunk_index INT NOT NULL, content TEXT NOT NULL, INDEX idx_doc (doc_id) );

roles.allowed_categories用 JSON 而不是关联表，是因为分类数量通常有限且不常变，JSON 查询在 MySQL 5.7+ 够用，省一张表。如果分类会动态增长到几百个，再拆成关联表。documents.status用枚举是为了让「上传了但没索引成功」这种中间态可见，这是排查双写问题的关键字段。

2.3 Streamlit 前端怎么接认证

Streamlit 每次交互都会重跑脚本，st.session_state是唯一能跨交互保存状态的地方。认证逻辑必须放在脚本最顶部，且要在任何数据加载之前。

import streamlit as st from auth import verify_login, get_user_permissions def require_login(): if "user" not in st.session_state: st.session_state.user = None if st.session_state.user is None: with st.form("login"): u = st.text_input("用户名") p = st.text_input("密码", type="password") if st.form_submit_button("登录"): user = verify_login(u, p) if user: st.session_state.user = user st.session_state.perms = get_user_permissions(user["role_id"]) st.rerun() else: st.error("用户名或密码错误") st.stop() # 关键：未登录直接终止本次渲染 require_login() # 以下才是业务代码

st.stop()是必须的，否则未登录状态下后面的业务代码照样执行，只是没显示而已，数据已经查出来了。这个坑我踩过，日志里能看到未登录请求打到了数据库。st.rerun()在登录成功后调用，是为了让页面用新状态重新渲染一遍，否则表单状态和 session 不同步。

提示：Streamlit 的 session_state 存在服务端内存，多副本部署时要配 sticky session 或改用外部 session 存储，否则用户会在副本间「掉登录」。

3. 文档解析与向量化：文件类型白名单和分块策略怎么定

3.1 文件类型限制不是前端 accept 就完事

前端st.file_uploader(type=["pdf","docx","txt"])只是 UI 提示，绕过它太容易。真正的白名单必须在后端按扩展名 + MIME + 文件头三重校验。

import magic # python-magic ALLOWED_EXT = {".pdf", ".docx", ".txt", ".md"} ALLOWED_MIME = { "application/pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "text/plain", "text/markdown", } MAX_SIZE = 20 * 1024 * 1024 # 20MB def validate_file(uploaded) -> tuple[bool, str]: import os ext = os.path.splitext(uploaded.name)[1].lower() if ext not in ALLOWED_EXT: return False, f"不支持的扩展名: {ext}" if uploaded.size > MAX_SIZE: return False, "文件超过 20MB" head = uploaded.read(2048) uploaded.seek(0) mime = magic.from_buffer(head, mime=True) if mime not in ALLOWED_MIME: return False, f"文件真实类型 {mime} 不在白名单" return True, ""

magic.from_buffer读文件头判断真实类型，防止把.exe改名成.pdf上传。uploaded.seek(0)必须调用，否则后续解析读到的是空内容——这是 Streamlit UploadedFile 对象的行为，读一次指针就到末尾了。MAX_SIZE限制在解析前做，别等加载进内存才判断，大文件能把内存打爆。

3.2 分块：按语义还是按长度

分块策略直接决定检索质量。我试过三种：

• 固定长度 512 token + 50 重叠：简单，但会把一句话从中间切断，检索出来的片段读不通。
• 按段落切：语义完整，但段落长度方差极大，有的 20 字有的 2000 字，向量质量不稳定。
• 递归字符切分（RecursiveCharacterTextSplitter）：优先按\n\n、再按\n、再按句号，最后才硬切。这是目前最稳的默认选择。

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""], length_function=len, ) chunks = splitter.split_text(raw_text)

chunk_size=500是按中文字符算的，对应大约 350-400 token，配合chunk_overlap=80保证跨块语义连续。separators里把中文标点放前面，是因为默认分隔符只认英文标点，中文文档会被硬切。如果你的文档以表格为主，这套切分不适用，得单独走表格提取逻辑，把表格转成 Markdown 再切。

3.3 向量化与双写：先写 MySQL 还是先写向量库

这是最容易出数据不一致的地方。正确顺序是：先写 MySQL 拿到doc_id，状态置pending，再向量化写入向量库，成功后回写status='indexed'和chunk_count。任何一步失败，状态停在pending或failed，可以重试。

def ingest_document(file_bytes, filename, category, uploader_id): doc_id = insert_document(filename, category, uploader_id, status="pending") try: text = parse_file(file_bytes, filename) chunks = splitter.split_text(text) ids = [] for i, c in enumerate(chunks): chunk_id = insert_chunk(doc_id, i, c) # 先落 MySQL ids.append(chunk_id) vectors = embed(chunks) vector_store.add(ids=ids, texts=chunks, embeddings=vectors) update_document(doc_id, status="indexed", chunk_count=len(chunks)) except Exception as e: update_document(doc_id, status="failed") raise return doc_id

insert_chunk先落 MySQL 再写向量库，是为了让向量库里的chunk_id一定能回查到内容。反过来先写向量库，MySQL 插入失败就会留下孤儿向量。embed批量调用而不是逐条，能省大量网络往返。失败时状态置failed而不是删除记录，保留现场方便排查。

注意：向量库写入通常不是事务性的，批量写一半失败会留下部分向量。重试前要先按doc_id删除该文档的旧向量，否则会重复。

4. 检索与问答：权限过滤、召回重排、流式响应

4.1 权限过滤必须在检索前完成

错误做法：先检索 top_k，再过滤掉无权访问的。这样 top_k 里可能一半被过滤，实际返回不足，且无权文档的内容已经进了内存，有泄露风险。

正确做法：先查权限，拿到allowed_doc_ids，再带过滤条件检索。

def get_allowed_doc_ids(user_id, role_id): perms = query_role(role_id) # allowed_categories cats = perms["allowed_categories"] if not cats: return [] placeholders = ",".join(["%s"] * len(cats)) sql = f"SELECT id FROM documents WHERE category IN ({placeholders}) AND status='indexed'" return [r["id"] for r in db.query(sql, cats)] def retrieve(query, allowed_ids, top_k=8): if not allowed_ids: return [] results = vector_store.similarity_search( query, k=top_k, filter={"chunk_id": {"$in": allowed_ids_chunks}} # 或按 doc_id 过滤 ) return results

如果向量库不支持$in大列表过滤（几千个 id 会拖慢），改用「按分类建多个 collection」或「在 MySQL 侧先缩小 doc_id 范围再传给向量库」。allowed_ids为空时直接返回空，不要走检索，否则等于全库搜。

4.2 召回之后要不要重排

top_k=8 直接喂给 LLM，效果通常一般，因为向量相似度高不等于答案相关。加一层重排能明显提升。轻量方案是用bge-reranker这类交叉编码器对 8 个候选重排取前 3。

from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-base") def rerank(query, candidates, top_n=3): pairs = [(query, c["content"]) for c in candidates] scores = reranker.predict(pairs) ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [c for c, _ in ranked[:top_n]]

重排模型比嵌入模型慢，但只对 8 个候选算，延迟可接受。top_n=3是经验值，给 LLM 的上下文不是越多越好，无关片段会稀释有效信息。如果候选本身就只有 3-4 个，重排收益不大，可以跳过。

4.3 流式响应怎么在 Streamlit 里落地

Streamlit 的st.write_stream能直接消费生成器，实现打字机效果。关键是后端要返回生成器而不是一次性字符串。

def answer_stream(query, user): allowed = get_allowed_doc_ids(user["id"], user["role_id"]) candidates = retrieve(query, allowed, top_k=8) if not candidates: yield "没有找到你有权限访问的相关文档。" return context = "\n\n".join(rerank(query, candidates)) prompt = build_prompt(query, context) for token in llm.stream(prompt): yield token # Streamlit 侧 if prompt := st.chat_input("提问"): st.session_state.messages.append({"role": "user", "content": prompt}) with st.chat_message("assistant"): response = st.write_stream(answer_stream(prompt, st.session_state.user)) st.session_state.messages.append({"role": "assistant", "content": response})

answer_stream是生成器，yield逐 token 返回。st.write_stream会边收边渲染，并返回完整字符串用于存历史。注意权限判定在生成器内部做，不要在外部算好传进来，否则生成器被多次迭代时会重复查询。用户中途关页面，生成器会被 GC，LLM 调用要支持取消，否则白烧 token。

提示：流式响应下，如果 LLM 输出到一半报错，前端已经显示了半截内容。建议在生成器里 try/except，出错时 yield 一句「生成中断，请重试」，而不是让异常冒到 Streamlit 层。

5. 避坑与排查：那些让系统半夜报警的细节

5.1 现象：用户能搜到别人的文档

原因：检索时只按 query 相似度召回，没带权限过滤，或者过滤条件写成了doc_id in allowed但allowed是空列表时被当成「不过滤」。

解决：retrieve函数入口先判断if not allowed_ids: return []，并且过滤条件用白名单而非黑名单。上线前写一个测试：用 A 用户登录，搜 B 用户文档里的独有关键词，必须返回空。

5.2 现象：文档显示 indexed 但搜不到

原因：向量写入了但chunk_id映射错位，或者嵌入时用了和检索时不同的模型。常见于中途换过 embedding 模型但没重建索引。

解决：查chunks表确认 chunk 存在，再直接拿某个 chunk_id 去向量库查是否存在。嵌入模型名要写进配置并在启动时校验，换模型必须全量重建，不能增量。

5.3 现象：MySQL 连接数暴涨，报 too many connections

原因：Streamlit 每次交互重跑脚本，如果在脚本层直接pymysql.connect()，每次重跑都新建连接且不关闭。

解决：用连接池，且池对象缓存在st.cache_resource里。

import streamlit as st from dbutils.pooled_db import PooledDB import pymysql @st.cache_resource def get_pool(): return PooledDB( creator=pymysql, maxconnections=10, mincached=2, host="127.0.0.1", user="app", password="***", database="ragdb" ) def get_conn(): return get_pool().connection()

st.cache_resource保证整个应用生命周期只建一个池。maxconnections=10按并发用户数估，别设太大，MySQL 默认max_connections才 151。

5.4 现象：上传大 PDF 后服务卡死

原因：解析和向量化在请求线程里同步做，大文件耗时几十秒，Streamlit 请求超时。

解决：上传后只落 MySQL 置pending，用后台任务（Celery 或简单的线程池）异步解析向量化，前端轮询status。别在 Streamlit 脚本里开threading.Thread做长任务，脚本重跑会丢引用。

5.5 现象：中文检索效果差，答非所问

原因：用了英文 embedding 模型，或者分块把中文句子切碎。

解决：换中文或多语言 embedding 模型，分块分隔符加中文标点，chunk_size按字符而非 token 估。检索前对 query 做一次改写（用 LLM 把口语问题转成检索友好的关键词）也能明显提升。

6. 进阶：把权限粒度做到分块级，以及一个验证检索质量的小技巧

到这一步系统能用了，但权限还停在文档级——一个用户要么能看整篇文档，要么完全看不到。真实场景里常有「同一份合同，销售能看金额，法务能看全部」的需求。做法是把权限下沉到分块：chunks表加一个sensitivity字段，roles里配max_sensitivity，检索时在向量库过滤条件里带上sensitivity <= user_max。

def retrieve_with_sensitivity(query, allowed_ids, user_max_sens, top_k=8): return vector_store.similarity_search( query, k=top_k, filter={ "$and": [ {"chunk_id": {"$in": allowed_ids}}, {"sensitivity": {"$lte": user_max_sens}}, ] } )

sensitivity在入库时由解析规则或人工标注赋值，比如含「报价」「底价」的块标 3，公开说明标 1。这套的代价是入库时要多一步分类，收益是权限模型能覆盖更细的场景。如果向量库不支持复合过滤，就在 MySQL 侧先按sensitivity筛出 chunk_id 再传给向量库。

验证检索质量，我有个土办法但很有效：准备 20 条「问题-标准答案所在 chunk」的对照表，跑一遍检索，看标准 chunk 有没有进 top_3，算命中率。低于 70% 就别急着调 LLM，先回去调分块和 embedding。这个对照表不用多，20 条足够暴露问题，而且每次改完分块策略都能快速回归。

def eval_retrieval(test_cases, retrieve_fn, top_n=3): hit = 0 for q, gold_chunk_id in test_cases: results = retrieve_fn(q) if gold_chunk_id in [r["chunk_id"] for r in results[:top_n]]: hit += 1 return hit / len(test_cases)

test_cases是(问题, 标准chunk_id)列表，retrieve_fn是你当前的检索函数。这个函数跑起来几秒钟，但能让你在改参数时心里有数，而不是凭感觉。我现在的习惯是：任何动到分块、embedding、重排的改动，先跑这个命中率，掉了就回滚，别等上线被用户骂了才发现。

希望帮到你。

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