lang

1. 本章目标 #

第 13 章我们把资料切成了 chunks.jsonl。现在有几百个 chunk 躺在文件里，问题变成：

用户问「滤芯多久换一次？」，怎么从几百个 chunk 里找出讲滤芯的那一个？

最容易想到的是关键词匹配——搜「滤芯」。但用户很可能问「过滤网什么时候要换」，一个「滤芯」的字都没有。这时关键词检索直接失效。

本章要装上的能力叫语义检索：把文字变成向量（一串数字），语义相近的文字在向量空间里距离也近。于是「过滤网什么时候要换」和「滤芯到期时需更换滤芯」即使用词不同，向量依然靠得很近。

第 12 章 Load     ingested.jsonl
   │
第 13 章 Split    chunks.jsonl
   │
   ▼
第 14 章 Embed    每个 chunk → 向量，存进向量库              ← 本章
   │
   ▼
第 15 章 Retrieve 按问题召回相关 chunk，交给 Agent 回答

这是整条 RAG 链路上第一次能看到效果的一章：跑完 §10 的实战，你就能对着自己的资料提问并看到命中的片段了。

本章目标：

用嵌入模型把 chunk 转成向量、存进本地向量库，并实现带元数据过滤的语义检索。

学完你应能：

解释「向量」「维度」「余弦相似度」在 RAG 里各自意味着什么
用统一的 get_embeddings() 工厂切换嵌入模型供应商（通义 / OpenAI / 本地）
说清 embed_documents 与 embed_query 的区别
用 Chroma 建库、持久化、按 chunk_id 增删改
用 FAISS 建库并存取本地文件，知道它与 Chroma 的取舍
用 metadata 过滤把检索限定在某个文件 / 某个章节里
用 MMR 解决「召回结果高度雷同」的问题
用 as_retriever() 把向量库接给第 15 章
看懂检索分数：方向、量纲，以及「top-1 与 top-2 的分差」这个置信度信号
产出本地向量知识库：建库脚本 + 检索脚本

前置依赖： 第 13 章（chunks.jsonl 与 chunk_id）。本章会调用云端嵌入 API，需要一个 API Key。

参考文档：

安装：

# langchain-chroma：Chroma 向量库集成
# faiss-cpu：FAISS 向量库（§6）
# dashscope：通义嵌入模型的官方 SDK
# numpy：手算相似度时用（§4）
# python-dotenv：从 .env 读 API Key（§3.2）
pip install langchain-chroma faiss-cpu dashscope numpy python-dotenv

1.1. 先看几条反直觉的结论 #

下面这些结论是过程中最容易踩的，先看一眼，读到对应小节时会更有感觉：

你可能以为	实际情况	小节
相似度 0.85 一定比 0.55 更像	绝对值不能跨模型比较。本地小模型给无关内容也打 0.5778，通义只给 0.2852	§3.3
`embed_documents` 和 `embed_query` 结果不同	LangChain 确实分别传了 `text_type="document"` / `"query"`，但通义 v4 对同一句话返回逐位完全相同的向量	§3.1
不写 `model=` 会用最新模型	默认是老版 `text-embedding-v1`，输出 1536 维而不是 1024	§3.2
Chroma 的分数越大越像	默认返回的是距离，越小越像。查库里一模一样的原文得到 `0.0000`	§5.2
向量库在做什么高深的事	就是在算你手算的那个余弦。`1 - 0.4661/2 = 0.7670`，与手算一字不差	§5.2
过滤条件写错会报错	静默返回 0 条。键名拼错、值类型不符（`page` 存的是 `11`，你传 `"11"`）都不报错	§5.3
FAISS 用同一个 id 重写会覆盖	Chroma 是覆盖；FAISS 抛 `ValueError`，而且这个 store 从此报废，之后每次检索都 `KeyError`	§6.2
FAISS 默认就是 HNSW 近似索引	`from_documents` 建的是 `IndexFlatL2`，暴力精确检索	§6.1
MMR 是「更好的检索」	是用相关性换覆盖面的旋钮。小库上开 MMR 会把无关内容拽进来	§7
集合名随便起就行	必须 3～512 个 `[a-zA-Z0-9._-]` 字符，中文名直接报错，`"kb"` 也太短	§5.1
检索不准就去调 `chunk_size`	实测把正确答案缩短后排名没变，改了纯属白折腾	§9.1

2. 从关键词匹配到按语义找 #

如果你已经清楚「向量」和「余弦相似度」是什么，可以直接跳到 §3；但 §2.3 末尾那个 $d = 2(1-\cos\theta)$ 的换算式建议看一眼，§5.2 会用实测数字验证它。

2.1. 关键词检索为什么不够 #

传统检索（如 SQL 的 LIKE、Elasticsearch 的分词匹配）比的是字面。中文场景下它有三个绕不过去的问题：

问题	例子
同义不同词	用户说「过滤网」，文档写「滤芯」
问法与陈述句不同	用户问「几天能退」，文档写「签收 7 日内可无理由退货」
需要分词	「退货政策」要先切成「退货 / 政策」，切错就搜不到

关键词检索并非一无是处——查订单号 A1002、查人名这类精确匹配它比向量检索强得多（第 15 章会讲把两者混合使用）。但回答「怎么退货」这类问题，需要的是理解意思。

2.2. 向量是什么 #

一句话：

嵌入模型（embedding model）是一个函数，把任意长度的文字映射成一串固定长度的数字，让「意思相近」变成「数字接近」。

"签收 7 日内可无理由退货。"  ──嵌入模型──►  [0.013, 0.071, 0.036, ...]  共 1024 个数字
"退货有几天的期限？"        ──嵌入模型──►  [0.015, 0.068, 0.041, ...]  ← 数字很接近
"今天天气很好。"            ──嵌入模型──►  [-0.22, 0.004, -0.15, ...]  ← 数字差很远

几个术语：

术语	含义	备注
向量（vector）	那串数字	也叫 embedding、嵌入
维度（dimension）	数字的个数	通义 v4 默认 1024，可选 64～2048
向量空间	所有向量所在的空间	语义相近 = 空间中位置相近

一个常见误解要先破除：向量不可读、也不能还原成原文。它只是拿来比相似度的中间表示，所以原文必须另外存着——这也是向量库要同时保存 page_content 和 metadata 的原因。

也就是说，嵌入是单向压缩：1024 个浮点数装不下一段话的全部信息，它只保留了「和别的句子比起来，这句话大概在讲什么方向」。所以不要指望向量能替代原文，也不要以为存了向量就等于存了资料——向量库丢了原文，检索出来的就只是一串数字。

维度不是越高越好：

维度	特点
低（64～256）	存储小、检索快，语义信息损失多
中（1024）	默认推荐，效果与成本平衡
高（1536～2048）	语义更细腻，存储与计算成本上升

小规模时维度怎么选都无所谓：1 万条 chunk × 1024 维 × 4 字节 ≈ 40 MB，怎么选都不心疼。真正要权衡的是几十万条以上的规模（§12 第 3 题会让你亲手对比 256 维和 1024 维）。新手直接用模型默认维度就好，别一上来就调。

2.3. 相似度怎么算 #

有了向量，「像不像」就变成了数学问题。RAG 里最常用的是余弦相似度（cosine similarity）——比较两个向量的方向夹角，而不管它们的长度：

$$ \text{sim}(\vec{a}, \vec{b}) = \cos\theta = \frac{\vec{a} \cdot \vec{b}}{\lVert \vec{a} \rVert \, \lVert \vec{b} \rVert} = \frac{\displaystyle\sum_{i=1}^{n} a_i b_i}{\sqrt{\displaystyle\sum_{i=1}^{n} a_i^{2}} \; \sqrt{\displaystyle\sum_{i=1}^{n} b_i^{2}}} $$

取值范围与含义：

余弦值	夹角	含义
接近 1	接近 0°	语义几乎相同
接近 0	接近 90°	无关
接近 −1	接近 180°	语义相反（实践中少见）

另一个常见度量是欧氏距离（L2），比较两点的直线距离：

$$ d(\vec{a}, \vec{b}) = \lVert \vec{a} - \vec{b} \rVert = \sqrt{\displaystyle\sum_{i=1}^{n} (a_i - b_i)^{2}} $$

两者的关键差别在方向：

度量	越像则值	常见于
余弦相似度	越大	语义检索的默认选择
L2 距离	越小	Chroma / FAISS 的默认打分

这个方向差异，是新手排查检索问题时最常踩的坑——看到「分数 0.87」时，先搞清它是相似度还是距离（§5.2 会实测）。

多数嵌入模型输出的是已归一化的向量（模长为 1），此时余弦相似度与 L2 距离可以互相换算，排序结果完全一致，所以实践中不必纠结选哪个。具体地，Chroma 与 FAISS 默认给的是平方 L2 距离，归一化向量下满足：

$$ d = \lVert \vec{a} - \vec{b} \rVert^2 = 2\,(1 - \cos\theta) $$

§5.2 会用实测数字验证这个等式——你手算的余弦，和向量库打印的距离，其实是同一件事。

3. Embeddings 接口 #

这一节做两件事：先看清接口长什么样（§3.1），再把嵌入模型收口到一个工厂函数里（§3.2）——后面每个示例都要 from embeddings_layer import get_embeddings，所以 §3.2 那段代码得先存成文件。

3.1. 两个方法 #

LangChain 把所有嵌入模型抽象成一个接口，只有两个核心方法：

方法	输入	输出	用在哪
`embed_documents(texts)`	字符串列表	向量列表	建库时批量转换 chunk
`embed_query(text)`	单个字符串	一个向量	检索时转换用户问题

为什么要拆成两个方法？因为有些模型对「文档」和「查询」会走不同处理（比如加不同前缀，或告诉服务端这次是查询还是入库），好抬高检索精度。

但有个实测结果值得知道。 同一句话分别走两个方法，比一下输出的向量：

embeddings_layer.py

"""统一嵌入层：换供应商只改这里，下游代码不动"""

# 让 str | None 这种新式类型注解在旧版 Python 上也能用
from __future__ import annotations

# 标准库：读环境变量
import os

# 从 .env 文件把配置读进环境变量
from dotenv import load_dotenv

# Embeddings 是所有嵌入模型的抽象基类，既用于类型注解，也用于自定义实现
from langchain_core.embeddings import Embeddings

# 读取项目根目录的 .env，把 DASHSCOPE_API_KEY 等注入环境变量
# 放在模块顶层，这样任何脚本 import 这个文件时都会自动加载
load_dotenv()


# 工厂函数：下游只认这一个入口，不直接 import 任何具体的模型类
def get_embeddings(provider: str | None = None) -> Embeddings:
    """按供应商名返回一个 Embeddings 实例。

    provider 取值：dashscope（默认）/ openai / local
    不传时读环境变量 EMBEDDING_PROVIDER，仍没有就用 dashscope。
    """
    # 三级兜底：函数参数优先，其次环境变量，最后硬编码默认值
    # 这样既能在代码里临时切换，也能靠改 .env 全局切换
    provider = provider or os.getenv("EMBEDDING_PROVIDER", "dashscope")

    # 分支一：通义千问，本文主线
    if provider == "dashscope":
        # 放在函数内部导入：没装 dashscope 的人用别的供应商也不会报错
        from langchain_community.embeddings import DashScopeEmbeddings

        # model 必须显式指定：这个类的默认值是老版的 text-embedding-v1（1536 维）
        # API Key 由这个类自己从环境变量 DASHSCOPE_API_KEY 读，不用手动传
        return DashScopeEmbeddings(model="text-embedding-v4")

    # 分支二：OpenAI，已有 OpenAI 生态时用
    if provider == "openai":
        # 同样延迟导入，需要 langchain-openai 包
        from langchain_openai import OpenAIEmbeddings

        # text-embedding-3-small 便宜且够用
        # dimensions=1024 把默认的 1536 维裁到 1024，和通义对齐便于对比
        return OpenAIEmbeddings(model="text-embedding-3-small", dimensions=1024)

    # 分支三：本地小模型，零 key 零成本，仅用于理解机制
    if provider == "local":
        # 注意：它的中文能力很弱，不要用于真实知识库（原因见 §3.3）
        return LocalOnnxEmbeddings()

    # 传了不认识的名字就立刻报错，而不是悄悄退回默认值
    # 「悄悄退回默认值」会让你误以为在用 A 模型，其实一直在用 B
    raise ValueError(f"未知的嵌入供应商：{provider}")


# 继承 Embeddings 基类，就能被所有向量库当成标准嵌入模型使用
class LocalOnnxEmbeddings(Embeddings):
    """把 Chroma 自带的 ONNX 小模型包装成 LangChain 的 Embeddings 接口。

    Chroma 安装时会附带 onnxruntime，首次调用会自动下载模型（约 80MB）。
    实现一个自定义 Embeddings 只需提供下面两个方法。
    """

    # 构造时准备好底层的嵌入函数，避免每次调用都重新加载模型
    def __init__(self) -> None:
        # chromadb 自带的嵌入函数集合，不需要额外装包
        from chromadb.utils import embedding_functions as ef

        # DefaultEmbeddingFunction 就是 all-MiniLM-L6-v2，384 维
        # 首次实例化会触发模型下载，之后有本地缓存
        self._fn = ef.DefaultEmbeddingFunction()

    # 必须实现的方法之一：批量转换（建库路径）
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        # 直接调用 chromadb 的嵌入函数，它接收字符串列表
        # 底层返回的是 numpy 数组，转成普通 float 列表以符合 LangChain 的接口约定
        # 不转的话下游某些序列化操作（比如写 json）会报 numpy 类型不可序列化
        return [list(map(float, vector)) for vector in self._fn(texts)]

    # 必须实现的方法之二：单条转换（检索路径）
    def embed_query(self, text: str) -> list[float]:
        # 查询只有一条，包成单元素列表复用批量方法，再取出第 0 个
        # 这个本地模型不区分 query 与 document，所以直接复用是安全的
        return self.embed_documents([text])[0]

# numpy 用来比较两个向量是否逐位相等
import numpy as np

# 复用 §3.2 的工厂函数拿到通义嵌入模型
from embeddings_layer import get_embeddings

# 默认供应商是通义（text-embedding-v4）
embeddings = get_embeddings()

# 一句普通的中文，两个方法都拿它去转
text = "签收 7 日内可无理由退货。"

# 走建库路径：注意 embed_documents 收列表、返回列表，所以要取 [0]
vector_doc = np.array(embeddings.embed_documents([text])[0])
# 走检索路径：embed_query 收单条字符串、直接返回一个向量
vector_query = np.array(embeddings.embed_query(text))

# allclose 判断两个数组在浮点误差范围内是否相等
print("两个向量是否相同：", np.allclose(vector_doc, vector_query))
# 逐元素求绝对差再取最大值，看最大偏差有多大
print("最大逐元素差：", f"{float(np.abs(vector_doc - vector_query).max()):.2e}")

实测输出：

