19 · 实战手敲五:企业级文档问答系统

手敲系列第五篇,也是第一个"生产级"项目:一堆真实 PDF 进去,一个带页码引用、拒绝幻觉、有缓存有限流有监控有评测有 CI 的问答服务出来。

全部 16 个文件、约 2000 行,每一行都在下面给出。预计 5~7 天。

核心新技能:PDF 解析与表格线性化、混合检索(BM25 + 向量 + RRF + 精排)、用数字证明你的 RAG 变好了、Postgres/Redis/Qdrant 三件套、质量闸门卡住 CI。

0. 为什么这一篇和前面 14 篇不一样

前面的手敲项目,知识库是写在 Python 字符串里的,答案对不对靠你自己看,挂了没人知道,改了参数也不知道变好还是变坏。那是demo。

这一篇要做的是能交付给公司用的东西,差别体现在四个地方:

这一篇是简历上最能打的一个项目。原因很简单:"我做过 RAG"现在人人都会说,但"我有标注评测集,单路向量 Hit@1 是 0.83,加 BM25 融合到 0.83、Recall@5 从 0.96 到 1.00,再加 rerank 把 Hit@1 提到 0.96,这套指标接在 CI 上,掉了就不许合并"——这句话能说出来的人,一百个里没有几个。

1. 成品长什么样

启动后往接口丢一个问题:

curl -X POST localhost:8000/api/ask \
  -H 'X-API-Key: your-key' -H 'Content-Type: application/json' \
  -d '{"question":"正式员工满5年有几天年假"}'

拿到的东西:

{
  "answer": "满 5 年不满 10 年的正式员工有 15 天年假 [1]。年假当年有效,未使用的最多可结转 5 天至次年 3 月 31 日 [2]。",
  "verdict": "answered",
  "citations": [
    {"n": 1, "doc": "员工手册", "page": 1, "heading": "第二章假期管理", "kind": "table"},
    {"n": 2, "doc": "员工手册", "page": 1, "heading": "第二章假期管理", "kind": "text"}
  ],
  "latency_ms": 1843,
  "cached": false,
  "trace_id": "7807f4bc25674fdb"
}

三个细节请注意,它们是这个项目的灵魂:

1. 每句话都有 [编号],编号对应到"哪份文档第几页哪一节"。企业场景里没有引用的答案等于没有答案——HR 不会因为一个 AI 说"15 天"就给你批 15 天假。
2. verdict 有三种值:answered / not_found / error。问知识库里没有的东西(比如"年终奖怎么发"),它会老老实实回答"文档中未提及",而不是编一个。
3. trace_id 能查到这次问答的耗时分解:缓存查了几毫秒、检索几毫秒、LLM 几毫秒。线上说"变慢了"的时候,你要能一秒回答"慢在哪一层"。

2. 架构:一条请求穿过几层

                       ┌──── 离线索引(独立任务, 不在 Web 进程里)────┐
PDF 目录 ──▶ 解析(版式/表格) ──▶ 分块(句子对齐+上下文前缀) ──▶ 向量化 ──▶ Qdrant
                                     │                                    ▲
                                     └──▶ documents 表(sha256 判断要不要重新索引)
                                                                          │
─────────────────────────────────────────────────────────────────────────┼──
                                                                          │
HTTP 请求                                                                 │
   │                                                                      │
   ▼                                                                      │
鉴权(X-API-Key) ──401──▶ 拒绝                                             │
   ▼                                                                      │
限流(Redis 令牌桶) ──429──▶ 拒绝                                          │
   ▼                                                                      │
缓存(Redis, key 带知识库版本) ──命中──▶ 直接返回, 一分钱不花               │
   ▼ 未命中                                                               │
检索: BM25 ──┐                                                            │
             ├──▶ RRF 融合 ──▶ cross-encoder 精排 ──▶ 阈值过滤 ───────────┘
   向量 ─────┘                        │
                                      ▼ 全部低于阈值
                                  返回"文档中未提及"(不调 LLM)
                                      │ 有材料
                                      ▼
                            LLM(带编号材料 + 强制引用 + temperature=0)
                                      ▼
                          落库(qa_logs: 问题/答案/引用/token/成本) + 写缓存
                                      ▼
                                  返回给用户
                          (整条链路每一层都记了一个 span)

16 个文件的分工:

顺序很重要:从 make_corpus.py 一路往下敲,每敲完一个文件都能单独 python xxx.py 跑出东西。这是本篇 16 个文件都写了 if __name__ == "__main__" 的原因——每个文件都是一个可以独立验证的小实验。

3. 建项目与起中间件(30 分钟)

这一篇第一次用到三个中间件,全部用 Docker 起,不要在本机装:

mkdir docqa_lab && cd docqa_lab
uv init --python 3.12
uv add "fastapi>=0.115" "uvicorn[standard]>=0.32" "pymupdf>=1.24" jieba \
       "qdrant-client>=1.12" "sentence-transformers>=3.0" \
       "sqlalchemy>=2.0" "psycopg[binary]>=3.2" "redis>=5.0" \
       "openai>=1.40" python-dotenv pytest
mkdir -p docqa/data tests

三个容器:

docker run -d --name docqa-qdrant -p 6333:6333 qdrant/qdrant:v1.16.1
docker run -d --name docqa-pg -p 5432:5432 \
  -e POSTGRES_PASSWORD=docqa -e POSTGRES_DB=docqa postgres:17-alpine
docker run -d --name docqa-redis -p 6379:6379 redis:7-alpine

✅ Checkpoint 1:docker ps 看到三个都是 Up:

NAMES          IMAGE                   STATUS
docqa-qdrant   qdrant/qdrant:v1.16.1   Up 2 hours
docqa-redis    redis:7-alpine          Up 2 hours
docqa-pg       postgres:17-alpine      Up 2 hours

再把 pyproject.toml 的两段配置补上。第一段是 lint 规则——很多人跳过这一步,后果是某天 ruff 升级、你一行代码没改,CI 却突然变红:

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "SIM"]
# E501(行太长)对中文注释不适用: 中文字符按 1 计数但显示占 2 列,
# 一行写满就"超长"。忽略它, 靠 review 控制可读性
ignore = ["E501"]

# 没有这一段, `uv run pytest tests` 会在 import docqa 时报 ModuleNotFoundError:
# 项目没被装进 venv(没写 build-system), pytest 也不会把 rootdir 加进 sys.path。
# 每次敲 PYTHONPATH=. 能绕过去, 但迟早有人忘记 —— 钉在配置里
[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]

[tool.pytest.ini_options] 那段是我写到第 19 节才补上的:前面所有 python docqa/xxx.py 我都手动加了 PYTHONPATH=.,到了跑 pytest 时忘了加,直接 ModuleNotFoundError: No module named 'docqa'。能钉在配置里的东西,不要靠每次记得敲命令——你会忘,CI 更会。

(顺带说明:本章后面的命令行仍然写着 PYTHONPATH=.,因为那些是直接跑脚本、不经过 pytest。想彻底免掉,给项目加个 [build-system] 并 uv pip install -e .。)

4. make_corpus.py —— 造一份"难"的语料(1 小时)

真实项目里语料是业务方给的。这里生成是为了让教程可复现,你换成自己公司的 PDF,后面所有代码一行都不用改。

但我不是随便造五份文档。请看这五份文档的设计:

• 员工手册.pdf —— 主文档,有三章正文 + 一张"年假天数对照表"
• 产品技术白皮书.pdf —— 有 max_concurrent_queries、P99、429 这类专有名词和数字
• 售后服务政策.pdf —— 有 SLA 数字和赔偿比例表
• 实习生管理办法.pdf —— 干扰文档 1:章节标题和员工手册一字不差(第一章考勤制度、第二章假期管理、第三章报销标准),内容全部冲突(9:30 上班 vs 弹性工作、不享受年假 vs 15 天年假)
• 售后服务政策-2024旧版.pdf —— 干扰文档 2:标题一样,SLA 数字全不一样(P1 响应 2 小时 vs 30 分钟)

这两份干扰文档才是本篇的难点所在。RAG 的难点从来不是"检索到相关内容",而是"在五份高度相似的文档里检索到对的那一份"。真实企业知识库里到处都是这种东西:新旧版本、不同人群适用、不同地区版本。没有干扰文档的评测集,分数会虚高得毫无意义。

"""生成三份带版式的中文 PDF 作为测试语料(含标题层级、正文、表格)。

真实项目里语料是业务方给的; 这里生成是为了让教程可复现——
你换成自己的年报/产品手册 PDF, 后面所有代码一行都不用改。
"""

from pathlib import Path

import pymupdf

DATA = Path(__file__).parent / "data"

DOCS = [
    {
        "file": "员工手册.pdf",
        "title": "XX 科技员工手册(2026 版)",
        "sections": [
            (
                "第一章 考勤制度",
                [
                    "公司实行弹性工作制, 核心工作时间为每个工作日的 10:00 至 16:00。",
                    "员工每月可申请不超过 2 次的远程办公, 需提前 1 个工作日在 OA 系统提交申请并获得直属主管批准。",
                    "迟到超过 30 分钟且未提前报备的, 计为迟到一次; 每月累计迟到 3 次以上, 当月绩效评级下调一档。",
                ],
            ),
            (
                "第二章 假期管理",
                [
                    "年假天数按入职年限计算, 具体见下表。年假当年有效, 未使用的年假最多可结转 5 天至次年 3 月 31 日。",
                    "病假需提供二级甲等以上医院的诊断证明; 单次病假超过 3 天的, 需额外提供住院或复诊记录。",
                    "婚假为 10 个自然日, 需在领证后 12 个月内使用完毕。",
                ],
            ),
            (
                "第三章 报销标准",
                [
                    "国内出差住宿标准: 一线城市每晚不超过 600 元, 其他城市不超过 400 元。",
                    "报销单据须在费用发生后 30 天内提交, 超期需附部门负责人书面说明。",
                    "打车费用单笔超过 200 元的, 须在报销说明中写明事由。",
                ],
            ),
        ],
        "table": {
            "caption": "表 2-1 年假天数对照表",
            "header": ["入职年限", "年假天数", "是否可结转"],
            "rows": [
                ["满 1 年不满 3 年", "5 天", "否"],
                ["满 3 年不满 5 年", "10 天", "是"],
                ["满 5 年不满 10 年", "15 天", "是"],
                ["满 10 年以上", "20 天", "是"],
            ],
        },
    },
    # ... 产品技术白皮书 / 售后服务政策 同结构, 内容见下方说明
]

另外四份文档同结构,把 sections 和 table 换成对应内容即可。关键数字我列出来,你照着填(故意让它们冲突):

白皮书里必须放这几个东西(它们专门用来验证 BM25 的价值):max_concurrent_queries 默认值为 128、P99 查询延迟低于 200 毫秒、超时返回 429 状态码、RTO 在 1TB 数据量下约为 25 分钟。

排版函数是本节唯一有技术含量的部分:

# macOS 自带的中文字体; Linux 换成 /usr/share/fonts/... 下的任意 CJK 字体
FONT_CANDIDATES = [
    "/System/Library/Fonts/Supplemental/Songti.ttc",
    "/System/Library/Fonts/Hiragino Sans GB.ttc",
    "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
    "/usr/share/fonts/truetype/arphic/uming.ttc",
]


def _font_path() -> str:
    for p in FONT_CANDIDATES:
        if Path(p).exists():
            return p
    raise SystemExit("找不到中文字体, 请把你系统里的 CJK 字体路径加进 FONT_CANDIDATES")


def build(doc_spec: dict, font: str) -> None:
    pdf = pymupdf.open()
    page = pdf.new_page()
    page.insert_font(fontname="cn", fontfile=font)
    y = 60

    def write(text: str, size: int, gap: int) -> None:
        nonlocal y, page
        # 简易折行: 按可容纳字数切
        per_line = int((page.rect.width - 100) / (size * 0.95))
        lines = [text[i : i + per_line] for i in range(0, len(text), per_line)] or [""]
        for ln in lines:
            if y > page.rect.height - 70:
                page = pdf.new_page()
                page.insert_font(fontname="cn", fontfile=font)
                y = 60
            page.insert_text((50, y), ln, fontname="cn", fontsize=size)
            y += size + 4
        y += gap

    write(doc_spec["title"], 18, 16)
    for idx, (heading, paras) in enumerate(doc_spec["sections"]):
        write(heading, 14, 8)
        for p in paras:
            write(p, 11, 6)
        if idx == 1:  # 表格放在第二节, 制造"表格夹在正文中间"的真实版式
            t = doc_spec["table"]
            write(t["caption"], 11, 4)
            rows_all = [t["header"], *t["rows"]]
            col = int((page.rect.width - 100) / len(t["header"]))
            top = y - 12
            for row in rows_all:
                if y > page.rect.height - 70:
                    page = pdf.new_page()
                    page.insert_font(fontname="cn", fontfile=font)
                    y = 60
                    top = y - 12
                for ci, cell in enumerate(row):
                    page.insert_text((54 + ci * col, y), cell, fontname="cn", fontsize=10)
                y += 18
            # 画表格线: 真实企业 PDF 的表格几乎都有边框, 有线 find_tables() 才认得出来
            bottom = y - 12
            for r in range(len(rows_all) + 1):
                yy = top + r * 18
                page.draw_line((50, yy), (50 + col * len(t["header"]), yy), width=0.6)
            for c in range(len(t["header"]) + 1):
                xx = 50 + c * col
                page.draw_line((xx, top), (xx, bottom), width=0.6)
            y += 14

    out = DATA / doc_spec["file"]
    pdf.save(out)
    pdf.close()
    print(f"生成 {out.name}  页数={pymupdf.open(out).page_count}")


if __name__ == "__main__":
    DATA.mkdir(parents=True, exist_ok=True)
    f = _font_path()
    print(f"使用字体: {f}")
    for spec in DOCS:
        build(spec, f)

注意 page.insert_font(fontname="cn", fontfile=font) 必须每新建一页调一次——字体是页级资源,忘了这行,第二页的中文会全部变成空白方框。

✅ Checkpoint 2:uv run python docqa/make_corpus.py

使用字体: /System/Library/Fonts/Supplemental/Songti.ttc
生成 员工手册.pdf  页数=1
生成 产品技术白皮书.pdf  页数=1
生成 售后服务政策.pdf  页数=1
生成 实习生管理办法.pdf  页数=1
生成 售后服务政策-2024旧版.pdf  页数=1

用系统的 PDF 阅读器打开 docqa/data/员工手册.pdf,你应该看到:大字标题、三个小节、夹在第二节中间的带边框表格。表格有没有边框在下一节会变成一个真实的 bug,现在先记住这件事。

5. parser.py —— 最脏也最拉开差距的一步(2 小时)

docs/1/08 的知识库是手写在 Python 字符串里的。真实项目里它是一堆 PDF,而 PDF 是为打印设计的格式,不是为读取设计的格式。这一节要解决四个真实问题:

1. 哪一行是标题?(PDF 里没有"标题"这个概念,只有"某个字号的一串字")
2. 表格怎么变成检索得到的文本?
3. 表格和正文的先后顺序怎么保持?
4. 折行时插进中文里的空格怎么清掉?

