企业 AI 知识库实战：文档解析、切分、Qdrant 混合检索与重排

本文从一份真实业务文档开始，构建一个可以返回答案、证据和页码的企业知识库。实现使用 Python、Qdrant 和任意 OpenAI 兼容模型接口，重点不是“向量库跑起来”，而是文档版本、权限过滤、混合检索、引用校验、增量更新与可复测评估。

1. 定义目标、范围与目录

示例系统只处理已批准的 PDF/Markdown 文档。每次回答必须返回 evidence_id、document_id、version、page；资料不足就拒答。权限必须在检索阶段过滤，不能先取出所有文档再让模型“自觉不看”。

mkdir -p rag-kb/{app,data/raw,data/processed,eval,qdrant_storage}
cd rag-kb
python3 -m venv .venv
. .venv/bin/activate
pip install qdrant-client sentence-transformers pydantic pypdf fastapi uvicorn
pip freeze > requirements-lock.txt

把原始文档放入 data/raw，但敏感文件不要提交 Git。每次导入记录解析器、embedding 模型和依赖版本。

2. 建立文档与权限数据契约

文档元数据至少包含稳定 ID、版本、哈希、租户、ACL、有效期和来源。正文块还要保存页码、章节路径和块序号。权限变更、文档删除和版本切换必须同步影响索引与缓存。

{"document_id":"policy-leave","version":3,"title":"休假制度","tenant_id":"acme","acl":["employee"],"effective_from":"2026-01-01","status":"active","source_uri":"s3://approved/policy-leave-v3.pdf","content_hash":"sha256:..."}

3. 解析文档并做质量门禁

PDF 可能是文本层、扫描件或混合文档。解析后不要直接索引：检查页数、总字符、乱码比例、空页、重复页眉页脚和表格顺序。扫描件需要 OCR，并抽查数字、日期和 0/O、1/l 等易错字符。

from pathlib import Path
from pypdf import PdfReader
import json

reader = PdfReader("data/raw/policy-leave-v3.pdf")
rows = []
for page_no, page in enumerate(reader.pages, start=1):
    text = (page.extract_text() or "").strip()
    if len(text) < 20:
        raise ValueError(f"page {page_no} has too little text; OCR/manual review required")
    rows.append({"document_id":"policy-leave","version":3,"page":page_no,"text":text})
Path("data/processed/pages.jsonl").write_text(
    "\n".join(json.dumps(x, ensure_ascii=False) for x in rows), encoding="utf-8")

预期：每页一行且页码连续。若抽取顺序错误，不要靠后续 embedding 修复，应替换解析器或人工校正。

4. 按结构切分并生成稳定 ID

优先按标题、段落、列表和表格边界切分，再用 token 上限兜底。块太大会引入无关内容；块太小会把条件与结论拆开。保留章节路径，并允许命中后扩展相邻块。

DOCUMENT = {
    "document_id": "policy-leave",
    "version": 3,
    "tenant_id": "acme",
    "acl": ["employee"],
}

def make_chunks(pages, max_chars=1200):
    chunks = []
    for page in pages:
        paragraphs = [p.strip() for p in page["text"].split("\n") if p.strip()]
        buffer = []
        for paragraph in paragraphs:
            if buffer and sum(len(x) for x in buffer) + len(paragraph) > max_chars:
                text = "\n".join(buffer)
                chunks.append({
                    **DOCUMENT,
                    "id": stable_id(DOCUMENT["document_id"], DOCUMENT["version"], len(chunks), text),
                    "page": page["page"],
                    "section": page.get("section", "未识别章节"),
                    "text": text,
                })
                buffer = []
            buffer.append(paragraph)
        if buffer:
            text = "\n".join(buffer)
            chunks.append({
                **DOCUMENT,
                "id": stable_id(DOCUMENT["document_id"], DOCUMENT["version"], len(chunks), text),
                "page": page["page"],
                "section": page.get("section", "未识别章节"),
                "text": text,
            })
    return chunks

chunks = make_chunks(rows)
assert chunks and all(c["id"] and c["text"] for c in chunks)

示例用字符数便于展示；正式环境用与生成模型匹配的 tokenizer 计算预算。统计块长度、空块和重复率，并人工抽查边界。

5. 启动 Qdrant 并创建集合

本地测试可以直接使用 Docker；生产环境必须增加认证、TLS、网络隔离、持久卷和备份。不要把 6333/6334 暴露到公网。

docker pull qdrant/qdrant
docker run -d --name qdrant \
  -p 127.0.0.1:6333:6333 -p 127.0.0.1:6334:6334 \
  -v "$PWD/qdrant_storage:/qdrant/storage" \
  qdrant/qdrant