两个向量是否相同： True
最大逐元素差： 0.00e+00

最大差是精确的 0，说明通义 v4 对 text_type 这个参数并不做区分处理。所以：

接口上的区分是真实存在的，但具体某个模型是否利用它，取决于服务端。 通义 v4 不区分，所以你哪怕不小心混用了，检索结果也不会变。

那还要不要守规矩？要，图的是可移植性：万一哪天换成 BGE、E5 这类明确要求加 "query: " / "passage: " 前缀的模型，混用就会实打实地掉效果，而且是静默的——不报错，只是排名变差。

结论：建库用 embed_documents，检索用 embed_query。 好消息是向量库会自动帮你调对（add_documents 内部调前者，similarity_search 内部调后者），只有 §4 那种手写检索才需要自己留意。

3.2. 供应商选型 #

供应商	模型	维度	中文能力	成本	适合
通义（DashScope）	`text-embedding-v4`	1024（可 64～2048）	强	有免费额度，之后按量极低	本文主线，国内直连
OpenAI	`text-embedding-3-small`	1536（可裁剪）	较强	低	已有 OpenAI 生态时
本地 BGE 系列	`bge-small-zh` 等	512～1024	强	免费，但要装 `torch`（约 2GB）	数据不能出内网
Ollama	`bge-m3` 等	1024	强	免费	已在用 Ollama
Chroma 自带 ONNX	`all-MiniLM-L6-v2`	384	很弱	免费、零配置	仅用于理解机制

关于最后一行，最好用数据说清楚，免得踩坑。同样三句话，分别让两个模型算相似度：

# numpy 提供点积（@）和求模长（linalg.norm）
import numpy as np

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 三句话构成一组最小对照实验
TEXTS = [
    # 第 0 句：当作「文档」
    "签收 7 日内可无理由退货。",
    # 第 1 句：与文档同主题的问句，期望与第 0 句相似度高
    "退货有几天的期限？",
    # 第 2 句：完全无关的闲聊，期望与第 0 句相似度低
    "今天天气很好，适合散步。",
]


# 把余弦公式单独封装成函数，后面要用两次
def cosine(a: np.ndarray, b: np.ndarray) -> float:
    """余弦相似度：点积除以两个向量的模长之积。"""
    # a @ b 是点积，即 §2.3 公式里的分子
    # np.linalg.norm 求模长，两者相乘是分母
    # 外层 float() 把 numpy 标量转成 Python 浮点，方便格式化打印
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))


# 同一批文本，分别交给两个模型，只比较各自内部的分数差距
for provider in ["local", "dashscope"]:
    # 每轮换一个供应商，其余代码完全不变
    embeddings = get_embeddings(provider)
    # 一次把三句话都转成向量，得到形状为 (3, 维度) 的矩阵
    vectors = np.array(embeddings.embed_documents(TEXTS))
    # 打印分隔标题，便于对照两组结果
    print(f"\n--- {provider} ---")
    # 第 0 句与第 1 句：同主题，期望分数高
    print(f"「退货政策」与「退货问句」的相似度：{cosine(vectors[0], vectors[1]):.4f}")
    # 第 0 句与第 2 句：完全无关，期望分数低
    print(f"「退货政策」与「天气闲聊」的相似度：{cosine(vectors[0], vectors[2]):.4f}")

实测输出：

--- local ---
「退货政策」与「退货问句」的相似度：0.5967
「退货政策」与「天气闲聊」的相似度：0.5778

--- dashscope ---
「退货政策」与「退货问句」的相似度：0.5426
「退货政策」与「天气闲聊」的相似度：0.2852

这组数字里藏着一个反直觉却极其重要的结论，值得单独拎出来：

	同主题	完全无关	差距
`local`	0.5967	0.5778	0.0189
`dashscope`	0.5426	0.2852	0.2574

注意通义在「同主题」上的分数（0.5426）其实比本地模型还低。只看单个绝对分，很容易得出「本地模型更好」的错结论。真正有意义的是差距：本地模型把无关的天气闲聊也判成 0.5778，跟同主题几乎并列，检索时自然分不清谁该靠前；通义把无关内容压到 0.2852，一眼就能分开。

记住这条判断准则：嵌入相似度的绝对值没有可比性，跨模型比较必须看「相关」与「无关」之间的区分度。 0.85 不一定比 0.55 更像——不同模型的分数分布天差地别。

所以 local 只适合没 key 时跑通流程；做中文知识库，必须用中文能力过关的模型。§9 会用完整检索场景，把这个差距量化成命中率。

3.3. 第一次调用 #

配好 key 后，先看一眼向量长什么样：

# numpy 用来求模长，验证向量是否已归一化
import numpy as np

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 不传参数即使用默认供应商（通义 text-embedding-v4）
embeddings = get_embeddings()

# embed_query 处理单条文本，返回一个向量（一维列表）
vector = embeddings.embed_query("退换货政策")

# 向量的长度就是这个模型的输出维度
print("维度：", len(vector))
# 切片取前 5 个并保留 4 位小数：向量里都是小数、正负都有，人是读不懂的
print("前 5 个数字：", [round(x, 4) for x in vector[:5]])
# 求模长（L2 范数）。多数嵌入模型输出已归一化的向量，模长约等于 1
print("向量模长：", round(float(np.linalg.norm(vector)), 4))

# embed_documents 处理一批文本，返回向量列表（二维）
vectors = embeddings.embed_documents(["签收 7 日内可退货。", "48 小时内发货。"])
# len(vectors) 是文本条数，len(vectors[0]) 是每个向量的维度
print("批量结果：", len(vectors), "个向量，每个", len(vectors[0]), "维")

实测输出：

维度： 1024
前 5 个数字： [0.019, -0.0669, 0.0137, -0.0512, 0.0523]
向量模长： 1.0
批量结果： 2 个向量，每个 1024 维

有三点值得留意：

1024 维：text-embedding-v4 的默认输出长度。这个数字要记住，因为换成维度不同的模型必须重建向量库（§11）。
前 5 个数字毫无意义：向量是不可读的，也无法还原成原文。这就是为什么原文必须另外存一份——向量库替你存了（§5）。
模长恰好是 1.0：通义输出的是已归一化的向量。这一点在 §5.2 会用到——模长为 1 时，余弦相似度和向量库给的 L2 距离可以互相换算（§2.3 那个 $d = 2(1-\cos\theta)$），你手算的分数和库里的分数才对得上。

顺便验证一下「忘写 model=」这个坑。 §3.2 强调过必须显式写模型名，实测后果是这样的：

# 直接实例化，不传 model 参数
from langchain_community.embeddings import DashScopeEmbeddings

# 不传 model 时会用类定义里的默认值
embeddings = DashScopeEmbeddings()

# 打印它实际用的模型名
print("实际模型：", embeddings.model)
# 打印输出维度
print("输出维度：", len(embeddings.embed_query("退换货政策")))

实测输出：

实际模型： text-embedding-v1
输出维度： 1536

默认是 2023 年的 text-embedding-v1，维度 1536。 它不报错、也能用，但中文不如 v4，维度还和 v4 不同——一部分脚本写了 model=、另一部分忘了，两边建的库就互不兼容，检索会直接撞上维度不匹配。所以工厂函数里那句显式的 model="text-embedding-v4" 不是啰嗦。

关于批量与计费，有个实现细节值得知道：通义单次请求最多 10 条（text-embedding-v3 / v4 都是 10），但你不用自己切。DashScopeEmbeddings 内部有一张批量上限表，会自动把长列表拆成合规小批，并带指数退避重试（默认 max_retries=5）：

# langchain_community/embeddings/dashscope.py 里的上限表
BATCH_SIZE = {
    "text-embedding-v1": 25,
    "text-embedding-v2": 25,
    "text-embedding-v3": 10,
    "text-embedding-v4": 10,
}

# 取不到时退回 25。所以用了表里没有的新模型名，可能会超限报错
batch_size = BATCH_SIZE.get(kwargs["model"], 25)

所以 §10 的 index.py 里写 batch_size = 50 不会报错——它只决定「多久打一次进度、失败时重试多大一块」，真正发给 API 的仍是每批 10 条。实测传 25 条可以一次拿回 25 个向量，正是内部分批的效果：

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 拿到通义 v4 嵌入模型，它的单请求上限是 10 条
embeddings = get_embeddings()

# 列表推导式生成 25 条互不相同的短句
batch = [f"这是第 {i} 条测试文本，用于验证内部分批。" for i in range(25)]
# 一次性传进去，内部会自动拆成 10+10+5 三个请求
vectors = embeddings.embed_documents(batch)
# 返回条数应与输入条数一致，说明分批对调用方是透明的
print(f"传入 25 条 -> 返回 {len(vectors)} 个向量，每个 {len(vectors[0])} 维")

实测输出：

传入 25 条 -> 返回 25 个向量，每个 1024 维

注意上面源码里 BATCH_SIZE.get(kwargs["model"], 25) 的兜底是 25——万一通义出了 v5 而这张表没更新，默认按 25 条发就会超限。遇到莫名的批量报错，先手动把批次调小验证一下。

一次调用就是一次 API 请求，按 token 计费。所以建库脚本别反复重跑——§10 的 index.py 会做增量判断。

4. 手算一遍相似度 #

cosine_scores

在用向量库之前，先手动做一遍完整检索，好弄清向量库到底替你做了什么。

整个语义检索其实只有四步：

① 把所有 chunk 转成向量（建库，一次性）
② 把用户问题转成向量（每次查询）
③ 逐个算相似度
④ 按相似度排序，取前 k 个

# numpy 负责向量点积、求模长与排序
import numpy as np

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 拿到默认供应商（通义）的嵌入模型
embeddings = get_embeddings()

# 模拟第 13 章切好的 chunk
CHUNKS = [
    "签收 7 日内可无理由退货，商品需保持完好。",
    "质量问题 15 日内可换货，需提供照片凭证。",
    "一般订单 48 小时内发出，节假日顺延。",
    "滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
    "开机后指示灯亮起，按下模式键可切换手动与自动档。",
]

# ① 建库：把所有 chunk 转成向量矩阵，形状是 (chunk 数, 维度)
# 注意这里用 embed_documents，因为这些是入库的文档（§3.1）
doc_vectors = np.array(embeddings.embed_documents(CHUNKS))
# .shape 打印出来应该是 (5, 1024)：5 条 chunk，每条 1024 维
print("向量矩阵形状：", doc_vectors.shape)


# 余弦相似度，和 §3.3 用的是同一个函数：点积除以两个向量的模长之积
def cosine(a: np.ndarray, b: np.ndarray) -> float:
    """余弦相似度，直译 §2.3 的公式。"""
    # a @ b 是点积，对应公式的分子 a·b
    # np.linalg.norm(x) 求模长（即 ||x||），两个模长相乘就是公式的分母
    # 外层 float() 把 numpy 标量转成 Python 浮点，方便格式化打印
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))


# 把算分逻辑封装成函数，就是对每条 chunk 各套一次上面的公式
def cosine_scores(query_vector: np.ndarray, matrix: np.ndarray) -> np.ndarray:
    """算出查询向量与矩阵中每一行的余弦相似度。"""
    # matrix 的每一行是一条 chunk 的向量，逐行和问题向量算余弦
    # 列表推导的结果长度等于 chunk 数，下标与 CHUNKS 一一对应
    # 转成 np.array 是为了后面能直接用 argsort 排序
    return np.array([cosine(query_vector, row) for row in matrix])


# ② 把用户问题转成向量
# 这个问句里没有「滤芯」二字，专门用来对比关键词检索
question = "过滤网什么时候要换？"
# 注意这里用 embed_query，而不是 embed_documents（§3.1）
query_vector = np.array(embeddings.embed_query(question))

# ③ 算分：得到一个长度为 5 的分数数组，下标与 CHUNKS 对应
scores = cosine_scores(query_vector, doc_vectors)
# ④ 排序：argsort 返回的是「从小到大」的下标，[::-1] 反转成从大到小，再取前 3
ranking = np.argsort(scores)[::-1][:3]

# 打印问题
print(f"\n问：{question}")
# enumerate 的 start=1 让名次从 1 开始而不是 0
for rank, index in enumerate(ranking, start=1):
    # index 是 CHUNKS 里的下标，用它同时取分数和原文
    print(f"  第{rank}名 {scores[index]:.4f}  {CHUNKS[index]}")

实测输出：

向量矩阵形状： (5, 1024)

问：过滤网什么时候要换？
  第1名 0.7670  滤芯到期时指示灯会闪红灯，此时需更换滤芯。
  第2名 0.4471  质量问题 15 日内可换货，需提供照片凭证。
  第3名 0.3854  签收 7 日内可无理由退货，商品需保持完好。

注意这个问句里没有「滤芯」二字，用的是「过滤网」——关键词检索完全搜不到，语义检索却稳稳排第一。这就是本章的价值所在。

另外留意第 1 名（0.7670）与第 2 名（0.4471）之间拉开了 0.32 的差距。这个差距本身就是信号：差距大说明模型「很确定」，差距小说明它其实在猜（§9 会看到一个反例）。

代码里的 cosine_scores 没有做任何优化，就是把 §2.3 的公式对每条 chunk 各算一遍：a @ b 是分子的点积，np.linalg.norm(a) * np.linalg.norm(b) 是分母的两个模长相乘。5 条 chunk 就循环 5 次，和你拿纸笔算的步骤完全一致。

这么写是为了让你看清「检索」到底在算什么。真实向量库不会这样逐条循环：它们把这一步做成底层批量矩阵运算，规模再大就换成近似索引，直接跳过大部分候选（Chroma 用的 HNSW 就是这类，§5.4）。但那是性能问题，不是原理问题——库里算出来的分数，和这段循环算出来的是同一个余弦值，§5.2 会用实测数字对上。

顺便记下 0.7670 这个数字，§5.2 会用它验证一件事：向量库返回的分数和你手算的余弦，是同一个东西的两种表达。

手算能跑通，为什么还要向量库？因为上面这段代码有三个致命缺陷：

缺陷	向量库怎么解决
每次重启都要重新调 API 转所有向量（花钱、慢）	持久化：向量存磁盘，重启直接用
几万个 chunk 逐个算相似度会很慢	索引：用 HNSW 等结构快速找近邻
没法按条件筛选（只搜某个文件）	元数据过滤：`filter={"source": ...}`

所以真实项目一定用向量库。手算这一节的意义是：检索结果不对时，你知道底层就是在算余弦，不会当成黑魔法。

5. 向量库：Chroma #

Chroma 装上就能用：pip install 即可，不必另起服务，也支持元数据过滤和持久化。

下面五个小节各自解决一个问题：怎么建库（§5.1）、分数怎么读（§5.2）、怎么限定范围（§5.3）、怎么持久化（§5.4）、资料更新时怎么改（§5.5）。

5.1. 建库 #

Chroma.from_documents() 一步做完三件事：调嵌入模型把每条 page_content 转成向量，把向量、原文和 metadata 一并写入集合，再建索引。