"""PDF 解析: 把 PDF 变成带元数据的纯文本块。

这是企业 RAG 里最脏、最耗时、也最能拉开差距的一步。
docs/1/08 的知识库是手写在 Python 字符串里的; 真实项目里它是一堆 PDF。
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path

import pymupdf


@dataclass
class Block:
    """解析出来的一个语义块。page/heading 是后面能做"带页码引用"的关键。"""

    doc: str
    page: int
    heading: str
    text: str
    kind: str = "text"  # text | table
    meta: dict = field(default_factory=dict)


# 中文文档常见的标题形态: "第一章 xx" / "1. xx" / "一、xx" / "1.2 xx"
HEADING_RE = re.compile(
    r"^\s*(?:第[一二三四五六七八九十百]+[章节条]|[一二三四五六七八九十]+、|\d+(?:\.\d+)*\.?)\s*\S"
)
TABLE_CAPTION_RE = re.compile(r"^\s*表\s*[\d\-\.]+\s*\S")
# PDF 折行会在中文之间插入空格("直属 主管"), 必须清掉,
# 否则 BM25 分词和向量都会被这些幽灵空格污染
CJK_GAP = re.compile(r"(?<=[一-鿿])\s+(?=[一-鿿])")


def clean(text: str) -> str:
    return CJK_GAP.sub("", text.strip())


def _is_heading(line: str, size: float, body_size: float) -> bool:
    """两条证据: 字号明显大于正文, 或者匹配中文标题编号模式且够短。"""
    if size > body_size * 1.15 and len(line) < 40:
        return True
    return bool(HEADING_RE.match(line)) and len(line) < 40

CJK_GAP 这三行是本文件最值钱的部分,请一定理解它:

PDF 折行时会把一个词切开并插入空格,直属主管 变成 直属 主管。这个幽灵空格有两个后果:jieba 会把它切成 直属 + 主管 两个词(而不是一个),向量模型也会把它编码成不同的东西。这是一个不报错、只让准确率悄悄变差的 bug,也是我在实际调试中真实踩到的第一个坑。

正则用了前后断言 (?<=[一-鿿])\s+(?=[一-鿿]):只删"中文和中文之间"的空格。中英之间的空格必须留着,否则 P99 延迟 会被粘成 P99延迟,而查询里写的是 P99 延迟,又对不上了。后面的测试会专门断言这一条。

接下来是主函数:

def parse_pdf(path: str | Path) -> list[Block]:
    path = Path(path)
    pdf = pymupdf.open(path)
    blocks: list[Block] = []

    # 先统计全文字号分布, 出现次数最多的字号就是正文字号——
    # 别硬编码 "12 号是正文", 每份 PDF 都不一样
    sizes: dict[float, int] = {}
    for page in pdf:
        for blk in page.get_text("dict")["blocks"]:
            for line in blk.get("lines", []):
                for span in line["spans"]:
                    if span["text"].strip():
                        s = round(span["size"], 1)
                        sizes[s] = sizes.get(s, 0) + len(span["text"])
    body_size = max(sizes, key=lambda k: sizes[k]) if sizes else 11.0

    current_heading = "(无标题)"

    def emit(buf: list[str], page_no: int, heading: str) -> None:
        """把攒着的正文行落成一个 Block。页码和标题显式传进来, 不靠闭包捕获。"""
        text = clean(" ".join(buf))
        if len(text) >= 10:  # 丢掉页眉页脚残渣
            blocks.append(Block(path.stem, page_no, heading, text))

    for page_no, page in enumerate(pdf, 1):
        # 1) 先用 PyMuPDF 内置的表格识别拿到表格区域
        try:
            found = page.find_tables()
            tables = found.tables if hasattr(found, "tables") else list(found)
        except Exception:
            tables = []
        table_rects = [pymupdf.Rect(tb.bbox) for tb in tables]

        # 2) 把这一页的所有元素(标题 / 正文行 / 表格)按纵坐标排到一个列表里。
        #    必须按版面顺序处理, 否则表格会被挂到错误的标题下——
        #    "先抽表格再抽正文" 是新手最容易犯的顺序错误
        items: list[tuple[float, str, object]] = []

        for tb in tables:
            rows = tb.extract()
            if rows:
                items.append((tb.bbox[1], "table", rows))

        for blk in page.get_text("dict")["blocks"]:
            if "lines" not in blk:
                continue
            brect = pymupdf.Rect(blk["bbox"])
            # 落在表格区域里的文字跳过, 否则表格内容会被索引两遍
            if any(brect.intersects(r) for r in table_rects):
                continue
            for line in blk["lines"]:
                text = "".join(s["text"] for s in line["spans"]).strip()
                if not text or TABLE_CAPTION_RE.match(text):
                    continue
                size = max((round(s["size"], 1) for s in line["spans"]), default=body_size)
                kind = "heading" if _is_heading(text, size, body_size) else "line"
                items.append((line["bbox"][1], kind, text))

        items.sort(key=lambda x: x[0])

三个设计决策,每一个都对应一个我真实踩过的坑:

① 正文字号靠统计,不靠硬编码。 你不知道业务方的 PDF 用 10 号还是 12 号做正文。按"出现字符数最多的字号就是正文"来判断,任何 PDF 都成立。注意统计的是字符数而不是行数——标题行数可能很多但字很少。

② items 按纵坐标排序。 我第一版是"先把所有表格抽出来,再走一遍正文",结果表格全部挂到了错误的标题下面(白皮书的性能表被挂到"3. 数据备份与恢复"上)。原因很简单:PDF 的文本块遍历顺序和视觉顺序无关。把表格、标题、正文行统一按 y 坐标放进一个列表再排序,才是"按人眼看到的顺序处理"。

③ 落在表格区域内的文字要跳过。 不跳过的话表格内容会被索引两遍(一次作为表格,一次作为散落的正文行),检索时你会看到同一个内容占掉两个位置。

然后是表格线性化,这是整个 RAG 里最容易被做错的一件事:

        # 攒正文行, 遇到下一个标题或表格就落一个块。
        # 这里刻意不用嵌套函数做 flush: 闭包捕获的是变量本身而不是当次的值,
        # current_heading 在循环里一直在变, 用闭包读它是经典的延迟绑定陷阱
        # (ruff 的 B023 规则专查这个)。写成显式传参, 读代码的人不用推理作用域
        buf: list[str] = []

        for _y, kind, payload in items:
            if kind == "heading":
                emit(buf, page_no, current_heading)
                buf = []
                current_heading = clean(payload)
            elif kind == "line":
                buf.append(payload)
            else:  # table
                emit(buf, page_no, current_heading)
                buf = []
                rows = payload
                # 表格必须"线性化"成自然语言才检索得到——
                # 直接把 markdown 表格塞进向量库, 检索效果会很差
                header = [clean(str(c or "")) for c in rows[0]]
                lines = []
                for row in rows[1:]:
                    cells = [clean(str(c or "")) for c in row]
                    # strict=False 是刻意的: 表格最后一行经常缺格,
                    # 长度不等时按短的截断, 不要抛异常炸掉整份文档的解析
                    pairs = [f"{h}是{v}" for h, v in zip(header, cells, strict=False) if h and v]
                    if pairs:
                        lines.append(", ".join(pairs) + "。")
                if lines:
                    blocks.append(
                        Block(
                            path.stem,
                            page_no,
                            current_heading,
                            "[表格] " + " ".join(lines),
                            kind="table",
                            meta={"columns": header, "rows": len(rows) - 1},
                        )
                    )
        emit(buf, page_no, current_heading)

    pdf.close()
    return blocks


def parse_dir(data_dir: str | Path) -> list[Block]:
    out: list[Block] = []
    for p in sorted(Path(data_dir).glob("*.pdf")):
        got = parse_pdf(p)
        print(f"[parse] {p.name}: {len(got)} 块 (表格 {sum(b.kind == 'table' for b in got)})")
        out.extend(got)
    return out


if __name__ == "__main__":
    blocks = parse_dir(Path(__file__).parent / "data")
    print(f"\n合计 {len(blocks)} 块\n")
    for b in blocks[:3]:
        print(f"--- {b.doc} p{b.page} [{b.kind}] <{b.heading}>")
        print(f"    {b.text[:90]}...")
    tb = next((b for b in blocks if b.kind == "table"), None)
    if tb:
        print(f"\n--- 表格块示例 ({tb.doc} p{tb.page}, {tb.meta})")
        print(f"    {tb.text[:180]}...")

表格线性化为什么这么重要:一个表格行 ["满 5 年不满 10 年", "15 天", "是"],如果你直接把 markdown 表格塞进向量库,模型看到的是一堆管道符和空格,语义几乎为零。线性化成 入职年限是满 5 年不满 10 年, 年假天数是15 天, 是否可结转是是。 之后,它变成了一句人话,而向量模型和 BM25 都是为人话训练的。

这就是为什么用户问"满5年员工年假多少天"能命中这个表格块——问题和线性化后的文本共享了"满 5 年""年假"这些词。

✅ Checkpoint 3:uv run python docqa/parser.py

[parse] 产品技术白皮书.pdf: 4 块 (表格 1)
[parse] 员工手册.pdf: 4 块 (表格 1)
[parse] 售后服务政策-2024旧版.pdf: 4 块 (表格 1)
[parse] 售后服务政策.pdf: 4 块 (表格 1)
[parse] 实习生管理办法.pdf: 4 块 (表格 1)

合计 20 块

--- 产品技术白皮书 p1 [text] <1. 系统架构>
    平台采用存算分离架构, 计算层无状态、可水平扩展, 存储层基于对象存储实现冷热分层。 元数据服务使用 Raft 协议保证强一致, 默认部署 3 副本, 可容忍 1 个节点故障。 所...
--- 产品技术白皮书 p1 [text] <2. 性能指标>
    在 32 核 128GB 的标准配置下, 单节点写入吞吐可达 12 万行每秒, P99 查询延迟低于 200 毫秒。 查询并发上限由 max_concurrent_queries ...
--- 产品技术白皮书 p1 [table] <2. 性能指标>
    [表格] 规格是8 核 32GB, 写入吞吐是3 万行/秒, P99 延迟是450 毫秒, 建议并发是3...

--- 表格块示例 (产品技术白皮书 p1, {'columns': ['规格', '写入吞吐', 'P99 延迟', '建议并发'], 'rows': 4})
    [表格] 规格是8 核 32GB, 写入吞吐是3 万行/秒, P99 延迟是450 毫秒, 建议并发是32。 规格是16 核 64GB, 写入吞吐是6 万行/秒, P99 延迟是300 毫秒, 建议并发是64。 规格是32 核 128GB, 写入吞吐是...

三件事必须对上,否则不要往下走:

• 每份文档 4 块、表格 1 个。如果表格数是 0,说明 find_tables() 没认出来——回去检查 make_corpus.py 里画表格线那几行。我第一版没画线,find_tables() 返回 0 个表格,更糟的是表格里的 5 天、10 天 被 _is_heading 误判成了标题,于是解析出一堆 <5 天> 这样的荒唐 heading。
• 表格块的 heading 是 2. 性能指标(表格所在的那一节),不是 3. 数据备份与恢复。挂错了说明你漏了按 y 坐标排序那一步。
• 正文里的 P99 延迟 中间的空格还在,而 直属主管 中间没有多余空格。

6. chunker.py —— 分块的三个讲究(1.5 小时)

docs/1/08 的分块是"每 300 字切一刀"。这里多做三件事,每一件都直接影响召回率:

1. 按句子边界切,不切断句子
2. 每个 chunk 前面拼上"文档名 · 标题"作为上下文前缀 —— 这一步对召回率提升极大
3. 表格块永不拆分 —— 拆开的表格行会丢失表头,变成无意义的碎片

"""分块: 把解析出的块切成适合检索的 chunk。

