知识库工程化:从 PDF 上传到智能检索的完整流程

系列第 5 篇 / 共 8 篇

第二周，有用户传了一份《项目A部署手册.pdf》，然后问助手："项目A怎么部署？"

助手说："根据我的训练数据……"

对方一脸懵："手册不是已经传了吗？"

我去查日志，发现根本没解析——后端只对 .txt 和 .md 做了处理，PDF 文件被存到了磁盘，但向量库里一个字都没进。

这是知识库最容易踩的坑："上传成功"≠"检索得到"。这篇讲清楚我们怎么把知识库从"能上传文件"做到"可管理的知识资产"。

一、解析器：把字节流变成 Document

文件：backend/app/services/knowledge/parsers.py

def parse_document(content: bytes, filename: str) -> list[Document]: suffix = Path(filename).suffix.lower() if suffix in (”.txt”, ”.md”, ”.markdown”): return parse_text(content, filename) if suffix == ”.pdf”: return parse_pdf(content, filename) if suffix == ”.docx”: return parse_docx(content, filename) raise ValueError(f”Unsupported file type: {suffix}”)

三种格式各一个函数：

def parse_pdf(content: bytes, filename: str) -> list[Document]: from pypdf import PdfReader reader = PdfReader(BinaryIO(content)) pages: list[Document] = [] for idx, page in enumerate(reader.pages, 1): text = page.extract_text() or ”” if text.strip(): pages.append(Document( page_content=text, metadata={”source”: filename, ”page”: idx}, )) return pages

两个细节：

pypdf 和 python-docx 都延迟 import（写在函数里），因为这两个库加起来 30MB，启动时不必加载。FastAPI 启动时间从 4s 降到 1.5s。

二、扫描版 PDF 怎么办

最坑的一类 PDF：扫描件，全是图片，extract_text 返回空。

我们的方案分两层：

文件：backend/app/services/knowledge/ocr.py

def ocr_available() -> bool: ”””渲染器和视觉模型都就位才返回 True。””” settings = get_settings() if not settings.ocr_enabled: return False if not (settings.ocr_api_key and settings.ocr_model): return False try: import fitz# noqa: F401 except ImportError: return False return True

设计要点：

pymupdf 惰性 import：不装这个包，其他功能完全不受影响；装了且配了 ocr_api_key / ocr_model（任何 OpenAI 兼容视觉模型），扫描件就自动走"渲染成 PNG → 视觉模型逐页转写"的流程。
OCR 失败兜底：provider 超时或报错，照样标红 unparseable，不会因为 OCR 挂了把整个上传搞死。
默认关闭：ocr_enabled=False 是出厂状态，文档里写清楚怎么开。

为什么不直接用 PaddleOCR？因为它要拉几百 MB 的模型文件，对"偶尔识别几份扫描件"的场景太重。视觉模型按次计费、零本地依赖，够用。真到每天几百份扫描件的时候，再换 PaddleOCR 也不迟——接口已经留好了。

三、分块：chunk_size 不是越大越好

文件：backend/app/services/knowledge/processor.py

def split_documents( documents: list[Document], chunk_size: int = 500, chunk_overlap: int = 50, separators: list[str] | None = None,) -> list[Document]: separators = separators or [”\n\n”, ”\n”, ”。”, ”.”, ” ”, ””] splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=separators, ) chunks = splitter.split_documents(documents) for idx, chunk in enumerate(chunks): chunk.metadata.setdefault(”source”, ”unknown”) chunk.metadata[”chunk_index”] = idx return chunks

我们用 RecursiveCharacterTextSplitter，按 \n\n → \n → 。 → . → 的顺序递归切，尽量保住段落完整性。

chunk_size 我们调到 500，不是越大越好：

太小（100）：上下文不完整，答案经常缺主语
太大（2000）： embedding 太稀释，相关段落排不到前面
500 是试出来的甜点，覆盖 90% 运维文档的段落长度

chunk_overlap=50 让相邻块有重叠，避免一个完整步骤被切两半。

注意 separators 里有。和 .——中文文档必须加中文句号，默认 splitter 只认英文标点，中文文档会被切得稀烂。

四、chunk_size 做成可配置

不同场景需求不一样：运维文档段落短，chunk_size=400 更好；技术设计文档段落长，chunk_size=800 更合适。

我们做了一张 KnowledgeConfig 表存全局配置，管理员后台可改：

class KnowledgeConfig(Base): __tablename__ = ”knowledge_config” id: Mapped[str] = mapped_column(String(36), primary_key=True, default=generate_uuid) chunk_size: Mapped[int] = mapped_column(Integer, default=500) chunk_overlap: Mapped[int] = mapped_column(Integer, default=50) separator: Mapped[str | None] = mapped_column(Text, nullable=True) updated_by_id: Mapped[str | None] = mapped_column(String(36), ForeignKey(”users.id”), nullable=True)

处理器在上传/重建时从这张表读配置。这个改动让团队自己调优了 3 次分块参数，没找过我——这就是工程化的价值。

五、FAISS 不支持删除：重建是唯一出路