# 标准库：递归删除目录，演示时用来清空旧库
import shutil
# 标准库：跨平台路径处理
from pathlib import Path

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma
# Document 是第 12、13 章一路传下来的数据结构
from langchain_core.documents import Document

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 模拟第 13 章的 chunks，metadata 里带着 chunk_id 与来源
CHUNKS = [
    Document(
        # page_content 是会被转成向量的正文
        page_content="签收 7 日内可无理由退货，商品需保持完好。",
        # metadata 不参与向量计算，但可以用来过滤（§5.3）和溯源
        metadata={"source": "policy/aftersale.md", "h2": "退换货", "chunk_id": "a-0000"},
    ),
    Document(
        page_content="质量问题 15 日内可换货，需提供照片凭证。",
        # 与上一条同来源、同章节，用来演示过滤会命中多条
        metadata={"source": "policy/aftersale.md", "h2": "退换货", "chunk_id": "a-0001"},
    ),
    Document(
        page_content="一般订单 48 小时内发出，节假日顺延。",
        # h2 换成「发货时效」，用来演示按章节过滤
        metadata={"source": "policy/aftersale.md", "h2": "发货时效", "chunk_id": "a-0002"},
    ),
    Document(
        page_content="滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
        # 换个来源文件，并带上 PDF 特有的 page 字段（注意值是整数 11）
        metadata={"source": "manual.pdf", "page": 11, "chunk_id": "c-0000"},
    ),
]

# 持久化目录，跑完这段会在当前目录下生成它
STORE_DIR = Path("chroma_store")
# 演示时先清空，保证每次运行结果一致；真实项目不要随手删库
# 注意：Windows 上如果同一进程里还有打开着的 Chroma 对象，这行会失败（见下文）
if STORE_DIR.exists():
    shutil.rmtree(STORE_DIR)

# from_documents 是建库的便捷入口：建集合 + 转向量 + 写入，一步完成
store = Chroma.from_documents(
    # 要入库的 Document 列表
    documents=CHUNKS,
    # 嵌入模型：建库时用它把 page_content 转成向量
    # 注意这里的参数名叫 embedding（单数，无 _function 后缀）
    embedding=get_embeddings(),
    # 集合名，一个库里可以有多个集合，互相隔离
    collection_name="kb_demo",
    # 指定目录即开启持久化；不传则只存在内存里，进程结束就没了
    # Path 对象要转成 str，Chroma 只接受字符串路径
    persist_directory=str(STORE_DIR),
    # 显式指定每条记录的主键，用第 13 章的 chunk_id
    # 列表推导式从每个 Document 的 metadata 里把 chunk_id 抽出来
    ids=[d.metadata["chunk_id"] for d in CHUNKS],
)

# 用公开 API 数条数（get() 不传参会返回全部记录）
# 返回的是字典，["ids"] 是主键列表，len() 即条数
print("入库条数：", len(store.get()["ids"]))
# 直接打印主键，确认用的是我们传进去的 chunk_id
print("主键：", store.get()["ids"])

实测输出：

入库条数： 4
主键： ['a-0000', 'a-0001', 'a-0002', 'c-0000']

主键就是我们传进去的 chunk_id，不是随机 UUID——这也是下面 §5.5 能精准更新的前提。

ids 参数是本章最该记住的细节之一：用 chunk_id 当主键，资料更新时才能精准替换（§5.5）。不传的话 Chroma 会生成随机 UUID，之后就没法定位某个 chunk。

store.get() 返回的字典比你想象的字段多，知道有什么便于排查：

# 打印 get() 返回字典的所有键
print(store.get().keys())

实测输出：

dict_keys(['ids', 'embeddings', 'documents', 'uris', 'included', 'data', 'metadatas'])

键	内容	备注
`ids`	主键列表	数条数、做增量判断都用它（§10.2）
`documents`	原文列表	与 `ids` 一一对应
`metadatas`	metadata 字典列表	排查过滤条件时打印它核对（§5.3）
`embeddings`	向量本体	默认是 `None`，要显式要求才返回
`included`	本次实际返回了哪些字段	用来确认上一条

想拿到向量本体得用 include 参数，但要注意它的行为：

# include 指定要返回哪些字段
data = store.get(include=["embeddings", "metadatas"])

# 打印实际非空的键，验证 include 的效果
print("非空字段：", [k for k, v in data.items() if v is not None])
# 取第一条向量的长度，确认维度
print("第一条向量维度：", len(data["embeddings"][0]))

实测输出：

非空字段： ['ids', 'embeddings', 'included', 'metadatas']
第一条向量维度： 1024

注意 documents 不见了：include 是「白名单」不是「追加」，没列进去的字段会变成 None。想同时要原文，就得写全 include=["embeddings", "documents", "metadatas"]，否则你会拿到一堆 None，愣半天。

集合名不能随便起

collection_name 有一套硬性规则，中文用户特别容易撞上：

# 逐个尝试，看哪些名字合法
for name in ["abc", "kb_demo", "kb", "退换货", "kb demo", "_kb", "kb-"]:
    try:
        # 只是创建集合对象，不写数据
        Chroma(collection_name=name, embedding_function=get_embeddings(),
               persist_directory="chroma_name_test")
        # 没抛异常就说明这个名字通过了校验
        print(f"  {name!r:<12} 合法")
    # 名字不合规时 Chroma 会在创建集合的那一刻就抛异常
    except Exception as error:
        # 只打印异常类名，完整消息很长
        print(f"  {name!r:<12} {type(error).__name__}")

实测输出：

  'abc'        合法
  'kb_demo'    合法
  'kb'         InvalidArgumentError
  '退换货'        InvalidArgumentError
  'kb demo'    InvalidArgumentError
  '_kb'        InvalidArgumentError
  'kb-'        InvalidArgumentError

完整报错说得很清楚：

Validation error: name: Expected a name containing 3-512 characters from
[a-zA-Z0-9._-], starting and ending with a character in [a-zA-Z0-9]. Got: kb

归纳成四条：

规则	反例
长度 3～512	`"kb"` 只有 2 个字符
只能用 `[a-zA-Z0-9._-]`	`"退换货"`（中文）、`"kb demo"`（空格）
必须以字母或数字开头	`"_kb"`
必须以字母或数字结尾	`"kb-"`

所以集合名别用中文，也别图短写成 "kb"。本章用的 kb_demo、company_kb 都合规。

5.2. 检索与分数 #

这一小节要钉死一件事：Chroma 返回的那个数字，到底是越大越像还是越小越像？ 方向搞反，你会把检索质量判反。下面用「拿库里的原文查自己」这个笨办法把它钉死。

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 打开已存在的库：注意参数名是 embedding_function（与建库时的 embedding 不同）
store = Chroma(
    # 必须与建库时一致，写错会静默打开一个空集合
    collection_name="kb_demo",
    # 必须与建库时是同一个模型，否则检索结果无意义
    embedding_function=get_embeddings(),
    # 必须与建库时一致，才能找到落盘的数据
    persist_directory="chroma_store",
)

# 最常用的检索方法：k 是返回条数
print("=== similarity_search ===")
# 内部会调 embed_query 把问题转成向量，再找最近的 k 条
for doc in store.similarity_search("过滤网什么时候要换？", k=2):
    # 返回的是 Document 列表，metadata 和 page_content 都还在
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# 带分数的版本，调参和排查时必用
print("\n=== similarity_search_with_score ===")
# 这个方法返回的是 (Document, 分数) 二元组的列表，所以要解包成两个变量
for doc, score in store.similarity_search_with_score("过滤网什么时候要换？", k=3):
    # :.4f 保留 4 位小数，便于对照
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}  {doc.page_content}")

# 拿一句和库里某条几乎一样的话去查，用来确认分数的方向
print("\n=== 方向验证 ===")
# 下面这句和 a-0000 的内容一字不差，是判断方向的关键实验
for doc, score in store.similarity_search_with_score(
    "签收 7 日内可无理由退货，商品需保持完好。", k=2
):
    # 第一名的分数如果是 0，就证明这个数字是「距离」而不是「相似度」
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

# 归一化版本：越大越像
print("\n=== relevance_scores ===")
# 这个方法把距离换算成 0~1 的相关度，方向与上面相反
for doc, score in store.similarity_search_with_relevance_scores("过滤网什么时候要换？", k=3):
    # 同一批数据、同一个问题，只是换了个表达单位
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

实测输出：

=== similarity_search ===
  c-0000 滤芯到期时指示灯会闪红灯，此时需更换滤芯。
  a-0001 质量问题 15 日内可换货，需提供照片凭证。

=== similarity_search_with_score ===
  0.4661  c-0000  滤芯到期时指示灯会闪红灯，此时需更换滤芯。
  1.1058  a-0001  质量问题 15 日内可换货，需提供照片凭证。
  1.2291  a-0000  签收 7 日内可无理由退货，商品需保持完好。

=== 方向验证 ===
  0.0000  a-0000
  0.6437  a-0001

=== relevance_scores ===
  0.6704  c-0000
  0.2181  a-0001
  0.1309  a-0000

关于分数的方向，这是最容易搞错的地方： Chroma 默认返回的是距离，所以数值越小越相似。

方向验证那一段是最有力的证据：拿一句和库里完全一样的话去查，分数是 0.0000——距离为零就是「一模一样」。要是返回的是相似度，这里该是 1.0。以后拿不准某个库的分数方向，用这个笨办法试一次，立刻清楚。

方法	返回	方向
`similarity_search`	只有文档	—
`similarity_search_with_score`	文档 + 距离	越小越像
`similarity_search_with_relevance_scores`	文档 + 归一化相关度	越大越像

现在把它和 §4 的手算对上。 Chroma 默认用平方 L2 距离，而通义输出的是归一化向量（§3.4 里模长恰好是 1.0）。这种情况下距离和余弦相似度是可以互相换算的：

$$ d = \lVert \vec{q} - \vec{v} \rVert^2 = 2\,(1 - \cos\theta) \qquad\Longleftrightarrow\qquad \cos\theta = 1 - \frac{d}{2} $$

代入实测数字：$1 - 0.4661 / 2 = 0.7670$——和 §4 手算的 0.7670 一模一样。

对得上意味着：向量库没做什么神秘的事，就是在算你手算过的那个余弦，只是换了个单位。relevance_scores 的 0.6704 是 LangChain 用 $1 - d/\sqrt{2}$ 归一化出来的，方向反过来，同样只是换算。

这套换算逻辑写在 Chroma._select_relevance_score_fn() 里，它做的事很简单：读出集合建的时候用的距离空间，然后挑对应的换算函数。

集合的距离空间	`with_score` 返回	`relevance_scores` 的换算
`l2`（默认）	平方 L2 距离	$1 - d/\sqrt{2}$
`cosine`	余弦距离，即 $1-\cos\theta$	$1 - d$

如果你不想在脑子里做换算，可以直接把集合建成 cosine 空间。 建库时多传一个参数即可：

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma
# 构造演示数据
from langchain_core.documents import Document

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 两条数据够验证分数了
DOCS = [
    # 第一条：后面用它「查自己」，看距离是不是 0
    Document(page_content="签收 7 日内可无理由退货，商品需保持完好。",
             metadata={"chunk_id": "a-0000"}),
    # 第二条：讲滤芯，是「过滤网」那个问题的正确答案
    Document(page_content="滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
             metadata={"chunk_id": "c-0000"}),
]

# 建库，其余参数与 §5.1 相同，只多了最后一个
cos_store = Chroma.from_documents(
    DOCS,
    get_embeddings(),
    # 换个集合名，不要和主线的 kb_demo 混在一起
    collection_name="cos_demo",
    # 也换个目录，避免互相干扰
    persist_directory="chroma_cos",
    # 照例用 chunk_id 作主键
    ids=[d.metadata["chunk_id"] for d in DOCS],
    # 关键参数：把底层 HNSW 索引的距离空间指定为 cosine
    # 可选值：'l2'（默认）/ 'cosine' / 'ip'
    collection_configuration={"hnsw": {"space": "cosine"}},
)

# 这次 with_score 返回的是余弦距离（1 - 余弦相似度）
print("=== cosine 空间的 with_score ===")
# 同一个问题，与 §5.2 的默认 l2 空间做对照
for doc, score in cos_store.similarity_search_with_score("过滤网什么时候要换？", k=2):
    # 值域变成 0~2，0 表示方向完全一致
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

# relevance_scores 用 1 - d 换算，结果就是余弦相似度本身
print("\n=== cosine 空间的 relevance_scores ===")
# 这一行的输出可以直接和 §4 手算的余弦对比
for doc, score in cos_store.similarity_search_with_relevance_scores("过滤网什么时候要换？", k=2):
    # 打印出来应该是 0.7669，即手算的那个 0.7670
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

实测输出：

=== cosine 空间的 with_score ===
  0.2331  c-0000
  0.6146  a-0000

=== cosine 空间的 relevance_scores ===
  0.7669  c-0000
  0.3854  a-0000

0.7669 就是 §4 手算的那个 0.7670（末位差是浮点误差）。所以在 cosine 空间下，relevance_scores 直接就是余弦相似度，不用再做任何心算。

排查检索问题时，先把分数的方向和量纲搞清楚。 把距离当相似度看，会得出「明明 1.2 分很高却召回了无关内容」这种完全错误的判断。

还有一个操作层面的提醒：距离空间只能在建集合时指定，之后改不了。 想换空间就得新建一个集合并重新嵌入全部资料，和换嵌入模型一样贵（§11）。

5.3. 元数据过滤 #

这是向量库比手算强的关键能力，也是第 12 章辛苦经营 metadata 的回报：先按条件把候选集缩小，再在里面算相似度。

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 打开 §5.1 建好的库
store = Chroma(
    collection_name="kb_demo",
    embedding_function=get_embeddings(),
    persist_directory="chroma_store",
)

# 只在 manual.pdf 这一个文件里检索
print("=== 只搜 manual.pdf ===")
# filter 的键必须与 metadata 里的键名完全一致，值也要类型一致
for doc in store.similarity_search("有什么规定？", k=2, filter={"source": "manual.pdf"}):
    # 传了 k=2 但库里符合条件的只有 1 条，所以只会打印一行
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# 只搜「退换货」这一章节：filter 的键就是 metadata 的键
print("\n=== 只搜 h2=退换货 ===")
# h2 是第 13 章 MarkdownHeaderTextSplitter 写进去的二级标题
for doc in store.similarity_search("有什么规定？", k=2, filter={"h2": "退换货"}):
    # 这次有 2 条符合条件，会打印两行
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# 组合条件用操作符，$and / $or / $in / $ne 等
print("\n=== 组合条件 ===")
# 这次要求同时满足「来自售后文件」和「属于退换货章节」
for doc in store.similarity_search(
    "有什么规定？",
    k=3,
    # $and 接收一个条件列表，要求全部满足；注意每个条件都是独立的字典
    filter={"$and": [{"source": "policy/aftersale.md"}, {"h2": "退换货"}]},
):
    # 结果应与只按 h2 过滤时一致，因为这两条本来就同源
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# get() 也支持过滤，用 where 参数（注意不叫 filter）；适合做统计而非检索
# 不传条件就是全部记录
print("\n库中总条数：", len(store.get()["ids"]))
# 带 where 条件就是符合条件的记录数，用来核对过滤条件写对没有
print("退换货章节共有：", len(store.get(where={"h2": "退换货"})["ids"]), "条")