和 docs/1/08 的 chunker.py 相比多了三件事:
1. 按句子边界切(中文标点), 不切断句子
2. 每个 chunk 前面拼上"文档名 + 标题"作为上下文前缀 —— 这一步对召回率提升极大
3. 表格块永不拆分 —— 拆开的表格行会丢失表头, 变成无意义的碎片
"""

from __future__ import annotations

import hashlib
import re
from dataclasses import dataclass

from docqa.parser import Block

SENT_END = re.compile(r"(?<=[。!?;;])")


@dataclass
class Chunk:
    id: str
    doc: str
    page: int
    heading: str
    kind: str
    text: str  # 送进向量/BM25 的文本(含上下文前缀)
    raw: str  # 原始文本(展示给用户看的)

    @property
    def cite(self) -> str:
        return f"{self.doc} p{self.page} · {self.heading}"


def chunk_id(doc: str, page: int, heading: str, raw: str) -> str:
    """内容哈希做 id, 而不是"第几个块-第几片"。

    坑: 用序号做 id 时, "一次解析整个目录" 和 "逐个文件解析" 算出来的序号不一样,
    于是 BM25 侧和向量侧对同一个 chunk 给出不同 id, RRF 去重失效 ——
    同一条内容被当成两条各拿一份分数, 融合效果反而比单路更差。
    内容哈希天然与遍历顺序无关。
    """
    h = hashlib.sha1(f"{doc}|{page}|{heading}|{raw}".encode()).hexdigest()
    return h[:16]


def split_sentences(text: str) -> list[str]:
    return [s.strip() for s in SENT_END.split(text) if s.strip()]

text 和 raw 分成两个字段是刻意的:送进检索的文本和展示给用户的文本不是一回事。text 带 员工手册 · 第二章假期管理\n 这个前缀,raw 不带。用户看答案时不需要看到前缀,但检索时非常需要它。

为什么上下文前缀这么有用?想象一个 chunk 的内容是"一线城市每晚不超过 600 元"。这句话里既没有"报销"也没有"住宿",用户问"出差住宿能报多少"时,单靠这句话很难召回。拼上 员工手册 · 第三章报销标准 之后,"报销"这个词进来了,BM25 和向量都能对上。这是一个几行代码换来显著召回提升的技巧,业界叫 contextual retrieval。

chunk_id 那段 docstring 记录的是本项目最隐蔽的一个 bug,第 8 节 RRF 部分会详细讲它是怎么暴露的。

def chunk_blocks(
    blocks: list[Block],
    size: int = 300,
    overlap: int = 60,
) -> list[Chunk]:
    chunks: list[Chunk] = []
    for b in blocks:
        prefix = f"{b.doc} · {b.heading}\n"

        if b.kind == "table" or len(b.text) <= size:
            chunks.append(
                Chunk(
                    id=chunk_id(b.doc, b.page, b.heading, b.text),
                    doc=b.doc,
                    page=b.page,
                    heading=b.heading,
                    kind=b.kind,
                    text=prefix + b.text,
                    raw=b.text,
                )
            )
            continue

        sents = split_sentences(b.text)
        cur: list[str] = []
        cur_len = 0
        for s in sents:
            if cur and cur_len + len(s) > size:
                body = "".join(cur)
                chunks.append(
                    Chunk(
                        chunk_id(b.doc, b.page, b.heading, body),
                        b.doc,
                        b.page,
                        b.heading,
                        b.kind,
                        prefix + body,
                        body,
                    )
                )
                # 回退若干句作为重叠, 保证跨 chunk 的指代不断裂。
                # 关键约束: 最多回退 len(cur)-1 句 —— 否则当 cur 只有一句时,
                # 会把整句留下, 下一个 chunk 完整包含上一个, 白白浪费一次向量化
                back, keep = 0, []
                for s2 in reversed(cur[1:]):
                    if back >= overlap:
                        break
                    keep.insert(0, s2)
                    back += len(s2)
                cur, cur_len = keep, back
            cur.append(s)
            cur_len += len(s)
        if cur:
            body = "".join(cur)
            chunks.append(
                Chunk(
                    chunk_id(b.doc, b.page, b.heading, body),
                    b.doc,
                    b.page,
                    b.heading,
                    b.kind,
                    prefix + body,
                    body,
                )
            )
    return chunks

reversed(cur[1:]) 里那个 [1:] 是我调试出来的:一开始写的是 reversed(cur),当一个 chunk 只装得下一句话时,重叠会把这唯一的一句整个留下,于是下一个 chunk 完整包含上一个,白白浪费一次向量化和一个检索位置。切片保证最多回退 len(cur)-1 句。

if __name__ == "__main__":
    from pathlib import Path

    from docqa.parser import parse_dir

    blocks = parse_dir(Path(__file__).parent / "data")
    chunks = chunk_blocks(blocks)
    lens = [len(c.raw) for c in chunks]
    print(f"\n{len(blocks)} 块 → {len(chunks)} chunk")
    print(f"长度: min={min(lens)} max={max(lens)} avg={sum(lens) // len(lens)}")
    print(f"表格 chunk: {sum(c.kind == 'table' for c in chunks)} 个(均未被拆分)")
    print("\n--- chunk 示例(注意开头的上下文前缀) ---")
    print(chunks[1].text[:200])
    print(f"\n引用格式: {chunks[1].cite}")

    # 本教程的语料每节都短于 300 字, 所以上面一个都没被切开。
    # 把 size 调小, 验证切分和重叠逻辑真的在工作:
    small = chunk_blocks(blocks, size=120, overlap=50)
    print(f"\n把 size 调到 120: {len(blocks)} 块 → {len(small)} chunk")
    same = [c for c in small if c.doc == "产品技术白皮书" and c.heading == "2. 性能指标"]
    for i, c in enumerate(same):
        print(f"  片段{i}: {c.raw}")
    if len(same) >= 2:
        first_sents = [s for s in same[0].raw.split("。") if s]
        dup = [s for s in first_sents if s in same[1].raw]
        print(f"  片段0 与 片段1 重复的句子数: {len(dup)}  (这就是 overlap 在起作用)")
    # 提示: size 再往下调到 100, 重叠会消失 —— 因为重叠以完整句子为单位回退,
    # 一个 chunk 只装得下一句时就没有可回退的句子了。这是句子对齐分块的固有代价,
    # 换来的是永不切断句子。真实项目里 size 一般取 300~500, 不会踩到这个边界

✅ Checkpoint 4:uv run python docqa/chunker.py

20 块 → 20 chunk
长度: min=61 max=213 avg=124
表格 chunk: 5 个(均未被拆分)

--- chunk 示例(注意开头的上下文前缀) ---
产品技术白皮书 · 2. 性能指标
在 32 核 128GB 的标准配置下, 单节点写入吞吐可达 12 万行每秒, P99 查询延迟低于 200 毫秒。 查询并发上限由 max_concurrent_queries 参数控制, 默认值为 128, 超出后请求进入排队队列。 排队超时时间默认 30 秒, 超时返回 429 状态码, 客户端应实现指数退避重试。

引用格式: 产品技术白皮书 p1 · 2. 性能指标

把 size 调到 120: 20 块 → 27 chunk
  片段0: 在 32 核 128GB 的标准配置下, 单节点写入吞吐可达 12 万行每秒, P99 查询延迟低于 200 毫秒。查询并发上限由 max_concurrent_queries 参数控制, 默认值为 128, 超出后请求进入排队队列。
  片段1: 查询并发上限由 max_concurrent_queries 参数控制, 默认值为 128, 超出后请求进入排队队列。排队超时时间默认 30 秒, 超时返回 429 状态码, 客户端应实现指数退避重试。
  片段2: [表格] 规格是8 核 32GB, 写入吞吐是3 万行/秒, P99 延迟是450 毫秒, 建议并发是32。 ...
  片段0 与 片段1 重复的句子数: 1  (这就是 overlap 在起作用)

注意 20 块 → 20 chunk:默认参数下一个块都没被切开,因为本教程语料每节都短于 300 字。这不是 bug,但它意味着你没有验证切分逻辑。所以 __main__ 里刻意加了 size=120 的第二次调用:27 个 chunk、片段 0 和片段 1 有 1 句重复,这才证明切分和重叠真的在工作。

"跑出来了"和"验证过了"是两件不同的事——这个习惯比本节的任何代码都值钱。

7. bm25.py —— 关键词检索的中文版(1 小时)

docs/1/08 手写过 BM25,公式一样。这里的区别只有三处,但都是中文场景的必需品:

1. 中文必须先分词,不能按字符切
2. 加停用词过滤,否则"的""了"会污染 IDF
3. 保留 idf 字典,排错时能看出"到底是哪个词把这条文档捞上来的"

"""BM25 关键词检索(中文用 jieba 分词)。"""

from __future__ import annotations

import math
from collections import Counter

import jieba

from docqa.chunker import Chunk

STOP = {
    "的", "了", "和", "与", "是", "在", "有", "我", "你", "他", "它", "也",
    "就", "都", "而", "及", "或", "对", "中", "为", "以", "及其", "这", "那",
    "个", "上", "下", "不", "会", "可", "要", "被", "把", "从", "到", "等",
    "一", "二", "三",
}  # fmt: skip


def tokenize(text: str) -> list[str]:
    """分词 + 去停用词。单字一律丢掉, 但纯数字例外 ——
    "5" "128" "429" 这些单字/短词恰恰是企业文档里最该被检索到的。
    括号必须写: and 的优先级高于 or, 不加括号语义会变
    """
    return [
        w
        for w in jieba.lcut(text.lower())
        if (w.strip() and w not in STOP and len(w) > 1) or w.isdigit()
    ]

len(w) > 1 丢单字是常规做法(单字的 IDF 几乎没有区分度),但 or w.isdigit() 这个例外是企业文档的刚需:用户问的就是 "5 年年假多少天"、"并发上限 128"、"返回 429 怎么办"。数字往往是问题里信息量最大的那个 token,把它当单字丢掉,BM25 这一路就废了一半。

括号也不能省:Python 里 and 优先级高于 or,不加括号会变成 w.strip() and w not in STOP and (len(w) > 1 or w.isdigit()),含义正好错开 —— 停用词表里的数字就漏出来了。

class BM25:
    def __init__(self, chunks: list[Chunk], k1: float = 1.5, b: float = 0.75) -> None:
        self.chunks = chunks
        self.k1, self.b = k1, b
        self.docs = [tokenize(c.text) for c in chunks]
        self.lens = [len(d) for d in self.docs]
        self.avg_len = sum(self.lens) / len(self.lens) if self.lens else 1.0
        self.tf = [Counter(d) for d in self.docs]
        df: Counter = Counter()
        for d in self.docs:
            df.update(set(d))
        n = len(self.docs)
        # 加了 +1 的平滑 idf: 保证永不为负, 否则高频词会给出负分把好文档挤下去
        self.idf = {w: math.log(1 + (n - c + 0.5) / (c + 0.5)) for w, c in df.items()}

    def search(self, query: str, top_k: int = 10) -> list[tuple[Chunk, float]]:
        q = tokenize(query)
        scored: list[tuple[Chunk, float]] = []
        for i, tf in enumerate(self.tf):
            s = 0.0
            for w in q:
                if w not in tf:
                    continue
                f = tf[w]
                s += (
                    self.idf.get(w, 0.0)
                    * f
                    * (self.k1 + 1)
                    / (f + self.k1 * (1 - self.b + self.b * self.lens[i] / self.avg_len))
                )
            if s > 0:
                scored.append((self.chunks[i], s))
        scored.sort(key=lambda x: -x[1])
        return scored[:top_k]

IDF 公式里那个 log(1 + ...) 的 1 + 值得单独说。教科书上常见的写法是 log((n - c + 0.5) / (c + 0.5)),当一个词出现在超过一半的文档里时,这个式子会算出负数。负分意味着:一篇文档因为包含了查询里的某个常见词,反而被扣分排到后面去。加上 1 + 之后最小值是 0,最坏情况是"这个词不加分",而不是"倒扣分"。

小知识库尤其容易踩到:20 个 chunk 里,"公司""服务"这种词出现在 11 篇以上是很平常的事。测试里有一条专门盯着这个:

def test_bm25_idf_never_negative():
    # 一个词出现在所有文档里时, 朴素 idf 公式会算出负数, 把好文档挤到后面
    chunks = [mk_chunk(str(i), "公司制度规定内容") for i in range(5)]
    assert all(v >= 0 for v in BM25(chunks).idf.values())

✅ Checkpoint 5:PYTHONPATH=. uv run python docqa/bm25.py

[parse] 产品技术白皮书.pdf: 4 块 (表格 1)
[parse] 员工手册.pdf: 4 块 (表格 1)
[parse] 售后服务政策-2024旧版.pdf: 4 块 (表格 1)
[parse] 售后服务政策.pdf: 4 块 (表格 1)
[parse] 实习生管理办法.pdf: 4 块 (表格 1)

索引 20 个 chunk, 词表 340 个词

Q: 满5年的年假有多少天
    2.58  员工手册 p1 · 第二章假期管理  [table]
    1.64  实习生管理办法 p1 · 第二章假期管理  [text]
    1.51  产品技术白皮书 p1 · 3. 数据备份与恢复  [text]

Q: P99 延迟
    7.09  产品技术白皮书 p1 · 2. 性能指标  [table]
    4.00  产品技术白皮书 p1 · 2. 性能指标  [text]

Q: 企业版故障多久响应
    7.68  售后服务政策-2024旧版 p1 · 一、服务等级  [text]
    7.63  售后服务政策 p1 · 一、服务等级  [text]
    2.53  售后服务政策 p1 · 二、赔偿条款  [table]

三个查询正好演示了 BM25 的性格:

• 第一个查询,满5年的年假 命中了年假对照表(2.58),对的。但第二名是实习生管理办法——实习生那份也讲"年假",词一样,BM25 分不清"谁的年假"。这是纯关键词检索的天花板。
• 第二个查询,P99 延迟 精准打到白皮书的性能表(7.09),而且只返回 2 条 —— 罕见词的 IDF 高,分数拉得开,干净。这种查询向量检索反而容易输,因为 P99 这种 token 在通用语料上训练出的向量里区分度不高。
• 第三个查询,企业版故障多久响应,新版 7.63、旧版 7.68 —— 旧版还排在前面。两份文档的措辞几乎一样,BM25 只看词频,天然不知道哪份废止了。这正是第 9 节 rerank 要解决的问题,也是第 15 节答案里必须同时列出两个版本并标注引用的原因:检索层不可能百分之百选对,那就让用户看见冲突。

8. vectorstore.py —— 从 numpy 点积换成真向量库(1.5 小时)

docs/1/08 的向量检索是内存里一个 numpy 点积。这一步换成 Qdrant,你会真实遇到三个只有用了向量库才会碰到的问题:

1. collection 的向量维度写错了,写入时才报错
2. 中文 bge 模型的 query 必须加指令前缀,不加召回率明显下降
3. payload 里要存什么 —— 存少了没法引用,存多了内存爆

"""向量检索: bge-small-zh 编码 + Qdrant 存储。"""

from __future__ import annotations

import uuid

from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer

from docqa.chunker import Chunk

MODEL_NAME = "BAAI/bge-small-zh-v1.5"
COLLECTION = "docqa"
# bge 中文系列官方要求: 查询侧加这个前缀, 文档侧不加。
# 漏掉它是最常见的静默性能损失——不报错, 只是准确率变差
QUERY_PREFIX = "为这个句子生成表示以用于检索相关文章:"


class Embedder:
    def __init__(self, name: str = MODEL_NAME) -> None:
        self.model = SentenceTransformer(name)
        self.dim = self.model.get_sentence_embedding_dimension()

    def encode_docs(self, texts: list[str]) -> list[list[float]]:
        return self.model.encode(texts, normalize_embeddings=True, show_progress_bar=False).tolist()

    def encode_query(self, text: str) -> list[float]:
        return self.model.encode(
            QUERY_PREFIX + text, normalize_embeddings=True, show_progress_bar=False
        ).tolist()

QUERY_PREFIX 是 bge 中文系列的官方要求:查询侧加,文档侧不加。这是一个非对称编码设计——模型训练时就是这么喂的,推理时不还原这个不对称,query 向量和 doc 向量就落在略微错开的空间里。

它是本章最典型的静默性能损失:不报错、不崩溃,只是准确率悄悄掉几个点,你永远不知道自己损失了什么。这类 bug 只有靠第 10 节的评测集才能抓出来。

self.dim 也别硬编码成 512。bge-small-zh 是 512 维,bge-base-zh 是 768,换模型时如果 collection 还按旧维度建着,Qdrant 会在写入的那一刻才报 Vector dimension error,而不是在创建 collection 时。问模型要维度,一行代码免掉一次半夜排错。

class VectorStore:
    def __init__(self, embedder: Embedder, url: str = "http://localhost:6333") -> None:
        self.client = QdrantClient(url=url)
        self.emb = embedder

    def recreate(self) -> None:
        if self.client.collection_exists(COLLECTION):
            self.client.delete_collection(COLLECTION)
        self.client.create_collection(
            COLLECTION,
            vectors_config=models.VectorParams(size=self.emb.dim, distance=models.Distance.COSINE),
        )

    def index(self, chunks: list[Chunk], batch: int = 32) -> int:
        total = 0
        for i in range(0, len(chunks), batch):
            part = chunks[i : i + batch]
            vecs = self.emb.encode_docs([c.text for c in part])
            self.client.upsert(
                COLLECTION,
                points=[
                    models.PointStruct(
                        # Qdrant 的 id 只接受 int 或 UUID, 不接受任意字符串;
                        # 用 uuid5 保证同一个 chunk 重复索引时是"覆盖"而不是"新增"
                        id=str(uuid.uuid5(uuid.NAMESPACE_URL, f"{c.doc}/{c.id}")),
                        vector=v,
                        payload={
                            "chunk_id": c.id,
                            "doc": c.doc,
                            "page": c.page,
                            "heading": c.heading,
                            "kind": c.kind,
                            "raw": c.raw,
                            "text": c.text,
                        },
                    )
                    for c, v in zip(part, vecs, strict=True)
                ],
            )
            total += len(part)
        return total

三个细节:

• uuid5 而不是 uuid4。Qdrant 的 point id 只接受整数或 UUID,不接受任意字符串,所以我们的 16 位哈希 id 不能直接用。uuid5 是确定性的:同样的输入永远得到同样的 UUID,于是重复索引同一个 chunk 是"覆盖"而不是"新增"。用 uuid4 的话,每次重建索引都会让知识库里堆满重复内容,检索结果里同一句话出现三遍。
• payload 同时存 raw 和 text。raw 给用户看,text 给 rerank 用(rerank 也需要上下文前缀)。多存一份文本是有内存代价的,但 Qdrant 的 payload 默认存在磁盘上、只有向量常驻内存,这点冗余换来的是"检索完不用再回原文档捞一遍"。
• zip(..., strict=True)。part 和 vecs 长度理论上必然相等,但如果哪天 encode_docs 因为截断少返回一条,strict=False 会静默丢掉最后一个 chunk。加上 strict=True,不一致就抛异常。

    def search(
        self, query: str, top_k: int = 10, doc: str | None = None
    ) -> list[tuple[Chunk, float]]:
        flt = (
            models.Filter(
                must=[models.FieldCondition(key="doc", match=models.MatchValue(value=doc))]
            )
            if doc
            else None
        )
        hits = self.client.query_points(
            COLLECTION,
            query=self.emb.encode_query(query),
            limit=top_k,
            query_filter=flt,
            with_payload=True,
        ).points
        out = []
        for h in hits:
            p = h.payload or {}
            out.append(
                (
                    Chunk(
                        p["chunk_id"],
                        p["doc"],
                        p["page"],
                        p["heading"],
                        p["kind"],
                        p["text"],
                        p["raw"],
                    ),
                    float(h.score),
                )
            )
        return out

query_filter 是向量库比 numpy 点积强的地方之一:带条件的向量检索。用户说"只在售后服务政策里找",你不需要先全库检索再过滤(那样 top_k 可能一条都不剩),Qdrant 在检索时就把范围收窄了。真实产品里这个能力用来做权限隔离——HR 的文档只有 HR 能检索到。

✅ Checkpoint 6:PYTHONPATH=. uv run python docqa/vectorstore.py

第一次跑会下载模型(约 100MB),需要联网,之后会缓存在 ~/.cache/huggingface。

模型 BAAI/bge-small-zh-v1.5, 向量维度 512
写入 20 个点, Qdrant 报告 points_count=20

Q: 工作时间是几点到几点
   0.5751  员工手册 p1 · 第一章考勤制度  [text]
   0.5626  实习生管理办法 p1 · 第一章考勤制度  [text]
   0.4560  员工手册 p1 · 第二章假期管理  [text]

Q: 配置越高延迟越低吗
   0.5456  产品技术白皮书 p1 · 2. 性能指标  [text]
   0.4738  产品技术白皮书 p1 · 2. 性能指标  [table]
   0.4314  产品技术白皮书 p1 · 1. 系统架构  [text]

Q: 赔多少钱
   0.5547  售后服务政策-2024旧版 p1 · 二、赔偿条款  [text]
   0.5043  售后服务政策 p1 · 二、赔偿条款  [text]
   0.4740  售后服务政策-2024旧版 p1 · 二、赔偿条款  [table]

带过滤的检索(只在售后服务政策里找):
   0.4847  售后服务政策 p1 · 一、服务等级
   0.3745  售后服务政策 p1 · 二、赔偿条款
   0.3628  售后服务政策 p1 · 二、赔偿条款

points_count=20 必须等于 chunk 数。如果比它大,说明你的 id 不是确定性的(检查 uuid5 有没有写成 uuid4);跑两次之后数字翻倍,就是这个毛病。

对比第 7 节的 BM25,向量的性格完全反过来:

• 配置越高延迟越低吗 —— 这句话里没有一个词出现在文档里("配置"勉强算),BM25 基本捞不到,向量却直接命中性能指标那一节。这就是语义检索的价值:用户不会用文档里的词提问。
• 赔多少钱 同理,文档里写的是"赔偿""返还服务信用点"。
• 但注意 赔多少钱 的第一名又是旧版(0.5547 > 0.5043)。和 BM25 一样,向量也分不清版本新旧——两份文档语义上确实几乎一样。这个问题两条路都解决不了,只能靠 rerank 和最终答案里的显式引用。

跑的时候会看到两条无害的告警:get_sentence_embedding_dimension 已改名(旧名仍可用),以及 qdrant client 1.19 与 server 1.16 的版本提示。功能不受影响,不用管。

9. retriever.py —— 混合检索三层楼(2 小时)

这一节是全章技术含量最高的地方,也是面试里"你的 RAG 怎么调优的"这个问题的标准答案。三层各司其职:

"""混合检索: BM25 + 向量, 用 RRF 融合, 再用 cross-encoder 精排。"""

from __future__ import annotations

from dataclasses import dataclass

from docqa.bm25 import BM25
from docqa.chunker import Chunk
from docqa.vectorstore import VectorStore


@dataclass
class Hit:
    chunk: Chunk
    score: float
    bm25_rank: int | None = None
    vec_rank: int | None = None
    rerank_score: float | None = None

    @property
    def source(self) -> str:
        if self.bm25_rank is not None and self.vec_rank is not None:
            return "both"
        return "bm25" if self.bm25_rank is not None else "vector"


def rrf_fuse(
    bm25_hits: list[tuple[Chunk, float]],
    vec_hits: list[tuple[Chunk, float]],
    k: int = 60,
) -> list[Hit]:
    """Reciprocal Rank Fusion: score = Σ 1/(k + rank)。

    k=60 是原论文的经验值, 作用是压低头部排名的权重差 ——
    k 越大, 两路的第 1 名和第 5 名差距越小(更"民主")。
    """
    table: dict[str, Hit] = {}

    def key(c: Chunk) -> str:
        return f"{c.doc}/{c.id}"

    for rank, (c, _s) in enumerate(bm25_hits, 1):
        h = table.setdefault(key(c), Hit(c, 0.0))
        h.bm25_rank = rank
        h.score += 1 / (k + rank)
    for rank, (c, _s) in enumerate(vec_hits, 1):
        h = table.setdefault(key(c), Hit(c, 0.0))
        h.vec_rank = rank
        h.score += 1 / (k + rank)

    hits = sorted(table.values(), key=lambda h: -h.score)
    return hits

RRF 的公式短到一行:score += 1 / (k + rank)。为什么选它而不是"BM25 分 × 0.3 + 向量分 × 0.7"这种加权求和?

因为两路的分数量纲根本不同。BM25 分数是 0 到十几,没有上界,还随知识库大小漂移;余弦相似度固定在 -1 到 1 之间,而且中文 bge 上大部分正样本挤在 0.4~0.6 这个窄区间里。要把它们相加,你得先归一化——而归一化又需要知道分数分布,分布又随查询变化。工程上这条路走不通。

RRF 直接扔掉分数、只用排名。"BM25 排第 1"和"向量排第 1"权重相等,不需要任何标定。代价是丢失了强度信息(BM25 分 100 和分 5 都只算"第 1 名"),但换来的是零调参、零维护。企业 RAG 里这笔交易几乎总是划算的。

k=60 来自原论文的经验值,作用是压平名次差:1/(60+1) = 0.0164 对 1/(60+5) = 0.0154,第 1 名只比第 5 名高 6%。如果把 k 设成 1,1/2 = 0.5 对 1/6 = 0.167,差 3 倍。k 越大越"民主"——更看重"有几路投票",而不是"排多前"。测试里有一条盯着这个性质:

def test_rrf_k_flattens_ranking():
    a, b = mk_chunk("a", "aa"), mk_chunk("b", "bb")
    bm = [(a, 9.0), (b, 8.0)]
    small = rrf_fuse(bm, [], k=1)
    large = rrf_fuse(bm, [], k=1000)
    gap_small = small[0].score - small[1].score
    gap_large = large[0].score - large[1].score
    assert gap_small > gap_large, "k 越大, 名次差异被压得越平"

我在这里踩的最深的一个坑

key(c) 返回 f"{c.doc}/{c.id}",这个去重键是 RRF 的命门。第一次跑评测时结果是这样的:

融合之后比两条单路都差,这在数学上不该发生——RRF 最坏情况也该接近较好的那一路。打开 retriever.py 的 demo 看,同一个 chunk 出现了 4 次,两次标 source=bm25、两次标 source=vector,bm25_rank 和 vec_rank 永远只有一个有值,both 一次都没出现过。

原因在 chunker.py:最初的 chunk id 是位置编号 f"{bi}-{part}"。而 BM25 那一路是一次性解析整个目录建的索引(bi 从 0 数到 19),向量那一路是逐个文件索引的(每个文件的 bi 都从 0 重新数)。同一条内容在两边拿到了不同的 id,去重键对不上,于是同一个 chunk 被当成两条不同文档,各拿一份 RRF 分数——分数被摊薄,真正该排第一的内容被挤下去,而且 top_k 里塞满了重复内容。

修复就是第 6 节那个 chunk_id():用内容哈希,天然与遍历顺序无关。改完之后 both 正常出现,混合+RRF 升到 0.83。

这个 bug 值得记住的地方在于:它不报错,不崩溃,只是让效果变差。如果没有第 10 节那个评测集,你只会觉得"混合检索没啥用啊",然后把它删掉——把一个正确的架构当成无效的优化扔了。评测集不是为了写在简历上,是为了让你的判断有依据。

class Retriever:
    def __init__(
        self,
        chunks: list[Chunk],
        store: VectorStore,
        reranker_name: str = "BAAI/bge-reranker-base",
        use_reranker: bool = True,
    ) -> None:
        self.bm25 = BM25(chunks)
        self.store = store
        self.reranker = None
        if use_reranker:
            from sentence_transformers import CrossEncoder

            self.reranker = CrossEncoder(reranker_name, max_length=512)

    def retrieve(
        self,
        query: str,
        top_k: int = 4,
        candidates: int = 20,
        min_rerank: float = 0.05,
    ) -> list[Hit]:
        bm = self.bm25.search(query, top_k=candidates)
        vec = self.store.search(query, top_k=candidates)
        fused = rrf_fuse(bm, vec)[:candidates]

        if not fused:
            return []
        if self.reranker is None:
            return fused[:top_k]

        scores = self.reranker.predict([(query, h.chunk.text) for h in fused])
        for h, s in zip(fused, scores, strict=True):
            h.rerank_score = float(s)
        fused.sort(key=lambda h: -(h.rerank_score or 0))
        # 阈值是防幻觉的第一道闸: 精排分都很低说明知识库里就没有答案,
        # 这时候要让上层回答"文档中未提及", 而不是硬编一个
        kept = [h for h in fused if (h.rerank_score or 0) >= min_rerank]
        return kept[:top_k]

candidates=20 和 top_k=4 的分工要理解清楚:召回 20 条是为了不漏,精排到 4 条是为了不吵。BM25 和向量各取 20 条,融合后仍取 20 条送进 rerank,最后只留 4 条给 LLM。rerank 是 cross-encoder,把 query 和文档拼在一起过一遍模型,能捕捉到双塔编码看不见的交互信息,精度高得多,代价是每条都要跑一次前向传播——所以只能对 20 条做,不能对全库做。

min_rerank=0.05 是防幻觉的第一道闸,也是本章最重要的一行工程判断。传统 RAG 无论如何都会返回 top_k 条,于是知识库里明明没有答案时,LLM 拿到 4 条不相关的材料,还是会硬编一个出来。加了阈值之后,精排分全部低于 0.05 就返回空列表,上层直接短路回答"文档中未提及",连 LLM 都不调用——省钱,而且比任何 prompt 技巧都可靠。

✅ Checkpoint 7:PYTHONPATH=. uv run python docqa/retriever.py

第一次跑会再下载 rerank 模型(约 1.1GB)。

Q: 满5年员工年假多少天
   rerank=0.9958 rrf=0.0325 [both   bm25=1 vec=2] 员工手册 p1 · 第二章假期管理
   rerank=0.9437 rrf=0.0325 [both   bm25=2 vec=1] 员工手册 p1 · 第二章假期管理
   rerank=0.1625 rrf=0.0156 [vector bm25=None vec=4] 实习生管理办法 p1 · 第二章假期管理
   rerank=0.1262 rrf=0.0315 [both   bm25=4 vec=3] 实习生管理办法 p1 · 第二章假期管理

Q: 几点上班
   rerank=0.8818 rrf=0.0164 [vector bm25=None vec=1] 员工手册 p1 · 第一章考勤制度
   rerank=0.6129 rrf=0.0161 [vector bm25=None vec=2] 实习生管理办法 p1 · 第一章考勤制度

Q: max_concurrent_queries 默认值
   rerank=0.9993 rrf=0.0328 [both   bm25=1 vec=1] 产品技术白皮书 p1 · 2. 性能指标

Q: 公司的年终奖怎么发
   (无命中: 精排分全部低于阈值 → 上层应回答'文档中未提及')

四个查询把三层楼的价值全演示到了:

• 满5年员工年假多少天 —— 前两名 [both],rerank 0.9958 / 0.9437,而干扰文档"实习生管理办法"虽然也被两路召回了(rrf 0.0315,排名并不低),rerank 却只给 0.1262。融合负责召回,rerank 负责判断。 注意 rrf 分数第 1 和第 4 只差 3%(0.0325 vs 0.0315),光看 RRF 排序根本区分不出对错,是 rerank 把 0.99 和 0.13 拉开的。
• 几点上班 —— 两条都是 [vector],bm25=None。BM25 一条都没召回:文档里写的是"核心工作时间为每个工作日的 10:00 至 16:00","上班"这个词根本不存在。这一条查询上没有向量检索就是零。
• max_concurrent_queries 默认值 —— [both bm25=1 vec=1],两路第 1 名一致,rerank 0.9993。这种查询谁都能做对。
• 公司的年终奖怎么发 —— 知识库里确实没有,返回空。这是唯一一个"正确答案是没有答案"的用例,也是最容易被忽略的一类。

如果你的输出里 both 一次都没出现,回去检查 chunker.py 的 chunk_id 是不是用了内容哈希——就是上面讲的那个坑。

10. evaluate.py —— 整套系统里最值钱的文件(2 小时)

如果这一章只能留一个文件,留这个。

面试时说"我做了混合检索和 rerank"是空话——所有人都这么说。说"我有 24 条人工标注集,纯向量 Hit@1 是 0.83,加 BM25 融合到 0.83 但 Recall@5 从 0.96 提到 1.00,再加 rerank 后 Hit@1 到 0.96、MRR 从 0.880 到 0.979"是硬话。差别不在做了什么,在能不能证明它有用。

指标定义和 docs/1/12 的评测框架是同一套思路:

• Recall@k:前 k 条里有没有捞到标准答案所在的 chunk(捞到算 1)
• Hit@1:第 1 条就是标准答案 —— 精排质量的直接体现
• MRR:1 / 标准答案的排名,取平均。同时体现"捞到没"和"排得多前"

"""检索评测: 用一个人工标注的小评测集, 量化每一层带来的提升。"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

from docqa.chunker import Chunk, chunk_blocks
from docqa.parser import parse_dir
from docqa.retriever import Retriever, rrf_fuse
from docqa.vectorstore import Embedder, VectorStore

