《AI Agent 开发入门》9.1 嵌入与向量存储：本地可离线方案

9.1 嵌入与向量存储：本地可离线方案

第 8 章的记忆检索是靠关键词命中的：把查询和记忆都分词，数一数重合的词。这在用户原话里出现记忆里的词时很好用，可一旦用户换了说法就立刻失灵——“工具报错"匹配不到写着"调用失败"的那条记忆。要把"说法不同、意思相同"拉近，就需要向量。

本节把 minagent 推进到 minagent/embeddings.py：定义 Embedder 协议与一个确定性的哈希嵌入，再配一个 numpy 支撑的极简 VectorStore，让记忆检索从"数重合词"升级为"算向量相似度”。

9.1.1 为什么记忆需要向量

关键词检索把每条记忆看成"一堆词的集合"，两条记忆要么共享某个词、要么毫无关系，中间没有"有点像"这个刻度。向量检索换了个表示：把一段文本映射成一个固定长度的浮点数组，语义越接近的文本，数组越接近。这个映射函数就叫嵌入（embedding）。

两种检索的差别可以列成一张表，后面几节会反复对照：

有了向量，检索就变成几何问题：把查询也变成向量，然后找记忆向量里跟它最接近的那些。度量"接近"最常用的工具是余弦相似度——两个向量的夹角余弦，值域 [-1, 1]，越接近 1 越相似：

cos(a, b) = (a · b) / (||a|| * ||b||)

如果我们在入库前就把每个向量除以其 L2 范数（单位化），那么余弦相似度就退化成一次点积 a · b。这就是为什么向量库几乎总在存储前做归一化：把昂贵的除法换成一次点积，检索时快得多。

9.1.2 Embedder 协议与哈希嵌入

先把接口定下来。无论背后是供应商 API 还是本地模型，对外都只需要一个方法：

from typing import Protocol

class Embedder(Protocol):
    dim: int
    def embed(self, text: str) -> list[float]: ...

dim 是向量维度，embed() 把文本变成 dim 个浮点数。接口这么窄，是为了让 9.2、9.3 以及后面第 10 章的检索全都只依赖这一件事。

真实生产里这个接口后面接的是供应商 embedding API 或本地小模型。但本机既没有 torch，也没有 HuggingFace 缓存，更不允许联网下载模型，所以这一章必须给出一个完全确定性、零依赖、可复现的替代方案：哈希嵌入（hashing trick）。

它的思路很朴素——把文本切成词元（token），每个词元用确定性哈希映射到 [0, dim) 里的一个桶，桶里累加计数；为了抵消哈希碰撞带来的偏差，再用哈希的最高位决定加 +1 还是 -1（这叫 signed hashing trick）。最后做 L2 归一化：

from __future__ import annotations
import hashlib, re
from dataclasses import dataclass
import numpy as np

_CJK = re.compile(r"[一-鿿]")
_WORD = re.compile(r"[a-zA-Z0-9_]+")

def tokenize(text: str) -> list[str]:
    tokens = _WORD.findall(text.lower())
    cjk = _CJK.findall(text)
    tokens += cjk
    tokens += [cjk[i] + cjk[i + 1] for i in range(len(cjk) - 1)]
    return tokens

@dataclass
class HashingEmbedder:
    dim: int = 256

    @staticmethod
    def _hash(token: str) -> int:
        return int.from_bytes(hashlib.md5(token.encode("utf-8")).digest()[:8], "little")

    def embed(self, text: str) -> list[float]:
        vec = np.zeros(self.dim, dtype=np.float64)
        for tok in tokenize(text):
            h = self._hash(tok)
            vec[h % self.dim] += 1.0 if (h >> 63) & 1 == 0 else -1.0
        norm = float(np.linalg.norm(vec))
        if norm > 0.0:
            vec /= norm
        return vec.tolist()

中文分词这里做了简化：因为不引入 jieba 之类的分词器，我们把每个汉字单独当一个词元，再补上相邻汉字组成的二元组（bigram）。单字保证"上下文"和"上"这种局部重合能被捕捉到，bigram 则让"上下文"“压缩"这类连续词有更强的信号。英文和数字直接按 [a-zA-Z0-9_]+ 切。

用 hashlib.md5 而不是内建 hash() 是有意的：Python 的 hash() 对字符串默认加了随机盐（PYTHONHASHSEED），进程重启后同一个词会落到不同桶，结果不可复现。md5 在任何机器、任何进程里对同一输入永远给出同一串字节，这正是"确定性"的保证。