实测输出：

=== 只搜 manual.pdf ===
  c-0000 滤芯到期时指示灯会闪红灯，此时需更换滤芯。

=== 只搜 h2=退换货 ===
  a-0001 质量问题 15 日内可换货，需提供照片凭证。
  a-0000 签收 7 日内可无理由退货，商品需保持完好。

=== 组合条件 ===
  a-0001 质量问题 15 日内可换货，需提供照片凭证。
  a-0000 签收 7 日内可无理由退货，商品需保持完好。

库中总条数： 4
退换货章节共有： 2 条

两处细节值得留意：

第一，过滤先生效，k 只是上限。第一段传了 k=2，却只返回 1 条——库里 source 为 manual.pdf 的本来就只有 1 条。所以「返回条数比 k 少」不是 bug，是过滤把候选集缩小了。

第二，过滤会强行改掉排序结果。这三段查询词都是「有什么规定？」，但限定文件后召回的内容完全不同。过滤是「硬约束」——范围外再相关也进不来。权限隔离靠的就是这个；过滤写太严，正确答案也会被挡在外面。

过滤条件写错不会报错，只会返回 0 条

这是本章最值得亲手验证一次的坑。三种写错方式，全部静默返回 0 条：

# 复用上面打开的 store

# 错法一：键名拼错（h2 写成了 h22）
print("键名拼错 h22：", len(store.similarity_search("规定", k=3, filter={"h22": "退换货"})))

# 错法二：键名对但值不存在（库里没有这个章节名）
print("值不存在：", len(store.similarity_search("规定", k=3, filter={"h2": "不存在的章节"})))

# 错法三：值的类型不对
# §5.1 里 page 存的是整数 11，这里传字符串 "11"
print("值类型不符：", len(store.similarity_search("规定", k=3, filter={"page": "11"})))

实测输出：

键名拼错 h22： 0
值不存在： 0
值类型不符： 0

三种情况都是 0 条、零报错、零警告。 第三种最阴险：page 在 metadata 里是整数 11，你从命令行参数或 JSON 里读出来的是字符串 "11"，看上去一模一样，检索却永远为空。

排查手法很固定，两步：

# 第一步：打印真实 metadata，看清键名和值的类型
# 按 id 精确取回那条带 page 的记录，比用下标猜更可靠
print(store.get(ids=["c-0000"])["metadatas"][0])

# 第二步：用 where 数一下该条件下到底有多少条
# 返回 0 就说明是过滤条件的问题，不是相似度的问题
print("符合条件的条数：", len(store.get(where={"page": 11})["ids"]))

实测输出：

{'source': 'manual.pdf', 'chunk_id': 'c-0000', 'page': 11}
符合条件的条数： 1

'page': 11 没有引号，说明它是整数——这就是第三种错法的根因。（键的顺序每次运行可能不同，这是正常的，见 §10.2。）

口诀：检索返回 0 条时，先怀疑 filter，再怀疑库空，最后才怀疑模型。 用 store.get(where=...) 数一下，一次就能定位。

典型用途：

需求	filter 写法
只搜某个文件	`{"source": "policy/aftersale.md"}`
只搜某章节	`{"h2": "退换货"}`
排除某文件	`{"source": {"$ne": "draft.md"}}`
多个来源之一	`{"source": {"$in": ["a.md", "b.md"]}}`
按部门做权限隔离	`{"department": "hr"}`

最后一条是第 15 章会用到的正经场景：普通员工检索时自动加上部门过滤，财务文档就不会被召回。这类需求靠 metadata，而 metadata 必须在第 12 章加载时就写好——现在该能体会那一章为什么反复强调它了。

5.4. 持久化 #

传了 persist_directory 的 Chroma 会自动落盘，不必再手动调 persist()。重启进程后，用同样的 collection_name 和目录打开即可：

# 标准库：遍历目录、查看文件大小
from pathlib import Path

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 关键：collection_name 与 persist_directory 都要和建库时一致，否则会打开一个空集合
store = Chroma(
    # 集合名写错不会报错，只会新建一个同名空集合——这是「库怎么空了」的常见原因
    collection_name="kb_demo",
    embedding_function=get_embeddings(),
    persist_directory="chroma_store",
)

# 条数与写入时一致，就说明数据确实落盘了
print("重新打开后条数：", len(store.get()["ids"]))

# 看看落盘目录里有什么
# rglob("*") 递归列出所有条目，sorted 保证输出顺序稳定
for path in sorted(Path("chroma_store").rglob("*")):
    # 只看文件，跳过目录本身
    if path.is_file():
        # relative_to 去掉前缀让输出更短；// 1024 转成 KB（整除，不足 1KB 显示 0）
        print(" ", path.relative_to("chroma_store"), f"{path.stat().st_size // 1024} KB")

实测输出（这段是在 §5.5 的增删改之后跑的，所以是 3 条）：

重新打开后条数： 3
  chroma.sqlite3 204 KB
  e5eedf9e-727a-4e66-897d-1f9695f8a2fc\data_level0.bin 413 KB
  e5eedf9e-727a-4e66-897d-1f9695f8a2fc\header.bin 0 KB
  e5eedf9e-727a-4e66-897d-1f9695f8a2fc\length.bin 0 KB
  e5eedf9e-727a-4e66-897d-1f9695f8a2fc\link_lists.bin 0 KB

条数与写入时一致（包含之前的增删改结果），说明落盘生效。目录结构也能看出 Chroma 是怎么存的：

文件	存什么
`chroma.sqlite3`	原文、metadata、主键——即「能读懂的部分」
`data_level0.bin`	向量本体与 HNSW 索引结构
那个 UUID 目录名	集合的内部 id，不用管，也不要手改

注意 data_level0.bin 有 413 KB，而库里只有 3 条记录。这是因为 HNSW 索引会预分配空间，小库的磁盘占用看起来不成比例是正常的，不代表出错。

有一个坑要提前说：打开时传的嵌入模型必须和建库时同一个。维度不一致会直接报错；更麻烦的是换了同维度的另一个模型——不报错，但检索结果会一塌糊涂（§11）。

5.5. 增删改：chunk_id 的价值 #

资料会更新，知识库就得跟着更新。有了 chunk_id 作主键，三种操作都很直接。

注意这段会修改和删除 §5.1 建的数据，所以整个 §5 里它要放在最后跑。如果你已经跑过 §5.5 又想复现 §5.2 的分数，重跑一次 §5.1 建库即可。

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma
# 构造要写入的新版本内容
from langchain_core.documents import Document

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 打开 §5.1 建好的库
store = Chroma(
    collection_name="kb_demo",
    embedding_function=get_embeddings(),
    persist_directory="chroma_store",
)

# 先记下操作前的条数，作为后面对比的基准
print("操作前条数：", len(store.get()["ids"]))

# 【改】用同一个 id 再写一次，效果是覆盖（upsert），不会产生重复记录
store.add_documents(
    [
        Document(
            # 内容改了：7 日变成 15 日，模拟政策更新
            page_content="签收 15 日内可无理由退货（政策已更新）。",
            # metadata 里的 chunk_id 保持不变，还是 a-0000
            metadata={"source": "policy/aftersale.md", "h2": "退换货", "chunk_id": "a-0000"},
        )
    ],
    # 关键：显式传入与旧记录相同的 id，Chroma 才知道这是「覆盖」而不是「新增」
    ids=["a-0000"],
)
# 条数应该不变（4 → 4），证明是覆盖而非新增
print("覆盖后条数：", len(store.get()["ids"]))
# 取回这条看内容有没有换成新版本
print("a-0000 的内容：", store.get(ids=["a-0000"])["documents"])

# 【删】按 id 删除，参数是 id 列表，可以一次删多条
store.delete(ids=["c-0000"])
# 条数应该减 1（4 → 3）
print("删除后条数：", len(store.get()["ids"]))

# 【查】按 id 精确取回，不做相似度计算，也不需要调嵌入 API
print("按 id 取回：", store.get(ids=["a-0001"])["documents"])

实测输出：

操作前条数： 4
覆盖后条数： 4
a-0000 的内容： ['签收 15 日内可无理由退货（政策已更新）。']
删除后条数： 3
按 id 取回： ['质量问题 15 日内可换货，需提供照片凭证。']

三件事一次看清：

操作	结果
用相同 `id` 再写入	条数不变（4 → 4），内容被替换成新版本
`delete(ids=[...])`	条数减少（4 → 3）
`get(ids=[...])`	精确取回指定记录，不做相似度计算

「覆盖后条数不变」是这里最关键的一条。如果不传 ids，同样的两次写入会变成 4 → 5，库里躺着两份内容矛盾的退货政策，检索时召回哪一份完全看运气——这是知识库最难排查的一类问题，因为它不报错。

资料更新的标准做法，正是第 13 章 chunk_id 用 doc_id 作前缀的原因：

某个文件被修改了
    │
    ▼
① 用 doc_id 前缀查出该文件的所有旧 chunk_id
② delete 掉它们
③ 重新切分该文件并写入新 chunk
    │
    ▼
只重算了一个文件，其他文件的向量一次都没重复付费

千万不要图省事每次全库重建——那意味着每次更新一个文件都要为全部资料重新付一遍嵌入费用。

6. 向量库：FAISS #

6.1. 建库与存取 #

FAISS 是 Meta 开源的向量检索库，特点是快、纯本地文件、无服务进程。它和 Chroma 最大的差别在定位：Chroma 是「带存储的数据库」，FAISS 更像「一个索引结构」——原文和 metadata 是 LangChain 额外用 Python 字典帮你存的。

# 标准库：递归删除目录
import shutil
# 标准库：路径处理
from pathlib import Path

# FAISS 的 LangChain 封装（和 DashScopeEmbeddings 同在 community 包里）
from langchain_community.vectorstores import FAISS
# Document 数据结构
from langchain_core.documents import Document

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 三条演示数据，故意比 §5 少一条，避免与 Chroma 的输出混淆
CHUNKS = [
    Document(
        page_content="签收 7 日内可无理由退货，商品需保持完好。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0000"},
    ),
    Document(
        page_content="一般订单 48 小时内发出，节假日顺延。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0002"},
    ),
    Document(
        page_content="滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
        metadata={"source": "manual.pdf", "page": 11, "chunk_id": "c-0000"},
    ),
]

# 只创建一次嵌入模型实例，后面存和读都要用同一个
embeddings = get_embeddings()

# 建库：FAISS 默认只在内存里，没有 persist_directory 这种参数
store = FAISS.from_documents(CHUNKS, embeddings)
# index 是底层的 FAISS 索引对象，ntotal 是里面的向量条数
# 注意这里不能像 Chroma 那样用 get()，FAISS 没有这个方法
print("入库条数：", store.index.ntotal)

# 检索：默认返回 L2 距离，越小越像
print("\n=== with_score（L2 距离，越小越像）===")
# 接口与 Chroma 完全一致，这是 LangChain 抽象层的价值
for doc, score in store.similarity_search_with_score("过滤网什么时候要换？", k=3):
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

# 归一化版本：越大越像，跨库比较时更直观
print("\n=== relevance_scores（越大越像）===")
# 换算公式与 Chroma 的 l2 空间相同：1 - d/sqrt(2)
for doc, score in store.similarity_search_with_relevance_scores("过滤网什么时候要换？", k=3):
    print(f"  {score:.4f}  {doc.metadata['chunk_id']}")

# 保存到本地目录
FAISS_DIR = Path("faiss_store")
# 演示时先清空，保证结果可复现
if FAISS_DIR.exists():
    shutil.rmtree(FAISS_DIR)
# save_local 需要手动调用——这是与 Chroma 最大的运维差别
store.save_local(str(FAISS_DIR))
# 列出目录里生成了哪些文件
print("\n保存的文件：", [p.name for p in FAISS_DIR.iterdir()])

# 读回来：必须传入同一个嵌入模型
# allow_dangerous_deserialization=True 是必需的，因为 index.pkl 用 pickle 存储
# 只对自己生成的文件开启，不要加载来源不明的 FAISS 文件
store2 = FAISS.load_local(str(FAISS_DIR), embeddings, allow_dangerous_deserialization=True)
# 条数一致说明存取成功
print("重新加载条数：", store2.index.ntotal)

实测输出：

入库条数： 3

=== with_score（L2 距离，越小越像）===
  0.4662  c-0000
  1.2292  a-0000
  1.2483  a-0002

=== relevance_scores（越大越像）===
  0.6704  c-0000
  0.1308  a-0000
  0.1173  a-0002

保存的文件： ['index.faiss', 'index.pkl']
重新加载条数： 3

save_local 生成两个文件，分工很清楚：

文件	存什么	实测大小（3 条数据）
`index.faiss`	向量本体与索引结构	12,333 字节
`index.pkl`	原文、metadata、id 映射表	809 字节

index.faiss 的大小基本就是 3 × 1024 × 4 字节 = 12,288 再加一点头部——FAISS 几乎没有额外存储开销，和 Chroma 那个预分配 413 KB 的 HNSW 索引形成对比（§5.4）。

顺便对照 §5.2 的 Chroma 结果：同样的问题、同样的 chunk，Chroma 给的距离是 0.4661，FAISS 给的是 0.4662，relevance_scores 两边都是 0.6704。两个库算的是同一个式子，差只在小数末位的浮点误差。这也说明：换向量库不会改检索效果，换嵌入模型才会（§9）。

allow_dangerous_deserialization=True 不是可选项而是必填。不传的话报错很直白：

ValueError: The de-serialization relies loading a pickle file. Pickle files can be
modified to deliver a malicious payload that results in execution of arbitrary code
on your machine. You will need to set `allow_dangerous_deserialization` to `True` to
enable deserialization.

参数名里的 "dangerous" 是认真的：pickle 文件加载即执行代码，所以只加载你自己生成的文件，绝不要加载别人发来的 FAISS 目录。

还有一个容易误解的点：from_documents 建的索引是什么类型？

# 打印底层索引的类名
print("索引类型：", type(store.index).__name__)

实测输出：

索引类型： IndexFlatL2

Flat 意思是暴力精确检索——每次查询都和全部向量算一遍距离，没有任何近似。所以「FAISS 更快」要加个限定：它快是因为底层 C++ 和 SIMD 指令优化，而不是因为用了近似算法。复杂度仍是 $O(n)$。

6.2. Chroma 还是 FAISS #

先看两个实测出来的关键差异，它们比性能更影响日常使用。

差异一：FAISS 也支持 ids，但同 id 重写会报错，而且报错之后这个库就废了。

