简介：AI知识库系统Python源码面向希望本地搭建知识管理工具的Python学习者与开发者，围绕文件上传、内容检索与问答记录等场景，提供一套可直接运行的小型系统实现。压缩包共22个文件、约53KB，以10个html模板页面和6个py源码为主，另有4个pyc字节码缓存、1个txt说明文档与1个db数据库文件；模板覆盖登录、文章编辑、文档管理、问答检索等界面，源码则承担配置、路由、数据模型与启动入口等职责，结构紧凑，适合通读前后端协作流程。目前已有66人学习，可作为课程设计、毕业项目原型或Web入门练手素材，帮助读者理解知识库从数据库初始化、页面渲染到检索功能落地的完整链路，并在此基础上扩展标签、权限或语义检索等能力。
1. 从一份三百页的产品手册开始：这套 AI 知识库系统解决什么问题
AI知识库系统 Python 源码拿到手，多数人第一反应是先把依赖装齐、把模型 Key 填上，然后跑起来问一句「售后政策怎么算」，结果返回一堆不相关段落。问题通常不在模型，而在文档怎么切、向量怎么存、检索怎么召回。这套系统解决的是把散落的 PDF、Word、Markdown 变成大模型可引用的私有知识，走的是 RAG 那条最成熟的链路：解析、切片、向量化、检索、拼 Prompt、生成。它适合想给团队搭内部问答的开发者，也适合拿它当骨架做行业问答后端。读完你该能判断它值不值得改，先从哪个文件下手。
2. 架构拆解：一次提问在六个模块之间怎么流转
这类项目看着文件不少，真正的主干只有一条线：文件进来、文本出去、向量落库、问题进来、片段召回、答案流出去。剩下的路由、模型定义、配置加载都是挂在主干上的枝节。先把主干认清楚，后面改任何一处你都知道会牵动谁。
我一般拿到源码先不读main.py，而是先看core/目录——那是业务真正发生的地方，入口文件只是把请求转交过去。
2.1 目录结构与模块边界：为什么切成四层
一份组织得比较规整的 Python 知识库源码，目录大概长这样：
ai-knowledge-base/ ├── app/ │ ├── main.py # FastAPI 入口，注册路由、CORS、异常处理器 │ ├── config.py # pydantic-settings 读 .env，全局唯一配置源 │ ├── api/ │ │ ├── documents.py # 上传、列表、删除 │ │ ├── chat.py # 问答、会话历史、流式接口 │ │ └── health.py # 健康检查，给容器探针用 │ ├── core/ │ │ ├── loader.py # 各种格式统一解析成纯文本 │ │ ├── splitter.py # 切片、重叠、元数据附加 │ │ ├── embedder.py # 向量化封装，屏蔽厂商差异 │ │ ├── vector_store.py # 向量库读写适配层 │ │ ├── retriever.py # 检索、过滤、重排 │ │ └── llm.py # 大模型调用与流式返回 │ ├── models/ # SQLAlchemy 表定义：文档、切片、会话 │ └── schemas/ # 请求响应模型，Pydantic 校验 ├── scripts/ │ ├── init_db.py # 建表 │ └── ingest_cli.py # 命令行批量灌库，不经过 HTTP ├── data/ # 原始文件目录 + 本地向量库落盘目录 ├── tests/ ├── .env.example └── requirements.txt四层边界很清楚：api/只负责收参数、做校验、调core/，不写业务逻辑；core/不依赖 FastAPI，可以脱离 Web 单独跑脚本；models/和vector_store.py是两类存储，一个存业务元数据，一个存向量；config.py是唯一的配置入口，所有魔法数字都从这里取。
这种分层最大的好处是可测试。你可以写一个 pytest，直接 importcore.splitter，喂一段文本进去断言切片数量，完全不用起服务。反过来，如果切片逻辑写在了路由函数里，测一次就得发一次 HTTP 请求，调参数会非常难受。
2.2 技术选型：三处必须做成可替换的地方
默认技术栈通常是这样一套组合：
| 环节 | 常见默认选择 | 可替换为 | 替换代价 |
|---|---|---|---|
| Web 框架 | FastAPI + Uvicorn | Flask、Litestar | 低，路由重写 |
| 文档解析 | pypdf、python-docx | pdfplumber、unstructured | 低，只改 loader |
| 向量化 | sentence-transformers 本地模型 | 任意 OpenAI 兼容的 embedding 接口 | 中，注意维度变化 |
| 向量库 | Chroma 本地持久化 | FAISS、Milvus、pgvector | 中，要重写适配层 |
| 大模型 | OpenAI 兼容接口 | 本地部署的推理服务 | 低，改 base_url 和模型名 |
关键不在于选哪个，而在于这三处是不是被抽象成了接口。合格的写法是在core/里定义基类，具体实现按配置加载：
# app/core/embedder.py from abc import ABC, abstractmethod class BaseEmbedder(ABC): """所有向量化实现都要满足这个契约：批量进，批量出，维度自报""" @property @abstractmethod def dimension(self) -> int: ... @abstractmethod def encode(self, texts: list[str]) -> list[list[float]]: ... class LocalEmbedder(BaseEmbedder): def __init__(self, model_name: str): from sentence_transformers import SentenceTransformer self._model = SentenceTransformer(model_name) @property def dimension(self) -> int: # 维度必须暴露出来，建集合时要用，换模型时能第一时间发现不一致 return self._model.get_sentence_embedding_dimension() def encode(self, texts: list[str]) -> list[list[float]]: vecs = self._model.encode(texts, normalize_embeddings=True, batch_size=32) return vecs.tolist()dimension这个属性看着多余，实际是保命的。向量库建集合时维度是写死的，你后来把 embedding 模型从 384 维换成 1024 维却没重建集合，写入会直接报维度不匹配，或者更糟——静默失败，检索永远返回空。把维度做成显式契约，启动时校验一次，能省掉一下午的排查。
normalize_embeddings=True也值得单独说一句。归一化之后余弦相似度和内积等价，很多向量库默认用内积检索，不归一化的话分数尺度会很怪，阈值调起来毫无手感。
2.3 配置项与数据模型：改代码前先看这几张表
.env.example里的参数决定了系统行为，改动之前最好心里有数：
| 变量 | 典型值 | 作用 | 调大的后果 |
|---|---|---|---|
CHUNK_SIZE | 500 | 单切片字符数 | 片段更完整，但召回精度下降 |
CHUNK_OVERLAP | 80 | 相邻切片重叠字符 | 边界信息不丢，但存储量上升 |
TOP_K | 5 | 检索返回片段数 | 上下文更全，但容易超长、噪声变多 |
SCORE_THRESHOLD | 0.25 | 相似度下限 | 调高会漏召回，调低会灌噪声 |
EMBEDDING_DIM | 384 / 1024 | 向量维度 | 与模型不匹配直接写库失败 |
业务侧一般有两张核心表：documents记录文件本身的 id、文件名、哈希、状态；chunks记录切片 id、所属文档 id、序号、文本、向量库里的主键。这两张表是向量库和关系库之间的桥，删除文档时靠doc_id去向量库里做过滤删除，靠chunks做级联清理。很多源码只做了前者，跑一段时间就会发现删掉的文件还能被检索出来——向量还在库里躺着。
提示：先确认
documents.status有没有pending / processing / ready / failed这几个状态。没有状态机的话，上传大文件时前端一直在等，用户会以为卡死了。
3. 入库链路：解析、切片与向量化落库
入库是整个系统里最脏的活，也是最值得花时间的地方。检索效果差，八成的原因在入库这一侧——切片切碎了、表格丢了、页眉页脚混进正文，后面模型再强也救不回来。
这条链路的顺序是固定的：文件落到磁盘、解析成文本、按规则切片、批量向量化、写入向量库、更新业务表状态。
3.1 文件解析层：把五种格式统一成一种结构
解析层唯一的目标是输出统一结构，让下游完全不用关心原始文件是什么格式：
# app/core/loader.py from pathlib import Path from pypdf import PdfReader from docx import Document def load_document(path: Path) -> list[dict]: """返回 [{text, metadata}]，metadata 保留来源信息供后续引用标注""" suffix = path.suffix.lower() if suffix == ".pdf": return _load_pdf(path) if suffix in (".docx", ".doc"): return _load_docx(path) if suffix in (".md", ".txt"): raw = path.read_text(encoding="utf-8", errors="ignore") return [{"text": raw, "metadata": {"source": path.name, "page": 0}}] raise ValueError(f"不支持的文件类型: {suffix}") def _load_pdf(path: Path) -> list[dict]: reader = PdfReader(str(path)) pages = [] for idx, page in enumerate(reader.pages): text = (page.extract_text() or "").strip() if not text: continue # 扫描件没有文本层，这里会拿到空串，直接跳过 pages.append({ "text": text, "metadata": {"source": path.name, "page": idx + 1}, }) return pages def _load_docx(path: Path) -> list[dict]: doc = Document(str(path)) paras = [p.text.strip() for p in doc.paragraphs if p.text.strip()] return [{"text": "\n".join(paras), "metadata": {"source": path.name, "page": 0}}]metadata里的page字段很容易被忽略，但它决定了回答能不能给出「见第 12 页」这种定位。做内部知识库时，用户对答案的信任度很大程度来自这个引用——没有出处，答得再对也像编的。所以解析阶段就要把页码带上，后面切片时原样继承下去。
扫描件那条注释是重点。PDF 分两类，一类是电子生成的，有文本层，extract_text()能拿到内容；另一类是扫描的图片，取出来是空字符串。有些源码在这里不判断，直接往下列，结果一份 200 页的扫描手册入库后 chunks 表里一条记录都没有，用户上传成功却永远问不出答案。要么接入 OCR 分支，要么在解析后判断空文本并返回明确的失败状态。
3.2 切片策略：按字符切还是按语义切
切片是最需要反复调的一步。常见做法是三级切分：先按标题和空行切段落，段落太长再按句子切，最后按固定窗口合并到目标长度：
# app/core/splitter.py import re def split_text(text: str, chunk_size: int = 500, overlap: int = 80) -> list[str]: """按段落优先、窗口兜底的顺序切片，尽量不把一句话从中间切断""" # 1. 先按连续空行切成自然段 paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] chunks, buf = [], "" for para in paras: # 2. 单段本身就超长，按句号补切 if len(para) > chunk_size: sentences = re.split(r"(?<=[。！？.!?])", para) for sent in sentences: if not sent.strip(): continue if len(buf) + len(sent) > chunk_size: chunks.append(buf.strip()) buf = buf[-overlap:] if overlap else "" buf += sent continue # 3. 正常段落，塞得下就塞，塞不下先落盘再留重叠 if len(buf) + len(para) > chunk_size: chunks.append(buf.strip()) buf = buf[-overlap:] if overlap else "" buf += para + "\n" if buf.strip(): chunks.append(buf.strip()) return [c for c in chunks if len(c) >= 20] # 丢掉碎片chunk_size=500是中文场景下一个比较稳的经验值，大致对应 300 到 400 个 token，既能装下一个完整条目，又不至于让检索变得模糊。英文资料可以放到 800 到 1000 字符。
overlap=80的作用是让相邻切片共享一小段尾巴，防止答案正好卡在切分点上。设成 0 的话，像「第三条 … 的例外情形如下」这种跨段表述，前半句在 chunk 7、后半句在 chunk 8，检索命中 chunk 7 时模型看到的是一句没说完的话，很容易答偏。重叠比例一般取 chunk_size 的 10% 到 15%，再大就是浪费存储和上下文预算。
最后那行len(c) >= 20的过滤别省。真实文档里全是「目录」「附录」「第 3 章」这种一两行的碎片，它们会参与向量化，然后以莫名的高分挤进检索结果，白白占掉 Top-K 名额。
3.3 批量向量化与写库：幂等和增量怎么做
入库脚本最怕两件事：重复灌库导致同一份文档出现两遍，以及中途失败留下半截数据。解决办法是给每个切片生成确定性的主键：
# app/core/ingest.py import hashlib from pathlib import Path def ingest_file(path: Path, doc_id: str, store, embedder, chunk_size: int, overlap: int) -> int: raw = path.read_bytes() file_hash = hashlib.md5(raw).hexdigest() # 文档级幂等：内容没变直接跳过，省掉一次全量向量化 if store.get_doc_hash(doc_id) == file_hash: return 0 pages = load_document(path) all_chunks, metas = [], [] for page in pages: for i, piece in enumerate(split_text(page["text"], chunk_size, overlap)): # 切片主键 = 文档 id + 页码 + 页内序号，重跑结果稳定 cid = f"{doc_id}:{page['metadata']['page']}:{i}" all_chunks.append(piece) metas.append({**page["metadata"], "doc_id": doc_id, "chunk_id": cid}) # 分批向量化，一次 64 条，避免长文本把内存打满 for start in range(0, len(all_chunks), 64): batch = all_chunks[start:start + 64] vectors = embedder.encode(batch) store.upsert( ids=[m["chunk_id"] for m in metas[start:start + 64]], vectors=vectors, documents=batch, metadatas=metas[start:start + 64], ) store.set_doc_hash(doc_id, file_hash) return len(all_chunks)三个设计点值得展开。第一，file_hash判重放在最前面，重新跑一遍入库脚本时未改动的文件直接返回 0，处理上千份文档时能省掉绝大部分时间。第二，切片 id 是确定性生成的，upsert语义下重复写入会覆盖而不是新增，即使中途失败重跑也不会产生脏数据。第三，分批大小 64 是内存和吞吐的折中，本地模型在 CPU 上跑时再大容易 OOM，接远程 embedding 接口时可以考虑放宽到 128 或 256。
删除路径要成对处理，这是最容易漏的一环：
def delete_document(doc_id: str, store, db): """先删向量再删业务记录，顺序反了会留下永远查不到的孤儿向量""" store.delete(where={"doc_id": doc_id}) db.execute("DELETE FROM chunks WHERE doc_id = :d", {"d": doc_id}) db.execute("DELETE FROM documents WHERE id = :d", {"d": doc_id}) db.commit()where={"doc_id": doc_id}依赖向量库支持元数据过滤。Chroma 和 pgvector 都支持，FAISS 原生不支持，得靠自己维护 id 到向量的映射表。如果源码用的是 FAISS，删文档这块大概率是缺失的，属于二次开发时要补的坑。
4. 问答链路：检索、拼装与流式返回
入库决定了下限，问答链路决定上限。这条链路短得多，但每一步都有能翻车的地方：检索召回了不相关片段、Prompt 里塞了太多上下文把关键信息挤掉、流式输出和前端对不上协议。
4.1 检索阶段：Top-K、阈值与去重
检索函数不该只做一次向量查询就返回，中间至少要加分数过滤和去重：
# app/core/retriever.py def retrieve(question: str, embedder, store, top_k: int = 5, score_threshold: float = 0.25, filters: dict | None = None) -> list[dict]: q_vec = embedder.encode([question])[0] # 先多取一些，过滤和去重之后再截断到 top_k，避免过滤后不够数 hits = store.search(q_vec, top_k=top_k * 3, filters=filters) seen, results = set(), [] for hit in hits: if hit["score"] < score_threshold: continue # 同一文档相邻切片内容高度重合，只留分数最高的那条 key = (hit["metadata"]["doc_id"], hit["metadata"].get("page")) if key in seen: continue seen.add(key) results.append(hit) if len(results) >= top_k: break return results先取top_k * 3再截断，是因为加了阈值过滤之后很可能不够数。如果你按top_k=5直接取，过滤掉两条低分的，最后只有 3 条进 Prompt，模型拿到的证据不足，答案就开始发虚。
filters参数是给多租户和权限控制留的口子。传入{"tenant_id": "t001"}，向量库会在检索阶段就做元数据过滤，而不是先查出来再在 Python 里筛——后者在大库上会把延迟拉高一个量级。
阈值 0.25 不是通用值，它跟 embedding 模型和归一化方式强相关。同一个问题在 A 模型上最高分 0.6，在 B 模型上可能只有 0.3。换模型之后必须重新标一遍阈值，最省事的办法是拿 20 条已知答案的问题跑一遍，看正确片段的分数分布落在哪个区间。
4.2 Prompt 组装：让模型只依据给定片段回答
Prompt 的结构比措辞重要。事实类问答最稳的写法是：系统提示词定死边界，用户消息里先给编号片段，再给问题：
# app/core/prompt.py SYSTEM_PROMPT = """你是一个严谨的知识库助手。规则： 1. 只能依据【参考资料】中的内容回答，不得引入外部知识； 2. 每个结论后面用 [编号] 标注来源； 3. 参考资料中没有相关信息时，直接回答「资料中未提及」，不要猜测。""" def build_messages(question: str, hits: list[dict]) -> list[dict]: blocks = [] for i, hit in enumerate(hits, start=1): src = hit["metadata"].get("source", "未知") page = hit["metadata"].get("page", 0) blocks.append(f"[{i}] 来源：{src} 第{page}页\n{hit['document']}") user = "【参考资料】\n" + "\n\n".join(blocks) + f"\n\n【问题】\n{question}" return [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user}, ]三条规则里第一条最关键。不写这句，模型遇到检索空白时会自动用训练数据补全，答得头头是道但跟你的文档毫无关系，用户一核对就崩。第三条给了它一个「不知道」的出口，比逼它硬答要好得多。
编号加来源的格式是为了可追溯。前端拿到答案后可以按[1]反查是哪份文件哪一页，这个体验差别很大——同样一句结论，带页码的用户敢用，不带的用户要自己去核。
上下文预算要卡住。top_k=5、每片 500 字，加上系统提示词大约 3000 字上下，对多数模型都富余。如果要放到 10 片，就得考虑截断或压缩，否则重要信息会被埋在中段——模型对长上下文中间部分的注意力本来就弱。
4.3 流式输出：SSE 接口与前端对齐
首字延迟直接决定体验，非流式接口在长答案上要等十几秒。用 SSE 推流，前端按data:行解析即可：
# app/api/chat.py import json from fastapi import APIRouter from fastapi.responses import StreamingResponse router = APIRouter() @router.post("/chat/stream") async def chat_stream(req: ChatRequest, retriever, llm): hits = retriever.retrieve(req.question, filters=req.filters) if not hits: # 没召回到东西就别调模型了，直接返回固定话术，省一次调用 async def empty(): yield f"data: {json.dumps({'delta': '资料中未提及'}, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(empty(), media_type="text/event-stream") messages = build_messages(req.question, hits) async def event_gen(): async for token in llm.stream(messages): payload = {"delta": token} yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" # 结束标记要单独发，前端靠它关闭连接和渲染引用区 yield "data: [DONE]\n\n" return StreamingResponse( event_gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, )X-Accel-Buffering: no这个响应头是在反向代理后面部署时的血泪经验。少了它，代理层会把 SSE 缓冲起来，前端表现就是「一直转圈，最后一次性刷出全部文字」，流式白做。
[DONE]标记也别省。有了它，前端可以明确区分「正常结束」和「连接断了」，不用靠超时判断。空召回分支提前返回同样重要，hits为空还去调模型，等于花一次调用让模型自由发挥。
5. 跑起来与排查：六个我踩过的坑
先把最小可跑路径走通，再去动代码。这套系统的依赖不算轻，尤其是本地 embedding 模型，第一次启动会下载几百兆权重，网络不稳的话很容易以为程序挂了。
5.1 本地启动的最小路径
按顺序执行，每一步都能验证：
# 1. 建虚拟环境，Python 建议 3.10 及以上 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 2. 装依赖，本地模型那套会拉 torch，比较慢，耐心等 pip install -r requirements.txt # 3. 复制配置，至少把模型名和 base_url 改对 cp .env.example .env # 4. 初始化数据库表 python scripts/init_db.py # 5. 命令行灌一份文档，先验证入库链路通不通 python scripts/ingest_cli.py --path ./data/sample.pdf --doc-id demo-001 # 6. 起服务 uvicorn app.main:app --reload --port 8000第 5 步是关键的自检点。直接走 HTTP 上传的话，出错了你分不清是解析问题、向量化问题还是接口问题；用 CLI 跑一遍，能确认loader → splitter → embedder → store这四段是否都正常。灌完之后去向量库目录看有没有生成文件，或者直接调一次检索函数，比猜快得多。
服务起来之后用 curl 验一下问答：
curl -N -X POST http://127.0.0.1:8000/api/chat/stream \ -H "Content-Type: application/json" \ -d '{"question":"这份文档讲了什么"}'-N是关掉 curl 自己的缓冲，能看到逐字输出说明 SSE 通了。
5.2 六个高频故障：现象、原因与解决
现象一：上传返回 200，但提问永远答「资料中未提及」。原因通常是向量没写进去，或者写进去的维度跟集合配置不一致。本地 embedding 模型换过一次、集合没重建，就会出现这种静默失败。 解决：先查chunks表有没有记录。有记录就直接数一下向量库的 count，对不上说明写入失败；再打印embedder.dimension和集合的维度设置比对，不一致就删掉集合重新灌库。
现象二：第一次提问要等一两分钟，之后正常。本地模型冷启动，权重不在页缓存里，首次 forward 特别慢。这不是 bug。 解决：启动时预热一次，在main.py的 startup 事件里调一次embedder.encode(["预热"])，顺便把大模型客户端也初始化掉。容器部署时把模型目录挂成卷，避免每次重建都重新下载。
现象三：表格类问题答得一塌糊涂，数字经常张冠李戴。固定窗口切片把表格切碎了，同一行的参数被分到两个 chunk 里。 解决：在loader阶段单独识别表格，整表作为独立片段入库，元数据打上type: table；检索时对表格片段提高权重。纯文本方案救不了这个问题，得在解析层动手。
现象四：答案越来越长，还重复同一段话。top_k调太大，加上相邻切片重叠，Prompt 里出现了三份几乎一样的内容，模型倾向于把它们都复述一遍。 解决：按doc_id + page去重，同时给 Prompt 加总字符数上限，超了就按分数从低到高裁掉尾巴。别指望模型自己忽略重复内容。
现象五：删掉文档之后，还能检索出它的内容。只删了documents和chunks表，向量库里还留着。或者用的是不支持元数据过滤的向量库，删除逻辑写了但没生效。 解决：删除走「先向量后业务表」的顺序，并且用where={"doc_id": ...}精确删。删完立刻检索一次验证，别等用户反馈。
现象六：接口偶发 422 或 500，日志里看不出所以然。多半是.env没读到——配置用相对路径、工作目录不是项目根，pydantic-settings就读了个空。或者必填的环境变量漏配，启动时没校验，跑到一半才炸。 解决：把路径统一转成基于__file__的绝对路径；配置类里给关键字段加默认值或设成必填，让它在启动阶段就报错，而不是请求阶段。
6. 二次开发：从通用问答改成能验收的业务助手
改造这套东西，我一般不重写链路，从换掉一个不合适的默认值开始，再用评测集兜底。
| 改造点 | 改动位置 | 验证方式 |
|---|---|---|
| 换 embedding 模型 | .env的模型名 + 重建集合 | 同一问题前后 Top-3 是否更相关 |
| 加租户隔离 | 上传时写tenant_id，检索传 filters | 租户 A 的问题搜不到 B 的文档 |
| 表格单独处理 | loader加表格分支 | 抽 10 个表格问题人工比对 |
| 引用外显 | 流结束时补发一份sources | 前端点击能跳到对应页码 |
权限那块最容易被做漏。很多源码是把tenant_id存进数据库，检索的时候忘了往filters里传，结果向量库把全库都搜了一遍，在 Python 侧再过滤——数据已经从库里出来了，隔离形同虚设：
# 正确做法：过滤条件下推到向量库，而不是查完再筛 hits = store.search(q_vec, top_k=top_k * 3, filters={"tenant_id": current_user.tenant_id})改完一定要建评测集。不用多，50 条就够：每条写清问题、标准答案的关键词、应该命中的文档 id。跑一遍能算出两个数——Top-5 召回率和答案命中率。前者低于 0.8 就去调切片和阈值，后者低但召回正常，说明问题在 Prompt。
从那以后我每次接这类知识库源码，都先把这 50 条问答跑通再动业务代码，不然改完根本说不清是变好了还是变差了。希望帮到你。
本文还有配套的精品资源，点击获取
