加向量数据库前，先用 NumPy 搞定 Embedding

向量数据库是解决真实问题的真实答案，但大多数项目在实际需要它之前大约两个数量级的时候就引入了它。在大约十万向量以下，NumPy 数组和矩阵乘法更简单、更精确、速度也足够——而解释这一点的算术只需一屏就能写完。

先说算术

两个量决定了暴力搜索是否可行：向量占用的内存，以及一次查询需要多少算术运算。两者都是可以精确计算的。以 100,000 个 1,536 维的向量为例，这是常见的 embedding 维度。

MEMORY
  values          = 100,000 x 1,536              = 153,600,000 floats
  float32         = 153,600,000 x 4 bytes        = 614,400,000 B  ~ 586 MiB
  float16         = 153,600,000 x 2 bytes        = 307,200,000 B  ~ 293 MiB
  int8            = 153,600,000 x 1 byte         = 153,600,000 B  ~ 147 MiB

ARITHMETIC PER QUERY (one dot product against every vector)
  multiply-adds   = 100,000 x 1,536              = 153,600,000 MACs
  FLOPs           = 2 x 153,600,000              = 307,200,000    ~ 0.31 GFLOP

MEMORY TRAFFIC PER QUERY
  every stored value is read exactly once        = 586 MiB (float32)

第三块才是预测实际耗时的关键，因为单次点积遍历是内存受限而非计算受限：0.31 GFLOP 对任何现代 CPU 来说都是舍入误差，而读取 614 MB 则不是。如果你的机器内存带宽是 B GB/s，一次查询大约需要 0.614 / B 秒（不计开销）。这就是全部模型，而接下来诚实的做法是在你自己的机器上测量 B，而不是相信别人给的数据。

同样的算术会得出三个结论。存储 float16 将流量减半，因此大致将查询时间减半。成本与向量数量呈线性关系，因此 1,000,000 个向量是十倍的时间和 5.7 GiB 内存——这就是暴力搜索开始变得不合理的地方。批量查询几乎免费：一百次查询作为一次矩阵乘法执行，只读取一次语料库而不是一百次，这是这里可用的最大优化。

Embedding 维度因模型而异差异很大——384、768、1024、1536 和 3072都很常见——内存随你选择的数字线性增长。将你自己的维度代入上面的计算，而不是复用总计。一些模型也支持截断向量；参见 Matryoshka embeddings。

一个 .npy 文件存矩阵，一个 JSONL 文件存元数据，以行索引作为连接键。这不花哨，但能挺过重启、版本控制，以及交给别人。

# build_index.py
import json

import numpy as np

def build(records: list[dict], embed: callable, path: str) -> None:
    """records: [{"id": ..., "text": ...}]. embed: list[str] -> list[list[float]]"""
    vectors: list[list[float]] = []
    with open(f"{path}.jsonl", "w", encoding="utf-8") as meta:
        for start in range(0, len(records), 256):        # batch the API calls
            batch = records[start:start + 256]
            vectors.extend(embed([r["text"] for r in batch]))
            for record in batch:
                meta.write(json.dumps(record, ensure_ascii=False) + "\n")

    matrix = np.asarray(vectors, dtype=np.float32)
    matrix /= np.linalg.norm(matrix, axis=1, keepdims=True)   # normalise ONCE
    np.save(f"{path}.npy", matrix)
    print(f"{matrix.shape[0]} vectors x {matrix.shape[1]} dims,"
          f" {matrix.nbytes / 2**20:.1f} MiB")

在构建时做归一化是让查询变廉价的技巧。一旦每一行都是单位长度，余弦相似度就是点积——无需除法，无需每次查询重新计算范式。一次性为 100,000 行做一次，而不是每次查询做 100,000 次。

Embedding 调用本身与这个集群里其他所有东西的 HTTP 形式一样：

def embed(texts: list[str]) -> list[list[float]]:
    response = client.post("/embeddings",
                           json={"model": EMBED_MODEL, "input": texts})
    response.raise_for_status()
    data = response.json()["data"]
    data.sort(key=lambda item: item["index"])   # do not assume input order
    return [item["embedding"] for item in data]

排序不是疑神疑鬼——响应携带 index 字段正是由于顺序无法保证，而错位的批次会产生一个索引，其中每个答案都微妙地错误，但没有任何报错。