# 标注方式: 写 query + 答案必须命中的"文档 + 标题"
# (不写 chunk id, 因为分块参数一变 id 就全乱了)
GOLD: list[dict] = [
    {"q": "正式员工满5年的有几天年假", "doc": "员工手册", "heading": "第二章假期管理"},
    {"q": "年假没用完能带到明年吗", "doc": "员工手册", "heading": "第二章假期管理"},
    {"q": "正式员工几点上班几点下班", "doc": "员工手册", "heading": "第一章考勤制度"},
    {"q": "正式员工远程办公一个月能申请几次", "doc": "员工手册", "heading": "第一章考勤制度"},
    {"q": "正式员工迟到多久算迟到一次", "doc": "员工手册", "heading": "第一章考勤制度"},
    {"q": "正式员工去北京出差住宿能报多少", "doc": "员工手册", "heading": "第三章报销标准"},
    {"q": "正式员工发票多久之内要交", "doc": "员工手册", "heading": "第三章报销标准"},
    # 干扰对照: 同样的问题问实习生, 答案必须落到另一份文档
    {"q": "实习生有年假吗", "doc": "实习生管理办法", "heading": "第二章假期管理"},
    {"q": "实习生几点上班", "doc": "实习生管理办法", "heading": "第一章考勤制度"},
    {"q": "实习生打车最多报多少", "doc": "实习生管理办法", "heading": "第三章报销标准"},
    {"q": "元数据服务能挂几个节点", "doc": "产品技术白皮书", "heading": "1. 系统架构"},
    {"q": "证书多久换一次", "doc": "产品技术白皮书", "heading": "1. 系统架构"},
    {"q": "max_concurrent_queries 默认是多少", "doc": "产品技术白皮书", "heading": "2. 性能指标"},
    {"q": "16核64G能跑多少写入", "doc": "产品技术白皮书", "heading": "2. 性能指标"},
    {"q": "排队超时了返回什么状态码", "doc": "产品技术白皮书", "heading": "2. 性能指标"},
    {"q": "备份留多久", "doc": "产品技术白皮书", "heading": "3. 数据备份与恢复"},
    {"q": "1TB数据恢复要多长时间", "doc": "产品技术白皮书", "heading": "3. 数据备份与恢复"},
    {"q": "2026版企业版P1故障多久响应", "doc": "售后服务政策", "heading": "一、服务等级"},
    {"q": "现行政策的可用性承诺是多少", "doc": "售后服务政策", "heading": "一、服务等级"},
    {"q": "服务信用点能提现吗", "doc": "售后服务政策", "heading": "二、赔偿条款"},
    {"q": "现在买了多久内可以无理由退款", "doc": "售后服务政策", "heading": "三、退款政策"},
    {"q": "促销买的能退吗", "doc": "售后服务政策", "heading": "三、退款政策"},
    # 干扰对照: 明确问旧版
    {
        "q": "2024年旧版政策企业版P1多久响应",
        "doc": "售后服务政策-2024旧版",
        "heading": "一、服务等级",
    },
    {"q": "旧版赔偿上限是多少", "doc": "售后服务政策-2024旧版", "heading": "二、赔偿条款"},
]


def is_gold(c: Chunk, g: dict) -> bool:
    return c.doc == g["doc"] and c.heading == g["heading"]

标注方式有三个讲究,都是被坑出来的:

1. 标"文档 + 标题",不标 chunk id。分块参数一改(size 从 300 调到 400),所有 id 全变,评测集当场作废。标到"哪一节"这个粒度,分块怎么改都不影响。
2. query 里刻意加限定词。正式员工满5年的有几天年假 而不是 满5年年假多少天——因为知识库里有实习生那一份,不加"正式员工"这个问题本身就是歧义的,标准答案不唯一。评测集里的歧义问题是纯噪音:模型答"实习生不享受年假"也没错,但会被算成失败,指标就失去意义了。
3. 专门设计干扰对照。7 条正式员工的问题,配 3 条实习生的同类问题;5 条现行政策的问题,配 2 条明确问旧版的问题。如果模型只会"看到年假就返回员工手册",这 5 条对照会立刻暴露它。 只标正样本的评测集会给你一个虚高的分数。

@dataclass
class Metrics:
    name: str
    n: int
    hit1: float
    recall3: float
    recall5: float
    mrr: float

    def row(self) -> str:
        return (
            f"| {self.name:22s} | {self.hit1:.2f} | {self.recall3:.2f} | "
            f"{self.recall5:.2f} | {self.mrr:.3f} |"
        )


def score(name: str, ranked_per_query: list[list[Chunk]]) -> Metrics:
    h1 = r3 = r5 = mrr = 0.0
    for g, ranked in zip(GOLD, ranked_per_query, strict=True):
        pos = next((i + 1 for i, c in enumerate(ranked) if is_gold(c, g)), None)
        if pos == 1:
            h1 += 1
        if pos and pos <= 3:
            r3 += 1
        if pos and pos <= 5:
            r5 += 1
        if pos:
            mrr += 1 / pos
    n = len(GOLD)
    return Metrics(name, n, h1 / n, r3 / n, r5 / n, mrr / n)

score() 只接受"排好序的 chunk 列表",不关心这个列表是怎么来的。这个签名设计是消融实验能做的前提——同一个打分函数喂四种不同的检索方案,数字才可比。如果打分逻辑和检索逻辑缠在一起,你就没法做 A/B 对比了。

def main() -> None:
    chunks = chunk_blocks(parse_dir("docqa/data"))
    store = VectorStore(Embedder())
    r = Retriever(chunks, store, use_reranker=True)

    bm_only, vec_only, fused_only, reranked = [], [], [], []
    for g in GOLD:
        q = g["q"]
        bm = r.bm25.search(q, top_k=10)
        vec = store.search(q, top_k=10)
        bm_only.append([c for c, _ in bm])
        vec_only.append([c for c, _ in vec])
        fused = rrf_fuse(bm, vec)
        fused_only.append([h.chunk for h in fused])
        # 精排: 对融合后的前 20 条重排(这里语料小, 就是全部)
        cand = fused[:20]
        sc = r.reranker.predict([(q, h.chunk.text) for h in cand])
        order = sorted(zip(cand, sc, strict=True), key=lambda x: -x[1])
        reranked.append([h.chunk for h, _ in order])

    rows = [
        score("① 纯 BM25", bm_only),
        score("② 纯向量", vec_only),
        score("③ 混合 + RRF", fused_only),
        score("④ 混合 + RRF + rerank", reranked),
    ]

    print(f"\n评测集 {len(GOLD)} 条 / 知识库 {len(chunks)} chunk\n")
    print("| 方案                   | Hit@1 | R@3  | R@5  | MRR   |")
    print("| ---------------------- | ----- | ---- | ---- | ----- |")
    for m in rows:
        print(m.row())

    base, best = rows[1], rows[3]
    print(
        f"\n纯向量 → 全套: Hit@1 {base.hit1:.2f}→{best.hit1:.2f}, "
        f"MRR {base.mrr:.3f}→{best.mrr:.3f}"
    )

    out = Path("eval_report.json")
    out.write_text(
        json.dumps([m.__dict__ for m in rows], ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"结果已写入 {out}")


if __name__ == "__main__":
    main()

注意这里四个方案共用同一次检索调用:bm 和 vec 各算一遍,后面三个方案都从它们派生。不要为每个方案重跑一次检索——不但慢 4 倍,而且如果中间有任何随机性,四个方案就不在同一个基线上了。

✅ Checkpoint 8:PYTHONPATH=. uv run python docqa/evaluate.py

评测集 24 条 / 知识库 20 chunk

| 方案                   | Hit@1 | R@3  | R@5  | MRR   |
| ---------------------- | ----- | ---- | ---- | ----- |
| ① 纯 BM25               | 0.75 | 0.96 | 0.96 | 0.847 |
| ② 纯向量                  | 0.83 | 0.88 | 0.96 | 0.880 |
| ③ 混合 + RRF             | 0.83 | 0.92 | 1.00 | 0.896 |
| ④ 混合 + RRF + rerank    | 0.96 | 1.00 | 1.00 | 0.979 |

纯向量 → 全套: Hit@1 0.83→0.96, MRR 0.880→0.979
结果已写入 eval_report.json

这张表是本章最重要的产出,值得逐行读:

• ① BM25 的 Recall@3 是 0.96,比向量的 0.88 还高。这条数字会让很多只用过向量库的人意外。BM25 不是"老掉牙的方案"——在专有名词密集的企业文档上,它非常能打。上来就只用向量的人,把免费的一半准确率扔掉了。
• ② 向量的 Hit@1 (0.83) 高于 BM25 (0.75),但 Recall@3 反而更低。两条路的强弱是互补的,不是"一个更好"。
• ③ 融合的价值主要在 Recall@5 = 1.00:只要给 5 个位置,标准答案一定在里面。Hit@1 没提升(仍是 0.83),这很正常——RRF 不判断内容对错,它只保证"该来的都来了"。
• ④ rerank 把 Hit@1 从 0.83 拉到 0.96,MRR 从 0.896 到 0.979。这一层才是真正做"判断"的。它能做到这一点的前提是③已经把正确答案捞进了候选集——没有③的 Recall@5=1.00,④再强也无从下手。

这就是分层的意义:融合负责"不漏",精排负责"排对",各自的指标不同,不能互相替代。

顺带说一句,如果你把 vectorstore.py 里的 QUERY_PREFIX 删掉再跑一遍这张表,就能亲眼看到那个"静默性能损失"值多少分。这才是评测集真正的用法——不是跑一次写进简历,而是每次改动都跑。

11. gate.py —— 让 CI 拦住变差的改动(20 分钟)

这个文件只有 51 行,但它是"做了个 demo"和"做了个工程"的分水岭。

没有它:你改分块大小、换 embedding 模型、调 RRF 的 k,全靠感觉,改完觉得"好像好点了"就合并。 有了它:每次改动都有数字说话,而且退步会被自动拦住。

"""质量闸门: 跑评测集, 指标低于阈值就以非 0 退出码结束(让 CI 变红)。"""

from __future__ import annotations

import argparse
import sys

from docqa.chunker import chunk_blocks
from docqa.evaluate import GOLD, score
from docqa.parser import parse_dir
from docqa.retriever import Retriever, rrf_fuse
from docqa.vectorstore import Embedder, VectorStore


def run() -> tuple[float, float]:
    chunks = chunk_blocks(parse_dir("docqa/data"))
    store = VectorStore(Embedder())
    r = Retriever(chunks, store)

    ranked = []
    for g in GOLD:
        q = g["q"]
        fused = rrf_fuse(r.bm25.search(q, top_k=10), store.search(q, top_k=10))[:20]
        sc = r.reranker.predict([(q, h.chunk.text) for h in fused])
        ranked.append(
            [h.chunk for h, _ in sorted(zip(fused, sc, strict=True), key=lambda x: -x[1])]
        )

    m = score("full", ranked)
    return m.hit1, m.mrr


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--min-hit1", type=float, default=0.90)
    ap.add_argument("--min-mrr", type=float, default=0.95)
    args = ap.parse_args()

    hit1, mrr = run()
    ok = hit1 >= args.min_hit1 and mrr >= args.min_mrr
    print(f"\nHit@1 = {hit1:.3f} (阈值 {args.min_hit1})")
    print(f"MRR   = {mrr:.3f} (阈值 {args.min_mrr})")
    print("闸门:", "通过" if ok else "不通过 —— 检索质量退步, 拒绝合并")

    # 失败必须非 0 退出, 否则 CI 会当成成功
    sys.exit(0 if ok else 1)

sys.exit(0 if ok else 1) 这一行是整个文件存在的理由。 打印"不通过"是给人看的,CI 只看退出码。少了这一行,你的闸门会在 GitHub Actions 里永远显示绿灯,而它其实什么都没拦住——这是一个特别隐蔽的失效,因为一切"看起来都在跑"。

阈值定 0.90 / 0.95 而不是 1.00,是因为闸门要拦住退步,不是追求满分。定成 1.00 的话,任何一条 gold 标注得稍微含糊一点就会让 CI 永久变红,团队第二天就会把这个检查关掉。定在"当前水平略下方"是合理的做法:当前 0.958 / 0.979,阈值 0.90 / 0.95,留出正常波动的余量,真掉下去了一定是出了事。

✅ Checkpoint 9:验证闸门两个方向都真的有效

# 1) 当前质量, 应该通过
PYTHONPATH=. uv run python docqa/gate.py

Hit@1 = 0.958 (阈值 0.9)
MRR   = 0.979 (阈值 0.95)
闸门: 通过

# 2) 把阈值抬到不可能达到的高度, 应该不通过
PYTHONPATH=. uv run python docqa/gate.py --min-hit1 0.99 --min-mrr 0.99

Hit@1 = 0.958 (阈值 0.99)
MRR   = 0.979 (阈值 0.99)
闸门: 不通过 —— 检索质量退步, 拒绝合并

# 3) 关键: 确认退出码真的是非 0
PYTHONPATH=. uv run python docqa/gate.py --min-hit1 0.99 --min-mrr 0.99 >/dev/null 2>&1; echo $?

1

第 3 步不能省。 我自己第一次验证时写的是 python docqa/gate.py ... | tail -4; echo $?,得到 0,差点以为闸门坏了——$? 报的是管道最后一个命令(tail)的退出码,不是脚本的。改成重定向到 /dev/null 再看 $?,才拿到真实的 1。

这是一个和 agent 开发无关、但会反复咬你的 shell 常识:管道里的退出码默认只反映最后一环。想让管道透传前面的失败,得用 set -o pipefail。

12. db.py —— 从 JSON 文件换成真数据库(1 小时)

docs/1 里所有项目都用 JSON 文件落盘,docs/1/13 自己也承认"多 worker 下会写坏"。这一节换成 Postgres,顺便学 SQLAlchemy 2.0 的现代写法。

两张表的职责要分清:

• documents:知识库里有什么。哈希用来判断"这份文件是否已经索引过",避免重复入库
• qa_logs:每一次问答的完整现场(问题/答案/引用了哪些 chunk/耗时/token/是否命中缓存)。这张表就是后面做 bad case 分析和成本核算的原料

"""Postgres 持久化: 文档登记表 + 问答日志表。"""

from __future__ import annotations

import hashlib
import os
from datetime import UTC, datetime

from sqlalchemy import (
    JSON, DateTime, ForeignKey, Integer, Numeric, String, Text,
    create_engine, func, select,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship

DB_URL = os.getenv("DB_URL", "postgresql+psycopg://postgres:docqa@localhost:5432/docqa")


class Base(DeclarativeBase):
    pass


class Document(Base):
    __tablename__ = "documents"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255), unique=True)
    sha256: Mapped[str] = mapped_column(String(64), index=True)
    pages: Mapped[int] = mapped_column(Integer)
    chunks: Mapped[int] = mapped_column(Integer)
    indexed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    logs: Mapped[list[QaLog]] = relationship(back_populates="document")


