生命不息，折腾不止。上一课那 80 行代码只够跑 Demo——文档一上千、一问到产品型号和英文缩写，就露馅了。今天给它装上三样升级件：按语义边界的切分、FAISS 落盘、BM25 混合检索，让你的私有知识库真正能吃下成百上千篇文档。

一、先给上一课的玩具挑挑毛病

上一课我们手搓的链路是：定长切分（300 字符 + 50 重叠）→ text-embedding-3-small 向量化 → 向量堆在 numpy 数组里 → 余弦相似度取 top-k → 喂给大模型。

跑个 Demo 完全没问题，但一上量就三个毛病：

  1. 内存扛不住。向量全堆在 numpy 数组里，几千篇文档、几十万个 chunk，内存直接打爆，进程一重启还全没了。
  2. 切分太粗暴。定长 300 字符，一句话可能被拦腰砍两半——前半块结尾是「承德在」，后半块开头是「河北省」，检索时两边都命不中，语义全断。
  3. 只认「意思」不认「字」。纯向量检索碰到精确词就跪：产品型号「B7-20X」、接口名「refresh_token」、报错码「ERR_500」，这些在向量空间里互相长得都差不多，但你的用户偏偏最爱搜这几种。

这三条，正好对应今天要装的三样零件。开工。

二、切分升级：别拦腰砍句子，按语义边界切

定长切分的本质问题就一句话：它不管语义边界在哪，到字数就切。正确做法是递归切分——优先按大边界（段落、换行）切，切不动再退到小边界（句号、逗号），最后才按字硬切。这样能尽量在自然停顿处断开，不拆散完整句子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
SEPARATORS = ["\n\n", "\n", "。", "！", "？", "，", " ", ""]

def recursive_split(text, target=300, seps=None):
    """递归切分：优先按大边界（段落/换行）切，切不动再降级到句子、逗号，最后按字硬切。"""
    seps = seps or SEPARATORS
    text = text.strip()
    if len(text) <= target:
        return [text] if text else []

    sep = seps[0]
    if sep == "":  # 兜底：无标点可切，按字硬切
        return [text[i:i + target] for i in range(0, len(text), target)]

    raw = text.split(sep)
    parts = [p + sep for p in raw[:-1]] + ([raw[-1]] if raw[-1] else [])  # 分隔符拼回，别丢标点

    chunks, buf = [], ""
    for p in parts:
        if buf and len(buf) + len(p) > target:
            chunks.append(buf)
            buf = p
        else:
            buf += p
    if buf.strip():
        chunks.append(buf)

    out = []
    for c in chunks:
        if len(c) > target and len(seps) > 1:
            out.extend(recursive_split(c, target, seps[1:]))  # 还超长，换下一级分隔符再切
        else:
            out.append(c)
    return out

几句话讲透它怎么干活：

  • 分隔符从大到小排队：\n\n（段落）→ \n（换行）→ 。！？（句子）→ ，（逗号）→ 空格 → 空串（按字硬切兜底）。
  • 先按最粗的 \n\n 切，攒到接近 300 字就收尾成一个 chunk；切完仍有超长的块，就换下一级分隔符递归再切。
  • 最后兜底：真碰到一个超长无标点串，就 text[i:i+target] 按字硬切。

光会切还不够，还得保留标题层级。一篇文档通常是一级标题、二级标题、正文层层嵌套，你切出来的每个小块，得知道自己是「挂在哪个标题下面」的。做法：先按 # 标题把文档切成「标题 + 正文」块，再对每块正文套上面的递归切分，最后把标题拼回每块开头——这样向量化时，块里自带「章节」上下文，问「部署怎么做」时，「部署」标题下的块天然更容易被命中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def split_markdown_with_heading(md, target=300):
    """按 # 标题切块，再把标题拼回每块开头，让向量自带「章节」上下文。"""
    blocks, cur, heading = [], "", ""
    for line in md.splitlines():
        if line.startswith("#"):
            if cur.strip():
                blocks.append({"heading": heading, "text": cur.strip()})
            heading = line.lstrip("#").strip()
            cur = ""
        else:
            cur += line + "\n"
    if cur.strip():
        blocks.append({"heading": heading, "text": cur.strip()})

    out = []
    for b in blocks:
        for c in recursive_split(b["text"], target):
            prefix = f"[{b['heading']}] " if b["heading"] else ""
            out.append(prefix + c)  # 标题拼进正文，向量化时带着章节信息
    return out