维度 dim 取 256 也是个折中：桶太少，不同词元撞车严重、噪声压过信号；桶太多，向量变稀疏、点积里绝大多数是零，既浪费内存又不提升效果。256 在这 8 条小记忆上够用，真实模型动辄 768 甚至 1536，是因为它们要区分的是几十万条文档、上万个概念。

9.1.3 极简 VectorStore：numpy 矩阵 + 余弦

有了嵌入，还需要一个存向量、算相似度的地方。VectorStore 把全部记忆向量摞成一个 (N, dim) 的 numpy 矩阵，检索时用一次矩阵乘法把所有相似度一起算出来：

from dataclasses import dataclass, field
import numpy as np

@dataclass
class VectorStore:
    dim: int
    _ids: list[str] = field(default_factory=list)
    _texts: list[str] = field(default_factory=list)
    _mat: np.ndarray = field(default_factory=lambda: np.zeros((0, 0)))

    def __post_init__(self) -> None:
        self._mat = np.zeros((0, self.dim), dtype=np.float64)

    def add(self, id: str, text: str, vector: list[float]) -> None:
        self._mat = np.vstack([self._mat, np.asarray(vector).reshape(1, -1)])
        self._ids.append(id)
        self._texts.append(text)

    def search(self, query_vector: list[float], k: int = 3):
        if not self._ids:
            return []
        q = np.asarray(query_vector, dtype=np.float64)
        scores = self._mat @ q
        order = np.argsort(-scores, kind="stable")[:k]
        return [(self._ids[i], self._texts[i], float(scores[i])) for i in order]

三个细节值得说：

这个 VectorStore 是教学版：每次 add 都 vstack 重建矩阵，是 O(N) 拷贝。真实场景要么用现成向量库（FAISS、Chroma、pgvector），要么预分配大矩阵。但接口就这三个方法，换实现不影响上层。

还有一点要注意：这个实现是暴力检索——每次查询都拿查询向量跟全部 N 条算点积，复杂度 O(N·dim)。记忆只有几十上百条时这完全够用，可一旦到百万级，就必须上近似最近邻（ANN）索引，比如 HNSW 或 IVF。本卷的记忆量级远没到这个坎，所以暴力检索是最诚实也最好懂的选择。

9.1.4 真跑：索引 8 条记忆

准备 8 条跨主题的记忆——有环境、有依赖、有偏好、有项目约定，让检索有得选。注意这里每条记忆的文本是"标签：正文”，标签会参与嵌入，等于给记忆加了一个很短的标题：

MEMORIES = [
    ("m1", "运行环境：macOS 上开发，Python 3.14，虚拟环境在 /tmp/agentbook/venv。"),
    ("m2", "依赖安装：pip 装包必须用中科大镜像源，官方源在本机不稳定。"),
    ("m3", "语言偏好：用户希望简体中文回复，代码注释保持英文。"),
    ("m4", "项目简介：minagent 是一个从零手写的 Agent 运行时库。"),
    ("m5", "工具错误处理：调用失败时不要抛异常，返回 error JSON。"),
    ("m6", "上下文压缩：模型窗口有限，超过 token 预算要做裁剪与摘要。"),
    ("m7", "开会日程：每天下午三点开会，这段时间别安排长任务。"),
    ("m8", "假模型：FakeModel 用脚本列表驱动，脚本耗尽就抛 RuntimeError。"),
]

emb = HashingEmbedder(dim=256)
store = VectorStore(dim=emb.dim)
for mid, text in MEMORIES:
    store.add(mid, text, emb.embed(text))

先确认索引规模，再拿两个查询试召回：

print("dim =", emb.dim, "| indexed =", len(store._ids))
for q in ["minagent 里工具报错了该怎么返回？", "安装 Python 包用哪个镜像源？"]:
    print("\nquery:", q)
    for mid, text, score in store.search(emb.embed(q), k=3):
        print(f"  {mid}  {score:+.4f}  {text.split('：')[0]}")

实测输出（Python 3.14.6，numpy 2.5.3）：

dim = 256 | indexed = 8

query: minagent 里工具报错了该怎么返回？
  m5  +0.2349  工具错误处理
  m2  +0.1637  依赖安装
  m6  +0.1155  上下文压缩

query: 安装 Python 包用哪个镜像源？
  m2  +0.3465  依赖安装
  m3  +0.1375  语言偏好
  m8  +0.0755  假模型