Chroma 上同 id 重写是覆盖（§5.5），FAISS 不是。更要紧的是它失败的方式——这是本章挖出来的最严重的一个坑，值得完整看一遍：

# 标准库：递归删除目录
import shutil

# 标准库：路径处理
from pathlib import Path

# FAISS 的 LangChain 封装（和 DashScopeEmbeddings 同在 community 包里）
from langchain_community.vectorstores import FAISS

# Document 数据结构
from langchain_core.documents import Document

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 三条演示数据，故意比 §5 少一条，避免与 Chroma 的输出混淆
CHUNKS = [
    Document(
        page_content="签收 7 日内可无理由退货，商品需保持完好。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0000"},
    ),
    Document(
        page_content="一般订单 48 小时内发出，节假日顺延。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0002"},
    ),
    Document(
        page_content="滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
        metadata={"source": "manual.pdf", "page": 11, "chunk_id": "c-0000"},
    ),
]
# 三条演示数据，故意比 §5 少一条，避免与 Chroma 的输出混淆
CHUNKS = [
    Document(
        page_content="签收 7 日内可无理由退货，商品需保持完好。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0000"},
    ),
    Document(
        page_content="一般订单 48 小时内发出，节假日顺延。",
        metadata={"source": "policy/aftersale.md", "chunk_id": "a-0002"},
    ),
    Document(
        page_content="滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
        metadata={"source": "manual.pdf", "page": 11, "chunk_id": "c-0000"},
    ),
]
# 只创建一次嵌入模型实例，后面存和读都要用同一个
embeddings = get_embeddings()
# 建库时同样可以指定 chunk_id 作主键
store3 = FAISS.from_documents(
    CHUNKS,
    embeddings,
    ids=[d.metadata["chunk_id"] for d in CHUNKS],
)

# index_to_docstore_id 是「索引下标 → 文档 id」的映射表，是 FAISS 的核心内部状态
# 打印第一阶段的标题
print("① 建库后")
# ntotal 是底层向量条数
print("   ntotal:", store3.index.ntotal)
# 映射表应与向量条数一一对应，这是 FAISS 能正常工作的前提
print("   映射表:", store3.index_to_docstore_id)

# 用已存在的 id 再写一次，模拟资料更新
print("\n② 尝试用相同 id 重写")
try:
    store3.add_documents(
        [Document(page_content="改过的内容", metadata={"chunk_id": "a-0000"})],
        ids=["a-0000"],
    )
# 用 try/except 包住，否则脚本会在这里直接崩掉
except Exception as error:
    # 打印异常类型与消息，看清 FAISS 的态度
    print(f"   {type(error).__name__}: {error}")

# 关键：检查报错之后的内部状态是否还自洽
print("\n③ 报错之后的状态")
# 重点看这个数字有没有变
print("   ntotal:", store3.index.ntotal)
# 再看映射表有没有跟着变
print("   映射表:", store3.index_to_docstore_id)

# 现在做一次最普通的检索
print("\n④ 此时检索")
try:
    # 一次最普通的检索，没有任何特殊参数
    hits = store3.similarity_search("有什么规定？", k=3)
    # 能走到这里说明库还是好的
    print("   成功:", [d.metadata.get("chunk_id") for d in hits])
# 同样要包 try，因为下面会看到它其实会炸
except Exception as error:
    print(f"   失败: {type(error).__name__}: {error}")

实测输出：

① 建库后
   ntotal: 3
   映射表: {0: 'a-0000', 1: 'a-0002', 2: 'c-0000'}

② 尝试用相同 id 重写
   ValueError: Tried to add ids that already exist: {'a-0000'}

③ 报错之后的状态
   ntotal: 4
   映射表: {0: 'a-0000', 1: 'a-0002', 2: 'c-0000'}

④ 此时检索
   失败: KeyError: np.int64(3)

看第 ③ 步：ntotal 变成了 4，映射表却还是 3 条。 也就是说 FAISS 在做重复检查之前，已经把向量写进底层索引了——报错只拦住了「登记 id」这一步，没有回滚已经写入的向量。于是索引里躺着一个第 3 号向量，映射表里查不到它，任何检索一旦命中它就抛 KeyError。

而且这个损坏是不可修复的，连删除都救不回来：

# 试着删掉一条，看能不能让映射表恢复自洽
print("delete:", store3.delete(["a-0000"]))
# 向量条数减了 1
print("ntotal:", store3.index.ntotal)
# 映射表也减了 1，但两者的差值依旧是 1，错位没有消失
print("映射表:", store3.index_to_docstore_id)

# 再检索一次
try:
    hits = store3.similarity_search("有什么规定？", k=2)
    # 这一行不会被执行到
    print("检索成功:", [d.metadata.get("chunk_id") for d in hits])
# 依旧会抛异常，只是错的下标号变了
except Exception as error:
    print(f"检索失败: {type(error).__name__}: {error}")

实测输出：

delete: True
ntotal: 3
映射表: {0: 'a-0002', 1: 'c-0000'}
检索失败: KeyError: np.int64(2)

错位依旧存在，只是下标从 3 变成了 2。

结论：在 FAISS 上，add_documents 抛出 Tried to add ids that already exist 之后，这个 store 对象必须丢弃重建，不能 try/except 接住然后继续用。 这是一个「响亮地报错，然后静默地损坏」的组合，比单纯报错危险得多。

正确的更新姿势是先删再加：

# 重新建一个干净的库
store4 = FAISS.from_documents(
    CHUNKS, embeddings, ids=[d.metadata["chunk_id"] for d in CHUNKS]
)

# 第一步：先把旧记录删掉
store4.delete(["a-0000"])
# 第二步：再写入新版本，此时 id 已经不存在，不会触发重复检查
store4.add_documents(
    [Document(page_content="签收 15 日内可无理由退货（已更新）。",
              metadata={"source": "policy/aftersale.md", "chunk_id": "a-0000"})],
    ids=["a-0000"],
)

# 条数回到 3，说明是替换而不是新增
print("ntotal:", store4.index.ntotal)
# 映射表自洽：注意 a-0000 被排到了末尾，下标会重新分配
print("映射表:", store4.index_to_docstore_id)
# 检索验证内容确实换成了新版本
print("检索:", store4.similarity_search("退货几天", k=1)[0].page_content)

实测输出：

ntotal: 3
映射表: {0: 'a-0002', 1: 'c-0000', 2: 'a-0000'}
检索: 签收 15 日内可无理由退货（已更新）。

一切正常。注意映射表里 a-0000 跑到了末尾——FAISS 删除时会重排下标，所以不要在代码里依赖下标的稳定性，只依赖 id。

差异二：FAISS 的 delete 和 metadata 过滤其实都能用，只是不如 Chroma 顺手：

# 用干净的 store4 演示，避免受上面损坏状态的影响
# metadata 过滤同样支持，语法比 Chroma 简单（不支持 $and / $in 这类操作符）
# 列表推导式只把 chunk_id 抽出来，输出更紧凑
print("过滤检索：", [
    d.metadata["chunk_id"]
    # 限定只搜 manual.pdf，与 §5.3 的第一个例子对应
    for d in store4.similarity_search("有什么规定？", k=2, filter={"source": "manual.pdf"})
])

实测输出：

过滤检索： ['c-0000']

所以下面这张表里的「较麻烦」「较弱」都是相对的，不是「不能用」：

	Chroma	FAISS
持久化	传 `persist_directory` 自动落盘	手动 `save_local` / `load_local`
元数据过滤	原生支持，有 `$and` / `$in` / `$ne` 等操作符	支持等值匹配，没有操作符语法
同 id 重写	覆盖（upsert）	抛 `ValueError`，且把库搞坏，必须先删再加
按 id 删除	`delete(ids=[...])`	`delete([...])`，可用但会重排内部下标
默认索引	HNSW（近似，预分配磁盘）	`IndexFlatL2`（暴力精确）
小库磁盘占用	400+ KB（预分配）	约等于向量本身大小
依赖	`chromadb`	`faiss-cpu`
适合	企业知识库	只读、大规模、追求速度

选择建议：

要按文件增量更新、要按部门/章节过滤，选 Chroma；数据基本只读、追求检索速度，选 FAISS。 拿不准就用 Chroma——本章实战用的就是它。

两者的 LangChain 接口高度一致（都有 similarity_search、as_retriever），所以切换成本不高。真正的迁移成本在重新嵌入全部资料。

7. MMR：结果太像怎么办 #

纯相似度检索有个副作用：召回的几条可能高度雷同。比如库里有三句都在讲退货期限，用户问「怎么退货」，前三名全是这三句的近义说法——占满了上下文，却没带来新信息。

MMR（Maximal Marginal Relevance，最大边际相关性） 在「与问题相关」和「彼此不同」之间做权衡：

普通相似度检索：           MMR：
  1. 退货期限是 7 天        1. 退货期限是 7 天       ← 最相关的先选
  2. 退货须在 7 日内        2. 退货需保持包装完好   ← 换一个角度
  3. 7 天内可无理由退       3. 退货运费由谁承担     ← 再换一个角度
  ↑ 三条几乎同义，浪费额度   ↑ 覆盖面更广

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 打开 §5.1 建好的库（这段要在 §5.5 的增删改之前跑）
store = Chroma(
    collection_name="kb_demo",
    embedding_function=get_embeddings(),
    persist_directory="chroma_store",
)

# 打印标题，区分下面两组结果
print("=== MMR ===")
# max_marginal_relevance_search 是 MMR 检索的入口
for doc in store.max_marginal_relevance_search(
    "退货有什么要求？",
    # k：最终返回这么多条
    k=2,
    # fetch_k：先按纯相似度粗召回这么多条作为候选池
    fetch_k=4,
    # lambda_mult：0~1，越大越偏相关性，越小越偏多样性，默认 0.5
    lambda_mult=0.5,
):
    # 注意 MMR 不返回分数，因为「边际相关性」不是一个可直接比较的量
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# 打印第二组的标题
print("\n=== 对照：普通相似度检索 ===")
# 同样的问题、同样的 k，只是不做多样性挑选，用来对比差别
for doc in store.similarity_search("退货有什么要求？", k=2):
    # 两组结果的第一名一定相同，差别在第二名
    print(" ", doc.metadata["chunk_id"], doc.page_content)

实测输出：

=== MMR ===
  a-0000 签收 7 日内可无理由退货，商品需保持完好。
  c-0000 滤芯到期时指示灯会闪红灯，此时需更换滤芯。

=== 对照：普通相似度检索 ===
  a-0000 签收 7 日内可无理由退货，商品需保持完好。
  a-0001 质量问题 15 日内可换货，需提供照片凭证。

这个结果如实暴露了 MMR 的代价，比漂亮结果更有教育意义：两者第一名相同（最相关的那条一定会被选中），但第二名 MMR 挑走了完全无关的滤芯说明——库里只有 4 条，「与已选内容不同」的候选本来就没剩几个，为了多样性只好去拿不相关的。

结论是：MMR 不是「更好的检索」，而是用相关性换覆盖面的旋钮。 库很小，或资料本身没有近义重复时，开 MMR 只会引入噪声。

lambda_mult 这个旋钮的手感值得实测一次。 从 0 扫到 1，看结果什么时候会变（仍然用 §5.1 的 4 条库）：

# 从纯多样性（0.0）扫到纯相关性（1.0）
for lambda_mult in [0.0, 0.3, 0.5, 0.7, 0.8, 0.9, 1.0]:
    # 其余参数保持不变，只改 lambda_mult
    hits = store.max_marginal_relevance_search(
        "退货有什么要求？", k=2, fetch_k=4, lambda_mult=lambda_mult
    )
    # 只打印 chunk_id，便于横向对比
    print(f"lambda_mult={lambda_mult}: {[d.metadata['chunk_id'] for d in hits]}")

# 最后打印普通相似度检索的结果作为基准
print(f"similarity  : {[d.metadata['chunk_id'] for d in store.similarity_search('退货有什么要求？', k=2)]}")

实测输出：

lambda_mult=0.0: ['a-0000', 'c-0000']
lambda_mult=0.3: ['a-0000', 'c-0000']
lambda_mult=0.5: ['a-0000', 'c-0000']
lambda_mult=0.7: ['a-0000', 'a-0002']
lambda_mult=0.8: ['a-0000', 'a-0001']
lambda_mult=0.9: ['a-0000', 'a-0001']
lambda_mult=1.0: ['a-0000', 'a-0001']
similarity  : ['a-0000', 'a-0001']

这个结果把旋钮的作用说得很清楚：第一名永远是 a-0000，第二名随着 lambda_mult 增大越来越相关。

`lambda_mult`	第二名	内容	与「退货要求」的关系
0.0～0.5	`c-0000`	滤芯更换	完全无关，纯粹为了不重复
0.7	`a-0002`	发货时效	沾一点边（都是售后规则）
0.8～1.0	`a-0001`	质量问题换货	最相关

三个观察：

lambda_mult=1.0 的结果和普通相似度检索完全一致。这符合定义：权重全给相关性，多样性项被彻底忽略，MMR 退化成普通检索。可以用它当 sanity check。
变化不是渐变的，而是分段跳变。MMR 是离散选择，只有当多样性惩罚大到足以翻转排名时结果才会变。所以别指望微调 0.05 能看出差别，要以 0.1～0.2 为步长试。
默认值 0.5 在这个小库上给出的是最差的结果。这再次说明默认值不是「推荐值」，只是一个中间点。

fetch_k 同样会改变结果，而且方向可能和你想的相反：

# 固定 lambda_mult=0.5，只改候选池大小
for fetch_k in [2, 3, 4]:
    hits = store.max_marginal_relevance_search(
        "退货有什么要求？", k=2, fetch_k=fetch_k, lambda_mult=0.5
    )
    # 只看 chunk_id，观察第二名怎么随候选池变化
    print(f"fetch_k={fetch_k}: {[d.metadata['chunk_id'] for d in hits]}")

实测输出：

fetch_k=2: ['a-0000', 'a-0001']
fetch_k=3: ['a-0000', 'a-0002']
fetch_k=4: ['a-0000', 'c-0000']

fetch_k 越大，MMR 能「跑」得越远。 fetch_k=2 时候选池只有两条，MMR 无从选择，结果等于普通检索；fetch_k=4 把全库都放进候选池，它就有机会挑到那条毫不相关的滤芯说明。

所以这两个参数是联动的：fetch_k 决定「能走多远」，lambda_mult 决定「愿不愿意走」。调 MMR 时两个都要动。

参数	作用	建议
`k`	最终返回条数	3～5
`fetch_k`	候选池大小，决定 MMR 能挑到多远的内容	`k` 的 3～5 倍，且要明显小于库的总条数才有意义
`lambda_mult`	相关性与多样性的权衡，越大越偏相关性	从 0.5 起调，步长 0.1～0.2；跑偏就往 0.8 调

什么时候用：