至于更进阶的「语义切分」（相邻两句向量突然不相似了就断开，说明话题变了），效果更好，但要反复调 embedding、也重得多。第一轮先用手里的递归切分，够用了。

三、FAISS 落盘：把向量从内存里请出来

上一课的向量是 numpy 数组，一关机就没、一多就爆。这次换成 FAISS（Facebook AI Similarity Search，Meta 开源），它专干「海量向量的快速相似度检索」这一件事，原生支持落盘，几千篇文档在它眼里是小菜。

先装依赖（全套一次装齐）：

1
pip install faiss-cpu openai numpy jieba rank_bm25

装完先理解一个关键点：FAISS 里最适合我们的是 IndexFlatIP。IP 是内积（Inner Product），而归一化之后的内积 = 余弦相似度。所以做法是：向量先做 L2 归一化，再丢进 IndexFlatIP，检索出来的分数就是余弦相似度，和上一课手写的逻辑完全对齐：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import os, json
import numpy as np
import faiss
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["RELAY_API_KEY"],      # 中转站 key
    base_url="https://ai.aklibk.com/v1",      # OpenAI 兼容地址
)
EMB_MODEL = "text-embedding-3-small"

def embed(texts):
    """批量向量化，返回 (n, 1536) 的 float32 矩阵。"""
    resp = client.embeddings.create(model=EMB_MODEL, input=texts)
    items = sorted(resp.data, key=lambda x: x.index)
    return np.stack([np.array(it.embedding, dtype="float32") for it in items])

def build_index(chunks, index_path="kb.faiss", texts_path="kb.json"):
    """一次性构建：向量化 → L2 归一化 → 塞进 IndexFlatIP → 落盘。"""
    vecs = embed(chunks)
    vecs = vecs / np.linalg.norm(vecs, axis=1, keepdims=True)  # 归一化，点积才 = 余弦

    dim = vecs.shape[1]
    index = faiss.IndexFlatIP(dim)          # IP = 内积；归一化后就是余弦相似度
    index.add(vecs)

    faiss.write_index(index, index_path)    # 向量落盘
    with open(texts_path, "w", encoding="utf-8") as f:  # 原文一起落盘，别只存向量
        json.dump(chunks, f, ensure_ascii=False, indent=2)
    return index

def load_index(index_path="kb.faiss", texts_path="kb.json"):
    index = faiss.read_index(index_path)
    with open(texts_path, encoding="utf-8") as f:
        chunks = json.load(f)
    return index, chunks

几个要点：

  • FAISS 只存「向量」，不存「文字」——所以原文必须自己存一份（这里用 JSON 落盘）。检索返回的是整数 id，拿它去 JSON 里找回原文。忘了存原文，检索出来你都不知道命中了啥。
  • 归一化用 np.linalg.norm(vecs, axis=1, keepdims=True) 按行归一，和上一课「先归一化再点积」是同一件事，别偷懒。
  • faiss.write_index / read_index 就是落盘和读回，一把梭。重启、换机器，向量不丢。

四、混合检索：BM25 关键词 + 向量两条腿走路

向量检索抓「意思」，但精确词是它的盲区。这时候请出 BM25——搜索引擎祖师爷级的关键词打分算法，Elasticsearch、Lucene 都靠它。它只认字面：文档里这个关键词出现得越多、别的文档里出现得越少，分就越高。专有名词、型号、报错码，它一抓一个准。

中文的关键在于分词，这里用 jieba：jieba.cut("全文检索") 会得到 ["全文", "检索"]：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import jieba
from rank_bm25 import BM25Okapi

# 建 BM25：把每个 chunk 用 jieba 分好词
tokenized = [list(jieba.cut(c)) for c in chunks]
bm25 = BM25Okapi(tokenized)