class QaLog(Base):
    __tablename__ = "qa_logs"

    id: Mapped[int] = mapped_column(primary_key=True)
    question: Mapped[str] = mapped_column(Text)
    answer: Mapped[str] = mapped_column(Text)
    citations: Mapped[list] = mapped_column(JSON, default=list)
    # 有答案 / 文档中未提及 / 出错 —— 后两种是要定期人工看的
    verdict: Mapped[str] = mapped_column(String(32))
    latency_ms: Mapped[int] = mapped_column(Integer)
    prompt_tokens: Mapped[int] = mapped_column(Integer, default=0)
    completion_tokens: Mapped[int] = mapped_column(Integer, default=0)
    cost_yuan: Mapped[float] = mapped_column(Numeric(10, 6), default=0)
    cached: Mapped[int] = mapped_column(Integer, default=0)
    document_id: Mapped[int | None] = mapped_column(ForeignKey("documents.id"), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    document: Mapped[Document | None] = relationship(back_populates="logs")

Mapped[int] / mapped_column() 是 SQLAlchemy 2.0 的写法,比 1.x 的 Column(Integer) 好在类型注解就是 schema 的一部分,mypy 和 IDE 都能看懂。Mapped[int | None] 自动推出 nullable=True(这里还是显式写了一遍,更清楚)。

三个字段设计值得说:

• cost_yuan 用 Numeric(10, 6) 而不是 Float。钱永远不要用浮点存。单次问答的成本是 0.000816 元这个量级,累加几万次之后浮点误差会真的出现在账上。
• verdict 是三值的:answered / not_found / error。这是本章反复强调的设计——把"没答案"和"出错"从"答了"里分出来,运营才知道该看哪些。全都记成"答了"的日志表,等于没记。
• cached 记进日志。命中缓存的请求成本是 0,不区分的话你的成本统计会虚高好几倍;反过来,缓存命中率也是从这个字段算出来的。

_engine = create_engine(DB_URL, pool_pre_ping=True, future=True)


def init_db() -> None:
    Base.metadata.create_all(_engine)


def session() -> Session:
    return Session(_engine)


def file_sha256(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for blk in iter(lambda: f.read(1 << 20), b""):
            h.update(blk)
    return h.hexdigest()


def register_document(name: str, sha: str, pages: int, chunks: int) -> tuple[int, bool]:
    """返回 (id, 是否需要重新索引)。哈希没变就跳过 —— 省掉一次全量向量化。"""
    with session() as s:
        doc = s.scalar(select(Document).where(Document.name == name))
        if doc and doc.sha256 == sha:
            return doc.id, False
        if doc:
            doc.sha256, doc.pages, doc.chunks = sha, pages, chunks
            doc.indexed_at = datetime.now(UTC)
        else:
            doc = Document(name=name, sha256=sha, pages=pages, chunks=chunks)
            s.add(doc)
        s.commit()
        return doc.id, True


def log_qa(**kw) -> int:
    with session() as s:
        row = QaLog(**kw)
        s.add(row)
        s.commit()
        return row.id

pool_pre_ping=True 是生产必备:连接池里的连接可能已经被数据库那边超时掉了,不 ping 一下就会在业务代码里抛 OperationalError。加上它,SQLAlchemy 会先发一个探测,坏连接自动重建。

register_document 返回的 (id, need_reindex) 是增量索引的全部逻辑:文件哈希没变就返回 False,上层直接跳过向量化。语料只有 5 份 PDF 时这没什么意义;真实项目里知识库有几千份文档,每次重启都全量重新 embedding 是既慢又烧钱的。file_sha256 按 1MB 分块读,是为了不把一个 500MB 的 PDF 整个读进内存。

✅ Checkpoint 10:PYTHONPATH=. uv run python docqa/db.py

建表完成
首次登记: id=1 需要索引=True
哈希未变再登记: id=1 需要索引=False  (应为 False)
哈希变了再登记: id=1 需要索引=True  (应为 True)
写入问答日志 id=46
日志总数=16 累计花费=0.001632 元

三行 需要索引=True/False/True 是这一节唯一需要盯的东西:同一个文件名、同一个哈希 → 跳过;哈希变了 → 重建。注意 id=1 三次都一样,说明是更新而不是插新行,name 上那个 unique=True 在起作用。

日志总数=16 是我这台机器上历史累积的结果,你的数字会不同,不用对齐。想清空重来:

docker exec -it docqa-pg psql -U postgres -d docqa -c "TRUNCATE qa_logs, documents RESTART IDENTITY CASCADE;"

表名是 qa_logs 复数。我第一次写成 qa_log 得到 ERROR: relation "qa_log" does not exist——__tablename__ 写的什么就是什么,SQLAlchemy 不会帮你转复数。忘了的话 \dt 一下就能看到。

13. cache.py —— 挡在 LLM 前面的两道墙(1 小时)

缓存和限流放在同一个文件里,因为它们属于同一层:都是为了让请求不要真的打到 LLM。

• 缓存:一样的问题不要再花一次钱(企业文档问答的重复提问率非常高——新人入职那一周,"年假几天"这个问题会被问几十遍)
• 限流:防止一个用户把当月预算烧光

"""Redis 缓存 + 令牌桶限流。"""

from __future__ import annotations

import hashlib
import json
import os
import time

import redis

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
_r = redis.from_url(REDIS_URL, decode_responses=True)

TTL = 3600  # 1 小时; 知识库变更时用 kb_version 让旧缓存自然失效


def kb_version(chunk_count: int, doc_shas: list[str]) -> str:
    h = hashlib.sha256()
    h.update(str(chunk_count).encode())
    for s in sorted(doc_shas):
        h.update(s.encode())
    return h.hexdigest()[:12]


def _key(question: str, version: str) -> str:
    q = question.strip().lower()
    return f"docqa:ans:{version}:{hashlib.sha256(q.encode()).hexdigest()[:16]}"


def get_cached(question: str, version: str) -> dict | None:
    raw = _r.get(_key(question, version))
    return json.loads(raw) if raw else None


def set_cached(question: str, version: str, payload: dict, ttl: int = TTL) -> None:
    # redis-py 5+ 把 setex 标记为 deprecated, 用 set(ex=...) 是现在的写法
    _r.set(_key(question, version), json.dumps(payload, ensure_ascii=False), ex=ttl)

缓存 key 里那个 version 是本节最重要的设计。只用问题做 key 是一个经典 bug:文档更新了,用户还在拿旧答案,而且 TTL 一小时意味着这个错误答案会持续一小时。加上知识库版本之后,只要有任何一份文档变了,kb_version 就变,所有旧 key 自然不再被命中——不需要主动清缓存,不需要遍历删除,旧数据自己就"过期"了。

kb_version 里 for s in sorted(doc_shas) 的 sorted 也不是装饰:目录遍历顺序在不同机器、不同文件系统上不一样,不排序的话同一个知识库在两台机器上会算出不同版本,缓存永远互相不命中。这和第 6 节 chunk id 的坑是同一类错误:让结果依赖了遍历顺序。

_key 里 question.strip().lower() 做归一化,所以 "满5年年假多少天" 和 " 满5年年假多少天 " 是同一个 key。这是一个廉价的命中率提升——用户复制粘贴问题时经常带空格。

class RateLimiter:
    """令牌桶。capacity 是瞬时可爆发的次数, refill 是每秒补多少个令牌。

    用 Lua 保证"读余额 → 补令牌 → 扣减"三步原子 ——
    分开写三条 Redis 命令在并发下会漏放请求, 这是限流实现最常见的错。
    """

    LUA = """
    local key = KEYS[1]
    local cap = tonumber(ARGV[1])
    local refill = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local tokens = tonumber(redis.call('HGET', key, 'tokens') or cap)
    local ts = tonumber(redis.call('HGET', key, 'ts') or now)
    tokens = math.min(cap, tokens + (now - ts) * refill)
    local allowed = 0
    if tokens >= 1 then
      tokens = tokens - 1
      allowed = 1
    end
    redis.call('HSET', key, 'tokens', tokens, 'ts', now)
    redis.call('EXPIRE', key, 3600)
    return {allowed, math.floor(tokens * 100)}
    """

    def __init__(self, capacity: int = 5, refill_per_sec: float = 0.5) -> None:
        self.cap = capacity
        self.refill = refill_per_sec
        self.script = _r.register_script(self.LUA)

    def allow(self, user: str) -> tuple[bool, float]:
        allowed, tokens = self.script(
            keys=[f"docqa:rl:{user}"], args=[self.cap, self.refill, time.time()]
        )
        return bool(allowed), tokens / 100

为什么要用 Lua? 因为令牌桶是"读余额 → 按时间补令牌 → 扣 1 个 → 写回"这四步。用 Python 分开发四条 Redis 命令,两个并发请求会这样交错:

请求 A: HGET tokens → 1
请求 B: HGET tokens → 1     ← B 读到了 A 还没扣的余额
请求 A: HSET tokens = 0
请求 B: HSET tokens = 0
结果: 余额只有 1, 却放过了 2 个请求

这是限流实现里最常见的错,而且在低并发下永远不会暴露——你本地手点几次全是对的,上线遇到突发流量就漏放。Redis 执行 Lua 脚本是单线程原子的,四步要么全做要么全不做。

math.floor(tokens * 100) 是因为 Redis 的 Lua 返回值只支持整数,而余额是小数(补令牌是按时间连续计算的)。乘 100 传出来、Python 端再除回去,是个常用的土办法。

refill 和 capacity 分开两个参数,表达的是两件不同的事:capacity 是允许多大的瞬时爆发,refill 是长期平均速率。capacity=5, refill=0.5 意思是"平时每 2 秒 1 次,但允许一下子来 5 次"——用户在页面上快速点几次不该被拦,持续刷才该被拦。

✅ Checkpoint 11:PYTHONPATH=. uv run python docqa/cache.py

Redis ping: True
知识库版本: 7a7f761098ef
命中: {'answer': '15 天。', 'citations': []}
大小写/空格归一化后仍命中: True
换个知识库版本就不命中: None

令牌桶 capacity=3, refill=0.5/s, 连打 5 次:
  第1次: 通过  余额=2.00
  第2次: 通过  余额=1.00
  第3次: 通过  余额=0.00
  第4次: 拒绝(429)  余额=0.00
  第5次: 拒绝(429)  余额=0.00
等 2.1 秒后再试: 通过  余额=0.05

换个知识库版本就不命中: None 这一行是缓存这部分的验收点——知识库变了必须失效。

令牌桶的输出正好演示了容量和速率的关系:capacity=3 所以前 3 次全通过、余额从 2 递减到 0,第 4、5 次被拒;等 2.1 秒、按 0.5/s 补了约 1.05 个令牌,第 6 次放行,余额剩 0.05。数字对得上,说明补令牌的时间计算是准的。

14. tracing.py —— 一次问答的时间花在哪(1.5 小时)

为什么自己写而不是直接上 Langfuse?因为你得先明白"trace / span / 属性"这三个概念是怎么串起来的。 想清楚之后,换成 Langfuse 或 OpenTelemetry 只是换个 exporter,概念完全一样。本节最后给出换 Langfuse 的方法。

目标是让一次问答的耗时长这样:

ask (2340ms)
├─ cache.get      (2ms)   hit=false
├─ retrieve       (410ms) candidates=20 kept=4
│  ├─ bm25        (12ms)
│  ├─ vector      (98ms)
│  └─ rerank      (295ms)
├─ llm            (1890ms) tokens=568 cost=0.000712
└─ persist        (18ms)

"""可观测性: 极简 span 追踪器(JSONL 落盘 + 控制台树形打印)。"""

from __future__ import annotations

import json
import time
import uuid
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from pathlib import Path

TRACE_FILE = Path("traces.jsonl")


@dataclass
class Span:
    name: str
    trace_id: str
    span_id: str
    parent_id: str | None
    start: float
    end: float | None = None
    attrs: dict = field(default_factory=dict)

    @property
    def ms(self) -> int:
        return int(((self.end or time.perf_counter()) - self.start) * 1000)


@dataclass
class Trace:
    trace_id: str
    spans: list[Span] = field(default_factory=list)

    def tree(self) -> str:
        by_parent: dict[str | None, list[Span]] = {}
        for s in self.spans:
            by_parent.setdefault(s.parent_id, []).append(s)

        lines: list[str] = []

        def walk(pid: str | None, depth: int) -> None:
            for s in by_parent.get(pid, []):
                pad = "   " * depth + ("└─ " if depth else "")
                extra = " ".join(f"{k}={v}" for k, v in s.attrs.items())
                lines.append(f"{pad}{s.name:12s} {s.ms:5d}ms  {extra}")
                walk(s.span_id, depth + 1)

        walk(None, 0)
        return "\n".join(lines)

Span 存的是 parent_id 而不是 children 列表,这样记录时不需要知道自己有没有子节点,树形结构在打印时才重建。这是所有分布式追踪系统的做法(OpenTelemetry 也一样),原因是 span 会跨进程、跨服务发送,发的时候子节点可能还没发生。

time.perf_counter() 而不是 time.time():后者是墙上时钟,会被 NTP 校时往回拨,算出负的耗时。测时间间隔一律用 perf_counter。

_current: ContextVar[Trace | None] = ContextVar("trace", default=None)
_parent: ContextVar[str | None] = ContextVar("parent", default=None)


@contextmanager
def trace(name: str):
    t = Trace(trace_id=uuid.uuid4().hex[:16])
    tok_t, tok_p = _current.set(t), _parent.set(None)
    try:
        with span(name):
            yield t
    finally:
        _current.reset(tok_t)
        _parent.reset(tok_p)
        with TRACE_FILE.open("a", encoding="utf-8") as f:
            for s in t.spans:
                f.write(
                    json.dumps(
                        {
                            "trace_id": s.trace_id,
                            "span_id": s.span_id,
                            "parent_id": s.parent_id,
                            "name": s.name,
                            "ms": s.ms,
                            **s.attrs,
                        },
                        ensure_ascii=False,
                    )
                    + "\n"
                )


@contextmanager
def span(name: str, **attrs):
    t = _current.get()
    if t is None:  # 不在 trace 里就静默降级, 不要因为埋点崩掉业务
        yield _Noop()
        return
    s = Span(
        name,
        t.trace_id,
        uuid.uuid4().hex[:8],
        _parent.get(),
        time.perf_counter(),
        attrs=dict(attrs),
    )
    t.spans.append(s)
    tok = _parent.set(s.span_id)
    try:
        yield s
    finally:
        s.end = time.perf_counter()
        _parent.reset(tok)


class _Noop:
    attrs: dict = {}

    def __setitem__(self, k, v) -> None:
        pass


def tag(sp, **kw) -> None:
    if hasattr(sp, "attrs"):
        sp.attrs.update(kw)

ContextVar 而不是全局变量,是这个文件唯一真正难的地方。

如果用普通的模块级全局变量存"当前 span 是谁",在 FastAPI 里会立刻出事:两个并发请求跑在同一个进程里,A 的 span 会挂到 B 的 trace 下面,树形结构变成一团乱麻。ContextVar 的语义是"每个执行上下文(线程 / asyncio task)各有一份",天然做到请求之间隔离,而且不需要把 trace 对象一层层当参数传下去——业务函数签名保持干净。

_parent.reset(tok) 那个 token 机制也要理解:ContextVar.set() 返回一个 token,reset(token) 精确回到设置前的值。不能写成 _parent.set(旧值),因为嵌套 span 退出的顺序和你以为的可能不一样(异常路径),token 机制是唯一可靠的还原方式。

if t is None: yield _Noop() 是刻意的降级:埋点绝不能崩掉业务。有人在单元测试里直接调一个内部函数、没包 trace(...),这时候 span() 静默变成空操作,而不是抛 NoneType has no attribute。_Noop 连 attrs 和 __setitem__ 都实现了,让 tag() 也安全。

✅ Checkpoint 12:PYTHONPATH=. uv run python docqa/tracing.py

trace_id=80e20410a37d4bc3

ask            212ms
   └─ cache.get        3ms  hit=False
   └─ retrieve       113ms  candidates=20 kept=4
      └─ bm25            15ms
      └─ vector          40ms
      └─ rerank          58ms
   └─ llm             90ms  tokens=568 cost=0.000712
   └─ persist          6ms

span 数=8, 已追加到 traces.jsonl

这个 demo 用 time.sleep 模拟耗时,所以数字比真实的小得多,但层级关系必须完全一致:bm25/vector/rerank 缩进在 retrieve 下面,retrieve 和 llm 是兄弟。如果你的输出全部平铺在同一层,是 _parent 那个 ContextVar 没正确嵌套。

span 数=8 = 1 个 ask + 7 个子 span。

看一眼落盘的 JSONL:

tail -3 traces.jsonl | jq -c .

{"trace_id":"80e20410a37d4bc3","span_id":"...","parent_id":"...","name":"llm","ms":90,"tokens":568,"cost":0.000712}

注意 attrs 是被 **s.attrs 摊平到顶层的,不是嵌在 attrs 字段里,而耗时字段叫 ms 不是 duration_ms。我自己后面用 jq '.duration_ms' 查这个文件,拿到一串 null,查了半天才想起来是自己定的 schema。自己写的 exporter 就要自己记住 schema——这也是用现成方案的一个理由。

换成 Langfuse 只需要改这里

真实项目里你大概不会自己维护追踪器。把 span() 换成 Langfuse 的装饰器就行,概念一一对应:

from langfuse.decorators import observe

@observe()
def retrieve(query: str): ...

trace → Langfuse 的 trace,span → span/generation,tag() → update_current_observation(metadata=...)。换掉的是 30 行代码,不是你的架构——这就是先手写一遍的价值:你知道它在做什么,所以换的时候不慌。

15. answerer.py —— 三条防幻觉的硬约束(1.5 小时)

到这里检索已经很准了,但准确的检索 + 随便写的 prompt = 依然会胡说。这一节的核心不是调 prompt,是三条结构性的约束:

1. 检索为空 → 直接返回"文档中未提及",一次 LLM 都不调(省钱,也不给模型编造的机会)
2. prompt 里每条材料带编号,要求答案用 [1][2] 标注;没有标注的答案视为不可信
3. temperature=0 —— 事实问答不需要创造力

"""生成答案: 只依据检索到的 chunk 回答, 强制带引用, 并统计 token 与成本。"""

from __future__ import annotations

import os
import re
import time
from dataclasses import dataclass, field

from docqa.retriever import Hit

SYSTEM = """你是企业知识库问答助手。严格遵守:
1. 只依据【参考材料】回答, 不使用任何材料之外的知识
2. 材料里没有的信息, 直接回答"文档中未提及", 不要推测
3. 每个结论后面用 [编号] 标注来源, 可以标多个
4. 材料之间冲突时(例如新旧版本、不同适用人群), 必须指出冲突并说明各自适用范围, 不要只挑一个
5. 回答简洁, 不复述问题"""

# 千分之一元每千 token 的量级; 换模型记得改这两个数
PRICE_IN = 0.001 / 1000  # 元 / token
PRICE_OUT = 0.002 / 1000


@dataclass
class Answer:
    text: str
    verdict: str  # answered | not_found | error
    citations: list[dict] = field(default_factory=list)
    prompt_tokens: int = 0
    completion_tokens: int = 0
    cost_yuan: float = 0.0
    latency_ms: int = 0
    cached: bool = False

SYSTEM 里第 4 条是本章从头到尾在铺垫的那件事:知识库里就是有冲突,模型的任务不是替用户拍板,是把冲突摊开。检索层再怎么调,也不可能一定选对"实习生"还是"正式员工"、"新版"还是"旧版"——第 7、8 节的实测里两条路都把旧版排在前面过。把这个问题上交给用户,比让模型猜一个然后自信地答出来要安全得多。

verdict 三值和第 12 节的日志表对齐:answered / not_found / error。这个字段是运营抓手:每周只看 not_found 就知道知识库缺什么文档,只看 error 就知道系统哪里在坏。

def build_prompt(question: str, hits: list[Hit]) -> tuple[str, list[dict]]:
    parts, cites = [], []
    for i, h in enumerate(hits, 1):
        c = h.chunk
        parts.append(f"[{i}] 出处: {c.cite}\n{c.raw}")
        cites.append({"n": i, "doc": c.doc, "page": c.page, "heading": c.heading, "kind": c.kind})
    return "参考材料:\n\n" + "\n\n".join(parts) + f"\n\n问题: {question}", cites


NOT_FOUND = "文档中未提及。知识库里没有检索到与该问题相关的内容, 请补充文档或换个说法提问。"

build_prompt 同时返回 prompt 和 citations 列表,这个设计是引用能对上的关键:编号 [1] 在 prompt 里和在 API 响应里是同一个 i,一次循环生成两边,不可能错位。如果分成两个函数各自遍历一遍,哪天有人在其中一个里加了过滤条件,编号就悄悄错开了——而用户看到的引用指向错误的文档,比没有引用更糟。

材料里给的是 c.raw 不是 c.text:上下文前缀是给检索用的,送进 LLM 会变成噪音(每条材料开头都重复一遍文档名和标题)。出处信息已经在 [i] 出处: 那一行里了。

def answer(
    question: str,
    hits: list[Hit],
    dry: bool = False,
    model: str = "deepseek-chat",
) -> Answer:
    t0 = time.perf_counter()

    if not hits:
        return Answer(NOT_FOUND, "not_found", latency_ms=int((time.perf_counter() - t0) * 1000))

    prompt, cites = build_prompt(question, hits)

    if dry:
        # 离线模式: 把命中材料的首句拼出来 + 自动补引用编号,
        # 用来验证"引用编号、verdict、耗时统计"这些非 LLM 逻辑。
        # token 一律记 0 —— 用字符数冒充 token 会污染成本统计, 这种"看起来有数据"
        # 的假指标比没有数据更危险
        sents = []
        for i, h in enumerate(hits, 1):
            first = re.split(r"(?<=[。;;])", h.chunk.raw.replace("[表格] ", ""))[0]
            sents.append(f"{first.strip()}[{i}]")
        return Answer(
            "(dry-run) " + " ".join(sents),
            "answered",
            cites,
            latency_ms=int((time.perf_counter() - t0) * 1000),
        )

if not hits 这三行是整套系统里性价比最高的防幻觉措施。它和第 9 节的 min_rerank=0.05 是一对:阈值让检索层敢返回空,这里让空结果直接短路。代价是三行代码,收益是"知识库里没有的东西永远不会被编出来"。

dry 模式的存在是为了让没有 API Key 的人也能跑通整条链路。它验证的是引用编号、verdict 判定、耗时统计这些确定性逻辑——占这个文件代码量的一大半。

dry 模式里 token 一律记 0,这一点我要单独强调:不要用字符数除以 1.5 之类的公式去"估算" token。看起来有数据、实际上是假的指标,比没有数据危险得多——你会拿它做成本预算,然后在真实账单面前傻掉。测试里也钉死了这条:

def test_dry_run_reports_zero_tokens():
    a = answer("问题", [Hit(mk_chunk("0", "内容甲。内容乙。"), 1.0)], dry=True)
    assert a.prompt_tokens == 0 and a.cost_yuan == 0, "dry-run 不能伪造 token 数"
    assert "[1]" in a.text

    try:
        from openai import OpenAI

        client = OpenAI(
            api_key=os.environ["LLM_API_KEY"],
            base_url=os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1"),
        )
        resp = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": SYSTEM},
                {"role": "user", "content": prompt},
            ],
            temperature=0,
        )
        text = resp.choices[0].message.content or ""
        u = resp.usage
        pt, ct = (u.prompt_tokens, u.completion_tokens) if u else (0, 0)
        verdict = "not_found" if "文档中未提及" in text else "answered"
        return Answer(
            text,
            verdict,
            cites,
            prompt_tokens=pt,
            completion_tokens=ct,
            cost_yuan=pt * PRICE_IN + ct * PRICE_OUT,
            latency_ms=int((time.perf_counter() - t0) * 1000),
        )
    except Exception as e:
        return Answer(
            f"生成失败: {type(e).__name__}: {e}",
            "error",
            cites,
            latency_ms=int((time.perf_counter() - t0) * 1000),
        )

几个细节:

• os.environ["LLM_API_KEY"] 用方括号,不是 .get()。Key 没配就应该抛 KeyError,而不是拿着 None 去调 API 然后收一个看不懂的 401。配置缺失要在最早的地方响亮地失败。
• token 数来自 resp.usage,不自己数。这是唯一权威的数字,而且 API 会把 prompt caching 的折扣反映在里面。
• verdict 从答案文本反推。模型按 SYSTEM 第 2 条回答"文档中未提及"时,虽然走的是 answered 这条代码路径,实际上是 not_found。这是一个不优雅但有效的判定——检索给了材料、模型仍然说不知道,这个 case 和检索为空一样值得运营去看。
• except Exception 兜底返回 verdict="error",而不是让异常冒到上层。API 超时、限流、余额不足都会走到这里;把错误变成一个正常的 Answer 对象,上层的日志、追踪、返回码逻辑就不需要写两套。

✅ Checkpoint 13:PYTHONPATH=. uv run python docqa/answerer.py

没配 LLM_API_KEY 时自动走 dry-run:

模式: dry-run(无 Key)

Q: 实习生有年假吗
   verdict=answered  引用=4  耗时=0ms
   (dry-run) 实习生不享受年假, 按实际出勤天数计算实习补贴。[1] 假期类型是事假, 额度是每月不超过 2 天, 是否带薪是否。[2] 实习生实行固定工作时间, 每个工作日 9:30 至 18:30, 不适用弹性工作制。[3] 入职年限是满 1 年不满 3 年, 年假天数是5 天, 是否可结转是否。[4]
     [1] 实习生管理办法 p1 · 第二章假期管理
     [2] 实习生管理办法 p1 · 第二章假期管理
     [3] 实习生管理办法 p1 · 第一章考勤制度
     [4] 员工手册 p1 · 第二章假期管理