场景	建议
资料里有大量近义重复内容（同一政策在多处重述）	用 MMR
问题需要多个角度的信息才能答全	用 MMR
资料本身已经很精简、条目互不重复	普通相似度检索即可
库里只有几十条	先别用，容易像上面一样跑偏

8. as_retriever：接给下游 #

第 15 章的 RAG 和 Agent 不会直接调 similarity_search，而是走统一的 Retriever 接口。任何向量库都能一行转成 Retriever：

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 打开 §5.1 建好的库
store = Chroma(
    collection_name="kb_demo",
    embedding_function=get_embeddings(),
    persist_directory="chroma_store",
)

# 默认按相似度检索，k 等参数通过 search_kwargs 字典传
# 注意 k 不是 as_retriever 的直接参数，写成 as_retriever(k=2) 会被忽略
retriever = store.as_retriever(search_kwargs={"k": 2})

# Retriever 是第 7 章讲过的 Runnable，所以用 invoke 调用
# 输入是字符串，输出是 Document 列表——注意没有分数
for doc in retriever.invoke("过滤网什么时候要换？"):
    print(" ", doc.metadata["chunk_id"], doc.page_content)

# 换成 MMR 检索：search_type 决定用哪种检索策略
mmr_retriever = store.as_retriever(
    # 可选 "similarity"（默认）/ "mmr" / "similarity_score_threshold"
    search_type="mmr",
    # MMR 特有的 fetch_k 也放在 search_kwargs 里
    search_kwargs={"k": 2, "fetch_k": 4},
)

# 带元数据过滤的 Retriever：把权限、范围固化在检索器里
scoped_retriever = store.as_retriever(
    # filter 与 §5.3 的写法完全一致，只是提前固定住了
    search_kwargs={"k": 2, "filter": {"source": "policy/aftersale.md"}}
)

# 打印标题，与上面第一段的输出区分开
print("--- 带过滤的 retriever（同一个问题）---")
# 同一个问题，但检索范围被限死在售后政策文件里
for doc in scoped_retriever.invoke("过滤网什么时候要换？"):
    # 正确答案在 manual.pdf 里，被过滤挡在门外，所以这两条都不相关
    print(" ", doc.metadata["chunk_id"], doc.page_content)

实测输出：

  c-0000 滤芯到期时指示灯会闪红灯，此时需更换滤芯。
  a-0001 质量问题 15 日内可换货，需提供照片凭证。
--- 带过滤的 retriever（同一个问题）---
  a-0001 质量问题 15 日内可换货，需提供照片凭证。
  a-0000 签收 7 日内可无理由退货，商品需保持完好。

第一段与 §5.2 的 similarity_search 结果完全一致——as_retriever() 只是换了调用接口，检索逻辑没变。

第二段再次印证 §5.3 的硬规则：问的是滤芯，检索范围却限死在售后政策文件里，于是正确答案根本没机会出现，返回的两条都不相关。把过滤固化进 Retriever 很适合做权限隔离，但也意味着范围外的问题会得到看似正常、实则无关的答案。第 15 章做 RAG 时要特别小心：过滤太严时，模型会拿着无关片段硬答。

as_retriever() 的价值在于统一接口：

好处	说明
是 `Runnable`	能用管道连接符组进链里（第 7 章），支持 `invoke` / `batch` / `astream`
屏蔽实现差异	换 Chroma → FAISS，下游代码不动
可包装增强	第 15 章的重排、混合检索都是在 Retriever 层加工

`search_type`	含义
`"similarity"`（默认）	纯相似度
`"mmr"`	兼顾多样性（§7）
`"similarity_score_threshold"`	只返回超过分数阈值的，需配 `score_threshold`

第三个选项有个陷阱要提醒：它用的是 relevance_score（越大越像，0～1），不是 with_score 返回的距离。所以阈值要按 §5.2 里 relevance_scores 那一栏的数量级来设——照着距离的数量级设成 score_threshold=1.0，结果会一条都不返回。

最后一点：as_retriever() 返回的 Retriever 不带分数，invoke 的结果就是纯 Document 列表。想拿分数做置信度判断（§9.1 的 top-1/top-2 分差），要么继续用 similarity_search_with_score，要么到第 15 章自己包一层 Retriever。

9. 嵌入质量决定检索质量 #

本节要核实一件事：同一套代码、同一份资料，只换嵌入模型，检索结果会差多少。

§3.3 已经用相似度数字说明了本地英文小模型在中文上的问题（同主题 0.5967 vs 无关 0.5778，几乎分不开）。这里换成完整检索场景做对照，用四个真实提问检验 top-1 是否命中。

# numpy 负责向量点积、求模长与取最大值
import numpy as np

# §3.2 定义的工厂函数
from embeddings_layer import get_embeddings

# 五条 chunk 构成一个迷你知识库，覆盖售后与说明书两个主题
CHUNKS = [
    "签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。",
    "质量问题 15 日内可换货，需提供照片凭证并联系客服登记。",
    "一般订单 48 小时内发出，节假日顺延。",
    "滤芯到期时指示灯会闪红灯，此时需更换滤芯。",
    "开机后指示灯亮起，按下模式键可切换手动与自动档。",
]

# 问题 → 期望命中的 chunk 序号，用来判断 top-1 对不对
# 这就是一份最小的「检索评测集」，第 15 章会把它做得更正式
CASES = [
    ("滤芯多久换一次？", 3),
    ("多少天内可以退货？", 0),
    ("发货要多久？", 2),
    ("怎么切换档位？", 4),
]


# 和 §4 完全相同的余弦公式：点积除以两个模长之积
def cosine(a: np.ndarray, b: np.ndarray) -> float:
    """余弦相似度，直译 §2.3 的公式。"""
    # 分子是点积，分母是两个向量的模长相乘
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))


# 把评测逻辑封装成函数，这样换模型只需换一个参数
def evaluate(provider: str) -> None:
    """用同一批问题检验某个嵌入模型的 top-1 命中率。"""
    # 按供应商名拿到对应的嵌入模型，其余逻辑完全不变
    embeddings = get_embeddings(provider)
    # 把 5 条 chunk 转成 (5, 维度) 的矩阵，每一行是一条 chunk 的向量
    matrix = np.array(embeddings.embed_documents(CHUNKS))

    # 命中计数器
    hit = 0
    # 打印分隔标题，区分两个模型的结果
    print(f"\n===== {provider} =====")
    # 逐题评测
    for question, expected in CASES:
        # 问题走 embed_query
        query = np.array(embeddings.embed_query(question))
        # 逐条套用余弦公式，算出问题与 5 条 chunk 的相似度（同 §4）
        scores = np.array([cosine(query, row) for row in matrix])
        # argmax 取分数最高那条的下标，即 top-1
        top = int(np.argmax(scores))
        # 与期望下标比对，得到中文标记
        ok = "命中" if top == expected else "未命中"
        # 布尔值参与加法时 True 算 1，用来累计命中数
        hit += top == expected
        # 打印判定结果与问题
        print(f"[{ok}] {question}")
        # 打印 top-1 的分数和内容，方便看它错在哪
        print(f"       top1({scores[top]:.4f}) {CHUNKS[top]}")
    # 汇总命中率
    print(f"top-1 命中率：{hit}/{len(CASES)}")


# 先看本地英文小模型的表现
evaluate("local")
# 再看通义中文模型的表现
evaluate("dashscope")

实测输出（两个模型跑同一批题）：

===== local =====
[未命中] 滤芯多久换一次？
       top1(0.6394) 一般订单 48 小时内发出，节假日顺延。
[未命中] 多少天内可以退货？
       top1(0.7237) 一般订单 48 小时内发出，节假日顺延。
[命中] 发货要多久？
       top1(0.5696) 一般订单 48 小时内发出，节假日顺延。
[命中] 怎么切换档位？
       top1(0.6275) 开机后指示灯亮起，按下模式键可切换手动与自动档。
top-1 命中率：2/4

===== dashscope =====
[命中] 滤芯多久换一次？
       top1(0.7777) 滤芯到期时指示灯会闪红灯，此时需更换滤芯。
[未命中] 多少天内可以退货？
       top1(0.5869) 一般订单 48 小时内发出，节假日顺延。
[命中] 发货要多久？
       top1(0.6560) 一般订单 48 小时内发出，节假日顺延。
[命中] 怎么切换档位？
       top1(0.6154) 开机后指示灯亮起，按下模式键可切换手动与自动档。
top-1 命中率：3/4

先看本地模型的错法：「一般订单 48 小时内发出」这一条被三个不同问题都排到了第一——包括问滤芯和问退货。这是弱模型的典型症状：它没真正理解中文语义，只是产出了一个「谁都有点像」的向量。

诊断口诀：如果某一条 chunk 总被各种无关问题召回，先怀疑嵌入模型，而不是急着调 k 或改切分参数。

9.1. 一个失败案例值得单独解剖 #

通义拿到 3/4，而不是满分。剩下那道「多少天内可以退货？」没答对，这个失败比满分更有价值，因为它是真实项目里最常见的一类错误。把完整排名打出来看：

问：多少天内可以退货？
  1. 0.5869  一般订单 48 小时内发出，节假日顺延。
  2. 0.5731  签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。   ← 正确答案
  3. 0.5686  质量问题 15 日内可换货，需提供照片凭证并联系客服登记。
  4. 0.3506  滤芯到期时指示灯会闪红灯，此时需更换滤芯。
  5. 0.2647  开机后指示灯亮起，按下模式键可切换手动与自动档。

关键信息是前三名挤在 0.5686～0.5869 之间，差距不到 0.02。对比一下答对的那题：滤芯问题的 top-1 是 0.7777，第二名只有 0.4343。

所以这两种情况的性质完全不同：

	答对的题（滤芯）	答错的题（退货）
top-1 分数	0.7777	0.5869
top-2 分数	0.4343	0.5731
分差	0.3434	0.0138，几乎并列
含义	模型很确定	模型其实在猜

分差差了 25 倍。这个对比说明：光看 top-1 的绝对分（0.7777 vs 0.5869）也能感觉出差别，但分差是更灵敏的信号——它不受「这个模型整体打分偏高还是偏低」的影响。

实用技巧：把「top-1 与 top-2 的分差」当作置信度信号。 分差很小说明检索没有把握，这时候把结果直接喂给模型生成答案是危险的——第 15 章会用它来决定「要不要追问用户」或「要不要走人工兜底」。

为什么会挤在一起？因为退货、换货、发货是「同族干扰项」：三条都在讲「某个时限规定」，句式几乎一样，差别只在一个词。向量把整句话压成一个点，句子的整体结构会盖过那一个关键词。这是向量检索的结构性弱点，不是模型不够好。

这里顺便验证了两个最常见的猜测，结论可能和你想的不一样：

猜测	实测	结论
正确答案那条太长，被无关从句稀释了	把它删成「签收 7 日内可无理由退货。」，分数从 0.5731 微升到 0.5825，名次仍是第 2，没能反超	猜测基本不成立，别急着改 `chunk_size`
换个问法结果会一样	问「签收后多久能退货？」→ 正确答案升到第 1（0.5910）	措辞影响很大，因为「签收」是原文里的强锚点

第一行的实测细节值得展开：删掉「商品需保持完好，包装与配件齐全」这半句后，分数确实涨了 0.0094，但干扰项「一般订单 48 小时内发出」是 0.5868，正确答案 0.5825，仍然差 0.004 排在第二。也就是说缩短 chunk 的方向对了，但收益远远不够翻盘。

遇到检索不准，很多人第一反应是回去调 chunk_size。这里的实测证明那不是主要原因，改了只是白折腾。 真正有效的是下面三条路。

第二行同样有实用价值：用户的措辞会显著影响召回。同一个意图，问「多少天内可以退货」失败，问「签收后多久能退货」成功，差别只在后者用了原文里的「签收」。这解释了为什么第 15 章要做「查询改写」——先让模型把口语问题换几种说法再检索。

那这类问题怎么真正解决？三条路，都留到第 15 章：

办法	原理
`k` 取 3～5 而不是 1	正确答案就在第 2 名。召回 3 条一起给模型，它能自己挑对
混合检索（向量 + BM25 关键词）	「退货」这个词的字面匹配能把正确答案顶上来，补上向量的弱点
重排（rerank）	用交叉编码器对候选重新打分，它会逐词比对，能分开同族干扰项

第一条现在就能用：别指望 top-1 永远对，让 k 承担一点冗余，这是 RAG 系统最廉价的鲁棒性来源。

9.2. 排查优先级 #

第 15 章会讲更系统的检索评测。现在先记住这条优先级：

排查顺序	检查什么
1	嵌入模型的语言能力是否匹配你的资料（本节）
2	切分是否切坏了、一个 chunk 是否混讲了多件事（第 13 章 §8）
3	加载是否丢内容（第 12 章 §9）
4	检索策略：`k` 是否太小、是否需要混合检索或重排（第 15 章）
5	最后才是调 `lambda_mult` 这类细节参数

10. 实战：本地向量知识库 #

10.1. 设计 #

本章实战产出两个脚本，一个建库、一个检索，接在第 13 章后面：

chunks.jsonl（第 13 章产物）
        │
        ▼  index.py    建库
① 读 chunks.jsonl
② 用 chunk_id 作主键写入 Chroma
③ 支持增量：已存在的 chunk_id 默认跳过，不重复付嵌入费用
④ 打印入库统计
        │
        ▼
kb_store/（持久化目录）
        │
        ▼  search.py   检索
交互式提问 → 语义检索 → 打印命中片段与来源

为什么拆成两个脚本：建库慢且花钱，检索快且免费。分开之后可以反复调检索参数，不必重新嵌入。

这个分工也决定了两个脚本的性格：index.py 要保守（默认不重复付费、不轻易删数据），search.py 要顺手（能连续试问、能临时改 k 和过滤条件）。

10.2. index.py #

"""建库脚本：chunks.jsonl → kb_store/"""

# 让 list[Document] 这类新式类型注解在旧版 Python 上也能用
from __future__ import annotations

# 标准库：解析命令行参数（--rebuild）
import argparse
# 标准库：解析 jsonl 的每一行
import json
# 标准库：路径处理与存在性判断
from pathlib import Path

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma
# 把 jsonl 记录还原成 Document
from langchain_core.documents import Document

# §3.2 定义的工厂函数，建库与检索必须用同一个
from embeddings_layer import get_embeddings

# 第 13 章的产物
INPUT_FILE = Path("chunks.jsonl")
# 向量库持久化目录
# 故意和 §5 演示用的 chroma_store 分开：那边的示例会 rmtree 整个目录，
# 放一起的话跑一遍演示就会把正式知识库删掉
STORE_DIR = Path("kb_store")
# 集合名，检索脚本要用同一个，写错会静默打开空集合
COLLECTION = "company_kb"


