ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

AI知识库系统实战:Python+FAISS+text2vec实现语义问答

AI知识库系统实战:Python+FAISS+text2vec实现语义问答

简介:这份人工智能知识库系统源码包面向需要搭建知识管理或信息检索应用的开发者,提供了基于网页的完整实现,可解决知识条目的存储、分类和快速查询等问题。压缩包共22个文件,其中10个HTML页面负责界面展示,6个Python脚本承担后端逻辑,4个缓存文件由解释器生成,另外还有1个数据库和1个说明文档,整体仅53KB,结构紧凑,适合学习轻量级知识管理系统的设计思路。资源中包含界面模板、数据库连接、配置初始化、系统入口等后端模块,并且涵盖用户注册登录、知识文档管理、问答检索等典型功能页面,支持本地部署后直接运行,用户可通过网页完成知识内容的录入与管理。目前已有66人学习下载,可作为入门参考。对于想了解Web应用或知识库检索机制的读者,该源码提供了清晰的目录划分和基础代码,便于二次开发,也可作为课程设计或毕业设计的起步素材。

1. AI知识库系统是干什么的:先把你手头文档变成能问答的资产

「AI知识库系统」这个名字看起来很大,但拆开看就是三件事:把散落的文档集中管理,用语义而不是关键词去检索内容,再把检索到的内容交给大模型组织成答案。这个源码包就是用 Python 把这三件事完整实现了一遍,跑通它只需要一台普通电脑、一套 Python 环境和几行命令,不需要 GPU 服务器。

适合谁呢?你手头有一堆 Markdown/TXT 文档,想给团队搭一个内部问答工具;或者正在做检索增强生成(RAG)入门,想找一份不是玩具代码、能直接改的项目——这个源码包刚好在你想要的区间。下面我从技术选型、核心实现到运行改造,把整个项目拆开讲一遍。

2. 技术选型与目录结构:为什么是 FAISS + text2vec 而不是别的组合

2.1 技术选型:四个关键组件背后的取舍

一个 AI 知识库系统,真正拆开要决定的事很多:文档存哪里、怎么分块、用什么模型转向量、向量存哪、问答用哪个大模型接口。我先说结论,再逐个解释。

嵌入模型选的是 text2vec-base-chinese。这个模型参数规模不大,CPU 上就能跑,输出向量是 768 维,中文语义质量在同类开源小模型里算稳的。如果你换成英文模型或者别的中文模型,向量维度会变,后面存储层的维度配置必须跟着改,这是整个项目里最容易翻车的一个耦合点。

向量存储用的是 FAISS 的 IndexFlatIP,没有上 Chroma 这类带独立服务的向量数据库。原因很直接:知识库的量级还没大到需要独立服务,IndexFlatIP 做全量暴力检索,在几万条 chunk 的规模下毫秒级返回。IndexFlatIP 是内积索引,配合归一化后的向量,算出来就是余弦相似度——这是关键词检索做不到的语义匹配。

后端用 FastAPI + Uvicorn。异步友好,上传文件、并发请求处理起来顺手,而且 FastAPI 自带的 /docs 页面能在浏览器里直接调试接口,省了不少事。

文档格式只支持 Markdown/TXT。不要小看这个限制,它替我们省掉了解析 PDF 的复杂度。PDF 在真实场景里问题很多,表格、扫描件、页眉页脚都会干扰切分质量,等核心链路跑通之后再按需接入就行。

组件选择理由
嵌入模型text2vec-base-chinese中文语义质量好,CPU 可跑,768 维
向量库FAISS IndexFlatIP轻量无服务,小规模知识库毫秒级返回
后端框架FastAPI异步友好,自带调试文档
文档格式Markdown/TXT切分干净,避免 PDF 解析复杂

2.2 目录结构:每个文件对应什么职责

代码组织遵循的是一个很常见的分层:API 层 -> 服务层 -> 存储层。这样拆的好处是,你换一个向量库或者换一个嵌入模型,改动范围被限制在某一个目录里,不会牵一发动全身。

