简介这份资源是一套基于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], mimeTrue) if mime not in ALLOWED_MIME: return False, f文件真实类型 {mime} 不在白名单 return True, ok逻辑说明先查后缀快速拒绝再查大小防止内存打爆最后用magic读前 2048 字节判断真实 MIME。参数MAX_SIZE按部署环境内存调整20MB 是单机 Streamlit 的保守值。注意python-magic在 Windows 上要额外装libmagicLinux 上apt install libmagic1即可这是部署时的高频翻车点。3. 文档解析与向量化流水线从 PDF 到可检索 chunk3.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, encodingenc) as f: return [{text: f.read(), page: 1}] except UnicodeDecodeError: continue raise ValueError(编码无法识别) if ext .csv: with open(path, encodingutf-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 建议加layoutTrue参数或换PyMuPDF。page字段保留是为了后面做引用溯源答案里能标出“来自第 3 页”。CSV 每行拼成一条文本适合表格型知识但行数太多时要限制否则 chunk 数量爆炸。3.2 切块策略固定长度 重叠但元数据要带全切块没有银弹但内部文档场景用“固定长度 重叠”最稳。RecursiveCharacterTextSplitter是 LangChain 里最常用的分隔符按[\n\n, \n, 。, , , , ]排优先在段落边界切。chunk_size 设 500800 字符overlap 设 50100。中文场景下 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_size600, chunk_overlap80, 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_size600是中文经验值英文文档可以调到 1000。chunk_overlap80保证跨块句子不被切断。注意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( namedocs, 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_embeddingsTrue).tolist() collection.add( ids[f{c[metadata][doc_id]}_{c[metadata][chunk_idx]} for c in batch], embeddingsembs, documentstexts, metadatas[c[metadata] for c in batch], )参数说明normalize_embeddingsTrue让向量单位化配合hnsw:spacecosine保证相似度计算正确。batch_size100是内存和速度的平衡点显存小的机器调到 32。ids用doc_id chunk_idx拼保证唯一且可追溯。注意 Chroma 的PersistentClient路径要挂持久化卷容器重启不丢数据。4. MySQL 与向量库的分工别把向量塞进 MySQL4.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_size5, max_overflow10, pool_recycle3600。# db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine( mysqlpymysql://rag:password127.0.0.1:3306/ragdb?charsetutf8mb4, pool_size5, max_overflow10, pool_recycle3600, pool_pre_pingTrue, ) SessionLocal sessionmaker(bindengine)参数说明pool_pre_pingTrue在取连接前先 ping 一下避免拿到已断开的连接这是解决MySQL server has gone away的关键。pool_recycle3600让连接每小时回收防止 MySQL 的wait_timeout把空闲连接掐掉。charsetutf8mb4必须带否则中文和 emoji 存进去变问号。热词里那个error 2002 (hy000): cant connect to local mysql server through socket /tmp/mysql.sock是经典问题连接串里写了localhostPyMySQL 会走 socket 而不是 TCP。改成127.0.0.1强制走 TCP 就能解决。这个坑我踩过不止一次血泪经验。5. Streamlit 前端与流式响应把等待变成体验5.1 流式响应的实现生成器 st.write_streamRAG 的答案生成要调 LLM几秒到十几秒的等待没有流式输出用户会以为卡死。Streamlit 从 1.31 开始有st.write_stream直接吃生成器。后端把 LLM 的流式输出包成生成器前端逐字渲染。# rag_chain.py from openai import OpenAI client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keynot-needed) def stream_answer(question: str, contexts: list[str]): prompt build_prompt(question, contexts) resp client.chat.completions.create( modelqwen2.5-7b-instruct, messages[{role: user, content: prompt}], streamTrue, temperature0.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智能文档检索, layoutwide) 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接收生成器并逐字渲染返回完整字符串用于落库。参数temperature0.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没设streamTrue。解决确认 LLM 服务支持 SSE客户端用streamTrue并逐行读。另一个原因是 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别设太大510 够用max_overflow给 10 缓冲。6.5 向量库和 MySQL 数据不一致现象MySQL 里文档状态是ready但向量库查不到或者文档已删除向量还在。原因是入库和删库是两步操作中间失败没回滚。解决入库流程做成“先写 MySQL 状态 processing → 向量化 → 写向量库 → 更新 MySQL 为 ready”任何一步失败都标failed并记录原因。删除时先删向量再删 MySQL 记录或者用软删除定期跑对账脚本清理孤儿向量。这个对账脚本我一般会写成定时任务每天凌晨跑一次。7. 进阶技巧用混合检索和重排序把召回率再提一档纯向量检索在专有名词、编号、缩写上经常翻车比如“GB/T 19001”这种向量模型未必能区分。进阶做法是混合检索向量检索 BM25 关键词检索两路召回后用 RRFReciprocal 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_k10): # 向量路 q_emb model.encode([query], normalize_embeddingsTrue).tolist() vec_res collection.query(query_embeddingsq_emb, n_resultstop_k, wherewhere) 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(), keylambda x: -x[1])[:top_k]参数说明jieba.cut做中文分词BM25 需要 token 列表。top_k10是每路召回数融合后取前 10 送进 LLM。RRF 的k60是论文里的经验值不用调。注意 BM25 索引要跟着文档增删同步更新否则新文档搜不到。如果嫌维护两套索引麻烦可以用支持混合检索的向量库比如 Qdrant 的 sparse vector但配置复杂度更高。重排序是另一档提升召回 top 20 后用 cross-encoder 模型如bge-reranker-base逐对打分取前 5 送 LLM。cross-encoder 比向量相似度准但慢只适合小候选集。我一般只在答案质量要求高的场景开重排序普通问答省掉这步。验证效果别靠感觉准备 50 条带标准答案的问题算召回率正确文档在 top-k 里的比例和答案准确率。每次改检索策略都跑一遍数据说话。我自己的习惯是任何检索参数改动先跑评测集涨了才合并跌了直接回滚。这套流程帮我省了无数次“感觉变好了其实变差了”的返工。希望帮到你。本文还有配套的精品资源点击获取
基于RAG的智能文档检索系统：权限控制与流式响应实战
简介这份资源是一套基于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], mimeTrue) if mime not in ALLOWED_MIME: return False, f文件真实类型 {mime} 不在白名单 return True, ok逻辑说明先查后缀快速拒绝再查大小防止内存打爆最后用magic读前 2048 字节判断真实 MIME。参数MAX_SIZE按部署环境内存调整20MB 是单机 Streamlit 的保守值。注意python-magic在 Windows 上要额外装libmagicLinux 上apt install libmagic1即可这是部署时的高频翻车点。3. 文档解析与向量化流水线从 PDF 到可检索 chunk3.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, encodingenc) as f: return [{text: f.read(), page: 1}] except UnicodeDecodeError: continue raise ValueError(编码无法识别) if ext .csv: with open(path, encodingutf-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 建议加layoutTrue参数或换PyMuPDF。page字段保留是为了后面做引用溯源答案里能标出“来自第 3 页”。CSV 每行拼成一条文本适合表格型知识但行数太多时要限制否则 chunk 数量爆炸。3.2 切块策略固定长度 重叠但元数据要带全切块没有银弹但内部文档场景用“固定长度 重叠”最稳。RecursiveCharacterTextSplitter是 LangChain 里最常用的分隔符按[\n\n, \n, 。, , , , ]排优先在段落边界切。chunk_size 设 500800 字符overlap 设 50100。中文场景下 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_size600, chunk_overlap80, 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_size600是中文经验值英文文档可以调到 1000。chunk_overlap80保证跨块句子不被切断。注意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( namedocs, 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_embeddingsTrue).tolist() collection.add( ids[f{c[metadata][doc_id]}_{c[metadata][chunk_idx]} for c in batch], embeddingsembs, documentstexts, metadatas[c[metadata] for c in batch], )参数说明normalize_embeddingsTrue让向量单位化配合hnsw:spacecosine保证相似度计算正确。batch_size100是内存和速度的平衡点显存小的机器调到 32。ids用doc_id chunk_idx拼保证唯一且可追溯。注意 Chroma 的PersistentClient路径要挂持久化卷容器重启不丢数据。4. MySQL 与向量库的分工别把向量塞进 MySQL4.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_size5, max_overflow10, pool_recycle3600。# db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine( mysqlpymysql://rag:password127.0.0.1:3306/ragdb?charsetutf8mb4, pool_size5, max_overflow10, pool_recycle3600, pool_pre_pingTrue, ) SessionLocal sessionmaker(bindengine)参数说明pool_pre_pingTrue在取连接前先 ping 一下避免拿到已断开的连接这是解决MySQL server has gone away的关键。pool_recycle3600让连接每小时回收防止 MySQL 的wait_timeout把空闲连接掐掉。charsetutf8mb4必须带否则中文和 emoji 存进去变问号。热词里那个error 2002 (hy000): cant connect to local mysql server through socket /tmp/mysql.sock是经典问题连接串里写了localhostPyMySQL 会走 socket 而不是 TCP。改成127.0.0.1强制走 TCP 就能解决。这个坑我踩过不止一次血泪经验。5. Streamlit 前端与流式响应把等待变成体验5.1 流式响应的实现生成器 st.write_streamRAG 的答案生成要调 LLM几秒到十几秒的等待没有流式输出用户会以为卡死。Streamlit 从 1.31 开始有st.write_stream直接吃生成器。后端把 LLM 的流式输出包成生成器前端逐字渲染。# rag_chain.py from openai import OpenAI client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keynot-needed) def stream_answer(question: str, contexts: list[str]): prompt build_prompt(question, contexts) resp client.chat.completions.create( modelqwen2.5-7b-instruct, messages[{role: user, content: prompt}], streamTrue, temperature0.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智能文档检索, layoutwide) 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接收生成器并逐字渲染返回完整字符串用于落库。参数temperature0.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没设streamTrue。解决确认 LLM 服务支持 SSE客户端用streamTrue并逐行读。另一个原因是 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别设太大510 够用max_overflow给 10 缓冲。6.5 向量库和 MySQL 数据不一致现象MySQL 里文档状态是ready但向量库查不到或者文档已删除向量还在。原因是入库和删库是两步操作中间失败没回滚。解决入库流程做成“先写 MySQL 状态 processing → 向量化 → 写向量库 → 更新 MySQL 为 ready”任何一步失败都标failed并记录原因。删除时先删向量再删 MySQL 记录或者用软删除定期跑对账脚本清理孤儿向量。这个对账脚本我一般会写成定时任务每天凌晨跑一次。7. 进阶技巧用混合检索和重排序把召回率再提一档纯向量检索在专有名词、编号、缩写上经常翻车比如“GB/T 19001”这种向量模型未必能区分。进阶做法是混合检索向量检索 BM25 关键词检索两路召回后用 RRFReciprocal 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_k10): # 向量路 q_emb model.encode([query], normalize_embeddingsTrue).tolist() vec_res collection.query(query_embeddingsq_emb, n_resultstop_k, wherewhere) 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(), keylambda x: -x[1])[:top_k]参数说明jieba.cut做中文分词BM25 需要 token 列表。top_k10是每路召回数融合后取前 10 送进 LLM。RRF 的k60是论文里的经验值不用调。注意 BM25 索引要跟着文档增删同步更新否则新文档搜不到。如果嫌维护两套索引麻烦可以用支持混合检索的向量库比如 Qdrant 的 sparse vector但配置复杂度更高。重排序是另一档提升召回 top 20 后用 cross-encoder 模型如bge-reranker-base逐对打分取前 5 送 LLM。cross-encoder 比向量相似度准但慢只适合小候选集。我一般只在答案质量要求高的场景开重排序普通问答省掉这步。验证效果别靠感觉准备 50 条带标准答案的问题算召回率正确文档在 top-k 里的比例和答案准确率。每次改检索策略都跑一遍数据说话。我自己的习惯是任何检索参数改动先跑评测集涨了才合并跌了直接回滚。这套流程帮我省了无数次“感觉变好了其实变差了”的返工。希望帮到你。本文还有配套的精品资源点击获取