检索质量:混合检索、rerank、query 改写

今天要解决的问题

Day 15 我说过 embedding 有两个短板:对否定词不敏感,对编号和专有名词不敏感。Day 16 和 Day 17 搭好了向量检索,今天来补这两个短板。

先看一个具体的失败。CloudLedger 帮助中心里有两篇长得很像的文章:《错误码 E1042:银行流水导入失败》和《错误码 E1043:文件编码错误》。用户问「导入的时候报 E1043」,向量检索很可能把 E1042 那篇排第一,因为两篇都在讲「导入出错」,话题几乎一样,而「1042」和「1043」在 embedding 眼里差别很小。客服 Agent 拿着 E1042 的解决办法(改日期格式)去回答一个编码问题,用户照着做了,问题没解决,然后投诉。

这就是检索质量的问题。RAG 的回答质量有一个很硬的上限:检索没找到的东西,模型不可能答对。所以生产系统里,检索这一层通常不止一步,而是一条流水线:多路召回 → 融合 → 精排。今天把它一块块搭起来。

今天结束时你应该能做到:

1. 解释 BM25 是怎么打分的,为什么它擅长精确匹配,以及中文为什么要先分词。
2. 用 RRF 把向量检索和关键词检索的结果融合,能手算 RRF 分数。
3. 讲清楚 rerank(cross-encoder)和 embedding(bi-encoder)的区别,调通 bge-reranker。
4. 用大模型做 query 改写和多路查询,知道 HyDE 是什么。
5. 对 6 个测试问题,做出「纯向量 / 纯 BM25 / 混合 / 混合 + rerank」的对比表。
6. 把 PDF(含表格和扫描页)解析成适合切块的 Markdown,知道表格怎么切块、什么时候该换 MinerU 这类解析工具。

预计用时 6 小时。前置:Day 16 的切块和数据,Day 17 的 pgvector 检索。

BM25:老派但好用的关键词检索

BM25 是搜索引擎用了几十年的打分算法,Elasticsearch 默认的相关性算法就是它。它不懂语义,只数词。但「只数词」恰好是向量检索缺的那部分能力。

它的打分逻辑用人话说是三条:

一个词在这篇文档里出现得越多,分越高,但有上限。 「退款」出现 1 次和出现 2 次差别很大,出现 20 次和 21 次差别就很小了。这叫词频饱和,由参数 k1 控制。

一个词在所有文档里越少见,越重要。 「的」「是」每篇都有,没有区分度;「E1043」只在一篇里出现,一旦匹配上就是强信号。这叫逆文档频率(IDF)。错误码、订单号、产品型号这些词,IDF 都很高,所以 BM25 对它们特别敏感。这正是我们要的。

文档越长,分数越要打折。 一篇万字长文偶然提到一次「退款」,不应该比一篇专讲退款的短文得分高。这叫长度归一化,由参数 b 控制。

公式写出来是这样,不用背,知道每部分对应上面哪一条就行:

score(D, Q) = Σ  IDF(q) × f(q,D) × (k1 + 1) / ( f(q,D) + k1 × (1 - b + b × |D| / avgdl) )
            q∈Q

f(q,D)  : 词 q 在文档 D 里出现的次数          → 第 1 条
IDF(q)  : 包含 q 的文档越少,值越大            → 第 2 条
|D|/avgdl: 文档长度 / 平均长度                 → 第 3 条
k1 通常 1.2–2.0,b 通常 0.75

中文要先分词

英文天然有空格,BM25 直接按空格切词。中文没有,「企业版年付中途取消能退多少」必须先切成词,BM25 才能数。分词质量直接决定 BM25 的效果。

Python 里最常用的中文分词库是 jieba。我拿它试了几个句子,发现一个典型问题:

jieba.lcut("企业版年付中途取消能退多少")
→ ['企业', '版年付', '中途', '取消', '能', '退', '多少']

「版年付」是错的。jieba 不认识「企业版」「年付」这些业务词,于是按统计规律切出了奇怪的组合。解决办法是加自定义词典,把你的业务术语告诉它。加了以后,「企业版」「年付」就能正确切出来。

这件事在真实项目里很重要:每个行业都有自己的术语,通用分词器一定会切错。我做过的项目都会维护一份业务词表,从产品文档里抽,定期更新。

顺带一提,Postgres 自带的全文检索(tsvector)默认的解析器不会切中文,一整句中文会被当成一个词。要在 Postgres 里做中文全文检索,得装 zhparser 或 pg_jieba 这类扩展。课程里为了简单,BM25 在 Python 内存里做,42 块数据一瞬间就建好了。生产上几十万块的话,可以用 Elasticsearch/OpenSearch(装 IK 分词插件)或者 Postgres + 中文分词扩展。

代码

uv add jieba rank-bm25

# agent-60-lab/w3/bm25_index.py
"""关键词检索:jieba 分词 + BM25。和向量检索互补,专治错误码、专有名词。"""
import logging
import re

import jieba
from rank_bm25 import BM25Okapi

from chunking import Chunk, load_chunks

jieba.setLogLevel(logging.WARNING)
# 业务词典:不加的话 jieba 会把「企业版年付」切成「企业 / 版年付」
for word in ["免费版", "专业版", "企业版", "年付", "月付", "红冲", "自动续费", "宽限期", "两步验证", "多币种", "账本"]:
    jieba.add_word(word)

# 写成集合字面量;写 set("... ".split()) 也能跑,但 Ruff 会报 SIM905 提示改成字面量
STOPWORDS = {
    "的", "了", "是", "吗", "呢", "啊", "吧", "在", "和", "与", "及", "或", "我", "你", "我们", "你们",
    "怎么", "怎么办", "什么", "如何", "可以", "能", "还", "要", "就", "都", "也", "把", "被", "让", "会",
    "有", "没有", "一个", "时候", "报",
}
WORD = re.compile(r"\w+")  # \w 在 Python 里包含汉字、字母、数字,用来去掉标点和空格


def tokenize(text: str) -> list[str]:
    return [t.lower() for t in jieba.lcut(text) if WORD.fullmatch(t) and t not in STOPWORDS]


class BM25Index:
    def __init__(self, chunks: list[Chunk]):
        self.chunks = chunks
        self.bm25 = BM25Okapi([tokenize(c.text) for c in chunks])

    def search(self, query: str, k: int = 20) -> list[tuple[str, float]]:
        scores = self.bm25.get_scores(tokenize(query))
        order = sorted(range(len(scores)), key=lambda i: -scores[i])[:k]
        return [(self.chunks[i].id, float(scores[i])) for i in order if scores[i] > 0]


if __name__ == "__main__":
    idx = BM25Index(load_chunks())
    for q in ["导入流水提示 E1042 怎么办", "E1043 是什么错误", "企业版年付中途取消能退多少"]:
        print(q, tokenize(q))
        for cid, s in idx.search(q, 3):
            print(f"   {s:6.2f}  {cid}")

uv run python w3/bm25_index.py

tokenize() 那一行是列表推导式,[输出 for 变量 in 列表 if 条件],相当于 TS 的 .filter(...).map(...) 合成一行。拆开就是:jieba 切词 → WORD.fullmatch 去掉空格和标点 → 去掉停用词 → 转小写。拿「导入的时候报 E1043」走一遍:

jieba.lcut   ['导入', '的', '时候', '报', ' ', 'E1043']
去空格标点   ['导入', '的', '时候', '报', 'E1043']
去停用词     ['导入', 'E1043']
转小写       ['导入', 'e1043']

转小写是为了让用户打的 e1043 和文档里的 E1043 算同一个词。建索引(BM25Index.__init__)和查询(search)都调同一个 tokenize,两边切法才一致。