ai_knowledge_base/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── api/ │ │ └── routes.py # /upload /ask /health 路由 │ ├── services/ │ │ ├── loader.py # 文档读取 │ │ ├── splitter.py # 文本分块 │ │ ├── embedder.py # 向量化 │ │ ├── retriever.py # 检索 │ │ ├── chat.py # 问答组装 │ │ └── llm_client.py # 大模型 API 通用客户端 │ ├── storage/ │ │ └── vector_store.py # FAISS 封装的增删查存 │ └── core/ │ ├── config.py # 全局配置 │ └── models.py # Pydantic 数据模型 ├── data/ │ ├── docs/ # 存放原始文档 │ └── index/ # 索引文件输出目录 ├── scripts/ │ └── build_index.py # 一次性构建索引脚本 ├── requirements.txt ├── run.py └── README.md

跑到最后你会发现,真正决定知识库好不好用的,不是大模型那个环节,而是 services 里 loader、splitter、retriever 这三个文件。它们决定了喂给大模型的上下文干不干净、准不准。

core/config.py 是整个项目的开关中枢。我把嵌入模型名、向量维度、分块大小、重叠长度、TopK、大模型 API 地址和密钥全放在这里,用环境变量做覆盖。你拿到源码后,80% 的定制需求都是改这个文件。

3. 核心实现拆解:从加载文档到检索出答案的完整链路

3.1 文档加载与分块:把整本手册拆成可检索的卡片

第一步是把磁盘上的文档读进来。loader 的做法很朴素:遍历 data/docs 目录,读取所有 .md 和 .txt,每条文档记录下文件名和正文内容。

# app/services/loader.py from pathlib import Path def load_documents(docs_dir: str) -> list[dict]: """读取 docs 目录下的 .md / .txt 文件,返回文档列表""" docs = [] for ext in ("*.md", "*.txt"): for file in Path(docs_dir).glob(ext): content = file.read_text(encoding="utf-8") if len(content.strip()) < 20: # 过滤掉空文件和只有几个字的占位文件 continue docs.append({"source": file.name, "content": content}) return docs

逻辑上注意两点:一是强制 UTF-8 读取,Windows 环境下容易出现编码问题;二是过滤短文档,避免后面把无意义的碎片也向量化。返回的 source 字段会一路带进检索结果,展示答案时能告诉用户「这段内容来自哪个文件」。

分块是知识库质量的第一道关口。我采用的是按空行分段再合并的策略,这样能尽量保住语义的完整性;如果不分段直接按字符硬切,很容易把一句话劈成两半。

# app/services/splitter.py import re def split_by_paragraph(content: str, chunk_size: int = 512, overlap: int = 64) -> list[str]: """按空行分段,再把段落拼成不超过 chunk_size 的块,块之间保留 overlap 字符重叠""" paragraphs = [p.strip() for p in re.split(r"\n\s*\n", content) if p.strip()] chunks = [] current = "" for para in paragraphs: if len(current) + len(para) > chunk_size and current: chunks.append(current) # 取当前块末尾 overlap 个字符作为下一块的上下文衔接 current = current[-overlap:] if overlap < len(current) else current current = (current + "\n" + para).strip() if current: chunks.append(current) return chunks

这里 chunk_size 默认 512 字符,overlap 默认 64。overlap 的作用是让相邻两个块之间有一小段重复内容,检索时不会因为切分边界把完整上下文丢掉。为什么默认 512?太大了 embedding 质量会下降,太小了单个块信息量不够;512 是中文场景下比较折中的值。

3.2 向量化与存储:让语义变成能计算的距离

分块完成之后,每一段文本要转成向量。Embedder 是对 sentence-transformers 库的薄封装,核心参数是 normalize_embeddings=True。

# app/services/embedder.py from sentence_transformers import SentenceTransformer class Embedder: def __init__(self, model_name: str): self.model = SentenceTransformer(model_name) def embed(self, texts: list[str]) -> list[list[float]]: """批量向量化,返回 list[list[float]]""" vectors = self.model.encode(texts, normalize_embeddings=True) return vectors.tolist()

normalize_embeddings=True 这行很关键。它把向量归一化成单位向量,配合 FAISS 的 IndexFlatIP 内积索引,计算结果就等于余弦相似度,分数越接近 1 表示语义越接近。

VectorStore 负责 FAISS 索引的读写。这里有一个容易忽略的设计:FAISS 只存向量,不存原始文本,所以必须单独把 chunks 和 sources 序列化到磁盘。