def hybrid_search(query, top_k=5, k=60):
    """向量路 + BM25 关键词路，各取 top_k*2 候选，再用 RRF 融合。"""
    # 向量路
    qvec = embed([query])[0]
    qvec = qvec / np.linalg.norm(qvec)               # 查询向量也要归一化
    _, I = index.search(np.expand_dims(qvec, 0), top_k * 2)
    dense_ids = I[0].tolist()

    # 关键词路
    qtok = list(jieba.cut(query))
    kw_scores = bm25.get_scores(qtok)
    kw_ids = np.argsort(kw_scores)[::-1][:top_k * 2].tolist()

    # RRF 融合：只看排名，不看分数
    fused = {}
    for rank, doc_id in enumerate(dense_ids):
        fused[doc_id] = fused.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    for rank, doc_id in enumerate(kw_ids):
        fused[doc_id] = fused.get(doc_id, 0.0) + 1.0 / (k + rank + 1)

    ranked = sorted(fused.items(), key=lambda x: -x[1])[:top_k]
    return [chunks[i] for i, _ in ranked]

两条路怎么合？用 RRF（Reciprocal Rank Fusion，倒数排名融合）。它不看分数绝对值，只看排名：某个 chunk 在向量路排第 1、在关键词路排第 3，分数就是 1/(60+1) + 1/(60+3)。k=60 是业界约定俗成的默认值。

RRF 为什么好？因为向量余弦分和 BM25 分量纲完全不同，直接加权相加根本没法调；RRF 只看排名，天然免疫量纲问题，还不用调参。这一手，就是「混合检索」的核心动作。

五、串起来：一分钟跑一遍升级版

把上面几节拼进一个 kb.py，主流程就是这几行——build_index 只跑一次，查询走 hybrid_search，生成沿用上一课的 ask：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def ask(question, top_k=5):
    ctx = "\n\n".join(hybrid_search(question, top_k))
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "你是知识库助手，只依据给定资料回答；资料里没有就明说不知道，不许编。"},
            {"role": "user", "content": f"资料：\n{ctx}\n\n问题：{question}"},
        ],
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    chunks = split_markdown_with_heading(load_your_docs())  # 你的文档集
    index = build_index(chunks)                             # 只跑一次
    for q in ["ERR_500 怎么处理？", "服务挂了怎么排查？"]:
        print(f"\nQ: {q}\nA: {ask(q)}")

跑起来后做个对比实验，就能直观感受到「两条腿」的价值：

  • 问「ERR_500 怎么处理」（精确词），BM25 路稳稳抓中那条写着报错码的文档；
  • 问「服务挂了怎么排查」（口语化，一个字都对不上），向量路负责捞出语义相关的那几段；
  • RRF 一融合，精确命中 + 语义相关同时进上下文，大模型答得有据可依、还给得出处。

六、第二课小结 + 这几个坑别踩

今天这三样零件的分工，一句话记住：递归切分负责「切得干净」，FAISS 负责「存得下、检得快」，混合检索负责「既认意思又认字」。

收尾前把几个坑点一点：

  • FAISS 索引和原文列表的顺序必须一一对应。按顺序 add 向量，文本 JSON 也得按同样顺序存，否则检索回的 id 对不上文字，全乱套。
  • 查询向量也必须归一化。存的时候归一化、查的时候不归一化，内积就不等于余弦，结果就漂了。
  • jieba 对专有名词会切错（比如「空缺的博客」可能切成「空缺/的/博客」）。业务里专有名词多，用 jieba.add_word("空缺的博客") 建一份自定义词典，命中率立竿见影。
  • BM25 和向量要共用同一套 chunk。两路检索的颗粒度不一致（一个按句、一个按段），融合出来的排名就是胡扯。
  • 切分的 target 别乱调，先沿用 256~512 token 这个甜区；真要调，改完记得重跑一遍向量化，索引和 chunk 要一起重建。

生命不息，折腾不止。今天的知识库终于「能存能查」了，但它还不会告诉你「自己答得靠不靠谱」。下一篇 RAG 实战第三课：给链路加一层 reranker 交叉排序（把召回结果再精排一遍，让最相关的稳坐头名），再教你怎么给 RAG 打分（召回率、命中率这些指标怎么算、用什么脚本测），让你的知识库从「能跑」变成「测得准、调得动」。回见。