简介面向需要构建知识管理与检索系统的Python开发者这份AI知识库系统源码提供了从数据存储到Web展示的完整实现思路可快速落地本地化知识库应用。压缩包共22个文件含10个HTML页面、6个Python脚本、4个pyc缓存、1个SQLite数据库及1份使用说明整体仅53KB结构紧凑。源码中app.db负责持久化存储知识条目与检索索引config.py管理数据库连接及上传参数init_db.py完成数据表初始化run.py作为启动入口协同routes.py实现路由分发templates目录下的登录、文章管理、问答统计等页面则覆盖了知识录入、文档管理、问答检索等核心场景。既适合初学者理解Flask项目的分层组织方式也可供进阶开发者参考文件上传、全文检索及日志追踪的落地写法。已有66人学习下载可直接对照源码二次开发。1. 这个AI知识库系统能解决什么问题一个文档散乱、问答靠人肉检索的团队最需要的不是什么智能音箱而是一条能把 PDF、Word、Markdown 吃进去、再按语义把答案捞出来的流水线。这个AI知识库系统 Python 源码本质就是拆好的这条流水线上传文档、自动切片、文本向量化、相似度检索、大模型生成回答每一步都能在本地私有化运行。它适合两类人一类是想在自己服务器上落地 RAG 检索增强生成做内部知识库的运维或后端开发另一类是刚接触向量检索的学生或初级工程师想找一个能跑通全流程的参考实现。整份代码没有复杂的分布式依赖解包后拿到的是完整的 FastAPI 服务加一套可替换的模型封装。2. 架构设计与技术选型一条从文档到答案的完整管线2.1 整体架构五段式流水线拿到这份源码我第一件事不是跑起来而是把代码里的调用关系画成数据流。整个项目设计的是一条单向流水线每一段相对独立替换其中一段不会牵动其他段。第一段是文档接入层接收上传的 PDF、Word、Markdown 文件做格式白名单校验、大小限制和临时存储。第二段是解析与清洗层把不同格式转成带页码标记的纯文本去掉页眉页脚和样式噪声。第三段是切片与向量化层先把长文本切成有重叠的块再逐块调用本地中文向量模型生成 embedding写入向量库。第四段是检索层用户问题先转成向量在库里做相似度查询取回 Top-K 个片段。第五段是问答生成层把片段和问题拼成限定上下文的 Prompt发给大模型做流式输出。这个分层最直接的好处是每一层都可以单独验证。文档能不能解析看切片文件就行切片有没有切碎查库里的文本块就行检索准不准单独调 search 函数就行。很多 RAG 项目翻车就是因为把所有环节揉在一个函数里出了问题不知道查哪一段。2.2 目录结构每个文件扮演什么角色源码解包后的目录结构如下ai-knowledge-base/ ├── app/ │ ├── main.py # FastAPI 入口注册路由与启动事件 │ ├── config.py # 全局配置模型路径、向量库目录、切块参数 │ ├── api/ │ │ ├── ingest.py # 文档上传与入库接口 │ │ └── chat.py # 问答与流式输出接口 │ ├── core/ │ │ ├── parser.py # 多格式文档解析器 │ │ ├── splitter.py # 文本切片器 │ │ ├── embeddings.py # Embedding 模型封装与向量生成 │ │ ├── retriever.py # 相似度检索封装 │ │ └── llm_client.py # 大模型 HTTP 客户端 │ ├── models/ │ │ ├── schema.py # Pydantic 请求/响应模型 │ │ └── db.py # SQLite 元数据表定义与操作 │ └── utils/ │ └── file_utils.py # 文件校验、临时目录清理 ├── data/ │ ├── vector_db/ # Chroma 持久化目录 │ └── uploads/ # 上传文件临时目录 ├── requirements.txt └── README.mdcore 目录是核心api 目录只是薄薄一层 HTTP 封装。models/db.py 里存的是文档 ID、文件名、上传时间、切片数量这些元数据向量本体在 data/vector_db 里。两者分开的好处是向量库要重建时元数据还在能反查原始文件。config.py 里集中管理所有可变参数切块大小、重叠大小、Top-K、相似度阈值、模型路径都在这里。我改造项目时习惯先改 config再改代码能少动很多地方。2.3 关键技术选型与依赖依赖清单集中在 requirements.txt核心选型如下依赖库用途选型理由fastapiWeb 框架原生异步流式输出方便uvicornASGI 服务器与 FastAPI 配套单进程够用chromadb向量库嵌入式持久化免单独部署sentence-transformers文本向量化本地跑模型数据不出网pymupdfPDF 文本抽取单文件解析速度快python-docxWord 解析保留段落和表格结构sqlmodel元数据 ORM配 SQLite表结构定义简洁这套选型明显是冲着“简单私有化”去的。没有引入 Elasticsearch因为内部知识库的数据量级通常在万级文档以内向量检索完全够用没有上 Milvus因为那是集群方案运维成本高。如果未来数据涨到百万级Chroma 的数据目录可以直接迁移到 FAISS 或 Milvus接口层改动不大。Embedding 模型默认配置的是本地路径第一次启动前要把下载好的中文向量模型放到 models 目录下再在 config.py 里指向它。英文模型跑中文检索很容易出现“一句话被切成两截”的尴尬所以代码里默认就要求中文模型这一点后面避坑章节还会再提。3. 核心模块拆解文档解析、切片与向量化的实现细节3.1 文档解析统一入口背后的多格式适配parser.py 暴露的是一个统一的 load_document 函数上层不问格式只要传路径和扩展名拿到的就是带页码的纯文本。def load_document(file_path: str, ext: str) - str: 统一入口按扩展名分发到对应解析器 if ext .pdf: return parse_pdf(file_path) if ext .docx: return parse_docx(file_path) if ext in (.md, .markdown, .txt): return parse_markdown(file_path) raise UnsupportedFormatError(f不支持的文件类型: {ext})PDF 解析是这里最需要小心的部分。代码用 pymupdf 的 get_text(text) 抽取文本同时把页码拼进文本块前面。import fitz def parse_pdf(file_path: str) - str: pages [] with fitz.open(file_path) as doc: for page_no, page in enumerate(doc): text page.get_text(text) if not text.strip(): continue pages.append(f【第{page_no 1}页】\n{text}) return \n\n.join(pages)页码标记是给检索结果做引用用的用户看到答案时能知道出自哪一页。这里有个容易踩的坑扫描版 PDF 用 get_text 抽出来是空字符串代码选择跳过而不是报错避免一个坏页拖垮整个入库流程。Word 解析用的 python-docx只提取段落和表格的纯文本样式全部丢弃。from docx import Document def parse_docx(file_path: str) - str: doc Document(file_path) lines [] for para in doc.paragraphs: if para.text.strip(): lines.append(para.text.strip()) for table in doc.tables: for row in table.rows: cells [cell.text.strip() for cell in row.cells] lines.append( | .join(cells)) return \n.join(lines)表格转成竖线分隔的文本是为了后面切片时不把一行表格切断。我实际跑过几个带复杂嵌套表格的 Word 文档这样处理后检索效果比直接丢 HTML 解析结果稳定得多。3.2 切片策略固定窗口与语义边界的取舍切片是 RAG 项目里最影响效果的环节之一。这份源码默认用的是固定窗口加重叠的切法简单、可控、不容易出错。def split_text(text: str, chunk_size: int 512, overlap: int 64) - list[str]: 固定窗口切分保留重叠区间避免关键信息被切断 chars list(text) chunks [] start 0 while start len(chars): end min(start chunk_size, len(chars)) chunks.append(.join(chars[start:end])) if end len(chars): break start start chunk_size - overlap return chunkschunk_size 取 512 字符是经过权衡的中文一个字大约占 1 到 2 个 token512 字在主流大模型上下文里约占 800 到 1000 token既保留了完整语义又不至于把 Prompt 撑爆。overlap 取 64让相邻块之间有两三句重合答案恰好跨在切点上时还能被另一块捞回来。这个方案有个明显短板完全按字符切会切断自然段落。我一般会在调用 split_text 之前先把文本按换行符粗分一次把短段落拼到接近 chunk_size 再切这样能显著减少“一句话被劈成两半”的情况。源码里也留了段落聚合接口改造时可以直接替换 splitter。3.3 向量化与存储embeddings.py 负责把文本块变成向量并写入向量库。模型路径从 config 读取默认指向本地目录避免每次启动都去联网下载。from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings model SentenceTransformer(./models/embedding-model) # 替换为本地中文向量模型 def embed_and_store(doc_id: str, chunks: list[str], metadatas: list[dict]): vectors model.encode(chunks, normalize_embeddingsTrue).tolist() client chromadb.PersistentClient( path./data/vector_db, settingsSettings(anonymized_telemetryFalse), ) collection client.get_or_create_collection(knowledge) collection.upsert( ids[f{doc_id}_{i} for i in range(len(chunks))], embeddingsvectors, documentschunks, metadatasmetadatas, )normalize_embeddingsTrue 这行很关键它让向量模长变为 1后续计算余弦相似度就变成了点积运算查询速度快一截。metadatas 里我传的是文档 ID 和页码检索结果展示来源时直接从这里取。顺带说一句入库接口 ingest.py 里我用的是 FastAPI 的 BackgroundTasks上传接口先返回“入库中”的状态解析和向量化在后台跑。千万别在请求线程里同步做完整条链路一个 20MB 的 PDF 能把请求拖到超时。4. 在线检索与问答相似度计算、Prompt 拼接与流式输出4.1 检索核心从问题到相关片段检索封装在 retriever.py核心就一个 search 函数。它接受用户问题转成向量后在 Chroma 里做查询再过滤掉距离过远的块。def search(query: str, top_k: int 5, threshold: float 0.3): query_vec model.encode([query], normalize_embeddingsTrue).tolist() client chromadb.PersistentClient(path./data/vector_db) collection client.get_collection(knowledge) result collection.query(query_embeddingsquery_vec, n_resultstop_k) hits [] for i in range(len(result[ids][0])): distance result[distances][0][i] if distance threshold: hits.append({ doc_id: result[metadatas][0][i].get(doc_id), page_no: result[metadatas][0][i].get(page_no), text: result[documents][0][i], score: round(1 - distance, 4), }) return hits注意 Chroma 返回的是 distance 而不是相似度距离越小越相关。threshold 用来过滤噪声distance 大于 0.3 的块基本是答非所问。score 用 1 - distance 转成“越接近 1 越相关”的直觉值方便前端展示。这里有个经验top_k 调大不代表效果变好。内部知识库的答案往往集中在一两个块里Top-5 里后三个都是陪跑塞进 Prompt 反而干扰大模型。我一般保持 Top-5但 threshold 稍微收紧。4.2 Prompt 拼接让模型只依据材料回答检索到的片段不能直接丢给大模型要拼成有结构的 Prompt。build_prompt 是 llm_client.py 里的一个纯函数不发起网络请求方便单测。def build_prompt(question: str, hits: list[dict]) - list[dict]: context_blocks [] for idx, hit in enumerate(hits, 1): context_blocks.append( f[材料{idx}]来源{hit[doc_id]} 第{hit[page_no]}页\n{hit[text]} ) context \n\n.join(context_blocks) system_prompt ( 你是一个严谨的知识库助手。只能依据下面的材料回答 材料中找不到答案时直接说不知道禁止编造。回答末尾列出用到的材料编号。 ) user_prompt f材料如下\n\n{context}\n\n问题{question} return [ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ]两条约束很关键禁止编造、末尾列材料编号。前者把大模型的幻觉限制在材料范围内后者让用户能回溯到原始文档。实际使用中大模型偶尔会忽略“直接说不知道”所以我在下游还加了一道校验回答里如果出现了材料里完全没有的专有名词就提示用户答案可能超出知识库范围。4.3 API 接口与流式输出chat.py 里定义了 SSE 流式接口。大模型生成是逐字的如果用普通 JSON 返回用户要等十几秒才看到第一个字体验很差。流式接口让每个 token 生成后立刻推给前端。from fastapi.responses import StreamingResponse import json app.post(/api/chat) async def chat(req: ChatRequest): hits search(req.question, top_kreq.top_k, thresholdreq.threshold) messages build_prompt(req.question, hits) async def event_stream(): async for chunk in llm_stream(messages): yield fdata: {json.dumps({delta: chunk}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)ChatRequest 里有 question、top_k、threshold 几个字段前端可以调。llm_stream 是在 llm_client.py 里封装的一个异步生成器内部用 httpx 的 AsyncClient 请求大模型 HTTP 接口逐段返回内容文本。流式协议用的是标准 SSE 格式前端用 EventSource 就能接不用自己解析 chunk。我最初图省事直接把流式响应塞进 JSON 里返回前端拿到的是乱码改了半天才发现要设置 media_type 为 text/event-stream。5. 避坑与排查跑通知识库系统常见的七个坑5.1 环境与依赖的坑现象pip install pymupdf 长时间卡住甚至报错重试几次都失败。原因pymupdf 的 wheel 包体积不小默认 PyPI 源在某些网络环境下连接不稳定经常下到一半断掉。解决安装时明确换源并固定版本命令改成 pip install pymupdf1.24.0 -i https://pypi.tuna.tsinghua.edu.cn/simple 这类镜像地址。装的时候建议用一个干净的虚拟环境别和系统 Python 混在一起。现象按 requirements.txt 装完后运行 SentenceTransformer 导入时报 module numpy has no attribute float。原因代码里有一段兼容逻辑调用了旧版 numpy 的 np.float新版本 numpy 已经把它移除了而环境中又没锁定 numpy 版本。解决把 requirements.txt 里的 numpy 固定到一个稳定版本比如 numpy1.26.4同时确认当前 Python 版本是 3.10 而不是 3.12。Python 3.12 下部分依赖还没有完整适配跑起来会莫名报一些底层错误。5.2 运行与入库的坑现象启动服务后第一个入库请求要等两三分钟才返回前端直接超时。原因向量模型没有预加载第一条请求才把几百 MB 的模型读进内存解析和向量化都被这个加载过程阻塞。解决在 main.py 的 startup 事件里主动调用一次 model.encode([])强制模型提前加载。这样请求到达时模型已经在内存里首查延迟从分钟级降到秒级。现象上传一个 50MB 的扫描版 PDF进程内存涨到几个 GB最后被系统杀掉。原因parse_pdf 把所有页的文本一次性拼成一个大字符串扫描版 PDF 虽然抽不出文字但页面对象全部驻留内存叠加起来就爆了。解决在上传接口限制文件大小超过 20MB 直接拒绝同时给解析器加一个分页流式处理每解析完一页就把文本追加到临时文件释放页对象再清空内存引用。现象第二天重启服务查询时报 cPickle.UnpicklingError 或 EOFError向量库数据查不出来。原因上次进程被强制 killChroma 持久化文件没写完整索引文件损坏。解决先停服务备份 data/vector_db 目录然后删除损坏的目录重启后重新入库。如果你不想再遇到这个坑可以在代码的关闭钩子里显式调用 collection.flush()把缓冲区的数据落盘。5.3 检索效果相关的坑现象检索到的片段有内容但明显和问题不相关比如问“报销流程”捞出来的全是“报销标准”。原因默认的向量检索只做语义近似没有关键词层面的精确匹配。问题换一种说法向量就偏了。解决给检索加一层混合召回。用 BM25 对同一批文档做关键词检索把向量结果和 BM25 结果按权重加权融合通常向量占 0.7、关键词占 0.3能明显改善精确匹配场景。现象中文文档切片后检索结果里经常出现半句话答案读起来断断续续。原因固定窗口切片没有感知中文语义边界切点正好落在句子中间把完整语义劈开了。解决切之前先按换行符和句号分句把句子聚合到接近 chunk_size 再切。聚合规则是遇到句号、问号、换行时作为自然边界下一句放不下就另起一块。这个改动工作量不大检索效果提升立竿见影。6. 二次开发从换格式到换模型的改造路径改造这套系统我推荐按“先加解析、再换模型、后调效果”的顺序来。比如要支持 HTML 文件只需在 parser.py 里加一个分支用 BeautifulSoup 抽取正文过滤掉 script 和 style 标签返回纯文本然后在 load_document 的白名单里加上 .html最后在 ingest.py 的 SUPPORTED_EXTS 里同步放行。三步走完HTML 文档就能走通整条入库链路其余代码一行不用动。替换向量模型同样简单config.py 里把模型路径换成一个新的中文向量模型目录启动时确认路径存在即可。但要注意新模型的向量维度可能和旧库不一致如果换了不同维度的模型Chroma 会报维度错误这时需要删掉 data/vector_db 目录重建索引。换大模型接口则只改 llm_client.py 里的请求地址和鉴权头保留流式接口不变前端不用动。验证方法我固定成三步入库后先去向量库查这条文档的切片总数对不对再拿一个肯定能命中的问题跑 search看 Top-5 里有没有包含答案的那段最后才发起完整问答看流式输出是否正常。三步都过了改动才算真跑通。改造这套系统我吃过一次亏一上来就改解析器改完直接重启结果检索效果还不如旧版回滚花了一个下午。事后我把验证流程固定成上面这三步每次动代码都强制走一遍跑通了再交出去。希望帮到你。本文还有配套的精品资源点击获取