# app/storage/vector_store.py import pickle import faiss import numpy as np class VectorStore: def __init__(self, dim: int): self.dim = dim self.index = faiss.IndexFlatIP(dim) self.chunks = [] self.sources = [] def add(self, vectors: list[list[float]], chunks: list[str], sources: list[str]): if len(vectors) == 0: return self.index.add(np.array(vectors, dtype="float32")) self.chunks.extend(chunks) self.sources.extend(sources) def search(self, query_vec: list[float], top_k: int = 5) -> list[dict]: scores, indices = self.index.search( np.array([query_vec], dtype="float32"), top_k ) results = [] for j, idx in enumerate(indices[0]): if idx < 0: continue results.append({ "chunk": self.chunks[idx], "source": self.sources[idx], "score": float(scores[0][j]), }) return results def save(self, index_path: str): faiss.write_index(self.index, index_path) meta_path = index_path + ".meta" with open(meta_path, "wb") as f: pickle.dump({"chunks": self.chunks, "sources": self.sources}, f) def load(self, index_path: str): self.index = faiss.read_index(index_path) meta_path = index_path + ".meta" with open(meta_path, "rb") as f: meta = pickle.load(f) self.chunks = meta["chunks"] self.sources = meta["sources"]

看出来了吗?save 的时候 FAISS 索引写进 .index 文件,raw 文本写进 .meta 文件。加载的时候两个都要读回来,缺一个服务就起不来。这是跑这个项目最容易踩的坑之一——很多人以为只要 .index 就够了。

3.3 检索与问答:从问题到答案的完整组装

Retriever 的逻辑很简单:把问题向量化,拿去 FAISS 里搜 top_k,返回带来源的结果。

# app/services/retriever.py class Retriever: def __init__(self, store: VectorStore, embedder: Embedder): self.store = store self.embedder = embedder def search(self, question: str, top_k: int = 5) -> list[dict]: vec = self.embedder.embed([question])[0] hits = self.store.search(vec, top_k) return hits

top_k 默认 5,这个值在小知识库上表现均衡;如果你的文档量大,可以考虑调到 8 甚至 10,但大模型能吃到的上下文有限,太多了反而会引入噪音。

问答组装是决定输出质量的一环。ChatService 把检索结果拼进 Prompt,明确要求模型只依据参考资料回答:

# app/services/chat.py class ChatService: def __init__(self, retriever: Retriever, llm_client): self.retriever = retriever self.llm = llm_client def answer(self, question: str, top_k: int = 5) -> str: hits = self.retriever.search(question, top_k) if not hits: return "知识库里暂时没有相关内容" context = "\n\n".join( f"【来源:{h['source']}】{h['chunk']}" for h in hits ) prompt = ( "你是企业内部知识库助手。请只根据下面参考资料回答," "不要编造。参考资料:\n" + context + "\n\n问题:" + question ) return self.llm.chat(prompt)

Prompt 结构是「角色设定 + 限定范围 + 参考资料 + 问题」四段式。为什么这么写?不限定参考范围,模型就会开始自由发挥;不标来源,你无法追溯答案可信度。

最后是 LLMClient。这个客户端走的是当前大多数大模型厂商通用的 /chat/completions 接口格式,换服务商只需要改三个配置:

# app/services/llm_client.py import requests class LLMClient: def __init__(self, api_base: str, api_key: str, model: str): self.api_base = api_base.rstrip("/") self.api_key = api_key self.model = model def chat(self, prompt: str) -> str: resp = requests.post( f"{self.api_base}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.2, }, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]

temperature 设成 0.2,是刻意压低随机性——知识库问答要的是稳定和可复现,不是创意写作。timeout=60 是给长回答留的余量,如果你用的是本地模型可以适当缩小。

4. 本地跑通全流程:依赖、索引、服务三步到位

4.1 环境准备与依赖安装

先准备 Python 3.10 以上的虚拟环境,然后安装依赖。requirements.txt 内容如下:

fastapi uvicorn[standard] sentence-transformers faiss-cpu numpy requests python-multipart

版本号我故意没锁。原因很简单:AI 生态更新太快,锁死了反而容易装不上。建议你在 x86_64 的 Linux 或 macOS 上先装 faiss-cpu,再装 sentence-transformers,这两个包的顺序有讲究——如果先装 sentence-transformers 再装 faiss-cpu,有时会因为 torch 的 numpy 版本冲突导致报错。

python -m venv venv source venv/bin/activate # Windows 下是 venv\Scripts\activate pip install -r requirements.txt

装完可以用一行命令验证:

python -c "from sentence_transformers import SentenceTransformer; m = SentenceTransformer('shibing624/text2vec-base-chinese'); print(m.encode(['测试']).shape)"