搜索，只需四行

# search.py
import json

import numpy as np

class Index:
    def __init__(self, path: str):
        self.matrix = np.load(f"{path}.npy")             # (n, d), unit rows
        with open(f"{path}.jsonl", encoding="utf-8") as fh:
            self.meta = [json.loads(line) for line in fh]
        assert self.matrix.shape[0] == len(self.meta), "index and metadata differ"

    def search(self, query_vector: list[float], k: int = 5) -> list[dict]:
        q = np.asarray(query_vector, dtype=np.float32)
        q /= np.linalg.norm(q)
        scores = self.matrix @ q                          # (n,) cosine similarities
        top = np.argpartition(-scores, min(k, len(scores) - 1))[:k]
        top = top[np.argsort(-scores[top])]               # order the k, not the n
        return [{"score": float(scores[i]), **self.meta[i]} for i in top]

    def search_many(self, query_vectors: np.ndarray, k: int = 5) -> np.ndarray:
        """(m, d) queries -> (m, k) row indices. Reads the corpus once."""
        q = query_vectors / np.linalg.norm(query_vectors, axis=1, keepdims=True)
        scores = q @ self.matrix.T                        # (m, n)
        top = np.argpartition(-scores, k, axis=1)[:, :k]
        ordered = np.take_along_axis(
            top, np.argsort(-np.take_along_axis(scores, top, axis=1), axis=1), axis=1
        )
        return ordered

self.matrix @ q 就是整个搜索。它是精确的——不像每个近似索引那样有召回率权衡——而且它会分派到你平台的 BLAS，那是高度优化的 C 代码，不是 Python。

init 第二行的 assert 节省的时间比这个类其余部分加起来都多。文件备份索引最常见的故障是矩阵重建了但元数据没跟着重建，这不会报错——它返回的是关于错误文档的自信结果。

在你自己的机器上计时

上面的算术给出了工作量的上限；但它无法告诉你具体秒数，因为那取决于你的内存带宽、你的 BLAS 构建，以及数组是否能放入缓存。运行这个脚本，你就能得到你的硬件的数值，那是唯一值得拥有的数字。

# bench.py
import time

import numpy as np

RUNS = 20

