企业官网建设流程全解析

简介：这份资源是一套基于RAG检索增强生成技术构建的智能文档检索系统完整源码，面向希望深入理解检索增强生成落地实践的Python开发者与AI应用学习者，可用于搭建具备用户认证、文档解析、向量存储与智能问答能力的文档管理平台。压缩包共62个文件，约175KB，以24个Python源码文件为核心，覆盖文档处理、向量存储、检索增强、用户与权限管理、邮件验证等模块，另含18个pyc编译文件、5个CSS与3个JS构成的前端样式与交互资源，以及XML配置、说明文档和项目架构图等辅助内容。系统采用Streamlit前端搭配Python后端，集成MySQL数据库，支持流式响应、角色权限控制与文件类型限制，并附带登录、密码重置、管理员面板、个人资料等页面实现。目前已有72人学习下载，适合作为课程设计、毕业项目或RAG工程实践的参考方案，帮助读者快速理解从文档向量化到智能问答的完整链路与模块划分。

1. 从一份带权限的 RAG 文档系统说起：为什么“能问答”只是及格线

很多团队第一次做 RAG 知识库，跑通“上传 PDF → 切块 → 向量化 → 问答”就以为收工了，结果一上线就翻车：实习生能搜到薪酬文档，外部账号能下载合同附件，几百页的扫描件解析出来全是乱码，MySQL 里存着向量却查不动。标题里这套“基于 RAG 检索增强生成技术的智能文档检索系统”，真正要解决的不是“让模型能回答”，而是让不同角色的人，只能检索到自己有权限的那部分文档，并且答案能流式吐出来。它适合已经跑通过 demo、准备把 RAG 落到内部文档场景的 Python 后端和全栈工程师。这一章先把边界划清楚：认证、文档处理、向量存储、MySQL、Streamlit 前端、流式响应、角色权限、文件类型限制，这八件事缺一件，系统就只是个玩具。下面按“先立骨架、再填血肉、最后排雷”的顺序拆开讲。

2. 认证、权限与文件类型限制：RAG 系统的第一道闸门

2.1 为什么权限必须做在检索层，而不是前端

新手最容易犯的错，是在 Streamlit 前端根据角色隐藏几个按钮，就以为做了权限控制。这是典型的“前端遮羞布”——后端接口一旦被直接调用，所有文档照样能被检索出来。正确的做法是：权限过滤必须发生在向量检索的候选集构造阶段，也就是在查询向量数据库之前，就把当前用户无权访问的文档 ID 排除掉。

常见做法是在文档入库时，给每个 chunk 打上owner_role或access_level元数据，检索时用过滤条件带上。以 Chroma 为例，元数据过滤是原生支持的；如果用的是 FAISS 这类纯向量库，就得自己在召回后做二次过滤，但那样会浪费 top-k 名额，召回质量下降。所以选型时优先选支持元数据过滤的向量库。

角色模型不用做太复杂，三档就够覆盖大多数内部场景：admin（全部文档）、editor（本部门文档 + 公共文档）、viewer（仅公共文档）。把角色和文档的可见范围映射成一张 MySQL 表，检索时先查这张表拿到允许的doc_id列表，再拼进向量库的where条件。

# permission.py # 角色到可见范围的映射，实际项目里从 MySQL 读取 ROLE_SCOPE = { "admin": {"dept": "*", "level": ["public", "internal", "confidential"]}, "editor": {"dept": "self", "level": ["public", "internal"]}, "viewer": {"dept": "self", "level": ["public"]}, } def build_where_clause(user: dict) -> dict: """根据用户角色生成向量库的元数据过滤条件""" scope = ROLE_SCOPE.get(user["role"], ROLE_SCOPE["viewer"]) conditions = [{"level": {"$in": scope["level"]}}] # editor 和 viewer 只能看本部门文档 if scope["dept"] == "self": conditions.append({"dept": user["dept"]}) # 多条件用 $and 组合，Chroma 的语法 return {"$and": conditions} if len(conditions) > 1 else conditions[0]

这段代码的关键在于build_where_clause返回的字典直接传给向量库的query(where=...)。参数说明：level是文档密级，dept是部门标识，两者都在文档入库时写入 chunk 的 metadata。注意$and的写法是 Chroma 的语法，换 Milvus 要改成expr字符串表达式，换 Qdrant 要改成must数组，迁移时这块最容易踩坑。