BM25Index 两个方法:

• __init__ 建索引,只跑一次。传给 BM25Okapi 的是「每块分好词的列表」组成的列表(42 个),它在内部把每个词的 IDF、每块长度、平均长度都预先算好。
• search 分三步。get_scores 返回 42 个分数,第 i 个对应第 i 块;sorted(range(len(scores)), key=lambda i: -scores[i])[:k] 排的是下标 0–41,按分数从高到低取前 k 个,相当于 TS 的 [...Array(n).keys()].sort((a, b) => scores[b] - scores[a]).slice(0, k);最后把下标换回 (块 id, 分数),丢掉 0 分的(一个词都没匹配上),float() 是把 numpy 的数字转成普通 Python float。纯 BM25 搜「把钱要回来」返回空,就是 42 块全是 0 分、在这一步被全部过滤了。

第一次运行会打印三条 SyntaxWarning: invalid escape sequence,是 jieba 源码里正则没用 r"..." 原始字符串,新版 Python 编译时会警告,不影响结果。第二次运行就没有了:Python 第一次导入时把源码编译成 .pyc 缓存,之后直接读缓存,不再重新编译,警告也就不再出现。

这段不调任何 API,我在本机跑过,输出是这样的(分数是 BM25 原始分,只在同一个查询内部可比):

导入流水提示 E1042 怎么办 ['导入', '流水', '提示', 'e1042']
     8.26  kb-010#2
     7.00  kb-010#1
     4.03  kb-009#1
E1043 是什么错误 ['e1043', '错误']
     8.06  kb-011#1
     6.31  kb-011#2
企业版年付中途取消能退多少 ['企业版', '年付', '中途', '取消', '退', '多少']
    12.02  kb-002#1
     7.75  kb-002#2
     4.18  kb-006#1

E1042 和 E1043 各自精确命中了自己那篇,这就是 BM25 的价值。第三个查询加了业务词典后,「企业版」「年付」都切对了。但查询里的「退」和文档里的「退款」是两个不同的词,BM25 匹配不上。它照样找对了,靠的是「企业版」「年付」「中途」「取消」这些词。这也说明了 BM25 的短板:用户说「退」、文档写「退款」,用户说「把钱要回来」、文档写「退款」,它都无能为力。这部分正好是向量检索擅长的。两者互补,所以要混合。

停用词表里的「时候」「报」是我实测踩坑后加的。最初的表里没有它们,结果「导入的时候报 E1043」这个问题,BM25 把《自动续费与扣款规则》排了第一。拆开看各个词的 IDF:「时候」在 42 块里只出现在 kb-004 一处,IDF 是 3.32;「e1043」出现在 kb-011 的两块里,IDF 只有 2.79。一个口语虚词的权重压过了错误码,原因就是它在语料里碰巧少见。所以 BM25 的停用词表不是网上抄一份就完事,要拿真实的用户问题跑一跑,看哪些口语词在捣乱。

还有一个 rank_bm25 的小坑:它的 IDF 计算在文档很少时会出现 0 甚至负数(一个词出现在一半以上的文档里时)。比如你用 2 篇文档测试,分数全是 0,别以为代码写错了。我们有 42 块,大部分词不会碰到这个问题。

RRF:把两路结果合起来

现在有两份排名:向量检索的 top-20 和 BM25 的 top-20。怎么合成一份?

最直觉的想法是把两边的分数加起来。但这行不通:余弦相似度在 0 到 1 之间,BM25 分数可能是 0 到 20,量纲完全不同。你可以做归一化,但不同查询的分数分布又不一样,很难调。

RRF(Reciprocal Rank Fusion,倒数排名融合)的做法很聪明:不看分数,只看名次。每一路里排第 r 名的结果,得 1 / (k + r) 分,k 是常数,通常取 60。一个结果在几路里的得分加起来,就是它的最终得分。