两个查询都把最相关的记忆顶到了第 1：第一个命中 m5（工具报错→工具错误处理），第二个命中 m2（镜像源→依赖安装）。但要注意后面的名次：查询里出现了 “minagent”，可 m4（项目简介）并没有进前三；第二个查询里的 “Python” 也没把 m1（运行环境）拉上来。分数还都挺低——+0.23、+0.34，说明这就是"有点像"，远没到"就是它"。

值得强调的一点是可复现：因为用的是 md5 而不是内建 hash()，同一段代码在任何机器上重跑，上面的分数会一位不差。这对教学和测试至关重要——第 16 章的确定性测试会大量依赖"同样的输入给出同样的检索结果"，而随机盐哈希会把这个前提直接毁掉。

9.1.5 哈希嵌入的局限

哈希嵌入能用，但它是词袋模型，没有任何语义理解。它只会数"桶撞没撞上"，所以有两个致命短板。

第一，同义词召回不到。换个说法，只要字面不重合，相似度立刻掉到噪声水平。拿一个故意用同义词的查询试：

for mid, text, score in store.search(emb.embed("对话太长怎么办"), k=3):
    print(f"{mid}  {score:+.4f}  {text.split('：')[0]}")

实测输出：

m7  +0.1214  开会日程
m4  +0.0483  项目简介
m5  +0.0456  工具错误处理

正确答案是 m6（上下文压缩：窗口有限要做裁剪与摘要），可查询里的"对话"“太长"跟 m6 里的"窗口"“裁剪"“摘要"一个字都不重，于是哈希嵌入完全抓瞎——m6 的相似度是 0.0，在 8 条里排到第 7，反倒是最不相关的 m7（开会日程）被顶到了第 1。这正是"无语义"的代价：它认的是字面，不是意思。

第二，哈希碰撞带来噪声。256 个桶要装下几千个词元，不同词撞进同一桶是常态。signed hashing 只把偏差的期望拉回零，并不能消除方差，于是偶尔会有不相关的记忆拿到一个虚高的分数。9.2 节的向量路线里就会看到这种假阳性。

所以本章的哈希嵌入定位很明确：它是教学用的占位嵌入。它让整条"嵌入 → 索引 → 检索"的流水线在零依赖下跑通、可复现，但它不该出现在生产里。真要上线，把 HashingEmbedder 换成供应商 embedding API，或本地跑一个 bge-small 这类小模型，embed(text) -> list[float] 的接口一个字都不用改。

9.1.6 把嵌入层做成可替换的插座

Embedder 协议的价值，就是让"用哪种嵌入"变成一个可以随时拔插的决定，而检索代码纹丝不动。三种典型选择各有取舍：

关键在于：上层代码永远只认 embed(text) -> list[float]。换成 API 版时，VectorStore 不用改、search() 不用改，甚至连 9.2 的混合检索也不用改——只要维度对得上。这也是为什么我们不把"分词"“归一化"泄漏到检索层：那些都是嵌入实现的内部细节。

一个常被忽略的坑是维度必须匹配。VectorStore 在 __post_init__ 里按 dim 建了 (0, dim) 的矩阵，如果换了个维度不同的嵌入器却没同步 dim，vstack 会直接抛形状错误。所以换嵌入器时，emb.dim 和 VectorStore(dim=...) 必须一起改——这正是把二者绑在同一个 emb 对象上的好处。

这个"插座"思路会贯穿整卷：模型（第 2 章）、工具（第 4 章）、嵌入（本节）、向量库（本节）都被收窄成一个协议，真实实现与教学实现互换时，上层逻辑一行不改。第 16 章的确定性测试之所以能跑，靠的正是每一层都有这么一个可替换的窄接口。

真实模型下会怎样：真实嵌入模型（text-embedding-3-small、bge-small-zh 等）是在海量文本上训出来的，把"语义近"的文本映射到相邻向量，所以"压缩对话历史"会真的靠近"上下文压缩”。差异不只是质量，还有维度（常见 384、768、1536，不是 256）和成本（每次调用要钱、要网络）。此外真实模型的相似度分布跟哈希嵌入完全不同：同一句话的自我相似度接近 1，语义相近的常在 0.7–0.9，而不相关的可能也有 0.2–0.4 的底噪，所以"阈值该设多少"必须按具体模型实测，不能照搬。换句话说，换掉嵌入器之后，9.2 里所有"多少分算命中"的经验值都要重新标定。

阅读导航：上一节：8.3 记忆写入策略 · 下一节：9.2 混合检索：关键词 + 向量 + 重排 。