能输出 [1, 768] 就说明基础环境没问题。如果卡在这一步,多半是网络问题导致模型下载不了,下面避坑章节会细说。

4.2 构建知识库索引

先往 data/docs 里放几篇测试文档,然后运行构建脚本:

# scripts/build_index.py from app.core.config import settings from app.services.loader import load_documents from app.services.splitter import split_by_paragraph from app.services.embedder import Embedder from app.storage.vector_store import VectorStore def main(): docs = load_documents(settings.DOCS_DIR) embedder = Embedder(settings.EMBED_MODEL) store = VectorStore(settings.VECTOR_DIM) all_chunks, all_sources, all_vectors = [], [], [] for doc in docs: chunks = split_by_paragraph( doc["content"], settings.CHUNK_SIZE, settings.CHUNK_OVERLAP, ) if not chunks: continue vectors = embedder.embed(chunks) all_chunks.extend(chunks) all_sources.extend([doc["source"]] * len(chunks)) all_vectors.extend(vectors) print(f"{doc['source']}: {len(chunks)} chunks") store.add(all_vectors, all_chunks, all_sources) store.save(settings.INDEX_PATH) print(f"索引完成,共 {len(all_chunks)} 个 chunk") if __name__ == "__main__": main()

执行:

python scripts/build_index.py

整个过程耗时的瓶颈在 embedding 阶段。text2vec 在纯 CPU 上大约每分钟处理几百到一千个文本块,你如果有十几篇长文档,等个几分钟很正常。第一次跑会更慢,因为要下载模型文件到本地缓存。

跑完之后 data/index 目录下会出现两个文件:faiss.index 和 faiss.index.meta。记住它们是成对存在的,拷贝目录时两个都要带上。

4.3 启动服务并验证

run.py 做了两个命令入口:构建索引和启动服务。

# run.py import argparse def main(): parser = argparse.ArgumentParser() parser.add_argument("--build-index", action="store_true", help="构建索引") parser.add_argument("--serve", action="store_true", help="启动服务") args = parser.parse_args() if args.build_index: from scripts.build_index import main as build build() elif args.serve: import uvicorn uvicorn.run("app.main:app", host="0.0.0.0", port=8000) if __name__ == "__main__": main()

启动服务:

python run.py --serve

在 app/api/routes.py 里,模块级初始化了 VectorStore、Embedder、Retriever 和 ChatService,加载已有索引后对外暴露三个接口。

# app/api/routes.py from fastapi import APIRouter, UploadFile from app.core.config import settings from app.services.chat import ChatService from app.services.retriever import Retriever from app.services.embedder import Embedder from app.services.splitter import split_by_paragraph from app.storage.vector_store import VectorStore router = APIRouter() store = VectorStore(settings.VECTOR_DIM) store.load(settings.INDEX_PATH) embedder = Embedder(settings.EMBED_MODEL) retriever = Retriever(store, embedder) chat_service = ChatService(retriever, get_llm_client()) @router.post("/upload") async def upload(file: UploadFile): content = (await file.read()).decode("utf-8") chunks = split_by_paragraph(content, settings.CHUNK_SIZE, settings.CHUNK_OVERLAP) vectors = embedder.embed(chunks) store.add(vectors, chunks, [file.filename] * len(chunks)) store.save(settings.INDEX_PATH) return {"chunks": len(chunks)} @router.post("/ask") async def ask(question: str): return {"answer": chat_service.answer(question)} @router.get("/health") async def health(): return {"status": "ok"}

启动后直接在浏览器打开 http://localhost:8000/docs ,用 Swagger 页面调接口最方便。也可以用 curl 验证:

curl -X POST http://localhost:8000/api/ask \ -H "Content-Type: application/json" \ -d '{"question": "数据库连接超时怎么排查?"}'

如果能返回一段带来源的答案,说明整条链路已经通了。如果 /ask 返回的是「知识库里暂时没有相关内容」,不要急着怪代码,先检查索引里到底有没有数据,再检查问题表述和文档内容是不是对得上。

5. 避坑排查:跑 AI 知识库容易翻车的五个地方

5.1 现象:启动时报 Index dimension mismatch

构建索引一切正常,但启动服务时 FAISS 直接报错,说查询向量维度和索引维度不一致。