手算一个例子。向量检索结果是 [kb-011#1, kb-010#1, kb-009#2],BM25 结果是 [kb-010#1, kb-010#2]:

• kb-010#1:向量第 2 名,得 1/62 = 0.01613;BM25 第 1 名,得 1/61 = 0.01639;合计 0.0325
• kb-011#1:只在向量里,第 1 名,1/61 = 0.0164
• kb-010#2:只在 BM25 里,第 2 名,1/62 = 0.0161
• kb-009#2:只在向量里,第 3 名,1/63 = 0.0159

kb-010#1 在两路里都排得靠前,所以融合后排第一。这正是 RRF 的思想:被多路同时认可的结果更可信。k=60 的作用是把分数压平,让第 1 名和第 5 名的差距没那么夸张,避免某一路的第一名一家独大。

# agent-60-lab/w3/fusion.py
from collections import defaultdict


def rrf(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """Reciprocal Rank Fusion:每一路排名第 r 的结果得 1/(k+r) 分,各路相加。只看名次,不看原始分数。"""
    scores: dict[str, float] = defaultdict(float)
    for ranking in rankings:
        for rank, cid in enumerate(ranking, start=1):
            scores[cid] += 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda kv: -kv[1])


if __name__ == "__main__":
    vector = ["kb-011#1", "kb-010#1", "kb-009#2"]
    keyword = ["kb-010#1", "kb-010#2"]
    for cid, s in rrf([vector, keyword]):
        print(f"{s:.4f}  {cid}")

跑一下 python3 w3/fusion.py,输出和上面手算的四行一致。

逐行看 rrf():

• defaultdict(float):访问不存在的 key 时自动当 0.0,所以能直接 +=,不用先判断 key 在不在。TS 里得写 scores[cid] = (scores[cid] ?? 0) + x。
• 外层循环遍历每一路(向量一路、BM25 一路),内层 enumerate(ranking, start=1) 同时给出名次和块 id,名次从 1 开始,相当于 TS 的 ranking.forEach((cid, i) => { const rank = i + 1 })。同一个块在两路里都出现,就会被 += 两次,这就是「多路认可的排前面」的来源。
• sorted(scores.items(), key=lambda kv: -kv[1]):items() 得到 (id, 分数) 元组,kv[1] 是分数,负号表示从大到小。
• 自己敲时常见两个笔误:返回类型写成 tuple(str, float)(类型标注只能用方括号 tuple[str, float],圆括号会在运行时报错);循环变量叫 rid、下一行却用 cid,会报 NameError。

RRF 简单到有点不像「算法」,但它非常好用,Elasticsearch、OpenSearch、各种向量数据库的混合检索默认都是它。它来自 2009 年 Cormack 等人在 SIGIR 发表的论文,论文的结论是这个简单方法打败了当时很多复杂的融合方法。

Rerank:先粗筛,再精排

混合检索解决了「找得全」的问题,但排序还不够精。这里要引入一个新角色:rerank 模型(重排序模型)。

要理解 rerank,先要分清两种模型结构:

bi-encoder(双塔),也就是 embedding 模型。 问题和文档分别编码成向量,再算余弦。好处是文档向量可以提前算好存起来,检索时只需要编码问题,再做一次向量比较,百万级文档也是毫秒级。坏处是问题和文档在编码时互相「看不见」,只能各自压缩成一个向量再比,细节丢得多。

cross-encoder(交叉编码),也就是 rerank 模型。 把问题和一篇文档拼在一起送进模型,模型同时看两者,直接输出一个相关性分数。因为能逐字对照,它判断得准得多,「E1043」和「E1042」这种差别它能看出来,「支持」和「不支持」也能。坏处是不能提前算,每个(问题,文档)对都要跑一次模型,对 100 万篇文档做 rerank 是不可能的。

所以标准做法是两阶段:先用便宜的方法(向量 + BM25)从全部文档里粗筛出 20–50 个候选,再用贵的 rerank 模型精排,取前 3–5 个给大模型。这和电商搜索、推荐系统的「召回 → 排序」是同一个思路,做过推荐的同学会很熟悉。

我们用的 rerank 模型是 BAAI/bge-reranker-v2-m3,和 bge-m3 同一家,中文好。2026-10-12 我查到硅基流动上它的标准版是免费的。API 不是 OpenAI 协议(OpenAI 没有 rerank 接口),而是 Cohere 风格的 /rerank,请求体是 model、query、documents、top_n,返回 results 列表,每项有 index(对应你传入的第几篇)和 relevance_score。格式以硅基流动 rerank 文档为准。

# agent-60-lab/.env  新增(不填就沿用 EMB_* 那组)
RERANK_BASE_URL=https://api.siliconflow.cn/v1
RERANK_API_KEY=sk-xxxx
RERANK_MODEL=BAAI/bge-reranker-v2-m3

# agent-60-lab/w3/rerank.py
"""调 /rerank 接口(Cohere 风格,硅基流动、Jina、Cohere 等都兼容这个形状)。"""
import os

import httpx
from dotenv import load_dotenv

load_dotenv()
BASE = os.environ.get("RERANK_BASE_URL", os.environ["EMB_BASE_URL"]).rstrip("/")
KEY = os.environ.get("RERANK_API_KEY", os.environ["EMB_API_KEY"])
MODEL = os.environ.get("RERANK_MODEL", "BAAI/bge-reranker-v2-m3")


def rerank(query: str, documents: list[str], top_n: int = 4) -> list[tuple[int, float]]:
    """返回 [(原列表下标, 相关性分数), ...],按分数从高到低。"""
    if not documents:
        return []
    resp = httpx.post(
        f"{BASE}/rerank",
        headers={"Authorization": f"Bearer {KEY}"},
        json={"model": MODEL, "query": query, "documents": documents,
              "top_n": min(top_n, len(documents)), "return_documents": False},
        timeout=30,
    )
    resp.raise_for_status()
    results = resp.json()["results"]
    return [(r["index"], float(r["relevance_score"])) for r in results]


if __name__ == "__main__":
    docs = [
        "错误码 E1042:银行流水导入失败。列名和模板不匹配,或者日期格式不是 YYYY-MM-DD。",
        "错误码 E1043:文件编码错误。文件不是 UTF-8 编码,Excel 另存 CSV 默认是 GBK。",
        "导入银行流水:支持 CSV 和 XLSX,单个文件不超过 10 MB。",
    ]
    for i, s in rerank("导入的时候报 E1043 怎么办", docs, top_n=3):
        print(f"{s:.3f}  {docs[i][:20]}")

uv add httpx
uv run python w3/rerank.py

开头读配置那三行:os.environ.get("RERANK_BASE_URL", os.environ["EMB_BASE_URL"]) 是「先读 RERANK_BASE_URL,没配就用 EMB_BASE_URL」;.rstrip("/") 去掉右边末尾的 /(r = right,相当于 TS 的 .replace(/\/+$/, ""))。因为下面是 f"{BASE}/rerank" 拼地址,.env 里如果写成 .../v1/,不去掉就会拼出 .../v1//rerank,有的服务直接 404。lstrip 去左边、strip 去两边,不传参数时去的是空格和换行。

rerank() 本身就是一次 POST,相当于 TS 的 fetch(url, { method: "POST", headers, body: JSON.stringify(...) })。几个细节:

• if not documents: return []:空列表直接返回,省一次请求,有的服务收到空列表会报 400。
• 用 httpx 而不是 openai SDK,因为 OpenAI 协议里没有 rerank 接口。json= 会自动序列化,timeout=30 防止卡死。
• top_n 取 min(top_n, len(documents)),不能超过文档数;return_documents: False 让响应里不带原文,只要下标和分数。
• resp.raise_for_status():状态码 4xx/5xx 时抛异常。fetch 遇到 4xx/5xx 不会抛,要自己判断 resp.ok,这行就相当于 if (!resp.ok) throw ...。
• 响应是 {"results": [{"index": 1, "relevance_score": 0.895}, ...]},已按分数从高到低排好。index 是你传进去的 documents 的下标。返回下标而不是文本,是因为调用方手里有完整的 Chunk,retrieve.py 最后用 candidates[i] 取回块 id、标题等信息。

你应该看到 E1043 那条排第一,而且分数和第二名拉开明显的差距。我实测的输出:

0.895  错误码 E1043:文件编码错误。文件不
0.038  错误码 E1042:银行流水导入失败。列
0.017  导入银行流水:支持 CSV 和 XLSX

embedding 眼里这两篇很像,rerank 直接拉开到 0.895 对 0.038。这就是 cross-encoder「逐字对照」的效果。rerank 分数的另一个用处是做「不相关」判断:如果精排后的最高分都很低,说明知识库里大概率没有答案,可以直接走「转人工」,不用浪费一次大模型调用。阈值同样要在 Day 19 的评估集上定。

rerank 的代价要心里有数。每次查询多一次网络调用,对 20–30 个候选打分,通常增加几十到几百毫秒延迟。如果自己部署 bge-reranker,需要 GPU 才快。客服这种场景,多 200 毫秒换准确率,非常值。

Query 改写:用户的问题往往不适合直接搜

用户的问题经常是这样的:

• 「那年付的呢?」(依赖上一轮对话,单独拿出来没法搜)
• 「我上周买的那个东西能退不」(口语,关键信息「年付/月付」「专业版/企业版」都没说)
• 「发票开错了,还有导入流水一直报错,怎么弄」(一句话两个问题)

直接拿这些去检索,效果都不好。query 改写就是在检索之前,让大模型把问题处理一下。常见的三种:

补全为独立问题。 结合对话历史,把「那年付的呢?」改写成「年付套餐能不能退款」。多轮对话的 RAG 几乎必做。

多路查询(multi-query)。 让模型从不同角度生成 2–3 个查询,比如「退款政策」「申请退款步骤」「已开发票如何退款」,每个都去检索,结果用 RRF 融合。适合一句话里包含多个问题,或者问题很模糊的情况。

HyDE(Hypothetical Document Embeddings)。 先让模型「假装」写一段答案,用这段假答案去做向量检索。原理是假答案的措辞更像文档,和真文档在向量空间里更近。它来自 2022 年的论文 Precise Zero-Shot Dense Retrieval without Relevance Labels。我在实际项目里用得不多,因为它多一次生成调用、增加延迟,而且假答案里编造的细节有时会把检索带偏。知道这个名字,面试能讲出原理和缺点就够了。

# agent-60-lab/w3/rewrite.py
"""用大模型改写查询:补全上下文,拆成 1–3 个适合检索的短查询。"""
import json

from llm import chat

PROMPT = """你负责把用户发给 CloudLedger 客服的话,改写成适合在帮助中心里检索的查询。
要求:
- 结合对话历史补全省略的信息(例如「那年付的呢」要补成完整问题)。
- 如果一句话里有多个问题,拆开,每个问题一条查询。
- 每条查询是简短的陈述或问题,使用帮助中心的说法(例如用「退款」而不是「把钱要回来」)。
- 只改写用户最新这句话问的内容。历史只用来补全省略和指代,历史里已经问过的问题不要再加进来。
- 最新的话如果是寒暄、致谢这类不需要查资料的,原样返回这句话。
- 最多 3 条。只输出 JSON:{"queries": ["...", "..."]}"""


def rewrite(question: str, history: list[dict] | None = None) -> list[str]:
    history_text = "\n".join(f'{m["role"]}: {m["content"]}' for m in (history or [])[-6:])
    user = f"对话历史:\n{history_text or '(无)'}\n\n用户最新的话:{question}"
    raw = chat(
        [{"role": "system", "content": PROMPT}, {"role": "user", "content": user}],
        temperature=0,
        response_format={"type": "json_object"},
    )
    try:
        queries = json.loads(raw).get("queries", [])
        return [q for q in queries if isinstance(q, str) and q.strip()][:3] or [question]
    except json.JSONDecodeError:
        return [question]  # 改写失败就用原问题,检索不能因为改写挂掉


if __name__ == "__main__":
    print(rewrite("发票开错了,还有导入流水一直报 E1043,怎么弄"))
    history = [{"role": "user", "content": "月付的能退款吗"},
               {"role": "assistant", "content": "月付套餐不支持退款……"}]
    print(rewrite("那年付的呢", history))

这里用了 Day 6 的 JSON mode(response_format={"type": "json_object"})。DeepSeek 和大部分 OpenAI 兼容服务都支持,如果你用的服务不支持,去掉这个参数,靠 prompt 约束加 try/except 兜底。注意兜底逻辑:改写失败就用原问题。检索链路上的任何增强步骤失败了,都应该降级,而不是让整个请求报错。

逐行看 rewrite():

• (history or [])[-6:]:没传历史就当空列表(TS 的 history ?? []),只取最后 6 条(约 3 轮,.slice(-6)),太早的对话和当前问题关系不大还费 token。每条变成 user: 月付的能退款吗 这样一行,再用 "\n".join 拼起来。
• history_text or '(无)':历史是空字符串时写「(无)」,相当于 TS 的 ||。
• json.loads(raw).get("queries", []):JSON.parse 后取 queries 字段,取不到就是空列表。
• 列表推导式只留「是字符串且不是空白」的项,防模型返回 null 或 "";[:3] 最多三条;or [question] 过滤完是空的就退回原问题。
• PROMPT 里「只改写最新这句话」和「寒暄原样返回」两条是 Day 21 做多轮问答时补的。没有这两条时,我在聊完发票之后问「你们有学生优惠吗」,改写出来是 ['学生优惠', '发票抬头错误如何处理', '退款申请流程'],把上一轮的话题又带了进来;说「好的谢谢」也会被改成两个发票问题。加上之后分别是 ['学生优惠'] 和 ['好的谢谢'],上面两个例子的输出不受影响。
• 小漏洞:模型如果返回 JSON 数组而不是对象,.get 会抛 AttributeError,except json.JSONDecodeError 接不住。要更稳就写 except (json.JSONDecodeError, AttributeError):。开了 JSON mode 后很少遇到。

我用 deepseek-flash 实测的输出(每次措辞会略有不同):

['发票开错了怎么处理', '导入流水报错 E1043 怎么解决']
['年付套餐是否支持退款']

改写不是免费的:多一次大模型调用,增加几百毫秒到一两秒的延迟,也多花钱。我的做法是:多轮对话里只要有历史就改写;单轮问题先不改写,等评估数据证明有收益再开。另外改写有时会「改坏」,比如把用户说的「E1043」改成了「编码错误」,错误码丢了,BM25 反而找不到了。所以多路查询里保留原问题作为其中一路,下面的代码就是这么做的。

把它们串成一条流水线

# agent-60-lab/w3/retrieve.py
"""完整检索流水线:(改写) → 向量 + BM25 多路召回 → RRF → (rerank) → top-k。"""
from bm25_index import BM25Index
from chunking import Chunk, load_chunks
from embed import embed_one
from fusion import rrf
from pg_store import connect
from pg_store import search as pg_search
from rerank import rerank
from rewrite import rewrite

CHUNKS: dict[str, Chunk] = {c.id: c for c in load_chunks()}
BM25 = BM25Index(list(CHUNKS.values()))
CONN = connect()


def vector_ids(query: str, pool: int) -> list[str]:
    return [h["id"] for h in pg_search(CONN, embed_one(query), k=pool)]


def bm25_ids(query: str, pool: int) -> list[str]:
    return [cid for cid, _ in BM25.search(query, pool)]


def retrieve(question: str, k: int = 4, mode: str = "hybrid+rerank",
             history: list[dict] | None = None, use_rewrite: bool = False,
             pool: int = 20) -> list[tuple[Chunk, float]]:
    queries = [question]
    if use_rewrite:
        queries += [q for q in rewrite(question, history) if q != question]

    if mode == "vector":
        return [(CHUNKS[c], 0.0) for c in vector_ids(question, k)]
    if mode == "bm25":
        return [(CHUNKS[c], s) for c, s in BM25.search(question, k)]

    rankings = []
    for q in queries:
        rankings.append(vector_ids(q, pool))
        rankings.append(bm25_ids(q, pool))
    fused = rrf(rankings)[:pool]

    if mode == "hybrid":
        return [(CHUNKS[c], s) for c, s in fused[:k]]

    candidates = [CHUNKS[c] for c, _ in fused]
    # 单轮用原话打分;多轮的原话常是半句(「那退款前要做什么准备吗」),改用补全后的问法
    rerank_q = "；".join(queries[1:]) if history and len(queries) > 1 else question
    ranked = rerank(rerank_q, [c.text for c in candidates], top_n=k)
    return [(candidates[i], s) for i, s in ranked]

rerank 拿什么去打分,分两种情况。单轮问题用用户的原话:改写是为了召回得更全,但判断「这块资料能不能回答用户」时应该对照原话,况且改写偶尔会把 E1043 这样的关键信息改丢。多轮追问就不行了。Day 21 实测,上一轮问完「年付专业版买了十天能退吗」,接着问「那退款前要做什么准备吗」,拿这半句原话去 rerank,第一名只有 0.037,被拒答阈值拦了下来;换成改写补全后的「年付专业版退款前需要准备什么」,第一名是 0.075,正常回答。所以有历史、而且改写出了新问法时,把这几条问法用「；」拼起来交给 rerank。多问题的情况也照顾到了:「发票抬头也写错了,一起处理」改写成发票和退款三条,拼起来打分,发票和退款两篇的块都排进了前两名(0.994、0.919)。

retrieve.py 逐段在干什么? 打个比方:两个图书管理员同时去找资料,一个懂意思(向量),一个只认字(BM25),各拿回 20 页;把两人的名单合并,两人都排得靠前的页往前放(RRF);最后请专家把这 20 页逐页和问题对照,挑出最好的 4 页(rerank)。

• 文件顶部那三行(CHUNKS、BM25、CONN)在 import 时就执行,只跑一次,之后每次查询都复用。代价是一 import retrieve 就会去连数据库、建索引,Postgres 没开的话 import 这一步就报错。
• CHUNKS 是 {id: 块} 字典,召回函数只返回 id,最后靠它把 id 换回块。
• vector_ids、bm25_ids 都只要 id、不要分数,因为 RRF 只看名次。向量距离和 BM25 分数量纲不同,本来就没法直接比。for cid, _ in ... 里的 _ 表示这个值不要。
• queries:默认只有原问题。开了 use_rewrite,LLM 会再给几种问法(和原问题一样的去掉)。history 只给改写用,处理「它的价格呢」这类指代。
• rankings:每种问法 × 两路各一份名单,1 个问法得到 2 份,改写成 3 个问法就是 6 份。rrf(...)[:pool] 合并后留前 20 个。
• pool=20 和 k=4:先多捞一些保证召回率,再精挑保证精度,这就是「召回 → 精排」漏斗。
• rerank_q:没有历史(单轮)、或者改写没给出新问法时,queries[1:] 为空,用原问题;多轮且有改写时,用 "；".join(queries[1:]),比如 '发票抬头填写错误如何修改；已开发票如何申请发票红冲；退款前发票红冲流程'。queries[0] 是原问题,所以从 1 开始取。
• rerank 只认文字,所以传 c.text。它返回的是 (下标, 分数),比如 [(7, 0.92), (2, 0.85)],再用 candidates[7] 把块取回来。

几个容易忽略的地方:

• 改写只在混合模式下生效。 vector、bm25 两个单路模式只用原问题,queries 算出来也没用上,所以 vector 加不加 use_rewrite,结果都一样。
• 四种模式返回的分数含义不同: vector 模式是 0.0 占位,bm25 是 BM25 分,hybrid 是 RRF 分(很小,约 0.01–0.05),rerank 是 rerank 模型分。不能跨模式比分数,也不能设一个统一阈值。
• 成本: 开改写后,假设改出 3 个问法,一次检索就是 1 次 LLM 改写 + 3 次 embedding + 3 次 pg 查询 + 1 次 rerank,延迟明显变大,所以 use_rewrite 默认关闭。

三路各自擅长什么:

再写一个对比脚本,把四种模式并排打出来:

# agent-60-lab/w3/compare_retrieval.py
from retrieve import retrieve

CASES = [
    ("导入的时候报 E1043", "kb-011"),
    ("E1042 是什么意思", "kb-010"),
    ("买错了能把钱要回来吗", "kb-002"),
    ("这个月被重复扣了两次钱", "kb-002"),
    ("发票抬头写错了", "kb-007"),
    ("API 调用返回 429", "kb-019"),
]

MODES = ["vector", "bm25", "hybrid", "hybrid+rerank"]

if __name__ == "__main__":
    hits = {m: 0 for m in MODES}
    for q, expect in CASES:
        row = []
        for m in MODES:
            top = retrieve(q, k=1, mode=m)
            got = top[0][0].doc_id if top else "-"
            hits[m] += got == expect
            row.append(f"{m}:{got}{'✓' if got == expect else '✗'}")
        print(f"{q:<16} 期望 {expect}  " + "  ".join(row))
    print("top-1 命中:", hits)

跑出来后把结果填进这张表(我不替你填,因为结果取决于你用的 embedding 模型):

跑一次要调 18 次 embedding、6 次 rerank,大约 8 秒,费用可以忽略。

我预期的规律是:纯 BM25 在错误码类问题上稳,在「把钱要回来」这种口语问题上容易失手;纯向量反过来;混合两类都不差;rerank 让排序再稳一点。如果你的结果和预期不一致,那就更有意思了,去看具体哪一块排在了前面、为什么。6 个问题太少,说明不了统计意义,这正是明天要建 30 条评估集的原因。

导入的时候报 E1043   期望 kb-011  vector:kb-011✓  bm25:kb-011✓  hybrid:kb-011✓  hybrid+rerank:kb-011✓
E1042 是什么意思     期望 kb-010  vector:kb-010✓  bm25:kb-010✓  hybrid:kb-010✓  hybrid+rerank:kb-010✓
买错了能把钱要回来吗  期望 kb-002  vector:kb-002✓  bm25:-✗       hybrid:kb-002✓  hybrid+rerank:kb-003✗
这个月被重复扣了两次钱 期望 kb-002  vector:kb-002✓  bm25:kb-002✓  hybrid:kb-002✓  hybrid+rerank:kb-002✓
发票抬头写错了       期望 kb-007  vector:kb-007✓  bm25:kb-007✓  hybrid:kb-007✓  hybrid+rerank:kb-007✓
API 调用返回 429    期望 kb-019  vector:kb-019✓  bm25:kb-019✓  hybrid:kb-019✓  hybrid+rerank:kb-019✓
top-1 命中: {'vector': 6, 'bm25': 5, 'hybrid': 6, 'hybrid+rerank': 5}

Contextual Retrieval:把上下文写进每一块

最后介绍一个 2024 年很有影响的做法。Day 16 我们在每块前面加了「文章标题 > 小节标题」,Anthropic 把这个思路推进了一大步,叫 Contextual Retrieval:在建索引时,对每一块,把整篇文档和这一块一起交给大模型,让它写一两句话说明「这一块在整篇文档里讲的是什么」,通常 50 到 100 个 token,然后把这句说明加在块的前面,再去做 embedding 和 BM25。

他们的实验结果(在代码库、小说、论文等多个数据集上,指标是 top-20 检索失败率):只做 contextual embedding,失败率下降 35%(5.7% → 3.7%);加上 contextual BM25,下降 49%;再加 rerank,下降 67%。成本方面,他们算出借助 prompt caching,生成这些上下文的一次性成本大约是每百万文档 token 1.02 美元。

这个方法的本质和我们加标题前缀一样:让每一块「自带上下文」。区别是让大模型来写上下文,比机械地拼标题更聪明,能写出「这一段讲的是年付套餐中途取消时按剩余整月退款的规则」这样的话。代价是建索引时每块多一次大模型调用。对我们 42 块的帮助中心,花不了几分钱,你可以作为练习 4 试试。

上游:文档解析(PDF、表格、扫描件)

前面所有的检索技巧,都默认了一个前提:文档已经是干净的文字。CloudLedger 的帮助中心是我们自己造的 Markdown,当然干净。真实项目里几乎从来不是这样。客户给你的知识库,往往是一堆 PDF 格式的产品手册和合同、Excel 价格表、Word 制度文件、扫描的发票和盖章文件,还有截图。国内 Agent 岗的 JD 里,「文档解析」「多格式文档处理」「表格和扫描件」出现得很频繁,原因就在这里:解析质量是检索质量的上游,解析错了,后面的 embedding、BM25、rerank 都是在垃圾上做功。

先分清 PDF 的三种情况,处理方法完全不同:

• 文字型 PDF:由 Word 或网页导出,里面有文字层,能直接抽出文字。大部分产品手册是这种。
• 扫描型 PDF:每一页其实是一张图片,没有文字层,抽文字得到空字符串。盖章合同、纸质档案、手机拍的单据都是这种。必须 OCR(光学字符识别)。
• 混合型:大部分页有文字层,夹着几页扫描的附件。要逐页判断。

判断方法很简单:逐页抽文字,抽出来是空的(而且页面上也没有表格),就当作扫描页走 OCR。

文字型 PDF 也有一堆坑,我按踩到的频率排:

表格被抽成散乱的文字。 一张四列的价格表,直接抽文字会变成「专业版 49 468 5」这样一行数字,哪个数是月付、哪个是年付,模型只能猜。表格要单独识别出来,转成 Markdown 表格保留结构。

页眉页脚每页重复。 「CloudLedger 产品手册 v3.2 第 7 页」在每一页都出现,切块后几十个块里都有这句话,BM25 和向量检索都会被它干扰。解析后要识别并去掉每页重复出现的行。

多栏排版的阅读顺序错乱。 双栏的论文或宣传册,按坐标从上到下抽,左栏第一行后面接着右栏第一行,语句全被打乱。需要版面分析(layout analysis)先识别出栏,再按栏的顺序读。

图片和图表里的信息。 流程图、截图里的文字,纯文本抽取拿不到。要么 OCR,要么用多模态模型给图片写一段描述。

下面是一个够用的最小版本:用 pdfplumber 逐页抽正文,识别有边框的表格转成 Markdown,没有文字层的页面渲染成图片交给 RapidOCR。RapidOCR 是基于 PaddleOCR 模型的轻量封装,中文识别不错,CPU 就能跑。

# agent-60-lab/w3/doc_parse.py
"""把 PDF 解析成适合切块的 Markdown:正文按页提取,表格转成 Markdown 表格,没有文字层的扫描页走 OCR。

依赖:uv add pdfplumber
OCR(可选):uv add rapidocr onnxruntime   第一次运行会下载模型
      Intel 芯片的 Mac 装不上新版 onnxruntime,改用:uv add rapidocr "onnxruntime<1.23"
"""
import sys
from pathlib import Path

import pdfplumber


def table_to_markdown(rows: list[list[str | None]]) -> str:
    # 换个变量名:rows 声明的是 str | None,洗成 str 后放进新变量,否则 join 会报类型红线
    clean = [[(c or "").replace("\n", " ").strip() for c in row] for row in rows if row and any(row)]
    if len(clean) < 2:
        return ""
    header, body = clean[0], clean[1:]
    lines = ["| " + " | ".join(header) + " |", "|" + " --- |" * len(header)]
    lines += ["| " + " | ".join(r + [""] * (len(header) - len(r))) + " |" for r in body]
    return "\n".join(lines)


_ocr_engine = None  # 模型加载慢,全局只建一次


def ocr_page(page) -> str:
    """扫描件:把这一页渲染成图片再 OCR。没装 rapidocr 就返回提示文字。"""
    global _ocr_engine
    try:
        import numpy as np
        from rapidocr import RapidOCR  # 没装时编辑器会标黄/红,正常
    except ImportError:
        return "(这一页是扫描图片,没有文字层;安装 rapidocr 后可 OCR)"
    if _ocr_engine is None:
        _ocr_engine = RapidOCR()
    img = page.to_image(resolution=200).original  # PIL 图片
    result = _ocr_engine(np.array(img))
    return "\n".join(result.txts or [])


def parse_pdf(path: Path, title: str | None = None) -> str:
    parts = [f"# {title or path.stem}"]
    with pdfplumber.open(path) as pdf:
        for i, page in enumerate(pdf.pages, start=1):
            tables = page.find_tables()
            # 正文里去掉落在表格区域内的字符,避免表格内容被当成散乱的句子再出现一遍
            boxes = [t.bbox for t in tables]

            def outside_tables(obj, boxes=boxes) -> bool:
                if "x0" not in obj:
                    return True
                x, y = (obj["x0"] + obj["x1"]) / 2, (obj["top"] + obj["bottom"]) / 2
                return not any(x0 <= x <= x1 and top <= y <= bottom for x0, top, x1, bottom in boxes)

            text = (page.filter(outside_tables).extract_text() or "").strip()
            if not text and not tables:
                text = ocr_page(page)
            parts.append(f"## 第 {i} 页\n{text}")
            for n, t in enumerate(tables, start=1):
                md = table_to_markdown(t.extract())
                if md:
                    parts.append(f"### 第 {i} 页 表 {n}\n{md}")
    return "\n\n".join(parts)


def table_rows_to_chunks(table_title: str, rows: list[list[str | None]]) -> list[str]:
    """大表格按行切块,每一块都带上表名和表头,单独检索到一行时也看得懂。"""
    clean = [[(c or "").strip() for c in row] for row in rows if row and any(row)]
    header, body = clean[0], clean[1:]
    return [f"{table_title}:" + ";".join(f"{h}={v}" for h, v in zip(header, row) if v) for row in body]


if __name__ == "__main__":
    print(parse_pdf(Path(sys.argv[1])))

我用 reportlab 生成了一个带表格的两页测试 PDF,跑出来是这样的(英文内容是因为生成 PDF 时没配中文字体,解析逻辑和中文一样):

# price

## 第 1 页
CloudLedger plans. Annual plans can be refunded within 14 days.

### 第 1 页 表 1
| Plan | Monthly | Annual | Ledgers |
| --- | --- | --- | --- |
| Free | 0 | 0 | 1 |
| Pro | 49 | 468 | 5 |
| Enterprise | 199 | 1908 | unlimited |

## 第 2 页
Second page text about invoices.

table_to_markdown() 在干什么? pdfplumber 的 t.extract() 抽出来的表格是 list[list[str | None]]:空单元格是 None,单元格里常带换行。这个函数把它转成 Markdown 表格。拿一个故意弄脏的输入试一下:

rows = [
    ["套餐", "价格", "备注"],
    ["基础版", "99", None],
    [None, None, None],            # 全空行
    ["专业版", "199", "含\n客服"],  # 带换行
    ["企业版"],                     # 列数不够
]

| 套餐 | 价格 | 备注 |
| --- | --- | --- |
| 基础版 | 99 |  |
| 专业版 | 199 | 含 客服 |
| 企业版 |  |  |

• clean = [[... for c in row] for row in rows if ...]:两层列表推导式,先看外层再看里层。外层 if row and any(row) 去掉空列表和整行全空的行(any 相当于 TS 的 row.some(Boolean))。里层逐个单元格处理:c or "" 把 None 变成空字符串(对 None 调 .replace 会报错);换行换成空格,因为 Markdown 表格的一行不能断开;.strip() 去首尾空格。
• if len(clean) < 2: return "":至少要有表头加一行数据,否则算不上表格。
• header, body = clean[0], clean[1:]:默认第一行是表头。
• "|" + " --- |" * len(header):字符串乘数字就是重复,3 列得到 | --- | --- | --- |。Markdown 靠这行分隔线识别表格。
• r + [""] * (len(header) - len(r)):列数不够就补空字符串。PDF 抽表常遇到合并单元格或缺列,补齐后每行竖线数量一致,表格才不会错位。
• 小漏洞:只处理了列数少的情况。某行比表头多一列时,[""] * -1 是 [],什么都不补,这一行多出一列,渲染时多出来那列通常被直接丢掉,内容就没了。更稳的写法是先补齐再截断:(r + [""] * len(header))[:len(header)]。

为什么要转 Markdown:表格直接抽成纯文本会变成「套餐 价格 备注 基础版 99 专业版 199」,行列对应关系全丢了,模型分不清 199 是哪个套餐的价格。转成 Markdown 后结构还在,大模型读 Markdown 表格很在行。

调这段代码时我踩了两个坑,都值得记下来。第一个:最开始我用 page.outside_bbox(表格范围) 去掉表格区域的文字,结果表格上方那一行正文被截成了「CloudLedger plans. Annua」,后半截不见了。原因是这个裁剪方法会把和表格范围有交叠的字符一起裁掉。改成按每个字符的中心点判断是否落在表格内,才正确。第二个:同一张表如果没有画边框线,pdfplumber 默认的表格识别(按线条找表格)根本识别不出来,整张表会被当成普通文字抽出来,又变回「Pro 49 468 5」。真实文档里无框表格非常多,这时候要么调 pdfplumber 的 table_settings 改成按文字对齐识别(容易误判),要么换下面说的版面分析工具。

表格怎么切块。 小表格整张作为一块就行。大表格(几十上百行的价格表、对照表)整张放一块,向量会被稀释;按行切又会丢掉表头,单独检索到「专业版 49 468 5」也看不懂。我的做法是按行切,每一块都带上表名和表头,也就是上面 doc_parse.py 里的这个函数:

def table_rows_to_chunks(table_title: str, rows: list[list[str | None]]) -> list[str]:
    """大表格按行切块,每一块都带上表名和表头,单独检索到一行时也看得懂。"""
    clean = [[(c or "").strip() for c in row] for row in rows if row and any(row)]
    header, body = clean[0], clean[1:]
    return [f"{table_title}:" + ";".join(f"{h}={v}" for h, v in zip(header, row) if v) for row in body]

CloudLedger 套餐价格表:套餐=免费版;月付(元)=0;年付(元)=0;账本数=1
CloudLedger 套餐价格表:套餐=专业版;月付(元)=49;年付(元)=468;账本数=5
CloudLedger 套餐价格表:套餐=企业版;月付(元)=199;年付(元)=1908;账本数=不限

这和 Day 16「块前加标题」、Contextual Retrieval 是同一个思想:每一块都要自带足够的上下文。用户问「专业版年付多少钱」,BM25 能精确命中「专业版」「年付」,向量也能匹配上,模型读到这一块也不会搞错是哪一列。如果表格的问题主要是「筛选和计算」(比如「哪些套餐年付低于 1000 元」),那就不该走 RAG,而是把表格导进数据库,做成 SQL 查询工具,这是 W4 的内容。

工具怎么选。 手写的 pdfplumber 版本适合格式规整的文字型 PDF。复杂文档建议直接用成熟的解析工具,我按自己的使用印象整理了一下,具体能力以各项目的文档为准:

我的习惯是:先用最简单的工具跑一遍,抽 20 页人工检查解析结果,尤其是表格和扫描页;哪类页面解析得差,再针对性地换工具。解析结果要和评估集联动,比如评估集里专门加几条「答案在表格里」的问题,解析方式改了就重跑一次检索评估。

代码库检索:代码不能按字数切

前面切的都是自然语言。如果你的 Agent 要回答「退款逻辑在哪个函数里」「改了 chat() 会影响谁」,知识库就变成了一个代码仓库。国内不少 JD 写「代码知识库」「研发助手」,指的就是这件事;Day 55 的代码审查 Agent 和 Day 89 的 coding agent 也离不开它。

代码按 500 字切块是灾难。一个函数被拦腰切成两半,前半块有函数名没有返回逻辑,后半块有逻辑却不知道属于谁。代码天然有结构,所以切块要按语法树来:一个函数一块,一个类一块(类太大再按方法切),每块前面带上文件路径。Python 自带 ast 模块就能做;多语言项目一般用 tree-sitter,它能解析几十种语言,是这类工具的标配。

光有块还不够。问「改了 chat() 会影响谁」,向量检索找到的是长得像 chat 的代码,而真正要的是调用关系。这里值得看一下 aider 的做法。aider 是一个开源的命令行 coding agent,它每次请求都会附带一张 repo map:列出仓库里的文件,以及每个文件里最重要的符号和它们的定义行。aider 用 tree-sitter 抽出每个文件定义了什么、引用了什么,把文件当节点、依赖当边建一张图,用图排序算法挑出被引用最多的那些符号。这张图默认只占 1k token 的预算(--map-tokens 参数),放不下就只留排名靠前的。模型看着这张地图,自己决定要打开哪几个文件细看。这些细节我是 2026 年 10 月对照 aider 文档 和它 2023 年那篇 repo map 博客 核对的。

下面是一个只用标准库的简化版:按函数和类切块,然后数每个名字在仓库里被引用了多少次,当作排名分数。真正的 aider 比这精细得多,但这几十行足够让你看到思路,以及简化在哪里会出问题。

# agent-60-lab/w3/code_chunks.py  代码按函数/类切块,再生成一张按被引用次数排序的 repo map
import ast
import sys
from collections import Counter
from pathlib import Path


def chunk_file(path: Path) -> list[dict[str, str]]:
    source = path.read_text(encoding="utf-8")
    tree = ast.parse(source)
    chunks = []
    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            code = ast.get_source_segment(source, node) or ""
            first_line = code.splitlines()[0]
            chunks.append({
                "id": f"{path}:{node.lineno}",
                "symbol": node.name,
                "signature": first_line.rstrip(":"),
                "text": f"# 文件 {path}\n{code}",
            })
    return chunks


def referenced_names(path: Path) -> Counter[str]:
    tree = ast.parse(path.read_text(encoding="utf-8"))
    names: Counter[str] = Counter()
    for node in ast.walk(tree):
        if isinstance(node, ast.Name):
            names[node.id] += 1
        elif isinstance(node, ast.Attribute):
            names[node.attr] += 1
    return names


def repo_map(root: Path, top: int = 8) -> str:
    files = sorted(p for p in root.rglob("*.py") if ".venv" not in p.parts)
    all_chunks = [c for f in files for c in chunk_file(f)]
    refs: Counter[str] = Counter()
    for f in files:
        refs.update(referenced_names(f))
    ranked = sorted(all_chunks, key=lambda c: refs[c["symbol"]], reverse=True)[:top]
    lines = [f"{len(files)} 个文件,{len(all_chunks)} 个函数/类,被引用最多的 {top} 个:"]
    for c in ranked:
        lines.append(f"{c['id']:<34} 引用 {refs[c['symbol']]:>2} 次  {c['signature']}")
    return "\n".join(lines)


if __name__ == "__main__":
    print(repo_map(Path(sys.argv[1] if len(sys.argv) > 1 else ".")))

在我自己的练习仓上跑 uv run python w3/code_chunks.py w2(只看 W2 目录),输出是:

23 个文件,32 个函数/类,被引用最多的 8 个:
w2/day09_demo.py:17                引用  7 次  def get_weather(
w2/weather.py:17                   引用  7 次  def get_weather(city: str, unit: str = "celsius") -> dict
w2/wf_common.py:41                 引用  6 次  def llm(prompt: str, system: str = "", model: str | None = None) -> str
w2/robust_agent.py:41              引用  5 次  class AgentResult
w2/calc.py:14                      引用  4 次  def calculate(expression: str) -> dict
w2/day09_demo.py:26                引用  4 次  def calculate(
w2/wf_common.py:25                 引用  4 次  def _params(
w2/wf_common.py:46                 引用  4 次  def llm_json(prompt: str, system: str = "", model: str | None = None) -> dict[str, Any]

看起来挺像样,llm、get_weather、calculate 确实是 W2 的核心。可是对整个仓库跑(.),第一名变成了 w3/llm.py:17 引用 38 次 def chat(...)。仓库里根本没有 38 处调用这个 chat,是所有 client.chat.completions.create 里的 chat 都被算到了它头上。再往下,w3/rag_numpy.py 的 answer 和 w3/rag_pg.py 的 answer 各显示 17 次,两个同名函数分不清谁是谁;main、run 也挤进了前 8,只因为名字常见。

这就是按名字计数的天花板:它不知道一个名字指向的是哪个定义。要做对,得知道「这一行的 chat 到底解析到哪个函数」,这需要真正的符号解析。aider 用 tree-sitter 的定义/引用查询加图排序,比计数好很多;更严格的做法是接语言服务器(LSP)或 SCIP 这类代码索引,拿到精确的跳转定义和查找引用结果。这也是为什么很多 coding agent 干脆不建向量库,直接给模型 grep、读文件、查引用这些工具,让它像人一样一层层往下翻。代码检索里,工具往往比 embedding 更重要,Day 89 读 coding agent 源码时你会看到这一点。

另一个方向:递归语言模型(RLM)

讲到这里你可能会想:既然模型能自己 grep、自己决定看哪些文件,那长文档是不是也可以不切块、不建索引,直接让模型自己去翻?2025 年底 MIT 的 Alex Zhang、Tim Kraska 和 Omar Khattab 发了一篇 Recursive Language Models(arXiv 2512.24601,2026 年 5 月更新到 v3),走的就是这个思路。超长输入不塞进 prompt,而是当成一个变量放在代码执行环境里,模型写代码去查看、切分它,对切出来的片段递归地调用自己处理,最后汇总。

论文报告的结果是:能处理超出模型上下文窗口两个数量级的输入;在四个长上下文任务上,GPT-5 加 RLM 的中位得分比上下文压缩高 26%,比带子调用的 CodeAct 高 130%,比 Claude Code 高 13%,成本相当;他们后训练的 8B 小模型 RLM-Qwen3-8B 比原版 Qwen3-8B 平均高 28.3%。作者开源了实现(alexzhang13/rlm)。这些数字是论文自己报的,我没有复现。

它和今天的 RAG 是两种对立的做法。RAG 是事先切好、建好索引,查询时一次性取回 top-k;RLM 是不预处理,查询时让模型现场决定怎么读。RLM 对「需要通读全文才能回答」的问题(比如总结一份 300 页合同里所有的违约条款)有优势,但每次查询都要跑很多次模型调用,延迟和成本比一次检索高得多。客服问答这种高频、短答案的场景,RAG 仍然是对的选择。面试里被问到「长上下文和 RAG 怎么选」,能把 RLM 当第三个选项讲出来,会显得你一直在跟进。

动手练习

1. 跑 bm25_index.py,把输出和上面我贴的对一下。然后去掉自定义词典那几行,再跑一次,看「企业版年付中途取消能退多少」的分词和检索结果有什么变化。
2. 跑 rerank.py,确认 E1043 排第一。把查询换成「导入报 E1042」再跑,看排序是否跟着变。
3. 跑 compare_retrieval.py,填完上面的对比表。挑一个混合检索也排错的问题,用 retrieve(q, k=5, mode="hybrid") 看前 5 名,分析原因。
4. (选做)实现简化版 contextual retrieval:改 chunking.py,对每一块调用一次大模型,输入整篇文章和这一块,让它写一句不超过 50 字的上下文说明,拼在块前面;重建 pgvector 和 BM25 索引,再跑一遍对比表。
5. 用 rewrite.py 试三个多轮对话的例子,至少找到一个「改写改坏了」的情况(比如丢了错误码),想想 prompt 怎么改。
6. 找一份真实的 PDF(产品说明书、带表格的报价单都行,注意别用含个人信息的文件),用 doc_parse.py 解析,人工检查表格和正文的解析质量。再找一张手机拍的文档照片转成 PDF,装上 rapidocr 试试 OCR。如果有时间,用 MinerU 或 Docling 解析同一份文件,对比效果。

验收标准: 对比表填完,并能用一两句话解释每一列的规律;retrieve() 的四种模式都能跑;能不看资料讲清楚 BM25、RRF、cross-encoder 三者各自解决什么问题。

名词解释

常见误区

把余弦相似度和 BM25 分数直接相加。 两者量纲不同、分布随查询变化,直接加等于随机加权。用 RRF,或者至少做归一化并在评估集上调权重。

中文直接按字切做 BM25。 按单字切,「退款」和「退出」都有「退」,噪音很大;按 jieba 默认切,业务词会切错。加业务词典是必须的。

对全部文档做 rerank。 cross-encoder 不能预先算,对每个候选都要跑一次模型。只对召回的几十个候选做。

改写后丢掉原问题。 改写可能丢失关键信息(错误码、订单号)。多路查询里保留原问题作为一路,rerank 也用原问题。

以为加了 rerank 就能救回召回阶段漏掉的文档。 rerank 只能重新排序它拿到的候选。正确答案没进 top-20,rerank 再强也没用。召回要宽,精排要准,两件事分开优化。

把 PDF 直接抽成纯文本就入库。 表格变成散乱的数字,页眉页脚重复几十遍,扫描页是空的。入库前至少人工抽查一部分解析结果。

大表格按行切块却不带表头。 检索到一行「专业版 49 468 5」,模型不知道哪个数是什么。每一块都要带表名和表头。

代码库也按字数切块。 函数会被拦腰切断,检索到的半块代码既不完整也不知道属于谁。按 AST 切,每块带文件路径;要回答调用关系类问题,还得有查引用的工具。

每加一个技巧都凭感觉判断好坏。 6 个问题的对比只是演示。真正的判断要靠评估集上的数字,这是明天的事。

学习资料

文档与文章

• Anthropic · Introducing Contextual Retrieval:今天最后一节的出处,把实验部分和成本计算读完,40 分钟。里面还附了一份 cookbook,演示了 BM25 + 向量 + rerank 的完整实现。
• Okapi BM25(维基百科):公式和参数的权威解释,10 分钟。
• Pinecone · Rerankers and Two-Stage Retrieval:bi-encoder 和 cross-encoder 的区别讲得很清楚,配图好。
• Elasticsearch · Reciprocal rank fusion:生产级搜索引擎里 RRF 的用法和参数,看看工业界怎么用。
• Sentence Transformers · Cross-Encoders:想自己本地跑 rerank 模型时看这里。
• HyDE 论文:读摘要和图 1。

B 站视频

其他经典视频

• Prompt Engineering 频道《The Best RAG Technique Yet? Anthropic’s Contextual Retrieval Explained!》(YouTube ID tmiBae2goJM,16 分钟):Contextual Retrieval 的讲解。
• Tejas Kumar《BM25 Algorithm and Hybrid Search: AI Explained》(YouTube ID lD3_VF5bmWg,10 分钟):BM25 公式的直观讲解。

书籍

• Christopher Manning 等《Introduction to Information Retrieval》(剑桥大学出版社,2008),作者提供免费在线版。第 6 章(打分、TF-IDF)和第 11 章(概率模型,BM25 的来历)。信息检索的经典教材,想打牢基础时读,4–6 小时。
• Chip Huyen《AI Engineering》第 6 章检索优化部分,讲了混合检索、rerank、query 改写的取舍。

开源项目与代码

• fxsjy/jieba:自定义词典、调整词频的用法都在 README 里。
• dorianbrown/rank_bm25:今天用的 BM25 实现,代码只有几百行,值得读一遍。
• BAAI/bge-reranker-v2-m3:模型卡,有本地运行的示例代码。
• jsvine/pdfplumber:今天文档解析用的库,README 里表格识别的 table_settings 参数值得细看。
• RapidAI/RapidOCR 和 PaddlePaddle/PaddleOCR:中文 OCR。
• opendatalab/MinerU、docling-project/docling、Unstructured-IO/unstructured:成熟的文档解析工具,复杂文档直接用它们。
• langchain-ai/rag-from-scratch:LangChain 的 RAG 从零系列 notebook,multi-query、HyDE、RAG-Fusion 都有实现,可以对照看框架版写法。

自测

1. 手算 RRF(k=60):向量检索结果 [A, B, C],BM25 结果 [C, D]。融合后的排名是什么?

2. 为什么 BM25 对错误码 E1043 特别敏感?

3. bi-encoder 和 cross-encoder 的根本区别是什么?为什么不能只用 cross-encoder?

4. 你在多路查询里把原问题去掉了,只用改写后的查询,可能出什么问题?

5. jieba 把「企业版年付」切成了「企业 / 版年付」,对检索有什么影响?怎么修?

6.(面试题)你的 RAG 系统召回率不错,但 top-1 经常不是最相关的那块,导致回答不准。你有哪些手段?怎么验证有效?

7.(面试题)要做一个能回答「改了这个函数会影响哪些地方」的研发助手,你会怎么做检索?

明天预告

Day 19 · RAG 怎么评估:召回率、MRR、忠实度,建第一份检索测试集。今天的对比表只有 6 个问题,结论全凭感觉。明天建一份 30 条的评估集(含答案所在的文档、还有几条「知识库里没有」的问题),写脚本算 recall@k 和 MRR,再用大模型当裁判检查回答是否忠于资料。然后用它回答几个今天悬而未决的问题:切块该多大、k 该取几、rerank 到底值不值。