# 第一步：把 jsonl 读回内存
def load_chunks(path: Path) -> list[Document]:
    """读回第 13 章切好的 chunk。"""
    # 累积结果的列表，显式标注类型便于编辑器提示
    docs: list[Document] = []
    # encoding 必须显式指定，否则中文 Windows 会用 GBK 读 UTF-8 文件而报错
    with path.open(encoding="utf-8") as f:
        # jsonl 是一行一个 JSON，逐行读即可，不用一次性加载整个文件
        for line in f:
            # 去掉行尾换行与多余空白
            line = line.strip()
            # 跳过空行，容忍文件末尾的空行
            if not line:
                continue
            # 把这一行解析成字典
            record = json.loads(line)
            # 还原成 Document：两个字段名与第 13 章写出时一致
            docs.append(
                Document(
                    page_content=record["page_content"],
                    metadata=record["metadata"],
                )
            )
    # 返回完整列表；几千条 chunk 放内存里毫无压力
    return docs


# 抽成函数是为了保证「建库」和「统计」用的是完全相同的连接参数
def open_store() -> Chroma:
    """打开（或创建）持久化的向量库。"""
    # 目录不存在时会自动创建，所以第一次运行不需要额外准备
    return Chroma(
        collection_name=COLLECTION,
        # 每次调用都会新建一个嵌入模型实例，开销很小（真正花钱的是 API 调用）
        embedding_function=get_embeddings(),
        # Path 要转成 str
        persist_directory=str(STORE_DIR),
    )


# 核心函数：* 之后的参数必须用关键字传，避免 index_chunks(docs, True) 这种难读的调用
def index_chunks(docs: list[Document], *, rebuild: bool = False) -> None:
    """把 chunk 写入向量库，默认跳过已存在的 chunk_id。"""
    # 打开（或创建）向量库
    store = open_store()

    # 库里已有的主键集合，用来做增量判断
    # 用 set 而不是 list：下面要频繁做「in」判断，set 是 O(1)
    existing = set(store.get()["ids"])
    # 打印已有条数，这是判断增量是否生效的第一个信号
    print(f"库中已有 {len(existing)} 条")

    # 只有显式传了 --rebuild 且库里确实有数据时才清空
    if rebuild and existing:
        # 全量重建：先清空再写，会重新付一遍嵌入费用
        store.delete(ids=list(existing))
        # 清空后把已存在集合也置空，否则下面会把所有 chunk 都当成已存在
        existing = set()
        # 明确提示这是一次全量重建，避免误操作后不知情
        print("已清空，准备全量重建")

    # 待写入的记录
    todo: list[Document] = []
    # 因已存在而跳过的条数
    skipped = 0
    # 逐条判断该不该写
    for doc in docs:
        # 用 get 而不是 [] 取值：缺字段时返回 None 而不是抛 KeyError
        chunk_id = doc.metadata.get("chunk_id")
        # 过滤掉缺少 chunk_id 的记录：没有主键就无法增量更新
        if not chunk_id:
            # 打印出来而不是静默丢弃，否则你永远不知道少了什么
            print(f"  跳过缺少 chunk_id 的记录：{doc.page_content[:20]}...")
            # 跳过这一条，继续处理下一条
            continue
        # 已经在库里的就跳过，这是「不重复付费」的关键一行
        if chunk_id in existing:
            # 计数用于最后汇报
            skipped += 1
            # 同样跳过
            continue
        # 剩下的才是真正要写的
        todo.append(doc)

    # 打印这次的工作量，方便判断增量是否生效
    print(f"待写入 {len(todo)} 条，跳过已存在 {skipped} 条")

    # 没有新内容就直接返回，一次 API 都不调
    if not todo:
        print("没有新内容，未调用嵌入 API")
        # 提前返回，一次 API 都不调，这行输出就是「没花钱」的凭据
        return

    # 分批写入：一次请求塞太多容易超限，也不利于失败重试
    # 注意通义内部还会再拆成每批 10 条（§3.4），这里的 50 只影响进度打印粒度
    batch_size = 50
    # range 的第三个参数是步长，用它把列表切成若干段
    for start in range(0, len(todo), batch_size):
        # 切片取出当前这一批；末尾不足 batch_size 时 Python 会自动截断
        batch = todo[start : start + batch_size]
        # 关键：ids 用 chunk_id，这样重复运行是覆盖而不是新增（§5.5）
        store.add_documents(batch, ids=[d.metadata["chunk_id"] for d in batch])
        # min 是为了让最后一批显示真实条数而不是超过总数
        print(f"  已写入 {min(start + batch_size, len(todo))}/{len(todo)}")


# 验收用的统计函数：它只读库、不调 API，所以可以随便重跑
def report() -> None:
    """打印入库结果统计。"""
    # 重新打开一次，确认数据确实落盘了（而不是只在内存里）
    store = open_store()
    # 取出全部记录
    data = store.get()
    # 总条数应与 chunks.jsonl 的行数一致，这是第一条验收标准
    print(f"\n向量库共 {len(data['ids'])} 条")

    # 按来源统计，确认每个文件都进库了
    by_source: dict[str, int] = {}
    # 遍历每条记录的 metadata
    for metadata in data["metadatas"]:
        # 缺 source 字段时归到 unknown，而不是报错
        source = metadata.get("source", "unknown")
        # 计数：get(key, 0) 是「没有就从 0 开始」的常用写法
        by_source[source] = by_source.get(source, 0) + 1
    # 打印小标题
    print("按来源统计：")
    # sorted 让输出顺序稳定，便于两次运行做对比
    for source, count in sorted(by_source.items()):
        print(f"  {source}: {count}")

    # 抽一条看看内容和 metadata 是否完好
    # 空库时跳过，避免索引越界
    if data["ids"]:
        print("\n示例记录：")
        # 主键应该是第 13 章的 chunk_id 格式：doc_id-序号
        print(" id:", data["ids"][0])
        # 截前 50 字，确认原文没有丢失或乱码
        print(" 内容:", data["documents"][0][:50])
        # 打印完整 metadata，确认第 12、13 章的字段都还在
        print(" metadata:", data["metadatas"][0])


# 只有直接运行这个文件才执行下面的逻辑，被 import 时不执行
if __name__ == "__main__":
    # 创建命令行参数解析器
    parser = argparse.ArgumentParser(description="把 chunks.jsonl 写入向量库")
    # action="store_true" 表示这是个开关：写了就是 True，不写就是 False
    # 加 --rebuild 才会清空重建，避免误操作导致重复付费
    parser.add_argument("--rebuild", action="store_true", help="清空后全量重建")
    # 解析实际传入的参数
    args = parser.parse_args()

    # 前置检查：输入文件不存在就给出可操作的提示，而不是让 open 抛底层异常
    if not INPUT_FILE.exists():
        # SystemExit 会以非 0 状态码退出，适合放进流水线
        raise SystemExit(f"找不到 {INPUT_FILE}，请先运行第 13 章的 split.py")

    # 读入全部 chunk
    chunks = load_chunks(INPUT_FILE)
    # 先报读到多少条，便于和下面的「待写入 / 跳过」对账
    print(f"读入 {len(chunks)} 个 chunk")

    # 写入向量库；rebuild 由命令行开关决定
    index_chunks(chunks, rebuild=args.rebuild)
    # 最后打印统计，作为验收依据
    report()

拿第 13 章产出的 chunks.jsonl（6 条）实跑，第一次建库：

读入 6 个 chunk
库中已有 0 条
待写入 6 条，跳过已存在 0 条
  已写入 6/6

向量库共 6 条
按来源统计：
  faq.txt: 1
  manual.pdf: 2
  policy/aftersale.md: 3

示例记录：
 id: a1b2c3d4e5f6-0000
 内容: 签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。质量问题 15 日内可换货，需提供照片凭
 metadata: {'source': 'policy/aftersale.md', 'char_len': 59, 'chunk_id': 'a1b2c3d4e5f6-0000',
            'chunk_index': 0, 'h1': '售后手册', 'doc_id': 'a1b2c3d4e5f6', 'start_index': 0,
            'h2': '退换货', 'file_name': 'aftersale.md',
            'loaded_at': '2026-08-12T14:00:00+00:00', 'file_type': 'md'}

紧接着原封不动再跑一次

读入 6 个 chunk
库中已有 6 条
待写入 0 条，跳过已存在 6 条
没有新内容，未调用嵌入 API

向量库共 6 条
按来源统计：
  faq.txt: 1
  manual.pdf: 2
  policy/aftersale.md: 3

示例记录：
 id: a1b2c3d4e5f6-0000
 内容: 签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。质量问题 15 日内可换货，需提供照片凭
 metadata: {'source': 'policy/aftersale.md', 'chunk_id': 'a1b2c3d4e5f6-0000', ...}

「未调用嵌入 API」意味着这次运行一分钱没花，而后面的 report() 照常执行（它只读库、不调 API）。这就是把 chunk_id 一路从第 13 章带过来的回报：脚本可以随便重跑，只有真正新增或变更的 chunk 才会付费。

顺便说一个容易忽略的细节：metadata 打印出来的键顺序每次运行都不一样。上面两次运行里 source 和 loaded_at 的先后就不同。这是 Chroma 内部存取字典时不保证顺序，不代表数据有问题——不要拿键顺序去做任何断言或比对。

report() 的两个输出都有实际用途：按来源统计能确认每个文件都进库了（对照第 12 章的加载报告，数量应该对得上）；打印一条完整 metadata 能确认第 12、13 章辛苦攒的字段——source、page、h2、chunk_id——一个都没在入库过程中丢掉。metadata 丢了不会报错，但下游的过滤和溯源会静默失效。

10.3. search.py #

"""检索脚本：对知识库提问，看命中哪些片段"""

# 让 str | None 这类注解在旧版 Python 上可用
from __future__ import annotations

# 标准库：解析命令行参数
import argparse

# Chroma 的 LangChain 封装
from langchain_chroma import Chroma

# §3.2 定义的工厂函数，必须与 index.py 用同一个模型
from embeddings_layer import get_embeddings

# 必须与 index.py 完全一致，否则会打开一个空集合（不报错，只是查不到东西）
STORE_DIR = "kb_store"
# 集合名同理，一个字符都不能差
COLLECTION = "company_kb"


# 与 index.py 里的同名函数保持一致，是「同一个模型、同一个库」的保证
def open_store() -> Chroma:
    """打开 index.py 建好的向量库。"""
    # 三个参数都要和 index.py 对齐：集合名、嵌入模型、目录
    return Chroma(
        collection_name=COLLECTION,
        embedding_function=get_embeddings(),
        persist_directory=STORE_DIR,
    )


# 溯源展示：把第 12、13 章攒下的 metadata 变成用户能核对的一行字
def format_source(metadata: dict) -> str:
    """把 metadata 拼成人能读的来源说明。"""
    # 先放文件名作为第一段；缺字段时用占位文字而不是报错
    parts = [metadata.get("source", "未知来源")]
    # PDF 的页码从 0 开始，展示时加 1（第 12 章 §5.1）
    if "page" in metadata:
        parts.append(f"第 {metadata['page'] + 1} 页")
    # Markdown 的标题层级（第 13 章 §6.1），txt 文件没有这个字段
    if "h2" in metadata:
        parts.append(metadata["h2"])
    # 用 > 连接成「文件 > 页码 > 章节」这种面包屑
    # str(p) 是因为 page 拼出来的可能不是字符串类型
    return " > ".join(str(p) for p in parts)


# 一次检索的完整流程：打开库 → 组装过滤条件 → 检索 → 格式化输出
def search(question: str, *, k: int, source: str | None) -> None:
    """检索并打印结果。"""
    # 每次调用都重新打开：交互模式下这样能反映出库的最新状态
    store = open_store()

    # 指定了 --source 就把检索范围限定在该文件内，否则传 None 表示不过滤
    filter_ = {"source": source} if source else None

    # 用带分数的版本，这样能顺便看出置信度（§9.1 的分差）
    results = store.similarity_search_with_score(question, k=k, filter=filter_)

    # 空结果单独处理，并给出可操作的排查方向（§5.3）
    if not results:
        print("没有召回任何内容。检查库是否为空，或过滤条件是否太严。")
        # 提前返回，跳过下面的打印逻辑
        return

    # 打印问题作为标题
    print(f"\n问题：{question}")
    # results 的元素是 (Document, 分数) 二元组，所以要用括号解包
    for rank, (doc, score) in enumerate(results, start=1):
        # Chroma 返回的是距离，越小越相似（§5.2）
        # 把方向直接写进输出里，避免自己或同事日后看错
        print(f"\n[{rank}] 距离 {score:.4f}（越小越相似）")
        # 来源信息，决定了用户能不能核对原文
        print(f"    来源：{format_source(doc.metadata)}")
        # 命中的正文片段
        print(f"    片段：{doc.page_content}")


# 直接运行时才执行，被 import 时不执行
if __name__ == "__main__":
    # 创建参数解析器
    parser = argparse.ArgumentParser(description="检索本地向量知识库")
    # nargs="?" 表示这个位置参数可有可无
    # 不传问题就进入交互模式，方便连续试问
    parser.add_argument("question", nargs="?", help="要查询的问题")
    # type=int 让 argparse 自动把字符串转成整数
    parser.add_argument("-k", type=int, default=3, help="返回条数")
    # 不传则为 None，表示不做过滤
    parser.add_argument("--source", help="只在指定来源文件里检索")
    # 解析命令行
    args = parser.parse_args()

    # 传了问题就查一次然后退出，适合脚本调用
    if args.question:
        search(args.question, k=args.k, source=args.source)
    # 没传问题就进入交互循环，适合人工连续试问
    else:
        print("进入交互模式，直接输入问题，回车查询；输入 q 退出。")
        # 无限循环，靠下面的 break 退出
        while True:
            # input 读一行；strip 去掉首尾空白
            text = input("\n问> ").strip()
            # 三种退出词都接受，用集合判断比连写三个 == 清晰
            if text in {"q", "quit", "exit"}:
                break
            # 直接回车（空字符串）时不查询，继续等下一次输入
            if text:
                search(text, k=args.k, source=args.source)

用法：

# 建库（第一次会调用嵌入 API）
python index.py

# 再跑一次：已存在的 chunk 会被跳过，不重复花钱
python index.py

# 确实需要全量重建时才加这个参数
python index.py --rebuild

# 单次查询
python search.py "滤芯多久换一次？"

# 返回 5 条，并限定只搜手册
python search.py "怎么退货" -k 5 --source manual.pdf

# 不带参数进入交互模式，连续试问
python search.py

实跑一下最关键的语义检索（问句里没有「滤芯」二字）：

问题：过滤网什么时候要换？

[1] 距离 0.7073（越小越相似）
    来源：manual.pdf > 第 12 页
    片段：开机后指示灯亮起，按下模式键可切换手动与自动档。选择自动档后，风速会根据空气质量自动调节。滤芯到期时指示灯会闪红灯，此时需更换滤芯。

[2] 距离 1.1063（越小越相似）
    来源：policy/aftersale.md > 退换货
    片段：签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。质量问题 15 日内可换货，需提供照片凭证并联系客服登记。