Q: 公司的年终奖怎么发
   verdict=not_found  引用=0  耗时=0ms
   文档中未提及。知识库里没有检索到与该问题相关的内容, 请补充文档或换个说法提问。

第一个问题验证了三件事:检索把实习生那份排在了前 3(干扰文档没抢到第一)、引用编号 1~4 和材料一一对应、[4] 确实指向员工手册(那是一条相关但不适用的材料,交给 LLM 去说明"那是正式员工的规定")。第二个问题验证了空检索短路:verdict=not_found、引用=0、没有调用 LLM。

真实 API 的验收标准(本文档不贴输出)

配了 Key 之后跑同样的命令会走真实 API。这一步的输出我没有实测,所以不在这里贴——本机的 .env 读取被权限规则挡住了,而伪造一段"看起来对"的模型输出违反本教程的基本原则。给你验收标准,自己跑完对照:

配置(.env,别提交进 git):

LLM_API_KEY=<你的 key>
LLM_BASE_URL=https://api.deepseek.com/v1

跑 实习生有年假吗,你的输出必须满足:

• verdict=answered,引用=4
• 答案里至少出现一个 [数字] 标注;一个都没有说明 SYSTEM 第 3 条没起作用,换个模型或把要求写进 user message 再试
• 答案的核心事实是"实习生不享受年假",并且提到了正式员工的年假规定不适用于实习生(这是 SYSTEM 第 4 条在起作用)
• prompt_tokens 在 400~700 之间(4 条材料的量级),cost_yuan 在 0.0005~0.001 元这个数量级
• 耗时 在 1000~4000ms

跑 公司的年终奖怎么发:输出必须和 dry-run 完全一样(verdict=not_found、引用=0),因为这条根本不会走到 API。如果配了 Key 之后这一条的行为变了,说明你的空检索短路写漏了。

再跑一条冲突问题 企业版P1故障多久响应,验收标准是:答案里同时出现 30 分钟和 2 小时两个数字,并说明前者来自现行政策、后者来自 2024 旧版(已废止)。只答一个数字的,就是把冲突吞掉了——哪怕答对了现行的那个,这个行为也不合格,因为用户无从判断。

16. indexer.py —— 把离线链路串起来(30 分钟)

前面九个文件各跑各的 __main__,这一节把它们串成一条流水线:PDF 目录 → 解析 → 分块 → 登记入库 → 写向量库。

真实项目里这是一个独立的离线任务(定时跑,或者上传文件触发),不放在 Web 进程里。

"""索引流水线: PDF 目录 → 解析 → 分块 → 登记入库 → 写向量库。"""

from __future__ import annotations

import argparse
from pathlib import Path

import pymupdf

from docqa.chunker import Chunk, chunk_blocks
from docqa.db import file_sha256, init_db, register_document
from docqa.parser import parse_pdf
from docqa.vectorstore import Embedder, VectorStore


def build_index(
    data_dir: str = "docqa/data", force: bool = False
) -> tuple[list[Chunk], list[str]]:
    init_db()
    emb = Embedder()
    store = VectorStore(emb)

    if force or not store.client.collection_exists("docqa"):
        store.recreate()
        print("[index] collection 已重建")

    all_chunks: list[Chunk] = []
    shas: list[str] = []
    for pdf_path in sorted(Path(data_dir).glob("*.pdf")):
        sha = file_sha256(str(pdf_path))
        shas.append(sha)
        blocks = parse_pdf(pdf_path)
        chunks = chunk_blocks(blocks)
        all_chunks.extend(chunks)

        pages = pymupdf.open(pdf_path).page_count
        _did, need = register_document(pdf_path.stem, sha, pages, len(chunks))
        if need or force:
            n = store.index(chunks)
            print(
                f"[index] {pdf_path.name}: {len(blocks)} 块 → {len(chunks)} chunk, 写入 {n} 个向量"
            )
        else:
            print(
                f"[index] {pdf_path.name}: 哈希未变, 跳过向量化({len(chunks)} chunk 仅载入内存)"
            )

    return all_chunks, shas

注意"跳过向量化"和"跳过解析"是两件不同的事。 即使哈希没变,我们仍然要解析和分块——因为 BM25 索引在内存里,进程重启后必须重建。跳过的只是 embedding 那一步(慢、烧钱),这是 build_index 返回 all_chunks 而不是只返回统计数字的原因。

返回值里的 shas 是给第 13 节 kb_version() 用的:知识库版本 = chunk 数 + 所有文件哈希。sorted(Path(...).glob(...)) 保证遍历顺序稳定,和 kb_version 里的 sorted(doc_shas) 是双重保险。

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--force", action="store_true", help="强制重建 collection 并全量重新索引")
    ap.add_argument("--data", default="docqa/data")
    args = ap.parse_args()

    chunks, shas = build_index(args.data, force=args.force)
    print(f"\n合计 {len(chunks)} chunk, {len(shas)} 份文档")
    print("提示: 不加 --force 再跑一次, 应该全部显示'哈希未变, 跳过'")

✅ Checkpoint 14:跑两次,第二次必须全部跳过

PYTHONPATH=. uv run python docqa/indexer.py --force

[index] collection 已重建
[index] 产品技术白皮书.pdf: 4 块 → 4 chunk, 写入 4 个向量
[index] 员工手册.pdf: 4 块 → 4 chunk, 写入 4 个向量
[index] 售后服务政策-2024旧版.pdf: 4 块 → 4 chunk, 写入 4 个向量
[index] 售后服务政策.pdf: 4 块 → 4 chunk, 写入 4 个向量
[index] 实习生管理办法.pdf: 4 块 → 4 chunk, 写入 4 个向量

合计 20 chunk, 5 份文档

PYTHONPATH=. uv run python docqa/indexer.py

[index] 产品技术白皮书.pdf: 哈希未变, 跳过向量化(4 chunk 仅载入内存)
[index] 员工手册.pdf: 哈希未变, 跳过向量化(4 chunk 仅载入内存)
[index] 售后服务政策-2024旧版.pdf: 哈希未变, 跳过向量化(4 chunk 仅载入内存)
[index] 售后服务政策.pdf: 哈希未变, 跳过向量化(4 chunk 仅载入内存)
[index] 实习生管理办法.pdf: 哈希未变, 跳过向量化(4 chunk 仅载入内存)

合计 20 chunk, 5 份文档

两次的 合计 20 chunk 必须一样——跳过向量化不等于跳过分块。第二次还在写向量的话,检查 register_document 的哈希比较;第二次 chunk 数变成 0 的话,你把 continue 写在了错误的位置。

17. main.py —— FastAPI 服务与生产配件(2.5 小时)

这一节把所有东西挂到一条请求链路上:

鉴权 → 限流 → 缓存 → 检索 → 生成 → 落库 → trace

每一层都能单独关掉验证,也都能在 /api/trace/{id} 里看到耗时。

"""FastAPI 服务: 带 API Key 鉴权、限流、缓存、SSE 流式、trace、DB 日志。"""

from __future__ import annotations

import json
import os
import queue
import threading
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy import func, select

from docqa.answerer import answer
from docqa.cache import RateLimiter, get_cached, kb_version, set_cached
from docqa.db import QaLog, init_db, log_qa, session
from docqa.indexer import build_index
from docqa.retriever import Retriever
from docqa.tracing import TRACE_FILE, span, tag, trace
from docqa.vectorstore import Embedder, VectorStore

STATE: dict = {}
limiter = RateLimiter(capacity=10, refill_per_sec=1.0)


def _keys() -> set[str]:
    raw = os.getenv("DOCQA_API_KEYS", "")
    return {k.strip() for k in raw.split(",") if k.strip()}


@asynccontextmanager
async def lifespan(app: FastAPI):
    if not _keys():
        raise RuntimeError("未配置 DOCQA_API_KEYS, 拒绝启动(不允许无鉴权运行)")
    init_db()
    chunks, shas = build_index()
    STATE["retriever"] = Retriever(chunks, VectorStore(Embedder()))
    STATE["version"] = kb_version(len(chunks), shas)
    STATE["chunks"] = len(chunks)
    print(f"[boot] 知识库就绪: {len(chunks)} chunk, version={STATE['version']}")
    yield
    STATE.clear()


app = FastAPI(title="企业文档问答", lifespan=lifespan)

if not _keys(): raise RuntimeError(...) 这一行是本章最重要的安全设计。

绝大多数教程会写成"没配 Key 就跳过鉴权",理由是"方便本地开发"。这是线上事故最常见的来源:开发时没配、代码走的是免鉴权分支,上线时环境变量名写错了一个字母、也走免鉴权分支,于是你的接口对全世界开放,而一切看起来都正常。

正确的做法是fail closed:配置缺失就拒绝启动。启动失败是刺眼的、5 秒钟就会被发现的错误;悄悄放开鉴权是几个月后才会被发现的错误。

lifespan 里做索引构建也是刻意的:模型加载和向量化都很慢(本机约 20~30 秒),必须在启动时一次做完,不能放在第一个请求里——否则第一个用户等 30 秒超时,而且并发的头几个请求会各自加载一遍模型,内存直接爆掉。

def auth(x_api_key: Annotated[str | None, Header()] = None) -> str:
    if not x_api_key or x_api_key not in _keys():
        raise HTTPException(401, "缺少或无效的 X-API-Key")
    return x_api_key


def rate_limit(user: Annotated[str, Depends(auth)]) -> str:
    ok, left = limiter.allow(user)
    if not ok:
        raise HTTPException(429, "请求过于频繁, 请稍后重试")
    return user


class AskRequest(BaseModel):
    question: str = Field(min_length=2, max_length=500)
    top_k: int = Field(default=4, ge=1, le=10)
    dry: bool = False


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "chunks": STATE.get("chunks", 0), "version": STATE.get("version")}

rate_limit 依赖 auth,这个依赖链是 FastAPI 的漂亮之处:业务函数只写 Depends(rate_limit),鉴权自动先跑,而且限流的 key 是鉴权返回的用户身份,不是 IP。按 IP 限流在企业内网里等于按整个公司限流(大家共用出口 IP),按 API Key 限流才有意义。

顺序也不能反:先鉴权再限流。反过来的话,未鉴权的请求也会消耗令牌桶,攻击者不需要有效 Key 就能把正常用户的配额刷掉。

AskRequest 的 min_length=2, max_length=500 和 le=10 是输入校验,不是装饰。没有上界的 top_k 意味着用户可以传 top_k=10000,让你把整个知识库塞进 prompt——一次请求烧掉几十块钱。所有来自外部的数值参数都要有上界,这是最基本的成本防护。

/health 不需要鉴权(负载均衡器要探活),但它只暴露 chunk 数和版本号,不暴露任何文档内容。

@app.post("/api/ask")
def ask(req: AskRequest, user: Annotated[str, Depends(rate_limit)]) -> dict:
    ver = STATE["version"]
    with trace("ask") as t:
        with span("cache.get") as s:
            hit = get_cached(req.question, ver)
            tag(s, hit=bool(hit))
        if hit:
            return {**hit, "cached": True, "trace_id": t.trace_id}

        with span("retrieve") as rs:
            hits = STATE["retriever"].retrieve(req.question, top_k=req.top_k)
            tag(rs, kept=len(hits))

        with span("llm") as ls:
            a = answer(req.question, hits, dry=req.dry)
            tag(
                ls,
                verdict=a.verdict,
                tokens=a.prompt_tokens + a.completion_tokens,
                cost=round(a.cost_yuan, 6),
            )

        payload = {
            "answer": a.text,
            "verdict": a.verdict,
            "citations": a.citations,
            "latency_ms": a.latency_ms,
        }
        with span("persist"):
            # 只缓存成功的回答; 出错的不缓存, 否则一次抖动会被记住一小时
            if a.verdict != "error":
                set_cached(req.question, ver, payload)
            log_qa(
                question=req.question,
                answer=a.text,
                citations=a.citations,
                verdict=a.verdict,
                latency_ms=a.latency_ms,
                prompt_tokens=a.prompt_tokens,
                completion_tokens=a.completion_tokens,
                cost_yuan=a.cost_yuan,
                cached=0,
            )
    return {**payload, "cached": False, "trace_id": t.trace_id}

if a.verdict != "error": set_cached(...) 这个判断很小但很关键:出错的答案绝不能进缓存。API 抖动一次、超时一次,如果被缓存了,这个"生成失败: Timeout"会被返回给之后一小时内问同样问题的所有人,而且你在日志里只会看到一次错误——排查时完全对不上。

not_found 反而应该缓存:知识库里确实没有,一小时内也不会有,重复检索是纯浪费。

trace_id 一定要返回给客户端。用户报"这个答案不对",你需要的第一件事就是 trace_id;没有它,你只能在几万条日志里靠问题文本去猜。

@app.post("/api/ask/stream")
def ask_stream(req: AskRequest, user: Annotated[str, Depends(rate_limit)]) -> StreamingResponse:
    q: queue.Queue = queue.Queue()

    def worker() -> None:
        try:
            with trace("ask.stream"):
                with span("retrieve") as rs:
                    hits = STATE["retriever"].retrieve(req.question, top_k=req.top_k)
                    tag(rs, kept=len(hits))
                q.put(
                    {
                        "type": "sources",
                        "items": [
                            {
                                "n": i,
                                "cite": h.chunk.cite,
                                "score": round(h.rerank_score or 0, 4),
                            }
                            for i, h in enumerate(hits, 1)
                        ],
                    }
                )
                with span("llm") as ls:
                    a = answer(req.question, hits, dry=req.dry)
                    tag(ls, verdict=a.verdict)
                q.put({"type": "answer", "text": a.text, "verdict": a.verdict})
                log_qa(...)  # 同上, 省略
        except Exception as e:
            q.put({"type": "error", "detail": f"{type(e).__name__}: {e}"})
        q.put(None)

    threading.Thread(target=worker, daemon=True).start()

    def events():
        while True:
            ev = q.get()
            if ev is None:
                break
            yield f"data: {json.dumps(ev, ensure_ascii=False)}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

流式接口的设计意图是先把检索到的出处推给前端,再推答案。因为检索只要几百毫秒,LLM 要几秒——用户能立刻看到"系统正在看这几份文档",体感完全不同,而且如果出处明显不对,用户不用等答案就知道该换个问法了。

为什么用 queue + 线程,而不是 async?因为 retrieve() 和 answer() 都是同步阻塞的(sentence-transformers 和 openai 同步客户端)。直接在 async 函数里调它们会阻塞整个事件循环,一个请求把所有并发请求卡住。丢到线程里跑、用队列把结果传回生成器,是同步库和异步框架之间最简单可靠的桥。

q.put(None) 是结束哨兵,而且它在 try/except 之外——无论成功还是异常都必须放,否则 events() 会永远卡在 q.get() 上,连接泄漏。

@app.get("/api/stats")
def stats(user: Annotated[str, Depends(auth)]) -> dict:
    with session() as s:
        total = s.scalar(select(func.count()).select_from(QaLog)) or 0
        spend = float(s.scalar(select(func.coalesce(func.sum(QaLog.cost_yuan), 0))) or 0)
        by_verdict = dict(
            s.execute(select(QaLog.verdict, func.count()).group_by(QaLog.verdict)).all()
        )
        p95 = s.scalar(select(func.percentile_cont(0.95).within_group(QaLog.latency_ms)))
    return {
        "total": total,
        "cost_yuan": round(spend, 6),
        "by_verdict": by_verdict,
        "latency_p95_ms": float(p95) if p95 is not None else None,
    }


@app.get("/api/trace/{trace_id}")
def get_trace(trace_id: str, user: Annotated[str, Depends(auth)]) -> dict:
    if not TRACE_FILE.exists():
        raise HTTPException(404, "无 trace 记录")
    spans = [
        json.loads(ln)
        for ln in TRACE_FILE.read_text(encoding="utf-8").splitlines()
        if ln and json.loads(ln)["trace_id"] == trace_id
    ]
    if not spans:
        raise HTTPException(404, "trace 不存在")
    return {"trace_id": trace_id, "spans": spans}

/api/stats 是运营视角的接口:总量、累计花费、三种 verdict 各多少、P95 延迟。percentile_cont(0.95).within_group(...) 是 Postgres 的窗口函数,直接在数据库里算分位数,不需要把几万行拉到 Python 里排序。

看 P95 而不是平均值,这一点值得强调:LLM 延迟的分布是长尾的,平均 1.5 秒可能意味着"大部分 0.8 秒,但有 5% 要 8 秒"。平均值会让你觉得一切良好,而那 5% 的用户已经在骂人了。

/api/trace/{trace_id} 这个实现是教学用的:每次请求都全文读一遍 JSONL、逐行 json.loads 两次,几万条之后就慢得没法用。生产上这里应该是 Langfuse 或 Jaeger 的查询接口。它的价值在于让你亲手体会"trace 数据是怎么被查出来的",不在于这段代码本身。

if not TRACE_FILE.exists() 的存在提醒一件事:这个方案在多副本部署下根本不工作——每个 Pod 写自己本地的 traces.jsonl,你查到的 trace 取决于负载均衡把请求打给了谁。这也是真实项目必须用集中式追踪后端的理由。

18. 起服务并把每一层都验一遍(1 小时)

代码写完了,但没验证过的生产配件等于没有。这一节把服务跑起来,逐层验证。

cd /tmp/docqa_lab
DOCQA_API_KEYS=demo-key-1,demo-key-2 PYTHONPATH=. \
  uv run python -m uvicorn docqa.main:app --host 127.0.0.1 --port 8010

启动要 25~30 秒(加载 embedding + reranker + 建索引),看到 [boot] 知识库就绪 才算好。

✅ Checkpoint 15:健康检查

curl -s http://127.0.0.1:8010/health

{"status":"ok","chunks":20,"version":"bc32f212de1a"}

version 是知识库版本哈希,你的值会不一样(取决于 PDF 生成时的字节),但同一份语料重复启动应该给出同一个值。变来变去说明 kb_version 依赖了不该依赖的东西。

✅ Checkpoint 16:鉴权三连

# 1) 不带 Key
curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8010/api/ask \
  -H 'Content-Type: application/json' -d '{"question":"满5年年假多少天","dry":true}'
# 2) 带错的 Key
curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8010/api/ask \
  -H 'X-API-Key: wrong-key' -H 'Content-Type: application/json' \
  -d '{"question":"满5年年假多少天","dry":true}'
# 3) 带对的 Key
curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8010/api/ask \
  -H 'X-API-Key: demo-key-1' -H 'Content-Type: application/json' \
  -d '{"question":"满5年年假多少天","dry":true}'

--- 1) 无 Key ---
401
--- 2) 错 Key ---
401
--- 3) 对 Key ---
200

我第一次跑这三条时三个全是 401,包括本该成功的第 3 条。原因是我复用了上一次会话遗留的进程(端口 8010 被 PID 占着),那个进程启动时用的是另一组 Key。杀掉旧进程重启才对上。

这个小插曲的教训:curl 打的是端口,不是你刚写的代码。改了配置一定要确认服务真的重启了(lsof -ti:8010 看 PID 变没变)。

✅ Checkpoint 17:完整响应体 + 缓存命中

curl -s -X POST http://127.0.0.1:8010/api/ask \
  -H 'X-API-Key: demo-key-1' -H 'Content-Type: application/json' \
  -d '{"question":"正式员工满5年有几天年假","dry":true}' | python3 -m json.tool

{
    "answer": "(dry-run) 入职年限是满 1 年不满 3 年, 年假天数是5 天, 是否可结转是否。[1] 年假天数按入职年限计算, 具体见下表。[2] 实习生不享受年假, 按实际出勤天数计算实习补贴。[3]",
    "verdict": "answered",
    "citations": [
        {"n": 1, "doc": "员工手册", "page": 1, "heading": "第二章假期管理", "kind": "table"},
        {"n": 2, "doc": "员工手册", "page": 1, "heading": "第二章假期管理", "kind": "text"},
        {"n": 3, "doc": "实习生管理办法", "page": 1, "heading": "第二章假期管理", "kind": "text"}
    ],
    "latency_ms": 0,
    "cached": false,
    "trace_id": "ba3c8d52dfb14e44"
}