2.2 文件类型限制：白名单 + 魔数校验，别只信后缀

标题里专门提了“文件类型限制”，这不是多余的功能。用户上传一个改了后缀的.exe冒充.pdf，解析库直接抛异常还是小事，如果解析库有漏洞就是安全事故。所以校验要分两层：后缀白名单 + 文件头魔数校验。

白名单建议只放这几类：.pdf、.docx、.txt、.md、.csv。.doc和.xls这些老格式解析库支持差，直接拒掉，让用户另存为新格式。魔数校验用python-magic或直接读前几个字节比对。

# file_guard.py import magic # pip install python-magic ALLOWED_EXT = {".pdf", ".docx", ".txt", ".md", ".csv"} ALLOWED_MIME = { "application/pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "text/plain", "text/markdown", "text/csv", } MAX_SIZE = 20 * 1024 * 1024 # 20MB def validate_upload(filename: str, content: bytes) -> tuple[bool, str]: ext = "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext not in ALLOWED_EXT: return False, f"不支持的后缀 {ext}" if len(content) > MAX_SIZE: return False, "文件超过 20MB" # 魔数校验，不信任后缀 mime = magic.from_buffer(content[:2048], mime=True) if mime not in ALLOWED_MIME: return False, f"文件真实类型 {mime} 不在白名单" return True, "ok"

逻辑说明：先查后缀快速拒绝，再查大小防止内存打爆，最后用magic读前 2048 字节判断真实 MIME。参数MAX_SIZE按部署环境内存调整，20MB 是单机 Streamlit 的保守值。注意python-magic在 Windows 上要额外装libmagic，Linux 上apt install libmagic1即可，这是部署时的高频翻车点。

3. 文档解析与向量化流水线：从 PDF 到可检索 chunk

3.1 解析器选型：不同格式走不同库，别指望一个库通吃

文档解析是 RAG 里最脏最累的活。PDF 分两种：文本型 PDF 用pypdf或pdfplumber直接抽文字；扫描型 PDF 必须先 OCR，用pytesseract或PaddleOCR。DOCX 用python-docx，能保留段落和表格结构。CSV 和 TXT 直接读，但要注意编码，中文环境优先试utf-8，失败回退gbk。

我一般会写一个统一入口，按扩展名分发：

# parser.py import pdfplumber from docx import Document import csv, io def parse_document(path: str, ext: str) -> list[dict]: """返回 [{"text": ..., "page": ...}, ...] 结构""" if ext == ".pdf": pages = [] with pdfplumber.open(path) as pdf: for i, page in enumerate(pdf.pages): txt = page.extract_text() or "" if txt.strip(): pages.append({"text": txt, "page": i + 1}) return pages if ext == ".docx": doc = Document(path) full = "\n".join(p.text for p in doc.paragraphs if p.text.strip()) return [{"text": full, "page": 1}] if ext in (".txt", ".md"): for enc in ("utf-8", "gbk"): try: with open(path, encoding=enc) as f: return [{"text": f.read(), "page": 1}] except UnicodeDecodeError: continue raise ValueError("编码无法识别") if ext == ".csv": with open(path, encoding="utf-8") as f: rows = list(csv.reader(f)) return [{"text": " | ".join(r), "page": i + 1} for i, r in enumerate(rows)] raise ValueError(f"未支持的格式 {ext}")

参数说明：pdfplumber的extract_text()对多栏排版支持一般，遇到双栏 PDF 建议加layout=True参数或换PyMuPDF。page字段保留是为了后面做引用溯源，答案里能标出“来自第 3 页”。CSV 每行拼成一条文本，适合表格型知识，但行数太多时要限制，否则 chunk 数量爆炸。

3.2 切块策略：固定长度 + 重叠，但元数据要带全

切块没有银弹，但内部文档场景用“固定长度 + 重叠”最稳。RecursiveCharacterTextSplitter是 LangChain 里最常用的，分隔符按["\n\n", "\n", "。", "！", "？", " ", ""]排，优先在段落边界切。chunk_size 设 500～800 字符，overlap 设 50～100。中文场景下 chunk_size 按字符算，别按 token 算，否则要额外引 tokenizer。

关键是每个 chunk 的 metadata 必须带全：doc_id、filename、page、dept、level、uploader、upload_time。这些字段是后面权限过滤和溯源的基础，少一个都要返工重灌。