curl -fsS http://127.0.0.1:6333/healthz

from qdrant_client import QdrantClient, models
client = QdrantClient(url="http://127.0.0.1:6333")
client.create_collection(
    collection_name="kb_v1",
    vectors_config=models.VectorParams(size=1024, distance=models.Distance.COSINE),
)

向量维度必须等于 embedding 输出。更换 embedding 模型时新建集合并重建，不要混用不同语义空间。

6. 生成向量并批量写入

from sentence_transformers import SentenceTransformer
from qdrant_client import models

encoder = SentenceTransformer("BAAI/bge-m3")
vectors = encoder.encode([c["text"] for c in chunks], normalize_embeddings=True)
points = []
for c, vector in zip(chunks, vectors):
    points.append(models.PointStruct(
        id=c["id"], vector=vector.tolist(), payload={
            "document_id":c["document_id"], "version":c["version"],
            "tenant_id":c["tenant_id"], "acl":c["acl"],
            "page":c["page"], "section":c["section"],
            "text":c["text"], "status":"active"}))
client.upsert(collection_name="kb_v1", points=points, wait=True)

写入后按 document_id 统计点数，并随机读取若干 point 与 processed 文件对照。embedding 失败的批次不能标记为成功。

7. 检索时强制租户与 ACL 过滤

from qdrant_client import models

def search(query_vector, tenant_id, roles):
    flt = models.Filter(must=[
        models.FieldCondition(key="tenant_id", match=models.MatchValue(value=tenant_id)),
        models.FieldCondition(key="acl", match=models.MatchAny(any=roles)),
        models.FieldCondition(key="status", match=models.MatchValue(value="active")),
    ])
    return client.query_points(
        collection_name="kb_v1", query=query_vector, query_filter=flt,
        limit=20, with_payload=True).points

tenant_id 和 roles 来自已认证会话，不接受模型或请求体自报。为 tenant_id、acl、status 创建合适的 payload index，并测试跨租户同名文档。

8. 加入混合检索与重排

向量检索擅长语义，稀疏/关键词检索擅长型号、错误码和专有名词。两路都使用同一权限过滤；各取 top 20，用 Qdrant prefetch 与 RRF 融合，再对前若干条重排。

result = client.query_points(
    collection_name="kb_hybrid_v1",
    prefetch=[
        models.Prefetch(query=sparse_vector, using="sparse", limit=20, filter=flt),
        models.Prefetch(query=dense_vector, using="dense", limit=20, filter=flt),
    ],
    query=models.RrfQuery(rrf=models.Rrf()),
    limit=10,
    with_payload=True,
)

具体 API 以锁定的 qdrant-client 版本为准。融合后去除近重复块，再用 reranker 选择 top 5–8。重排超时应降级到已评测的融合结果，不能无限等待。

9. 组装证据并校验引用

应用为本次证据分配 E1、E2，并保存 ID 到 payload 的映射。模型只能引用已有 ID。生成后解析引用，检查每个 ID 是否存在；数字、日期和关键实体可做规则校验。

import re

def validate_citations(answer, evidence_map):
    cited = set(re.findall(r"\[([A-Z]\d+)\]", answer))
    if not cited:
        return False, "missing citation"
    unknown = cited - set(evidence_map)
    if unknown:
        return False, f"unknown citations: {sorted(unknown)}"
    return True, "ok"

证据不足时返回“当前资料无法确认”，并说明缺少什么。不要让模型自行生成文档链接。

10. 增量更新、删除和回滚

计算新文档 content_hash；未变化则跳过。
以 version+1 写入 staging 点。
运行解析、点数、召回和 ACL 测试。
把新版本切为 active，并使缓存键使用新索引版本。
保留旧版本一段回滚窗口，验证稳定后清理。

删除时同时处理向量、缓存、派生摘要和原始文档映射。Qdrant 快照只是向量库的一部分，恢复还需要文档元数据、embedding 版本和解析配置。

11. 端到端验收与排错

现象	定位
正确证据未进入 top-k	解析、切分、query、embedding、权限
证据进入但排名靠后	混合权重、去重、重排
证据正确但答案错误	上下文结构、提示、生成模型
旧版本仍出现	status 过滤、缓存键、版本切换
跨租户泄漏	两路检索与缓存是否都带 ACL

检索层测 Recall@k、MRR、nDCG 和权限零泄漏；生成层测正确性、忠实度、引用精度和拒答。每次切分、embedding、融合或重排变化都重跑固定集合。

12. 总结

企业 RAG 的核心是可追踪与可控制：稳定 ID、版本、权限、混合召回、引用和增量更新必须在同一条链路中。只有能定位错误发生在解析、召回、排序还是生成，系统才真正具备维护价值。