kind: "table" 排第 1 —— 表格线性化那一步在起作用,答案就在表里。第 3 条引用是实习生那份(相关但不适用),交给 LLM 去说明差异。

一模一样的问题再打一次:

{"verdict": "answered", "cached": true, "trace_id": "a23c968a78914c78"}
answer 前40字: (dry-run) 入职年限是满 1 年不满 3 年, 年假天数是5 天, 是否

cached: true,答案文本完全一致,而 trace_id 换了新的——这是刻意的:缓存命中也是一次独立的请求,有自己的 trace。用同一个 trace_id 的话,你就没法统计"缓存命中的请求耗时多少"了。

✅ Checkpoint 18:SSE 流式,先出处后答案

curl -s -N -X POST http://127.0.0.1:8010/api/ask/stream \
  -H 'X-API-Key: demo-key-1' -H 'Content-Type: application/json' \
  -d '{"question":"企业版P1故障多久响应","dry":true}'

data: {"type": "sources", "items": [{"n": 1, "cite": "售后服务政策 p1 · 一、服务等级", "score": 0.9994}, {"n": 2, "cite": "售后服务政策-2024旧版 p1 · 一、服务等级", "score": 0.9993}]}

data: {"type": "answer", "text": "(dry-run) 标准版提供 5x8 工单支持, 首次响应时间承诺 8 小时;[1] 标准版提供 5x8 工单支持, 首次响应时间承诺 24 小时;[2]", "verdict": "answered"}

两条 event、顺序固定:sources 先到,answer 后到。

这条输出也把本章的主题演示得很清楚:rerank 给现行版 0.9994、旧版 0.9993 —— 差 0.0001。检索层根本分不出这两份文档谁是现行的,它俩内容太像了。所以答案里必须同时列出 8 小时和 24 小时并标注 [1]/[2],由用户结合引用去判断。

这不是系统做得不够好,这是正确的设计。 一个把冲突吞掉、只给你一个数字的系统,才是真正危险的。

✅ Checkpoint 19:限流在并发下真的原子

for i in $(seq 1 20); do
  curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8010/api/ask \
    -H 'X-API-Key: demo-key-2' -H 'Content-Type: application/json' \
    -d "{\"question\":\"并发测试问题编号$i\",\"dry\":true}" >> /tmp/rl_out.txt &
done; wait; sort /tmp/rl_out.txt | uniq -c

  10 200
  10 429

服务里配的是 capacity=10,20 个并发精确地放过 10 个、拒绝 10 个。这就是第 13 节 Lua 脚本原子性的价值——如果那四步是分开的 Redis 命令,这里会出现 11、12 甚至 15 个 200。

这个测试必须用 & 真并发,串行 for 循环跑 20 次会因为补令牌而放过更多,验证不到原子性。

✅ Checkpoint 20:运营视角的 stats

curl -s http://127.0.0.1:8010/api/stats -H 'X-API-Key: demo-key-1' | python3 -m json.tool

{
    "total": 31,
    "cost_yuan": 0.001632,
    "by_verdict": {
        "not_found": 9,
        "answered": 22
    },
    "latency_p95_ms": 616.9999999999984
}

数字会和你的不同(取决于你打了多少请求)。要看的是结构:by_verdict 把 9 条 not_found 单列出来了——这 9 条就是"知识库缺内容"的清单,真实项目里每周看一遍这个列表,就知道该补哪些文档。

cost_yuan 这么小是因为全都走的 dry-run(token 记 0),只有第 12 节 db.py 的 demo 写入了一条模拟成本。latency_p95_ms 那串小数是 percentile_cont 的浮点插值,正常。

✅ Checkpoint 21:用 trace_id 追一次请求

TID=$(curl -s -X POST http://127.0.0.1:8010/api/ask \
  -H 'X-API-Key: demo-key-1' -H 'Content-Type: application/json' \
  -d '{"question":"备份保留多少天","dry":true}' \
  | python3 -c "import json,sys;print(json.load(sys.stdin)['trace_id'])")
curl -s "http://127.0.0.1:8010/api/trace/$TID" -H 'X-API-Key: demo-key-1'

整理后:

trace_id=c3d6e0417cb64f13

ask            406ms
cache.get        0ms  {'hit': False}
retrieve       395ms  {'kept': 1}
llm              0ms  {'verdict': 'answered', 'tokens': 0, 'cost': 0.0}
persist         10ms

耗时分布一目了然:406ms 里 395ms 在检索(而且这是 dry-run,LLM 是 0)。真实调 API 时 llm 会涨到 1500~3000ms,retrieve 那 395ms 反而变成小头——但 retrieve 里的 rerank 是 CPU 密集的,并发上来之后它会成为瓶颈。没有这棵树,你只会看到"接口慢",不知道慢在哪。

kept=1 说明这个问题只有 1 条材料过了 rerank 阈值。答案质量不受影响(备份保留 30 天这个事实就在那一条里),但这个数字如果经常是 0 或 1,可以考虑把 min_rerank 调低一点。

19. tests/test_docqa.py —— 只 mock LLM(1.5 小时)

Mock 的边界要选对。 我们 mock 掉 openai 客户端,但保留分块、检索、引用、缓存、落库的真实行为。

只 mock LLM,是 agent 测试的黄金分割线——mock 得太多,测的就是 mock 自己了。

"""测试: 不花一分钱、不连一次网, 验证整条链路的确定性部分。"""

from __future__ import annotations

import pytest

from docqa.answerer import NOT_FOUND, answer, build_prompt
from docqa.bm25 import BM25, tokenize
from docqa.cache import RateLimiter, get_cached, kb_version, set_cached
from docqa.chunker import Chunk, chunk_blocks
from docqa.parser import Block, clean
from docqa.retriever import Hit, rrf_fuse


def mk_block(text: str, heading: str = "第一章测试", kind: str = "text") -> Block:
    return Block("测试文档", 1, heading, text, kind=kind)


def mk_chunk(cid: str, text: str, doc: str = "测试文档") -> Chunk:
    return Chunk(cid, doc, 1, "第一章测试", "text", text, text)

两个 mk_* 工厂函数省掉了几十行重复构造。测试代码也要重构——一堆位置参数的 Chunk(...) 散落在 16 个测试里,哪天 dataclass 加个字段就全炸了。

# ---------- 解析层 ----------


def test_clean_removes_cjk_gap():
    assert clean("直属 主管批准") == "直属主管批准"
    # 中英之间的空格必须保留, 否则 "P99 延迟" 会被粘成一个词
    assert clean("P99 延迟") == "P99 延迟"


# ---------- 分块层 ----------


def test_table_block_never_split():
    long_table = mk_block("[表格] " + "列是值。" * 200, kind="table")
    chunks = chunk_blocks([long_table], size=100)
    assert len(chunks) == 1, "表格块被拆开会丢表头, 必须整块保留"
    assert chunks[0].kind == "table"


def test_chunk_has_context_prefix_and_overlap():
    text = "".join(f"这是第{i}句话内容填充。" for i in range(1, 40))
    chunks = chunk_blocks([mk_block(text)], size=120, overlap=40)
    assert len(chunks) > 1
    for c in chunks:
        assert c.text.startswith("测试文档 · 第一章测试\n"), "缺上下文前缀会明显降低召回"
    # 相邻 chunk 必须有重叠, 否则跨块的句子关系会断
    assert any(s in chunks[1].raw for s in chunks[0].raw[-40:].split("。") if s)


def test_cite_format():
    assert mk_chunk("0-0", "x").cite == "测试文档 p1 · 第一章测试"

注意每个断言都带中文失败信息,而且这些信息说的是"为什么这条必须成立",不是"哪里不等于哪里"。半年后 CI 红了,你看到的是 AssertionError: 表格块被拆开会丢表头, 必须整块保留,而不是 assert 3 == 1。

# ---------- BM25 ----------


def test_tokenize_drops_stopwords():
    toks = tokenize("这是的了公司制度")
    assert "的" not in toks and "了" not in toks
    assert any("公司" in t or "制度" in t for t in toks)


def test_bm25_ranks_exact_term_first():
    chunks = [
        mk_chunk("0", "查询并发上限由 max_concurrent_queries 参数控制, 默认值为 128"),
        mk_chunk("1", "公司实行弹性工作制, 核心工作时间 10:00 至 16:00"),
        mk_chunk("2", "年假天数按入职年限计算"),
    ]
    top = BM25(chunks).search("max_concurrent_queries 默认值", top_k=1)
    assert top and top[0][0].id == "0"


def test_bm25_idf_never_negative():
    # 一个词出现在所有文档里时, 朴素 idf 公式会算出负数, 把好文档挤到后面
    chunks = [mk_chunk(str(i), "公司制度规定内容") for i in range(5)]
    assert all(v >= 0 for v in BM25(chunks).idf.values())


# ---------- RRF 融合 ----------


def test_rrf_prefers_doc_found_by_both_paths():
    a, b, c = mk_chunk("a", "aa"), mk_chunk("b", "bb"), mk_chunk("c", "cc")
    bm = [(a, 9.0), (b, 1.0)]  # a 在 BM25 第 1
    vec = [(c, 0.9), (a, 0.8)]  # a 在向量第 2
    fused = rrf_fuse(bm, vec)
    assert fused[0].chunk.id == "a", "两路都召回的应该排第一"
    assert fused[0].source == "both"
    assert fused[0].bm25_rank == 1 and fused[0].vec_rank == 2

test_rrf_prefers_doc_found_by_both_paths 就是如果早写出来,第 9 节那个 bug 根本不会发生的那条测试。它断言 source == "both"——那个 bug 的表现恰恰是 both 从不出现。

这是一个很值得记住的教训:bug 修完之后,补一条会在它复发时立刻变红的测试。不然半年后有人为了"性能"把内容哈希换回序号,你会再踩一遍。

# ---------- 生成层(mock LLM) ----------


def test_empty_hits_short_circuits_without_calling_llm(monkeypatch):
    def boom(*a, **k):
        raise AssertionError("检索为空时不应该调用 LLM")

    monkeypatch.setattr("openai.OpenAI", boom, raising=False)
    a = answer("知识库里没有的问题", [], dry=False)
    assert a.verdict == "not_found"
    assert a.text == NOT_FOUND
    assert a.cost_yuan == 0


def test_prompt_numbers_every_source():
    hits = [Hit(mk_chunk("0", "材料甲"), 1.0), Hit(mk_chunk("1", "材料乙"), 0.9)]
    prompt, cites = build_prompt("问题", hits)
    assert "[1] 出处:" in prompt and "[2] 出处:" in prompt
    assert [c["n"] for c in cites] == [1, 2]
    assert prompt.rstrip().endswith("问题: 问题")

test_empty_hits_short_circuits_without_calling_llm 用了一个漂亮的技巧:把 openai.OpenAI 替换成一个会抛断言错误的函数。这样"不该调 LLM"这件事不是靠间接推断(检查 token 数是 0),而是一旦调了就立刻炸。

测"某件事没有发生"通常很难,这种"地雷式 mock"是最直接的办法。

def test_answer_with_mocked_llm(monkeypatch):
    class FakeUsage:
        prompt_tokens, completion_tokens = 500, 40

    class FakeResp:
        choices = [
            type("C", (), {"message": type("M", (), {"content": "实习生不享受年假[1]。"})()})()
        ]
        usage = FakeUsage()

    class FakeClient:
        def __init__(self, **kw):
            self.chat = type("Chat", (), {"completions": self})()

        def create(self, **kw):
            assert kw["temperature"] == 0, "事实问答必须 temperature=0"
            return FakeResp()

    monkeypatch.setenv("LLM_API_KEY", "fake")
    monkeypatch.setattr("openai.OpenAI", FakeClient)
    a = answer("实习生有年假吗", [Hit(mk_chunk("0", "实习生不享受年假"), 1.0)])
    assert a.verdict == "answered"
    assert "[1]" in a.text
    assert a.prompt_tokens == 500 and a.completion_tokens == 40
    assert a.cost_yuan == pytest.approx(500 * 1e-6 + 40 * 2e-6)

FakeClient.create 里那句 assert kw["temperature"] == 0 是把设计约束写进测试:哪天有人为了"让答案更自然"把 temperature 调到 0.7,这条测试会拦住。比写在注释里管用。

pytest.approx 而不是 ==:浮点相等在成本计算上会翻车。

def test_dry_run_reports_zero_tokens():
    a = answer("问题", [Hit(mk_chunk("0", "内容甲。内容乙。"), 1.0)], dry=True)
    assert a.prompt_tokens == 0 and a.cost_yuan == 0, "dry-run 不能伪造 token 数"
    assert "[1]" in a.text


def test_llm_error_becomes_error_verdict(monkeypatch):
    monkeypatch.setenv("LLM_API_KEY", "fake")

    class Boom:
        def __init__(self, **kw):
            raise RuntimeError("连接超时")

    monkeypatch.setattr("openai.OpenAI", Boom)
    a = answer("问题", [Hit(mk_chunk("0", "内容"), 1.0)])
    assert a.verdict == "error" and "连接超时" in a.text


# ---------- 缓存与限流(要求 Redis 在跑) ----------


def test_cache_roundtrip_and_version_isolation():
    ver = kb_version(3, ["s1", "s2"])
    set_cached("测试问题", ver, {"answer": "答案"}, ttl=30)
    assert get_cached("  测试问题 ", ver) == {"answer": "答案"}
    assert get_cached("测试问题", kb_version(4, ["s1", "s2"])) is None, "知识库变了必须失效"


def test_rate_limiter_blocks_then_refills():
    import time

    from docqa.cache import _r

    user = f"pytest-{time.time()}"
    rl = RateLimiter(capacity=2, refill_per_sec=5.0)
    _r.delete(f"docqa:rl:{user}")
    assert rl.allow(user)[0] and rl.allow(user)[0]
    assert not rl.allow(user)[0]
    time.sleep(0.35)
    assert rl.allow(user)[0], "补充令牌后应该放行"

最后两条测试连的是真的 Redis,不是 fakeredis。理由和 CI 里挂 service 容器一样:Lua 脚本的原子性、set(ex=...) 的行为、Redis 返回值的类型转换,这些东西 mock 出来的假 Redis 保证不了。用真的,才能在本地就发现"我的 Lua 写错了"。

user = f"pytest-{time.time()}" 用时间戳做 key,是为了测试之间不互相污染。共用一个固定 key 的话,连续跑两遍 pytest 第二遍就会失败(令牌还没补满)。这类"单独跑能过、连着跑就挂"的测试极其耗人,一开始就把 key 隔离掉。

✅ Checkpoint 22:uv run pytest tests -q

Redis 和 Postgres 必须在跑。

................                                                         [100%]
16 passed in 3.42s

16 passed,不允许 skip。 有 skip 说明中间件没连上——那两条缓存测试被跳过了,而它们恰恰是最容易出错的部分。

20. Dockerfile / compose / CI —— 交付这件事(1.5 小时)

代码能跑不等于能交付。这三个文件是"我做了个项目"和"我交付了一个服务"的差别。

Dockerfile:多阶段 + 非 root

# 多阶段构建: 第一阶段装依赖, 第二阶段只拷 venv 和代码
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS builder

WORKDIR /app
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy

# 先只拷依赖清单 —— 代码改了不会让依赖层缓存失效, 这是 Docker 层缓存的关键顺序
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-install-project --no-dev

COPY docqa ./docqa
RUN uv sync --frozen --no-dev

FROM python:3.12-slim-bookworm AS runtime

# 非 root 运行: 容器逃逸时的第一道防线
RUN useradd -m -u 10001 appuser
WORKDIR /app

COPY --from=builder --chown=appuser:appuser /app/.venv /app/.venv
COPY --from=builder --chown=appuser:appuser /app/docqa /app/docqa

ENV PATH="/app/.venv/bin:$PATH" \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    HF_HOME=/models

RUN mkdir -p /models && chown appuser:appuser /models
USER appuser

EXPOSE 8000
HEALTHCHECK --interval=15s --timeout=3s --start-period=180s --retries=5 \
  CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/health',timeout=2).status==200 else 1)"