# chunker.py from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""], ) def build_chunks(pages: list[dict], meta: dict) -> list[dict]: chunks = [] for p in pages: for i, piece in enumerate(splitter.split_text(p["text"])): chunks.append({ "text": piece, "metadata": {**meta, "page": p["page"], "chunk_idx": i}, }) return chunks

逻辑说明：meta是文档级元数据（doc_id、dept、level 等），每个 chunk 都继承一份，再叠加页码和块序号。参数chunk_size=600是中文经验值，英文文档可以调到 1000。chunk_overlap=80保证跨块句子不被切断。注意separators里中文标点要放在空格前面，否则中文长句会被空格切碎。

3.3 向量化与入库：模型选型 + 批量写入

Embedding 模型选型看两点：中文效果和部署成本。内部系统优先用bge-large-zh-v1.5或text-embedding-3-small（如果允许调外部 API）。本地部署用sentence-transformers加载，显存不够就用bge-small-zh。维度 512 或 1024，入库前统一 normalize，用余弦相似度检索。

批量写入向量库时，别一条一条 insert，攒够 100 条批量提交，速度差十倍。Chroma 用collection.add()，Milvus 用insert()后flush()。

# embedder.py from sentence_transformers import SentenceTransformer import chromadb model = SentenceTransformer("BAAI/bge-small-zh-v1.5") client = chromadb.PersistentClient(path="./chroma_db") collection = client.get_or_create_collection( name="docs", metadata={"hnsw:space": "cosine"} ) def index_chunks(chunks: list[dict], batch_size: int = 100): for i in range(0, len(chunks), batch_size): batch = chunks[i:i + batch_size] texts = [c["text"] for c in batch] embs = model.encode(texts, normalize_embeddings=True).tolist() collection.add( ids=[f"{c['metadata']['doc_id']}_{c['metadata']['chunk_idx']}" for c in batch], embeddings=embs, documents=texts, metadatas=[c["metadata"] for c in batch], )

参数说明：normalize_embeddings=True让向量单位化，配合hnsw:space=cosine保证相似度计算正确。batch_size=100是内存和速度的平衡点，显存小的机器调到 32。ids用doc_id + chunk_idx拼，保证唯一且可追溯。注意 Chroma 的PersistentClient路径要挂持久化卷，容器重启不丢数据。

4. MySQL 与向量库的分工：别把向量塞进 MySQL

4.1 为什么向量存储不放在 MySQL

热词里 MySQL 出现频率很高，很多人第一反应是“MySQL 也能存向量，何必再引一个库”。MySQL 8.0 确实没有原生向量类型，5.7 更没有。用BLOB存向量、用JSON存元数据，检索时全表扫描算余弦相似度，几万条数据就慢到不可用。向量检索需要 ANN 索引（HNSW、IVF），这是专用向量库的活。

正确的分工是：MySQL 存业务数据，向量库存向量和检索元数据。MySQL 里建三张表：users（用户和角色）、documents（文档元信息、权限字段、状态）、chat_logs（问答记录，用于审计和效果分析）。向量库只存 chunk 的向量和过滤用元数据，两边用doc_id关联。