原因:config.py 里的 VECTOR_DIM 默认是 768,对应 text2vec-base-chinese 的输出维度。如果你换了嵌入模型,比如换成一个输出 512 维的模型,但没改 VECTOR_DIM,FAISS 创建索引时用的是旧维度,检索时就炸了。

解决:换模型之前先确认新模型的输出维度。最简单的方法是在 Python 里跑一次 embedding,打印 shape。然后在 config 里同步修改。还有一个小细节:改维度之后,旧的 .index 文件也不能用了,需要删掉重建。

5.2 现象:检索结果和问题完全对不上

TOP5 返回的文本块看起来和你问的问题八竿子打不着,或者像随机抽取的。

原因:八成出在分块策略上。如果你图省事直接按字符长度切块,语义单元被切断的概率极高;另外,overlap 要是设成 0,块与块之间没有任何承接,检索时容易漏掉跨边界的信息。

解决:用 split_by_paragraph 这种按段落聚合的策略,并把 overlap 至少设到 50 字符。中文不像英文按空格分词,字符切分天然容易把词和短语砍断,overlap 是补偿这种断裂最直接的手段。

5.3 现象:首次构建索引卡在模型加载

运行 build_index 看起来像死机了,终端停在一行日志上十几分钟没动静。

原因:sentence-transformers 首次使用需要下载模型权重到本地缓存,这一步受网络环境影响很大。卡住不一定是死了,是下载慢。

解决:最稳妥的办法是在网络好的机器上先跑一次 embedding,让模型下载完成,然后把缓存目录整个拷贝到目标机器;或者在传参时直接用绝对路径指向本地模型目录,跳过在线拉取。从那以后我不管在哪个环境跑,都会先检查本地缓存,不会再傻等。

5.4 现象:答案全来自同一篇文档

问题问得很具体,结果返回的 TOP5 全是同一篇长文档的不同切片,其他文档的信息完全没被召回。

原因:长文档被切成多个 chunk 后,彼此语义相似度高,全部挤占了 top_k 的名额。知识库里要是有几篇超长文档,这种现象尤其明显。

解决:retrieve 之后对结果按 source 做去重,限制同一个来源最多出现两个 chunk。还有一种思路是用 MMR(最大边际相关性)对候选结果做多样性排序,兼顾相关性和信息冗余。对大部分场景,简单的按 source 限流已经够用了。

5.5 现象:服务重启后新上传的知识丢失

通过 /upload 接口传了文档,当时问答正常;重启服务后再问,新文档相关的问题全部答不上来。

原因:/upload 只往内存里的 FAISS 索引加了向量,没有触发持久化。虽然路由里调了 store.save,但如果你改了代码或者没走这个分支,重启后索引就会回到构建时的状态。

解决:确认上传逻辑里必须调用 store.save(我在 routes.py 里保留了这行)。更稳妥的做法是每次上传后检查一下 data/index 下 .meta 文件的修改时间,确认真的写进去了。另外,正式使用时建议把 upload 接口做成写一个临时目录,再触发重建索引,避免并发上传时内存索引和磁盘文件不同步。

6. 二次开发实战:三个方向把知识库接进自己的业务

6.1 替换成大模型 API

源码默认的 LLMClient 直接可用,你只需要在 config.py 里填上 api_base、api_key、model 三项。如果你用的是本地开源模型,也可以部署一个兼容 /chat/completions 的服务,然后把地址指向 localhost 端口,业务代码一行都不用动。

6.2 给知识库加业务权限过滤

在 VectorStore 里给每个 chunk 增加 metadata 字段,比如部门、权限级别。检索时先按问题携带的上下文过滤 metadata,再进 FAISS 计算相似度。这块我有一个习惯:metadata 过滤必须发生在向量检索之前,否则结果集会掺杂大量无权限内容。

6.3 按 Markdown 标题自适应分块

默认分块策略按空行聚合,已经能覆盖大多数场景。如果你的文档结构很强,比如操作手册、故障手册,可以改 splitter 按#标题层级切分,保证一个块就是一段完整的主题。实现思路是先用正则匹配标题位置,再决定段落归属哪个块。

我刚做这个项目第二版的时候,整个服务上线第二天就被吐槽检索不准,查到最后就是分块太粗暴、metadata 也没过滤干净。从那以后我每次改知识库,都会先准备一组真实问题和对应文档,离线跑一遍 TOP5 召回率,确认准了再放出去用。希望帮到你。

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

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进