# 注意 workers=1: 模型加载在内存里, 多 worker 会把 embedding + reranker 各加载一份,
# 内存直接翻倍。要扩容就横向加容器, 别加 worker
CMD ["uvicorn", "docqa.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

五个要点:

• 先 COPY pyproject.toml uv.lock,后 COPY docqa。这个顺序是 Docker 层缓存的全部诀窍:改一行业务代码,依赖层命中缓存、构建 3 秒;顺序反过来,每次都重装几百兆依赖、构建 3 分钟。
• 多阶段:builder 阶段有 uv 和编译工具链,runtime 阶段只有 Python + venv + 代码。镜像小几百兆,而且攻击面小得多(容器里没有编译器,拿到 shell 也不好做事)。
• useradd -u 10001 + USER appuser。默认 root 运行是最常见的容器安全问题。加两行,容器逃逸的成本立刻高一个量级。
• start-period=180s。这个服务启动要下模型 + 建索引,启动期长达几分钟。不设 start-period,编排系统会在它还在加载模型时判定"不健康"然后杀掉重启,永远起不来。这是部署 AI 服务最常见的翻车点。
• --workers 1。embedding 和 reranker 模型常驻内存,4 个 worker 就是 4 份模型。这类服务要横向扩容器,不要纵向加 worker。

HF_HOME=/models 把模型缓存路径指到一个独立目录,配合 compose 里的 volume,重建镜像不用重新下载。

docker-compose.yml:服务名 DNS 与必填变量

services:
  qdrant:
    image: qdrant/qdrant:v1.16.1
    ports: ["6333:6333"]
    volumes: ["qdrant_data:/qdrant/storage"]
    healthcheck:
      # Qdrant 镜像里没有 curl, 用 bash 的 /dev/tcp 探活
      test: ["CMD-SHELL", "bash -c 'echo > /dev/tcp/127.0.0.1/6333'"]
      interval: 5s
      retries: 20

  postgres:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: docqa
      POSTGRES_DB: docqa
    ports: ["5432:5432"]
    volumes: ["pg_data:/var/lib/postgresql/data"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d docqa"]
      interval: 5s
      retries: 20

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      retries: 20

  api:
    build: .
    ports: ["8000:8000"]
    environment:
      # 容器之间用服务名互相访问, 不是 localhost —— 这是 compose 里最常踩的坑
      DB_URL: postgresql+psycopg://postgres:docqa@postgres:5432/docqa
      REDIS_URL: redis://redis:6379/0
      QDRANT_URL: http://qdrant:6333
      # 没配 key 服务会拒绝启动; 生产环境从密钥管理服务注入, 不要写在 compose 里
      DOCQA_API_KEYS: ${DOCQA_API_KEYS:?请先 export DOCQA_API_KEYS=你的key}
      LLM_API_KEY: ${LLM_API_KEY:-}
      HF_HOME: /models
    volumes:
      - ./docqa/data:/app/docqa/data:ro
      - hf_models:/models          # 模型缓存挂出来, 否则每次重建镜像都要重新下载
    depends_on:
      qdrant: {condition: service_healthy}
      postgres: {condition: service_healthy}
      redis: {condition: service_healthy}

volumes:
  qdrant_data:
  pg_data:
  hf_models:

四个坑:

• @postgres:5432 而不是 @localhost:5432。这是 compose 里最常踩的坑:容器内的 localhost 指的是容器自己。服务名就是 DNS 名。
• ${DOCQA_API_KEYS:?...} —— 冒号问号语法表示"这个变量必填,没有就报错退出并显示这句话"。配合第 17 节的 fail-closed 设计,形成两道防线:compose 层拦一次,应用层再拦一次。
• depends_on 用 condition: service_healthy 而不是裸 depends_on。后者只保证"容器启动了",不保证"数据库能连了"。Postgres 容器起来到能接受连接有几秒空档,api 会在这个空档里连接失败退出。
• ./docqa/data:...:ro —— 语料目录只读挂载。服务没有任何理由去改知识库源文件,给它写权限就是给自己留隐患。

✅ Checkpoint 23:验证必填变量真的拦得住

cd /tmp/docqa_lab && unset DOCQA_API_KEYS && docker compose config >/dev/null

error while interpolating services.api.environment.DOCQA_API_KEYS:
required variable DOCQA_API_KEYS is missing a value: 请先 export DOCQA_API_KEYS=你的key

连 docker compose config 都过不去,更不可能起一个没鉴权的服务。这就是我们想要的效果。

CI:lint + 真中间件测试 + 质量闸门 + 构建镜像

name: ci

on:
  push: {branches: [main]}
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest

    # 测试要连真的 Redis 和 Postgres —— 用 service 容器, 不要 mock 掉中间件,
    # 否则"Lua 脚本原子性""SQL 语法"这类问题只会在线上暴露
    services:
      redis:
        image: redis:7-alpine
        ports: ["6379:6379"]
        options: >-
          --health-cmd "redis-cli ping" --health-interval 5s --health-retries 10
      postgres:
        image: postgres:17-alpine
        env:
          POSTGRES_PASSWORD: docqa
          POSTGRES_DB: docqa
        ports: ["5432:5432"]
        options: >-
          --health-cmd "pg_isready -U postgres" --health-interval 5s --health-retries 10

    steps:
      - uses: actions/checkout@v5

      - uses: astral-sh/setup-uv@v6
        with:
          enable-cache: true

      - run: uv sync --frozen

      # 模型缓存单独存: 每次重新下载 bge 会让 CI 慢好几分钟
      - uses: actions/cache@v4
        with:
          path: ~/.cache/huggingface
          key: hf-${{ hashFiles('uv.lock') }}

      - name: Lint
        run: |
          uvx ruff check docqa tests
          uvx ruff format --check docqa tests

      - name: Test
        env:
          DB_URL: postgresql+psycopg://postgres:docqa@localhost:5432/docqa
          REDIS_URL: redis://localhost:6379/0
        run: uv run pytest tests -q

      # 检索质量回归: 指标掉了就让 CI 红, 这样"改了分块参数导致召回下降"
      # 会在合并前被拦住, 而不是上线后靠用户投诉发现
      - name: Retrieval regression gate
        env:
          DB_URL: postgresql+psycopg://postgres:docqa@localhost:5432/docqa
        run: |
          docker run -d --name qdrant -p 6333:6333 qdrant/qdrant:v1.16.1
          for i in $(seq 1 30); do curl -sf localhost:6333/healthz && break; sleep 2; done
          uv run python docqa/make_corpus.py
          uv run python docqa/indexer.py --force
          uv run python docqa/gate.py --min-hit1 0.90 --min-mrr 0.95

      - name: Build image
        run: docker build -t docqa:${{ github.sha }} .

"Retrieval regression gate" 这一步是这个 CI 与普通 Python 项目 CI 的唯一区别,也是全章最想让你带走的东西。

普通 CI 保证的是"代码没坏"(lint 通过、测试通过)。但 RAG 系统有一类改动不会让任何测试变红,却会让效果明显变差:调了分块大小、换了 embedding 模型、改了 RRF 的 k、删掉了 query 前缀。这些改动在单元测试眼里全部合法。

只有把评测集跑进 CI、把指标当断言,这类退步才拦得住。这就是第 10、11 节那两个文件的最终归宿。

几个实现细节:

• service 容器给 Redis 和 Postgres,Qdrant 用 docker run。因为 Qdrant 只在 gate 这一步需要,而 service 容器在整个 job 期间都占着资源。
• curl -sf localhost:6333/healthz 轮询等就绪,而不是 sleep 30。固定 sleep 要么白等要么不够。
• 模型缓存的 key 用 hashFiles('uv.lock'):依赖没变就复用缓存。省下的是每次 CI 三四分钟的模型下载。
• uv sync --frozen 而不是 uv sync:--frozen 要求 lock 文件和 pyproject 一致,不一致就报错而不是悄悄更新。CI 里装的必须和你本地装的是同一批版本,否则"我本地是好的"这句话就没有意义了。
• Build image 放在最后:构建镜像最慢,前面任何一步失败都没必要构建。

✅ Checkpoint 24:本地把 CI 的三道关全过一遍

cd /tmp/docqa_lab
uvx ruff check docqa tests
uvx ruff format --check docqa tests
uv run pytest tests -q

All checks passed!
---
16 files already formatted
---
16 passed, 5 warnings in 4.72s

5 warnings 是第三方库的 DeprecationWarning(swigvarlink,来自 PyMuPDF 的 SWIG 绑定),不是我们的代码。

在本地把 CI 的每一步跑过一遍再推,这个习惯能省掉大量"推上去红了、改一行再推、又红了"的往返。

✅ 最终验收:16 个文件全部敲完后,下面每一条都要能自己打勾。打不上的那条,回到对应小节。

离线链路:

• [ ] PYTHONPATH=. python docqa/make_corpus.py 生成 5 份 PDF,其中 2 份是内容冲突的干扰文档
• [ ] PYTHONPATH=. python docqa/parser.py 解析出的 kind=table 块里,表格已被线性化成 规格是32核128GB, P99延迟是200毫秒 这样的完整句子
• [ ] PYTHONPATH=. python docqa/chunker.py 每个 chunk 都带 文档名 · 标题 前缀,chunk_id 是内容哈希而不是序号
• [ ] PYTHONPATH=. python docqa/indexer.py --force 写入 20 个向量;不加 --force 再跑一次,5 份文档全部"哈希未变, 跳过向量化"

检索质量:

• [ ] PYTHONPATH=. python docqa/retriever.py 中 满5年员工年假多少天 的第一条来自员工手册的表格,rerank 分 > 0.99,且实习生那份被压到 0.2 以下
• [ ] 同一脚本里 公司的年终奖怎么发 返回空(阈值起作用了),不是硬凑 4 条
• [ ] PYTHONPATH=. python docqa/evaluate.py 打印四方消融表,混合+rerank 的 Hit@1 明显高于纯向量
• [ ] PYTHONPATH=. python docqa/gate.py --min-hit1 0.99 返回退出码 1(用 echo $? 单独验,别接管道)

服务与生产配件:

• [ ] 不设 DOCQA_API_KEYS 启动 → 直接 RuntimeError 拒绝启动
• [ ] 不带 / 带错 X-API-Key → 401;带对 → 200
• [ ] 同一个问题问两次,第二次 "cached": true 且 trace_id 是新的
• [ ] 20 个并发请求 → 精确 10 个 200 + 10 个 429
• [ ] /api/ask 的答案里每个结论都带 [n],且 citations 能对上
• [ ] 问知识库里没有的问题 → verdict 为 not_found,不是编一段
• [ ] 问新旧版政策冲突的问题 → 答案同时列出两个版本并说明适用范围
• [ ] /api/ask/stream 先收到 sources 事件,再收到 answer
• [ ] /api/trace/{id} 能看到 cache / retrieve / llm / persist 四段耗时
• [ ] /api/stats 的 latency_p95_ms 和 cost_yuan 都有值

交付:

• [ ] uvx ruff check + uvx ruff format --check + uv run pytest tests -q 三条全绿
• [ ] docker compose config 在没有 DOCQA_API_KEYS 时退出码非 0 并给出中文提示
• [ ] docker compose up 起来后 /health 返回 status: ok 且 chunks 不为 0

21. 踩坑与专家提示

这一节全部是我在写这个项目时真的踩到的坑,按"造成的损失"排序。

坑 1:chunk id 用序号,RRF 静默失效(损失最大)

现象:混合检索 Hit@1 = 0.71,比纯 BM25(0.75)和纯向量(0.83)都低。

根因:BM25 那一路一次性解析整个目录建索引,向量那一路逐文件索引,位置编号 f"{bi}-{part}" 在两边算出来不一样。RRF 的去重键 f"{doc}/{id}" 永远对不上,同一条内容被当成两条各拿一份分数。

修复:chunk id 改成内容哈希。

教训:任何跨模块共享的标识符都不能依赖遍历顺序。 这个坑还有个孪生兄弟——kb_version() 里如果不 sorted(doc_shas),两台机器算出的知识库版本不同,缓存永远互不命中。同一类错误,不同表现。

而且它不报错。要不是有评测集,我会得出"混合检索没用"的结论然后删掉它。

坑 2:find_tables() 认不出没有边框的表格

现象:解析出 0 个表格,而且更糟的是表格里的 5 天、10 天 被 _is_heading 误判成标题,产出一堆 <5 天> 这样的荒唐 heading,污染了整份文档的层级结构。

根因:make_corpus.py 第一版只画文字不画线,PyMuPDF 靠线框识别表格。

教训:如果你的语料是扫描件或纯文字排版的表格,find_tables() 帮不了你,得上 pymupdf_layout 或版面分析模型。表格是企业文档里信息密度最高的部分,也是最容易在解析阶段被丢掉的部分。

坑 3:ruff B023 —— 闭包捕获的是变量不是值

现象:parser.py 里的嵌套 flush() 函数读 current_heading,而这个变量在循环里一直变。B023 报错。

第一次修复失败:用默认参数 def flush(page_no: int = page_no) 冻结了 page_no,但 current_heading 仍然报错——它必须是循环里最新的值,冻结反而错。

最终修复:删掉闭包,改成 emit(buf, page_no, heading) 显式传参。

教训:Python 闭包捕获的是变量,不是值。这是一个所有人都会栽一次的经典陷阱。发现自己在跟闭包作用域搏斗时,把参数显式传进去通常是最干净的解法——读代码的人也不用在脑子里模拟作用域。

坑 4:$? 报的是管道最后一环的退出码

现象:验证质量闸门时跑 python docqa/gate.py --min-hit1 0.99 | tail -4; echo $? 得到 0,以为闸门坏了。

根因:$? 是 tail 的退出码。

教训:验证退出码时别接管道,或者用 set -o pipefail。"我以为我验证了"比"我没验证"更危险——后者你知道自己不知道。

坑 5:改了配置但服务没重启

现象:鉴权三连测试全部 401,包括本该 200 的那条。

根因:8010 端口被上一次会话遗留的进程占着,那个进程用的是另一组 Key。

教训:curl 打的是端口,不是你刚写的代码。lsof -ti:<port> 确认 PID 变了再测。

坑 6:表名写错、字段名记错

• TRUNCATE qa_log → ERROR: relation "qa_log" does not exist,表名是 qa_logs
• jq '.duration_ms' → 一串 null,自己定的 JSONL schema 里字段叫 ms,而且 attrs 是摊平到顶层的

教训:自己写的东西也会忘。这是"用现成的可观测性方案"的一个真实理由——不是因为你写不出来,是因为你不想维护一套只有自己知道 schema 的东西。

坑 7:pytest 找不到自己的包

现象:所有脚本都加了 PYTHONPATH=. 跑得很好,uv run pytest tests 直接 ModuleNotFoundError: No module named 'docqa'。

修复:[tool.pytest.ini_options] pythonpath = ["."]。

教训:能钉在配置文件里的,不要靠每次记得敲命令。

专家提示:三个"不做会后悔"的设计

1. min_rerank 阈值 + 空检索短路。三行代码,拦掉最大一类幻觉,而且省钱。绝大多数 RAG 教程没有这个,因为它们从不测"知识库里没有的问题"。
2. verdict 三值化。把"没答案"和"出错"从"答了"里分出来。一个月后你想做质量分析时,会感谢当初多加的这个字段。
3. 评测集从第一天就写。24 条标注花了我不到一小时,但它:抓出了坑 1、给了消融表、成了 CI 闸门、还让我敢改分块参数。先写评测再调优,顺序反了就是在盲调。

专家提示:什么时候不要用这套架构

诚实一点——这套东西不是万能的:

• 知识库只有几十条 QA:直接把全部内容塞进 prompt 更简单,长上下文模型现在便宜到这么干很合理。检索是为了省 token,不是为了正确。
• 需要跨文档推理("对比一下新旧政策的所有差异"):RAG 一次只取 4 条材料,做不了全局比较。这类问题要么全量喂,要么上 GraphRAG。
• 答案需要计算("我明年能休几天假"):检索能找到规则,算不出结果。要接工具调用。

能说清一个架构的边界,比能说清它的优点更能体现你真的用过它。

22. 面试视角

一句话版本

"我做了一个企业文档问答系统:PyMuPDF 解析版式和表格并把表格线性化成句子,句子对齐分块加文档标题前缀;检索是 BM25 + bge 向量双路 RRF 融合再 cross-encoder 精排,带阈值兜底防幻觉;有 24 条人工标注评测集,纯向量 Hit@1 0.83 / MRR 0.880,全套上到 0.96 / 0.979,这套指标做成质量闸门接在 CI 上,掉到阈值以下就不许合并;服务侧有 API Key 鉴权、Redis 令牌桶限流、按知识库版本失效的答案缓存、span 追踪和成本统计,Docker 多阶段非 root 打包。"

高频追问与答法

Q:为什么用 RRF 而不是加权求和?

BM25 分数无上界且随知识库规模漂移,余弦相似度固定在窄区间,两者量纲不可比。要加权就得先归一化,归一化又需要知道分数分布,而分布随查询变化。RRF 只用排名,零调参。代价是丢失强度信息,企业场景下这笔交易划算。k=60 控制名次差被压平的程度,k 越大越看重"几路投票"而非"排多前"。

Q:怎么防幻觉?

三层,而且都不是 prompt 技巧:精排阈值让检索层敢返回空;检索为空直接短路,连 LLM 都不调;prompt 里每条材料带编号、强制 [n] 标注,没有标注的答案视为不可信。另外知识库里有冲突时(新旧版本、不同适用人群),要求模型摊开冲突而不是替用户拍板——我的实测里 rerank 给现行版 0.9994、旧版 0.9993,差 0.0001,检索层根本分不出来,那就必须让用户看见。

Q:怎么证明你的优化有效?

24 条标注集,标到"文档 + 标题"粒度而不是 chunk id(分块参数一改 id 全变)。四方消融:BM25 / 向量 / +RRF / +rerank。有意思的是 BM25 的 Recall@3 (0.96) 比向量 (0.88) 高——只用向量会把免费的一半准确率扔掉。RRF 的贡献主要在 Recall@5 到 1.00(保证不漏),rerank 的贡献在 Hit@1 从 0.83 到 0.96(负责排对)。分工不同,不能互相替代。

Q:线上说变慢了,你怎么查?

每次响应带 trace_id,span 树给出 cache / retrieve(内含 bm25 / vector / rerank)/ llm / persist 各自耗时。/api/stats 看 P95 而不是平均——LLM 延迟长尾很重,平均 1.5 秒可能是"多数 0.8 秒但 5% 要 8 秒"。

Q:知识库更新了,缓存怎么办?

缓存 key 是"问题 + 知识库版本哈希",版本 = chunk 数 + 所有文件 sha256(排序后)。任何文档变了版本就变,旧 key 自然不再命中,不需要主动清理。只用问题做 key 是经典 bug:文档更新了用户还在拿旧答案,而且要错一整个 TTL。另外出错的答案绝不缓存,否则一次网络抖动会被记住一小时。

Q:为什么限流用 Lua?

令牌桶是"读余额 → 按时间补 → 扣减 → 写回"四步。分开发命令在并发下会交错,两个请求读到同一个余额,漏放请求。这个 bug 在低并发下永远不暴露,本地手点几次全对。我用 20 个并发实测过:capacity=10,精确放过 10 个、拒绝 10 个。

Q:部署上有什么坑?

--workers 1 —— embedding 和 reranker 常驻内存,多 worker 会各加载一份,内存翻倍;要扩容就横向加容器。HEALTHCHECK --start-period=180s —— 启动要下模型建索引,不设的话编排系统会在加载期判定不健康然后杀掉,永远起不来。这是部署 AI 服务最常见的翻车点。

Q:如果让你继续优化,下一步做什么?

先扩评测集(24 条太少,置信区间宽),再上 query 改写(用户问"几点上班",先让小模型改写成文档里的说法)。不会先去换更大的 embedding 模型——那是在没有足够评测数据的情况下盲调。

23. 划重点

1. chunk_id 用内容哈希,不用序号。 跨模块共享的标识符不能依赖遍历顺序,否则 RRF 去重静默失效——不报错,只是效果变差。
2. 上下文前缀(文档名 · 标题\n)是几行代码换来的显著召回提升。 用户提问用的词往往不在正文里,而在标题里。
3. 表格必须整块保留 + 线性化成句子。 规格是32核128GB, P99延迟是200毫秒 这种句子,BM25 和向量都能对上;拆散的表格行是无意义的碎片。
4. BM25 不是过时方案。 在专有名词密集的企业文档上,它的 Recall@3 比向量还高。只用向量等于扔掉一半准确率。
5. RRF 只看排名不看分数,所以两路量纲不同也能融合,零调参。k=60 越大越"民主"。
6. 融合负责"不漏"(Recall@5→1.00),精排负责"排对"(Hit@1→0.96)。 指标不同,不能互相替代,也不能跳过融合直接精排。
7. 精排阈值 + 空检索短路是最有效的防幻觉措施,三行代码,还省钱。它比任何 prompt 技巧都可靠。
8. 知识库里的冲突要摊开,不要替用户拍板。 检索层区分不出新旧版本(实测差 0.0001),这是设计约束不是 bug。
9. 评测集是这套系统里最值钱的文件。 它抓出了本项目最隐蔽的 bug,给了消融表,成了 CI 闸门。先写评测再调优。
10. sys.exit(0 if ok else 1) 那一行是质量闸门存在的理由。 少了它,CI 永远绿灯而什么都没拦住。
11. 配置缺失要 fail closed。 没配 API Key 就拒绝启动,绝不留"没配就放开"的后门——那是线上事故最常见的来源。
12. 只 mock LLM,中间件用真的。 Lua 原子性、SQL 语法这些东西,假 Redis 保证不了。
13. dry-run 里 token 一律记 0,不要估算。 假指标比没有指标危险。
14. 看 P95 不看平均值。 LLM 延迟长尾很重。
15. --workers 1 + start-period=180s 是 AI 服务部署的两个必备参数。

24. 进阶练习

按投入产出排序:

1. 扩评测集到 100 条(半天)。24 条的置信区间太宽,0.96 和 0.92 的差别可能只是噪音。加入更多"知识库里没有"的负样本,专门测阈值。
2. query 改写(1 天)。用户问"几点上班",先用一次便宜的小模型调用改写成"核心工作时间",再去检索。测一下 Hit@1 能提多少——先测再决定要不要留。
3. 按文档权限过滤(半天)。vectorstore.search() 已经支持 doc 过滤了,把它接到用户身份上:HR 文档只有 HR 能检索到。这是企业落地的硬需求。
4. 把 tracing 换成 Langfuse(半天)。第 14 节给了对应关系。做完你会发现自己手写那 157 行的价值:换的时候你知道每个概念对应什么。
5. 增量更新单份文档(1 天)。现在改一份 PDF 要 --force 全量重建。真实项目里要能只重建那一份的向量(documents 表的哈希已经具备判断能力了,缺的是按 doc 删除旧 point)。
6. 加 Reflection 环节(1 天)。生成答案后,让模型自己检查"每个结论是否都有引用支撑",没有的就重写。用第 10 节的评测集验证它到底有没有用。

25. 下一章预告

这一篇解决的是数据平面的工程化:文档进来、答案出去、质量可测、成本可控。

但整条链路是固定的——鉴权、检索、生成、落库,一条直线走完。真实的 agent 需要分支、循环、暂停等人批准、出错重来,而这些东西用 if/else 和 try/except 硬写,写到第三个分支就会失控。

下一篇把 docs/2/13 那个深度研究 agent 用 LangGraph 重写,补上控制平面的工程化:

• 状态机代替函数调用链:节点、边、条件跳转
• Postgres checkpointer:进程挂了能从中断的节点继续,而不是从头再来
• human-in-the-loop:关键步骤前暂停等人批准,批完继续
• replay:回放任意一次执行,改一个节点重跑后面

学完这两篇,你手上就有了一个 RAG 生产系统 + 一个有状态可恢复的 agent 工作流——数据平面和控制平面各一个。

下一篇:20 · 实战手敲六:用 LangGraph 重写研究智能体 —— 已完成,四个中断/续跑/回放能力全部有实测输出 | 返回目录:README