-- schema.sql CREATE TABLE users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(64) UNIQUE NOT NULL, password_hash VARCHAR(255) NOT NULL, role ENUM('admin','editor','viewer') DEFAULT 'viewer', dept VARCHAR(64) DEFAULT 'public', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE documents ( doc_id VARCHAR(36) PRIMARY KEY, filename VARCHAR(255) NOT NULL, ext VARCHAR(10) NOT NULL, dept VARCHAR(64) NOT NULL, level ENUM('public','internal','confidential') DEFAULT 'public', uploader INT, status ENUM('processing','ready','failed') DEFAULT 'processing', chunk_count INT DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (uploader) REFERENCES users(id) ); CREATE TABLE chat_logs ( id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT, question TEXT, answer TEXT, cited_docs JSON, latency_ms INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

参数说明：documents.status字段很关键，文档上传后先置processing，解析和向量化完成再改ready，失败改failed。前端只展示ready的文档，避免用户搜到半成品。chat_logs.cited_docs存 JSON 数组，记录答案引用了哪些 doc_id 和页码，方便后续做效果评估和审计。level用 ENUM 而不是 VARCHAR，查询快且防脏数据。

4.2 连接池与常见连接错误

Streamlit 每次交互都会重跑脚本，如果每次新建 MySQL 连接，很快就too many connections。必须用连接池。SQLAlchemy的create_engine自带池，配pool_size=5, max_overflow=10, pool_recycle=3600。

# db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine( "mysql+pymysql://rag:password@127.0.0.1:3306/ragdb?charset=utf8mb4", pool_size=5, max_overflow=10, pool_recycle=3600, pool_pre_ping=True, ) SessionLocal = sessionmaker(bind=engine)

参数说明：pool_pre_ping=True在取连接前先 ping 一下，避免拿到已断开的连接，这是解决MySQL server has gone away的关键。pool_recycle=3600让连接每小时回收，防止 MySQL 的wait_timeout把空闲连接掐掉。charset=utf8mb4必须带，否则中文和 emoji 存进去变问号。

热词里那个error 2002 (hy000): can't connect to local mysql server through socket '/tmp/mysql.sock'是经典问题：连接串里写了localhost，PyMySQL 会走 socket 而不是 TCP。改成127.0.0.1强制走 TCP 就能解决。这个坑我踩过不止一次，血泪经验。

5. Streamlit 前端与流式响应：把等待变成体验

5.1 流式响应的实现：生成器 + st.write_stream

RAG 的答案生成要调 LLM，几秒到十几秒的等待，没有流式输出用户会以为卡死。Streamlit 从 1.31 开始有st.write_stream，直接吃生成器。后端把 LLM 的流式输出包成生成器，前端逐字渲染。

# rag_chain.py from openai import OpenAI client = OpenAI(base_url="http://127.0.0.1:8000/v1", api_key="not-needed") def stream_answer(question: str, contexts: list[str]): prompt = build_prompt(question, contexts) resp = client.chat.completions.create( model="qwen2.5-7b-instruct", messages=[{"role": "user", "content": prompt}], stream=True, temperature=0.2, ) for chunk in resp: delta = chunk.choices[0].delta.content if delta: yield delta

# app.py import streamlit as st from rag_chain import stream_answer st.set_page_config(page_title="智能文档检索", layout="wide") if "user" not in st.session_state: st.warning("请先登录") st.stop() question = st.chat_input("输入你的问题") if question: st.chat_message("user").write(question) with st.chat_message("assistant"): contexts = retrieve(question, st.session_state.user) answer = st.write_stream(stream_answer(question, contexts)) log_chat(st.session_state.user["id"], question, answer, contexts)

逻辑说明：stream_answer是生成器，yield每个 token。st.write_stream接收生成器并逐字渲染，返回完整字符串用于落库。参数temperature=0.2让答案稳定，RAG 场景不需要创造性。retrieve函数内部先查权限、再查向量库、再拼 context，这一步是权限控制真正生效的地方。

5.2 登录态与角色在 Streamlit 里的管理

Streamlit 没有内置 session 管理，用st.session_state存用户信息。登录页用st.form收账号密码，后端查 MySQL 校验password_hash（用bcrypt，别用 MD5）。登录成功后把user字典写进session_state，后续每个页面先检查session_state.user是否存在。

注意st.session_state在浏览器刷新后会丢，这是 Streamlit 的机制。要持久化得用 cookie 或 URL 参数，但内部系统场景下刷新重登可以接受，不必过度设计。角色控制体现在：admin能看到“文档管理”页，editor能看到“上传”按钮，viewer只能问答。但记住，这些只是 UI 层面的，真正的权限在retrieve里。

6. 避坑与排查：那些让系统上线即翻车的问题

6.1 检索结果为空但文档明明在

现象：用户提问后回答“未找到相关文档”，但管理员确认文档已入库。原因通常是权限过滤条件写错，把用户自己的文档也过滤掉了。比如editor的dept是tech，但文档入库时dept写的是技术部，字符串不匹配，过滤后候选集为空。解决：统一dept的取值来源，从users表读，别让上传时手填。排查时先把where条件去掉查一次，确认向量库里有数据，再逐步加条件定位。

6.2 中文 PDF 解析出来全是乱码

现象：pdfplumber抽出来的文字是\x00\x01之类。原因是 PDF 内嵌字体没有 ToUnicode 映射，属于扫描件或特殊编码。解决：先判断extract_text()返回的字符里可打印字符占比，低于 60% 就走 OCR 分支。OCR 用PaddleOCR，中文效果好但慢，建议异步任务队列处理，别阻塞上传接口。这个坑的后悔药是：上传时就做质量检测，解析失败的文档直接标failed并提示用户，别让它进向量库污染检索。

6.3 流式响应在 Streamlit 里断流

现象：答案输出到一半停住，或者整段一次性蹦出来。原因通常是 LLM 服务端的流式没开，或者中间用了requests没设stream=True。解决：确认 LLM 服务支持 SSE，客户端用stream=True并逐行读。另一个原因是 Streamlit 的write_stream要求生成器 yield 字符串，如果 yield 的是 dict 或 None 会中断。排查时先在命令行单独跑stream_answer，确认能逐字输出，再接到前端。

6.4 MySQL 连接数暴涨导致服务不可用

现象：系统跑一段时间后所有请求超时，MySQL 报Too many connections。原因是 Streamlit 每次交互重跑脚本，如果retrieve或log_chat里每次SessionLocal()新建 session 却没关，连接池很快耗尽。解决：用with SessionLocal() as session:上下文管理，确保释放。或者用st.cache_resource缓存 engine，全局一个。参数上pool_size别设太大，5～10 够用，max_overflow给 10 缓冲。

6.5 向量库和 MySQL 数据不一致

现象：MySQL 里文档状态是ready，但向量库查不到；或者文档已删除，向量还在。原因是入库和删库是两步操作，中间失败没回滚。解决：入库流程做成“先写 MySQL 状态 processing → 向量化 → 写向量库 → 更新 MySQL 为 ready”，任何一步失败都标failed并记录原因。删除时先删向量再删 MySQL 记录，或者用软删除，定期跑对账脚本清理孤儿向量。这个对账脚本我一般会写成定时任务，每天凌晨跑一次。

7. 进阶技巧：用混合检索和重排序把召回率再提一档

纯向量检索在专有名词、编号、缩写上经常翻车，比如“GB/T 19001”这种，向量模型未必能区分。进阶做法是混合检索：向量检索 + BM25 关键词检索，两路召回后用 RRF（Reciprocal Rank Fusion）融合。BM25 用rank_bm25库，对 chunk 文本建索引，查询时算分。RRF 的公式是score = Σ 1/(k + rank)，k 取 60。

# hybrid.py from rank_bm25 import BM25Okapi import jieba def build_bm25(chunks: list[dict]): corpus = [list(jieba.cut(c["text"])) for c in chunks] return BM25Okapi(corpus), chunks def hybrid_retrieve(query: str, bm25, chunks, collection, where, top_k=10): # 向量路 q_emb = model.encode([query], normalize_embeddings=True).tolist() vec_res = collection.query(query_embeddings=q_emb, n_results=top_k, where=where) vec_ids = vec_res["ids"][0] # BM25 路 tokens = list(jieba.cut(query)) scores = bm25.get_scores(tokens) bm25_ids = [chunks[i]["metadata"]["doc_id"] + "_" + str(chunks[i]["metadata"]["chunk_idx"]) for i in scores.argsort()[::-1][:top_k]] # RRF 融合 fused = {} for rank, cid in enumerate(vec_ids): fused[cid] = fused.get(cid, 0) + 1 / (60 + rank) for rank, cid in enumerate(bm25_ids): fused[cid] = fused.get(cid, 0) + 1 / (60 + rank) return sorted(fused.items(), key=lambda x: -x[1])[:top_k]

参数说明：jieba.cut做中文分词，BM25 需要 token 列表。top_k=10是每路召回数，融合后取前 10 送进 LLM。RRF 的k=60是论文里的经验值，不用调。注意 BM25 索引要跟着文档增删同步更新，否则新文档搜不到。如果嫌维护两套索引麻烦，可以用支持混合检索的向量库，比如 Qdrant 的 sparse vector，但配置复杂度更高。

重排序是另一档提升：召回 top 20 后用 cross-encoder 模型（如bge-reranker-base）逐对打分，取前 5 送 LLM。cross-encoder 比向量相似度准，但慢，只适合小候选集。我一般只在答案质量要求高的场景开重排序，普通问答省掉这步。

验证效果别靠感觉，准备 50 条带标准答案的问题，算召回率（正确文档在 top-k 里的比例）和答案准确率。每次改检索策略都跑一遍，数据说话。我自己的习惯是：任何检索参数改动，先跑评测集，涨了才合并，跌了直接回滚。这套流程帮我省了无数次“感觉变好了其实变差了”的返工。

希望帮到你。

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