def bench(n: int, d: int, dtype=np.float32) -> None:
    rng = np.random.default_rng(0)
    matrix = rng.standard_normal((n, d)).astype(dtype)
    matrix /= np.linalg.norm(matrix, axis=1, keepdims=True)
    query = matrix[0].copy()

    matrix @ query                                   # warm up: page in, plan BLAS

    single = []
    for _ in range(RUNS):
        start = time.perf_counter()
        scores = matrix @ query
        np.argpartition(-scores, 5)[:5]
        single.append(time.perf_counter() - start)
    single.sort()

    batch = rng.standard_normal((100, d)).astype(dtype)
    batch /= np.linalg.norm(batch, axis=1, keepdims=True)
    start = time.perf_counter()
    batch @ matrix.T
    batched = time.perf_counter() - start

    gib = matrix.nbytes / 2**30
    median = single[len(single) // 2]
    print(f"n={n:,} d={d} dtype={np.dtype(dtype).name}")
    print(f"  resident      {matrix.nbytes / 2**20:8.1f} MiB")
    print(f"  1 query p50   {median * 1e3:8.2f} ms"
          f"   -> {gib / median:6.1f} GiB/s effective")
    print(f"  100 queries   {batched * 1e3:8.2f} ms"
          f"   ({batched / 100 * 1e3:.3f} ms each)")

if __name__ == "__main__":
    for n in (10_000, 100_000, 1_000_000):
        bench(n, 1536)

从输出中读出三件事。单次查询延迟在你的界面中是否可接受。有效的 GiB/s 数值，你可以用它与你的机器规格比较，看看 NumPy 是否达到了硬件性能。以及批量查询和单次查询的每次耗时比率，这是通过批处理你能获得多少收益——这个数字通常很大，而且会影响你设计调用代码的方式。

Top-k 无需完整排序

np.argsort(-scores)[:k] 对所有 100,000 个分数排序，只保留五个。这是 O(n log n) 的工作用于 O(n) 的问题。np.argpartition 做的是部分选择：它把最佳的 k 个放入前 k 个位置，顺序任意，时间线性。然后只需对这 k 个排序。

# wrong, and gets slower as the corpus grows
top = np.argsort(-scores)[:k]

# right: partition n, then sort k
top = np.argpartition(-scores, k)[:k]
top = top[np.argsort(-scores[top])]

第二种形式有两个边缘情况：argpartition 如果 k 不小于数组长度会抛异常，所以要做边界限制；而且在第二行运行之前，得到的索引是无序的，所以在之前不要使用它们。

元数据过滤器是一个布尔掩码应用到分数上，它保持精确性，而近似索引必须在搜索前或搜索后过滤之间做选择：

mask = np.array([m["lang"] == "en" for m in self.meta])
scores = np.where(mask, scores, -np.inf)

保持索引最新

文档会变化，每晚重新 embedding 所有内容既慢也会产生不必要的账单。两个操作覆盖几乎所有情况，而且都是普通的数组操作。

# update.py
import json

import numpy as np

def append(path: str, records: list[dict], embed) -> None:
    """Add new documents without touching the existing vectors."""
    matrix = np.load(f"{path}.npy")
    fresh = np.asarray(embed([r["text"] for r in records]), dtype=np.float32)
    fresh /= np.linalg.norm(fresh, axis=1, keepdims=True)
    np.save(f"{path}.npy", np.vstack([matrix, fresh]))
    with open(f"{path}.jsonl", "a", encoding="utf-8") as meta:
        for record in records:
            meta.write(json.dumps(record, ensure_ascii=False) + "\n")

def rewrite(path: str, keep: np.ndarray) -> None:
    """Drop rows. keep is a boolean mask over the current rows."""
    matrix = np.load(f"{path}.npy")
    with open(f"{path}.jsonl", encoding="utf-8") as fh:
        meta = [json.loads(line) for line in fh]
    np.save(f"{path}.npy", matrix[keep])
    with open(f"{path}.jsonl", "w", encoding="utf-8") as fh:
        for row, keeping in zip(meta, keep):
            if keeping:
                fh.write(json.dumps(row, ensure_ascii=False) + "\n")

删除必须同时重写两个文件，这是需要小心谨慎的操作：中断的重写会让矩阵和元数据描述不同的文档集合，而加载器中的 assert 只有在数量恰好不同时才能捕获这种情况。先写到临时路径，最后再重命名两个文件，或者接受崩溃意味着完全重建。

将 embedding 模型 ID 和内容哈希与每条记录一起存储。模型 ID 告诉你哪些行需要在模型更换后重新 embedding——来自两个不同模型的向量不可比较，混合它们会悄悄降级每个结果，不只是新行。内容哈希让你能增量 job 只 embedding 文本实际变化了的文档，而不是所有修改时间都是新的文档。

更换 embedding 模型意味着重新 embedding 整个语料库，而不是追加到它。不存在部分迁移：用新模型 embedding 的查询与用旧模型 embedding 的行评分是胡说。Re-embedding migrations 涵盖了如何无停机地做到这一点，embedding drift 涵盖了注意到你需要这么做。

当你确实需要数据库时

暴力搜索在相当具体的边界处不再是正确答案，知道你碰到了哪一个而不是因为不安而迁移是值得的。

语料库不再能放入 RAM。这是硬性的。在 float32 的 1,536 维情况下，每百万向量约 5.7 GiB；一旦数组开始交换，每次查询都要付出磁盘延迟，上面的算术就不再适用。

你所在规模的延迟超出了你的预算。用上面的工具测量，不是假设。如果查询必须在 10ms 内返回但实际需要 60ms，近似索引如 HNSW 以少量召回率换取大幅度提速。

写入是连续的。每晚重建 .npy 文件没问题，但如果文档每秒都在到达就力不从心了。实时插入和删除是真正的索引给你的东西。

多个进程必须查询它。每个进程一个数组意味着每个进程一份 586 MiB 的副本。这就是共享服务器开始合算的临界点。

在那之前，保持索引为两个文件意味着可以用脚本从零重建，提交到对象存储，可以 diff，可以丢弃。Do you need a vector database 从操作角度论证了同样的观点。