FAISS 是个内存索引，不支持高效删除单个向量（只有 IndexFlatL2 这种最简单的支持 remove_ids，但慢）。文档更新场景很常见：手册改了，旧向量得删掉。

文件：backend/app/services/knowledge/vector_store.py

async def delete_source(source: str) -> int: ”””FAISS does not support true deletion; rebuild the index without the source.””” store = await get_vector_store() loop = asyncio.get_running_loop() def _rebuild(): embeddings = get_embedding_model() all_docs = list(store.docstore._dict.values()) kept_docs = [d for d in all_docs if d.metadata.get(”source”) != source] if not kept_docs: kept_docs = [Document(page_content=”knowledge base placeholder”, metadata={”source”: ”__init__”})] new_store = FAISS.from_documents(kept_docs, embeddings) new_store.save_local(str(_get_index_dir())) return new_store, len(all_docs) - len(kept_docs) new_store, deleted = await loop.run_in_executor(None, _rebuild) global _vector_store _vector_store = new_store return deleted

策略是重建整个索引：把所有文档读出来，过滤掉要删的 source，重新构建 FAISS，原子替换。代价是一次全量重建（具体耗时取决于文档总量和你的 embedding 服务速度），对中小规模知识库完全可接受——如果你的文档量到了几十万 chunk 级别，那应该一开始就用支持真删除的向量库（Milvus/Qdrant/pgvector），别用 FAISS。

两个细节：

六、单文档 reindex：用户自己救自己

文档经常更新，每次都让管理员重建全量索引不现实。我们做了一个接口：

POST /api/v1/knowledge/documents/{doc_id}/reindex

逻辑：从磁盘读最新文件 → 用当前 KnowledgeConfig 分块 → 调 reindex_source()。

文件：backend/app/services/knowledge/vector_store.py

async def reindex_source(source: str, documents: list[Document]) -> int: await delete_source(source) if documents: await add_documents(documents) return len(documents)

配套前端在文档列表加了"重建索引"按钮。内容维护者改完导出的 MD 重新上传，自己点一下重建，10 秒搞定，不用找我。

这是知识库工程化最值钱的一个功能——把运维能力下放给内容维护者。

七、分类和标签：让"找文档"比"问模型"还快

模型再聪明，用户也经常想直接看原文。我们给 KnowledgeDocument 加了 category 和 tags：

class KnowledgeDocument(Base):# ... category: Mapped[str | None] = mapped_column(String(100), nullable=True) tags: Mapped[str | None] = mapped_column(Text, nullable=True)# JSON 数组

category：单选，比如"运维手册"、"故障案例"、"SOP"
tags：多选，JSON 数组，比如 ["k8s", "网络", "故障"]

后台文档列表支持按 category/tag 筛选，前端有个树形导航。很多用户后来发现：直接翻文档比问模型更快。这是好事——知识库本来就不只是 RAG 的燃料，它本身就是资产。

八、踩过的坑

做知识库踩了不少坑。第一个是 embedding 模型选错，成本翻倍——第一版用的是 text-embedding-ada-002，后来切到国内的 embedding 服务（阿里百炼 text-embedding-v3），价格差接近一个数量级，中文场景效果还更好。选 embedding 前一定要算全量成本：文档总数 × 平均 chunk 数 × chunk token 数 × 单价。

第二个坑是上传时没存原始文件——第一版只存了向量，没存原文件。后来要 reindex，发现没源头，只能让用户重新上传。原文件一定要存磁盘，路径记在 KnowledgeDocument.source 里。

第三个坑是搜索分数不可比——FAISS 返回的 score 是 L2 距离，不同 query 之间不可比。第一版用 score < 0.5 过滤，结果某些 query 命中率为 0。改成"取 top K 不过滤"才正常。后来又加了 rerank，用向量距离 + 关键词重叠做混合排序，"关键词完全对得上但向量距离一般"的结果终于能排上来。

第四个坑是向量库并发写——两个用户同时上传文档，FAISS 同时 save_local 会写坏文件。我们用了 asyncio.Lock 全局串行化写入，牺牲一点并发换稳定。

九、这套知识库的验收清单

上线前建议跑一遍这些检查：四种格式上传（.txt/.md/.pdf/.docx）都能正确解析并检索到；上传扫描版 PDF 后，文档列表标红"无法解析"；改完原文件点重建索引，新内容能被检索到、旧内容不再命中；后台能按 category 和 tag 过滤文档列表；改 KnowledgeConfig 后新上传的文档用新参数分块；多人同时上传，FAISS 索引文件不损坏；rerank=true 时返回 rerank_score，关键词精确匹配的结果排上来；贴一个 Wiki 链接能入库，重复贴返回 409；reindex 后能看到版本快照，回滚后旧内容重新命中。

这些检查在本仓库都有对应的实现和测试，具体怎么用后面章节展开。

十、上线后补的三块板

前九节是第一版就有的。上线跑了一段时间后，又补了三块——都是用户真实提出来的。

10.1 rerank：向量距离不是唯一裁判