[3] 距离 1.1810（越小越相似）
    来源：manual.pdf > 第 12 页
    片段：更换完成后长按复位键三秒，计时器归零。

命中了，而且来源精确到 PDF 第 12 页——这就是第 12 章保留 page 字段的价值。

但这里有个数字值得细想一下：同一个问题，在 §4 和 §5.2 的距离是 0.4661，这里却是 0.7073。 区别在哪？§5.2 里那条 chunk 只讲滤芯一件事；而这里的 chunk 是第 13 章按长度切出来的，把「开机指示灯」「自动档风速」「滤芯更换」三件事装在了一起。无关内容拉着整句话的向量偏离了「滤芯」这个主题，相关度就掉了 0.24。

这是第 13 章「一个 chunk 只讲一件事」那条原则的量化证据。 切分质量不会让检索报错，只会让分数悄悄变差、排名悄悄变乱。

再试试过滤参数：

# python search.py "怎么退货" -k 2 --source policy/aftersale.md

问题：怎么退货

[1] 距离 0.7563（越小越相似）
    来源：policy/aftersale.md > 退换货
    片段：跨境订单以页面公示为准，退回运费由买家承担。

[2] 距离 0.8656（越小越相似）
    来源：policy/aftersale.md > 退换货
    片段：签收 7 日内可无理由退货，商品需保持完好，包装与配件齐全。

过滤生效了（两条都来自指定文件），但排名又一次出现了 §9.1 说的同族干扰：讲跨境运费的细则排在了主条款前面，两者只差 0.11。再次印证那条结论——k 不要取 1。这里 k=2 就把正确答案带上了，交给模型自己去挑。

最后验证一下空结果分支，这也是 §5.3 那个坑在本章实战里的体现：

# python search.py "怎么退货" --source nonexist.md

没有召回任何内容。检查库是否为空，或过滤条件是否太严。

传了一个库里不存在的文件名，没有任何报错，只是一条都召回不到。脚本里那句提示是有意为之——如果只是静默打印空列表，你很容易去怀疑嵌入模型，而真正原因只是文件名写错了一个字。

10.4. 验收清单 #

建库成功：index.py 打印的条数等于 chunks.jsonl 的行数
增量生效：再次运行 index.py 显示「没有新内容，未调用嵌入 API」
持久化生效：关掉终端重开，search.py 无需重新建库即可检索
语义命中：用不含原文关键词的问法提问（如「过滤网什么时候换」），仍能命中滤芯那一条
来源可溯：每条结果都能打印出文件名，PDF 还能显示页码
过滤生效：--source 限定后，结果只来自该文件
更新可控：改一个文件后重跑第 13 章切分，只有该文件的 chunk 被重写
分数可解释：能说清打印出来的距离是什么方向，以及 top-1 与 top-2 的分差意味着什么（§9.1）

第 4 条是本章真正的验收点——它证明系统在按语义检索，而不是在做关键词匹配。第 8 条则决定你在第 15 章能不能判断「这次检索到底靠不靠谱」。

11. 实用约定与坑 #

11.1. 该守的约定 #

约定	说明
嵌入模型收口到 `get_embeddings()`	换供应商只改一处（§3.2）
建库与检索必须用同一个嵌入模型	否则检索结果无意义，且可能不报错
显式写 `model=`	不写会用老版 `text-embedding-v1`（1536 维），与 v4 不兼容（§3.4）
用 `chunk_id` 作向量库主键	才能按文件增量更新（§5.5）
建库脚本要支持增量	嵌入 API 按 token 计费，别重复付钱（§10.2）
分批写入	一次几千条容易超限或超时
API Key 放 `.env`	不要写进代码、不要提交 Git
换模型 = 全量重建	维度不同直接报错，维度相同则静默失效，更危险
记录建库用的模型名	建议存进集合名或旁边的元信息文件
演示库与正式库分目录	示例代码里的 `rmtree` 不会误删你的知识库（§10.2）
`k` 不要取 1	正确答案排第 2 是常态，留点冗余交给模型挑（§9.1）
把分数方向写进日志输出	像 `search.py` 那样打印「越小越相似」，省掉日后的反复确认

11.2. 静默失败：最危险的一类 #

这些问题不报错、不警告，只是结果悄悄变坏。它们占了排查时间的绝大部分。

现象	原因	处理
检索结果完全不相关	建库与检索用了不同嵌入模型（维度恰好相同时不会报错）	确认 `get_embeddings()` 两边返回一致；必要时重建（§11.1）
过滤后返回 0 条	filter 键名拼错、值不存在、或值类型不符（`page` 是 `11` 而你传 `"11"`）	用 `store.get(where=...)` 数一下，并打印真实 metadata 核对（§5.3）
重开进程后库是空的	没传 `persist_directory`，或 `collection_name` 写错——写错等于新建空集合	两者都要与建库时字符一致（§5.4）
同一条内容重复出现多次	没传 `ids`，每次写入都新建记录，库里躺着多份矛盾内容	用 `chunk_id` 作 `ids`，重写即覆盖（§5.5）
某条 chunk 被各种无关问题召回	嵌入模型语言能力不足，产出了「谁都有点像」的向量	换中文能力强的模型（§9）
`get()` 拿到一堆 `None`	`include` 是白名单，没列进去的字段会变成 `None`	把需要的字段全部写进 `include`（§5.1）
返回条数少于 `k`	过滤是硬约束，先把候选集缩小了；不是 bug	确认该条件下本来有多少条（§5.3）
明明维度是 1024，却拿到 1536 维	忘了写 `model=`，用成默认的 `text-embedding-v1`	工厂函数里显式指定模型名（§3.4）

11.3. 会报错的：反而好处理 #

报错	原因	处理
`dimension mismatch`	换了维度不同的模型	全量重建，或新建一个集合
`Tried to add ids that already exist`	FAISS 不支持同 id 覆盖	先 `delete` 再 `add`；或换 Chroma。捕获这个异常后必须丢弃整个 store（§6.2）
FAISS 检索抛 `KeyError: np.int64(N)`	之前有一次失败的 `add_documents`，向量写进了索引但 id 没登记，`ntotal` 与映射表错位	无法修复，重建 store（§6.2）
pickle 相关的 `ValueError`	FAISS 加载缺 `allow_dangerous_deserialization=True`	加上；且只加载自己生成的文件（§6.1）
`extra_forbidden`（传 `dimensions`）	`DashScopeEmbeddings` 不支持该参数	走 OpenAI 兼容模式，见 §12 第 3 题
`Did not find dashscope_api_key`	没配 `.env` 或没 `load_dotenv()`	检查 `.env` 位置与内容（§3.2）
嵌入 API 限流	批量太大或并发太高	减小 `batch_size`（`DashScopeEmbeddings` 自带退避重试，默认 5 次）
`InvalidArgumentError: Validation error: name`	集合名不合规：太短、含中文或空格、首尾不是字母数字	改成 3 个以上的 `[a-zA-Z0-9._-]` 字符（§5.1）
`PermissionError: [WinError 32]`	Windows 上 `rmtree` 时还有 Chroma 对象占着文件	用 `delete_collection()` 清空集合，别删目录（§5.1）

11.4. 概念上容易搞反的 #

容易搞反	事实
分数越大越像	Chroma / FAISS 默认返回距离，越小越像；`relevance_scores` 才是越大越像（§5.2）
0.85 比 0.55 更像	跨模型不可比。要看「相关」与「无关」的区分度（§3.3）
MMR 是更好的检索	是用相关性换覆盖面的旋钮，小库上开它会引入噪声（§7）
FAISS 默认用近似索引	`from_documents` 建的是 `IndexFlatL2`，暴力精确检索（§6.1）
检索不准就调 `chunk_size`	实测缩短正确答案后名次没变，先查模型再查切分（§9.1）
`embed_query` 和 `embed_documents` 结果不同	接口上区分，但通义 v4 返回逐位相同的向量（§3.1）

口诀：

嵌入模型定上限，向量库定运维。建库与检索必须同一个模型；主键必须是 chunk_id；看到分数先问方向。

12. 练习 #

12.1. 机制验证类 #

分数方向：拿库里某条原文一字不改地去查，确认距离是 0.0000；再把 §5.2 的距离用 $\cos\theta = 1 - d/2$ 换算成余弦，看是否与 §4 手算的一致。
过滤的三种错法：照 §5.3 分别制造「键名拼错」「值不存在」「值类型不符」，确认三者都返回 0 条且都不报错。然后用 store.get(where=...) 把原因定位出来。
同 id 重写：在 Chroma 上用相同 ids 写两次，确认条数不变；在 FAISS 上做同样的事，确认抛 ValueError，并打印 index.ntotal 与 index_to_docstore_id 看它们怎么错位的（§6.2）。
lambda_mult 与 fetch_k 的联动：照 §7 把 lambda_mult 从 0 扫到 1，找出你自己库上的跳变点，并验证 lambda_mult=1.0 与普通相似度检索结果一致；再固定 lambda_mult 只改 fetch_k，观察候选池大小怎么影响结果。
忘写 model=：直接 DashScopeEmbeddings() 不传模型名，确认拿到的是 1536 维；再想一想如果两个脚本一个写了一个没写会发生什么。
include 白名单：用 store.get(include=["embeddings"]) 取数据，确认 documents 变成了 None（§5.1）。

12.2. 能力构建类 #

语义 vs 关键词：用三种不含原词的问法（如「过滤网」「换网子」「多久保养」）查滤芯那一条，记录是否命中。
过滤实战：给 search.py 加 --section 参数，按第 13 章的 h2 过滤。注意处理「拼错章节名返回 0 条」的提示。
维度实验：把通义模型的维度改成 256，重建一个新集合（不要覆盖原集合），对比检索效果与目录体积。写法见下面的补充说明。
换库：把 index.py 改成 FAISS 版本。注意 FAISS 不支持同 id 覆盖，增量逻辑要改成「先 delete 再 add」。
增量更新：修改一个源文件，跑通「重切分 → 删旧 chunk → 写新 chunk」全流程，确认其他文件的向量没被重算（看 index.py 打印的「跳过已存在」条数）。
置信度信号：给 search.py 加一个判断——当 top-1 与 top-2 的距离差小于某个阈值时，打印「本次检索置信度低」。这个想法出自本章 §9.1，也是拒答机制的雏形——第 15 章 §6 会用实测数据说明它为什么不够可靠。
（扩展）模型对照：用 §9 的评测脚本，给自己的资料写 10 个真实问题，比较两个嵌入模型的 top-1 命中率，并记录每题 top-1 与 top-2 的分差。

13. 本章小结 #

这一章第一次让「按意思找」跑起来。要记住的是：向量库负责运维（增删改查、过滤、持久化），嵌入模型负责效果上限——两件事要分开评估。

嵌入模型把文字映射成向量，让「语义相近」变成「数字接近」；向量不可读、不能还原原文，所以原文要另存。
余弦相似度比较方向、越大越像；L2 距离比较远近、越小越像。Chroma / FAISS 默认给的是距离，向量归一化时两者可换算：$\cos\theta = 1 - d/2$（§5.2 用实测数字验证过：1 - 0.4661/2 = 0.7670，与手算一字不差）。不想心算的话，可以把集合建成 cosine 空间，relevance_scores 直接就是余弦。
相似度的绝对值不能跨模型比较，要看「相关」与「无关」之间的区分度；同一次检索内，top-1 与 top-2 的分差是可用的置信度信号——实测答对的题分差 0.3434，答错的题只有 0.0138（§3.3、§9.1）。
embed_documents 建库、embed_query 检索。LangChain 确实分别传了 text_type，但通义 v4 对两者返回相同向量；守这个规矩是为了将来换模型时不出问题（§3.1）。
嵌入模型收口到 get_embeddings()：换供应商只改一处，因为换模型意味着全量重建。别忘了显式写 model=，默认值是 1536 维的老模型。
Chroma 是主线：persist_directory 自动落盘，元数据过滤有操作符语法，同 chunk_id 重写即覆盖。FAISS 的 save_local 要手动调、默认索引是暴力精确的 IndexFlatL2、同 id 重写不仅抛 ValueError 还会让整个 store 报废（必须先删再加）。实测两者算的是同一个式子，换库不改变检索效果，换模型才会。
chunk_id 作主键是增量更新的前提：同 id 重写即覆盖，改一个文件只重算一个文件。§10 实测第二次运行「未调用嵌入 API」，一分钱没花。
元数据过滤是第 12 章经营 metadata 的回报：限定文件、章节、部门权限都靠它。它是硬约束——范围外的正确答案再相关也进不来；而且写错条件只会静默返回 0 条，键名、值、类型三处都要核对。
MMR 用相关性换覆盖面，不是「更好的检索」。lambda_mult 的变化是跳变而非渐变，调它要用 0.1～0.2 的步长；as_retriever() 把向量库统一成 Runnable 交给下游，但它不返回分数。
嵌入模型的语言能力决定检索质量：排查检索问题时，先看模型，再看切分，最后才调参数。实测「缩短 chunk」并没能救回一个排第 2 的正确答案。
向量检索对「同族干扰项」天生不擅长（退货 / 换货 / 发货），这不是模型不够好，而是结构性弱点——解法是 k 取 3～5、混合检索与重排（第 15 章）。用户的措辞也会显著影响召回，这是第 15 章要做查询改写的原因。
本章产出：本地向量知识库 + index.py（增量建库）+ search.py（检索）。

下一章：Retrievers 与 RAG 实战——把检索结果交给模型生成带引用的回答，并做成 Agent 能调用的检索工具，最终完成企业知识库问答系统。

导航菜单

1. 本章目标 #

1.1. 先看几条反直觉的结论 #

2. 从关键词匹配到按语义找 #

2.1. 关键词检索为什么不够 #

2.2. 向量是什么 #

2.3. 相似度怎么算 #

3. Embeddings 接口 #

3.1. 两个方法 #

3.2. 供应商选型 #

3.3. 第一次调用 #

4. 手算一遍相似度 #

5. 向量库：Chroma #

5.1. 建库 #

5.2. 检索与分数 #

5.3. 元数据过滤 #

5.4. 持久化 #

5.5. 增删改：chunk_id 的价值 #

6. 向量库：FAISS #

6.1. 建库与存取 #

6.2. Chroma 还是 FAISS #

7. MMR：结果太像怎么办 #

8. as_retriever：接给下游 #

9. 嵌入质量决定检索质量 #

9.1. 一个失败案例值得单独解剖 #

9.2. 排查优先级 #

10. 实战：本地向量知识库 #

10.1. 设计 #

10.2. index.py #

10.3. search.py #

10.4. 验收清单 #

11. 实用约定与坑 #

11.1. 该守的约定 #

11.2. 静默失败：最危险的一类 #

11.3. 会报错的：反而好处理 #

11.4. 概念上容易搞反的 #

12. 练习 #

12.1. 机制验证类 #

12.2. 能力构建类 #

13. 本章小结 #