向量检索有个经典失败模式：关键词完全对得上，但 embedding 距离一般，被排到很靠后。比如用户问"项目A部署"，有一段文档标题就叫《项目A部署手册》，但因为它只有标题没有正文，向量分数平平。

文件：backend/app/services/knowledge/reranker.py

def rerank_results(query: str, results: list[dict], top_k: int | None = None) -> list[dict]: ”””向量距离 + 关键词重叠的混合排序。score 是 FAISS 距离（越小越好）。””” ranked = [] for item in results: distance = float(item.get(”score”, 0.0)) vector_score = 1.0 / (1.0 + distance) keyword = _keyword_score(query, item.get(”content”, ””))# 查询词命中率 combined = vector_score * 0.6 + keyword * 0.4 ranked.append({**item, ”rerank_score”: combined}) ranked.sort(key=lambda x: x[”rerank_score”], reverse=True) return ranked[:top_k] if top_k else ranked

关键词打分用的是正则 \W+ 分词（中英文都能切），统计查询词在文档内容中的命中率。

接入两个地方：

/knowledge/search 接口：rerank=true 时先取 2 倍候选，混合排序后截断。
search_knowledge_base 工具：LLM 拿到的检索结果也是排过序的，并重排 index——不然 [citation:N] 会和模型看到的顺序错位。

为什么不用 cross-encoder？多一个几百 MB 的模型依赖，对"几十个候选里重排"的场景收益有限。向量+关键词的混合排序 30 行代码、零新依赖，先把 80 分拿到。

10.2 URL 直接抓取：少一步"导出再上传"

知识库运营里最高频的抱怨是："Wiki 上更新了，我还得导出 MD 再传一遍？" 于是加了 URL 导入：

POST /api/v1/knowledge/documents/url{”url”: ”https://wiki.example.com/deploy-a”, ”category”: ”运维手册”}

流程：抓取 → HTML 转纯文本 → 分块 → 嵌入。几个设计点：

文件：backend/app/services/knowledge/crawler.py（抓取）+ backend/app/api/v1/knowledge.py（端点）。

10.3 版本快照：重建也有后悔药

reindex 是把双刃剑：新内容生效了，但如果新文件本身有问题（比如导出时少了一半），旧向量已经被覆盖，回不去了。

文件：backend/app/services/knowledge/versions.py

async def snapshot_document_version(db, doc, *, created_by_id=None): ”””把当前原始文件复制到 versions/ 目录，落一行 KnowledgeDocumentVersion。””” src = upload_dir / doc.source next_version = (max_existing_version or 0) + 1 storage_path = versions_dir / f”{doc.id}_v{next_version}” shutil.copy2(src, storage_path)# ... insert KnowledgeDocumentVersion row

规则：

每次 reindex 前先快照当前状态，版本号递增。
POST /documents/{id}/versions/{vid}/rollback 回滚：回滚前也先给当前状态做快照——所以回滚本身也能再撤销，没有不可逆操作。
前端文档列表加"版本"按钮，弹窗列版本，点一下确认回滚。

这三块板加上去之后，知识库从"能上传能检索"变成了"敢让非管理员的内容维护者自己玩"——改错了能回滚，Wiki 链接直接贴，检索不准有 rerank 兜底。

十一、聊点分块与检索策略

知识库的核心是"存得好、搜得到"，简单聊两句分块和检索。

分块就是把长文本拆成小块，方便向量化和检索。为什么要分块？Embedding 模型有最大 token 限制，文本太长会稀释关键信息，检索效果差。常见的分块策略有固定长度、按段落、按章节、语义分块，各有优缺点。我们用的是 RecursiveCharacterTextSplitter，按段落、换行、中文句号、英文句号的顺序递归切分，尽量保住段落完整性。

chunk_size 我们调到 500，太小上下文会断裂，太大信息会稀释，500 是试出来的甜点。chunk_overlap=50 让相邻块有重叠，避免一个完整步骤被切两半。注意中文文档必须加中文句号作为分隔符，默认 splitter 只认英文标点，中文文档会被切得稀烂。

检索方面，向量检索语义理解好，但不支持关键词精确匹配；关键词检索精确匹配、速度快，但不理解语义。我们用的是混合检索：先向量检索取 2 倍候选，再用向量距离 + 关键词重叠做混合排序，最后截断到 top_k。

Rerank 是在向量检索之后做精排。向量检索只考虑向量相似度，不考虑关键词匹配，很多结果相关性不高。我们用的是简单的混合策略：60% 向量分数 + 40% 关键词分数，30 行代码、零新依赖，效果还不错。

为了提升性能，我们还加了缓存机制，用 Redis 缓存检索结果，TTL 设为 1 小时。同一个查询短时间内多次请求，直接返回缓存，不用重复计算。

写在最后

知识库工程化核心是三件事：解析要做扎实，扫描件、空段落、中文标点，每个都能让你翻车；分块要可调，一份 chunk_size 走天下肯定不合适；管理要自助，reindex、分类、标签、版本回滚都要让内容维护者自己玩。

下一篇讲多租户与数据安全——上线前必须想清楚的三件事：SSO 怎么接、租户数据怎么隔离、PII 怎么脱敏。