向量数据库：语义检索的底座

ANN 索引

精确最近邻（KNN）在亿级向量上太慢，工程上用近似最近邻（ANN）换取速度。两种主流结构：

• HNSW：基于多层可导航小世界图，查询快、内存占用高，适合中高并发在线服务。
• IVF：先聚类再在最近簇内搜索，内存友好、可量化压缩，适合超大规模。

理解 ANN 的关键指标只有三个：召回率（近似结果里有多少是真正的最近邻）、QPS/延迟（每秒能扛多少查询）、内存占用。所有索引结构与参数，本质上都是在这三者之间做交换。生产环境常见目标是「召回率 95% 以上、P99 延迟 50ms 以内」，先定住这个预算，再回头调参数。

还有一个常被忽视的点：距离度量必须与嵌入模型匹配。多数句向量模型（如 BGE、text-embedding-3）推荐余弦相似度。工程上惯用做法是先对向量做 L2 归一化，然后用内积（Inner Product）代替余弦，两者在归一化后数学等价，而内积计算更快。

IVF 与 HNSW 图解

一图看懂两种索引的搜索方式：IVF 把全库向量先聚成若干簇，查询时只进最近的几个簇里精搜；HNSW 则像高速公路网，从顶层稀疏图快速逼近目标区域，再逐层下沉到底层稠密图精确定位。

直观结论：IVF 的召回损失来自「目标向量落在没被搜到的簇里」，所以增大 nprobe 能提召回；HNSW 的召回损失来自「贪心搜索提前收敛」，所以增大 ef_search 能提召回。两者都是用更多计算换更高召回。

FAISS 实战

FAISS 是 Meta 开源的向量检索库，是理解一切向量数据库的「白盒教具」：没有服务端、没有网络层，纯粹的索引与搜索。先用最朴素的精确索引 IndexFlatIP 跑通全流程：

# pip install faiss-cpu sentence-transformers
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

# 1. 文本转向量（bge-small-zh 输出 512 维）
model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
docs = [
    "RAG 通过检索外部知识增强大模型回答",
    "向量数据库为语义检索提供索引底座",
    "LoRA 是一种参数高效的微调方法",
    "HNSW 是基于图结构的近似最近邻索引",
]
emb = model.encode(docs).astype("float32")

# 2. L2 归一化后，内积 == 余弦相似度
faiss.normalize_L2(emb)

# 3. 建精确内积索引并入库
dim = emb.shape[1]          # 512
index = faiss.IndexFlatIP(dim)
index.add(emb)
print("库内向量数:", index.ntotal)   # 4

# 4. Top-K 相似度搜索
query = model.encode(["什么是近似最近邻搜索"]).astype("float32")
faiss.normalize_L2(query)
scores, ids = index.search(query, k=2)
for s, i in zip(scores[0], ids[0]):
    print(f"score={s:.4f}  doc={docs[i]}")
# 排第一的应是 HNSW 那条文档

数据量到十万级以上时，Flat 索引的暴力扫描开始吃不消，换 IndexIVFFlat：先训练聚类，再分簇搜索。

# IVF 索引：10 万条 512 维向量的典型写法
import faiss
import numpy as np

dim, n = 512, 100_000
data = np.random.rand(n, dim).astype("float32")
faiss.normalize_L2(data)

nlist = 1024                      # 簇数，经验值约 4*sqrt(n)
quantizer = faiss.IndexFlatIP(dim) # 用于找最近簇的粗量化器
index = faiss.IndexIVFFlat(quantizer, dim, nlist,
                           faiss.METRIC_INNER_PRODUCT)

# IVF 必须先 train（跑 k-means 学出簇中心）再 add
index.train(data)
index.add(data)

# nprobe：查询时探测的簇数，默认 1，生产常设 8 到 64
index.nprobe = 16
query = data[:1]                  # 拿第一条自查，应召回自己
scores, ids = index.search(query, k=5)
print(ids[0], scores[0])          # ids[0][0] 应为 0

# 索引落盘与加载
faiss.write_index(index, "docs.ivf.faiss")
index2 = faiss.read_index("docs.ivf.faiss")

四大方案对比

• Chroma：轻量、嵌入式优先，开发体验极佳，适合原型与中小规模。
• Milvus：分布式、水平扩展强，支持十亿级向量，适合生产大规模。
• Qdrant：Rust 实现、过滤与量化优秀，云原生友好，性价比高。
• pgvector：直接跑在 Postgres 里，事务与关系数据一体，无需额外组件。

把 FAISS 也放进来横向对照，一张表看清各自的定位与适用场景：

pgvector 与 Milvus 实战

pgvector 的哲学是「别新增一个系统」：向量就是 Postgres 里的一种列类型，检索就是一条 SQL，能和业务表 JOIN、能进事务、能被备份工具统一管理。建表、插入、建 HNSW 索引、Top-K 查询四步走：

-- 1. 启用扩展（Postgres 14+，需已安装 pgvector）
CREATE EXTENSION IF NOT EXISTS vector;

-- 2. 建表：512 维向量列 + 业务字段
CREATE TABLE docs (
  id        bigserial PRIMARY KEY,
  content   text NOT NULL,
  category  text,
  embedding vector(512)
);

-- 3. 插入向量（实际由程序把嵌入结果拼进参数）
INSERT INTO docs (content, category, embedding)
VALUES ('RAG 通过检索外部知识增强回答', 'rag', '[0.12, -0.03, ...]');

-- 4. 建 HNSW 索引，用余弦距离算子类
CREATE INDEX ON docs
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- 5. Top-5 语义检索 + 元数据过滤，<=> 是余弦距离算子
SET hnsw.ef_search = 40;
SELECT id, content, embedding <=> '[0.09, 0.11, ...]' AS dist
FROM docs
WHERE category = 'rag'
ORDER BY dist
LIMIT 5;

Milvus 则走「专业分布式」路线：存储与计算分离，索引构建、查询、数据落盘由不同组件分担，水平扩容即可支撑十亿级。用 pymilvus 的 MilvusClient 可以几行跑通（本地开发可用 Milvus Lite，无需起集群）：

# pip install pymilvus  （milvus-lite 会作为本地后端自动安装）
from pymilvus import MilvusClient

# 本地文件模式，零部署；生产改为 uri="http://milvus:19530"
client = MilvusClient("./local.db")

# 建集合：512 维，metric 用内积（向量需已归一化）
client.create_collection(
    collection_name="docs",
    dimension=512,
    metric_type="IP",
)

# 插入：字典列表，vector 为归一化后的 float 列表
rows = [
    {"id": 1, "vector": emb[0].tolist(), "text": docs[0], "tag": "rag"},
    {"id": 2, "vector": emb[1].tolist(), "text": docs[1], "tag": "infra"},
]
client.insert(collection_name="docs", data=rows)

# Top-3 检索 + 标量过滤，output_fields 指定回带的字段
hits = client.search(
    collection_name="docs",
    data=[query[0].tolist()],
    limit=3,
    filter='tag == "rag"',
    output_fields=["text", "tag"],
)
for h in hits[0]:
    print(h["distance"], h["entity"]["text"])

Milvus 完整流程：指定索引类型

上面的例子用了默认索引。生产环境通常显式指定索引类型并调整参数。以下是带 index_type 的完整流程：

# Milvus 完整流程：显式指定索引类型与参数
from pymilvus import MilvusClient, DataType

# 1. 连接（本地文件模式；生产换成 uri + token）
client = MilvusClient("./milvus_demo.db")

# 2. 定义 schema：字段类型、主键、向量维度与索引类型
collection_name = "docs_v2"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = client.create_schema(
    auto_id=False,
    enable_dynamic_field=True,
)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=512)
schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=512)
schema.add_field(field_name="tag", datatype=DataType.VARCHAR, max_length=64)

# 3. 创建 collection：指定 metric + index_type
client.create_collection(
    collection_name=collection_name,
    schema=schema,
    metric_type="IP",            # 内积（需向量已归一化）
)

# 4. 建索引：HNSW 适合中小规模高性能检索
index_params = {
    "field_name": "vector",
    "index_type": "HNSW",
    "metric_type": "IP",
    "params": {"M": 16, "efConstruction": 200},
}
client.create_index(collection_name, index_params)
client.load_collection(collection_name)    # 索引加载到内存

# 5. 批量插入
data = [
    {"id": i, "vector": emb[i].tolist(),
     "text": docs[i], "tag": tags[i]}
    for i in range(len(docs))
]
client.insert(collection_name, data)

# 6. 搜索（指定 ef 参数控制召回范围）
hits = client.search(
    collection_name=collection_name,
    data=[query_vec.tolist()],
    limit=5,
    search_params={"ef": 64},     # HNSW 搜索范围，越大越准越慢
    filter='tag == "index"',
    output_fields=["text", "tag"],
)
for h in hits[0]:
    print(f"score={h['distance']:.4f}  text={h['entity']['text'][:40]}")

Qdrant 实战

Qdrant 用 Rust 编写，单机即可支撑千万级向量检索。API 设计上介于 Chroma 的简单与 Milvus 的强结构化之间：需要声明向量配置，但用 upsert 语义（插入或更新）简化操作。

# pip install qdrant-client
from qdrant_client import QdrantClient
from qdrant_client.http import models as qmod

# 1. 连接（本地模式，生产用 host + port 连远端）
client = QdrantClient(path="./qdrant_store")

# 2. 创建 collection：显式声明向量维度与距离度量
collection_name = "docs"
vector_config = qmod.VectorParams(
    size=512,                       # 向量维度
    distance=qmod.Distance.COSINE, # 余弦距离
    hnsw_config=qmod.HnswConfigDiff(m=16, ef_construct=200),
)
client.create_collection(collection_name, vectors_config=vector_config)

# 3. Upsert：不存在则插入，存在则更新（幂等操作）
points = [
    qmod.PointStruct(
        id=i,
        vector=emb[i].tolist(),
        payload={"text": docs[i], "tag": tags[i]},
    )
    for i in range(len(docs))
]
client.upsert(collection_name, points)

# 4. 搜索：带 payload 过滤 + 指定 ef 控制召回
hits = client.search(
    collection_name=collection_name,
    query_vector=query_vec.tolist(),
    limit=5,
    search_params=qmod.SearchParams(hnsw_ef=64),
    query_filter=qmod.Filter(
        must=[qmod.FieldCondition(
            key="tag", match=qmod.MatchValue(value="index")
        )]
    ),
    with_payload=True,
)
for h in hits:
    print(f"score={h.score:.4f}  text={h.payload['text'][:40]}")

选型维度

按三个维度决策：规模（百万级以内 Chroma/pgvector 足够，十亿级上 Milvus）、运维成本（不想多养一个系统选 pgvector）、性能与过滤（强元数据过滤选 Qdrant）。

再补两条实战经验：一是先按「团队已有什么」选，而不是按跑分选，已有 Postgres 就先上 pgvector，跑分差距远小于多运维一套系统的隐性成本；二是预留迁移路径，把嵌入生成、入库、查询封装在自己的一层薄接口后面，未来从 Chroma 换到 Milvus 只需改适配层，业务代码不动。

光有数据库还不够，索引类型的选择同样影响性能。下面这张决策图从使用场景出发，帮你快速定位该用哪种索引：

混合检索

纯向量检索擅长语义，但弱于精确匹配（专有名词、编号）。混合检索把向量召回与 BM25 关键词召回用 RRF 等策略融合，既懂语义又认字面，是生产环境的推荐基线。

RRF（Reciprocal Rank Fusion）不需要调分数权重，只看名次，实现只有几行：

# RRF：融合向量召回与 BM25 召回的两个排名列表
def rrf_fuse(rank_lists, k=60, top_n=10):
    """rank_lists: 若干个按相关性降序的 doc_id 列表"""
    scores = {}
    for ranking in rank_lists:
        for pos, doc_id in enumerate(ranking):
            # 名次越靠前贡献越大，k 平滑头部权重
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + pos + 1)
    fused = sorted(scores, key=scores.get, reverse=True)
    return fused[:top_n]

vector_hits = ["d3", "d1", "d7", "d2"]   # 向量召回结果
bm25_hits   = ["d7", "d3", "d9", "d4"]   # BM25 召回结果
print(rrf_fuse([vector_hits, bm25_hits]))
# d3、d7 因在两路都靠前而排到最前

Chroma 上手

import chromadb
client = chromadb.Client()
col = client.create_collection("docs")
col.add(ids=["1"], documents=["RAG 让回答有据可依"])
res = col.query(query_texts=["什么是 RAG"], n_results=1)
print(res["documents"])

Chroma 完整 CRUD + 查询

上面是最小化示例。完整 CRUD 操作覆盖创建 collection、批量添加、更新、删除、带元数据过滤的查询，以及错误处理：

# Chroma 完整 CRUD：创建、增删改查 + 元数据过滤
import chromadb
from chromadb.utils import embedding_functions

# 1. 初始化（持久化到本地目录，生产可换 HTTP 客户端连远程实例）
ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-small-zh-v1.5"
)
client = chromadb.PersistentClient(path="./chroma_store")

# 2. 创建 collection（若已存在则报错，用 get_or_create 规避）
try:
    col = client.create_collection(
        name="kb",
        embedding_function=ef,
        metadata={"description": "技术知识库"},
    )
except chromadb.db.base.UniqueConstraintError:
    col = client.get_collection("kb", embedding_function=ef)

# 3. 批量添加（ids + documents + metadatas 三者长度必须一致）
docs = [
    "HNSW 用多层图结构加速 ANN 搜索",
    "IVF 先 K-Means 聚类再在最近簇内精搜",
    "RAG 结合检索与生成提高回答准确性",
    "LoRA 用低秩矩阵高效微调大模型",
]
metas = [
    {"topic": "index", "level": "advanced"},
    {"topic": "index", "level": "intermediate"},
    {"topic": "rag",   "level": "basic"},
    {"topic": "finetune", "level": "intermediate"},
]
col.add(ids=["d1", "d2", "d3", "d4"], documents=docs, metadatas=metas)

# 4. 查询：元数据过滤 + 语义搜索
res = col.query(
    query_texts=["图结构的 ANN 索引"],
    n_results=3,
    where={"topic": "index"},        # 只搜 topic=index 的文档
)
for doc, meta, dist in zip(
    res["documents"][0], res["metadatas"][0], res["distances"][0]
):
    print(f"score={dist:.4f}  topic={meta['topic']}  doc={doc[:40]}")

# 5. 更新：修改文档文本或元数据（根据 id 原地覆盖）
col.update(
    ids=["d3"],
    documents=["RAG 通过检索增强上下文来减轻幻觉"],
    metadatas=[{"topic": "rag", "level": "advanced"}],
)

# 6. 删除：按 id 精确删除
col.delete(ids=["d4"])

# 7. 浏览：列出前几条（验证数据状态）
first = col.peek(limit=3)
print("剩余文档 ids:", first["ids"])

上面是最小可运行示例。实际项目再加三件事：持久化（PersistentClient 落盘到目录）、元数据过滤（where 条件）、指定嵌入函数（默认用内置模型，中文建议换 BGE）：

# 持久化 + 元数据过滤 + 自定义嵌入模型
import chromadb
from chromadb.utils import embedding_functions

ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-small-zh-v1.5"
)
client = chromadb.PersistentClient(path="./chroma_store")
col = client.get_or_create_collection("kb", embedding_function=ef)

col.add(
    ids=["a1", "a2"],
    documents=["HNSW 用多层图加速搜索", "IVF 先聚类再进簇搜索"],
    metadatas=[{"topic": "index"}, {"topic": "index"}],
)
res = col.query(
    query_texts=["图结构的 ANN 索引"],
    n_results=1,
    where={"topic": "index"},
)
print(res["documents"][0])   # 应命中 HNSW 那条

它是 RAG 模块里最常见的起步选择。需要长期记忆支撑 Agent 时，也可复用同一套向量库。

三大向量库性能基准

在选型时不能只看文档，实际跑一遍 benchmark 能让你对性能和资源消耗有直观认知。下面用 10 万条 768 维随机向量（模拟 bge-base-zh 输出），在同等硬件上对比 Chroma、Milvus、Qdrant 的表现。

以下 benchmark 脚本可直接在终端运行，生成随机向量并测量四个核心指标：

# benchmark_vectordb.py -- 三大向量库性能基准脚本
# pip install chromadb qdrant-client pymilvus numpy
import time
import numpy as np
import chromadb
from chromadb.utils import embedding_functions
from qdrant_client import QdrantClient
from qdrant_client.http import models as qmod
from pymilvus import MilvusClient, DataType

# 参数：10 万条 768 维随机向量（模拟 bge-base-zh 输出）
N, DIM = 100_000, 768
vectors = np.random.rand(N, DIM).astype("float32")
# L2 归一化，使内积等价于余弦相似度
vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
queries = vectors[:1000]  # 1000 条查询用于测 QPS
ids = [str(i) for i in range(N)]

# ---- Chroma benchmark ----
t0 = time.time()
c_client = chromadb.PersistentClient(path="./bench_chroma")
c_col = c_client.create_collection("bench")
# Chroma 按 id 分批插入（直接传 embedding）
c_col.add(ids=ids, embeddings=vectors.tolist())
t_build = time.time() - t0

t0 = time.time()
for q in queries[:100]:  # 取 100 条测吞吐
    c_col.query(query_embeddings=[q.tolist()], n_results=10)
t_query = time.time() - t0
print(f"[Chroma]  构建={t_build:.1f}s  QPS={100/t_query:.0f}")

# ---- Milvus benchmark ----
t0 = time.time()
m_client = MilvusClient("./bench_milvus.db")
m_schema = m_client.create_schema(auto_id=False)
m_schema.add_field("id", DataType.INT64, is_primary=True)
m_schema.add_field("vector", DataType.FLOAT_VECTOR, dim=DIM)
m_client.create_collection("bench", m_schema, metric_type="IP")
data = [{"id": i, "vector": vectors[i].tolist()} for i in range(N)]
m_client.insert("bench", data)
m_client.create_index("bench", {
    "field_name": "vector", "index_type": "HNSW",
    "metric_type": "IP", "params": {"M": 16, "efConstruction": 200}
})
m_client.load_collection("bench")
t_build = time.time() - t0

t0 = time.time()
for q in queries[:100]:
    m_client.search("bench", data=[q.tolist()], limit=10,
                    search_params={"ef": 64})
t_query = time.time() - t0
print(f"[Milvus]  构建={t_build:.1f}s  QPS={100/t_query:.0f}")

# ---- Qdrant benchmark ----
t0 = time.time()
q_client = QdrantClient(path="./bench_qdrant")
q_client.create_collection("bench", vectors_config=qmod.VectorParams(
    size=DIM, distance=qmod.Distance.COSINE,
    hnsw_config=qmod.HnswConfigDiff(m=16, ef_construct=200),
))
points = [qmod.PointStruct(id=i, vector=vectors[i].tolist()) for i in range(N)]
q_client.upsert("bench", points)
t_build = time.time() - t0

t0 = time.time()
for q in queries[:100]:
    q_client.search("bench", query_vector=q.tolist(), limit=10,
                    search_params=qmod.SearchParams(hnsw_ef=64))
t_query = time.time() - t0
print(f"[Qdrant]  构建={t_build:.1f}s  QPS={100/t_query:.0f}")

动手练习

三个练习由浅入深，全部可以在一台笔记本上完成。建议按顺序做，每个都跑出验收标准里的数字再进下一个。

练习一：索引 1000 条文档并做 Top-K 查询（FAISS）

• 任务：准备 1000 条中文短文本（可用维基摘要、新闻标题，或让大模型批量生成），用 bge-small-zh 编码并归一化，分别建 IndexFlatIP 与 IndexIVFFlat（nlist=32）两个索引，对同一批 20 条查询做 Top-5 检索。
• 验收标准：两个索引都能返回结果；以 Flat 为真值计算 IVF 的 Recall@5，nprobe=1 时记录一次、nprobe=8 时记录一次，后者召回率应明显更高（通常可达 0.95 以上）；索引成功 write_index 落盘并能 read_index 加载后复现同样结果。

练习二：pgvector 语义检索 + SQL 过滤

• 任务：用 Docker 起一个带 pgvector 的 Postgres（镜像 pgvector/pgvector:pg16），建本页 docs 表结构，把练习一的 1000 条文本及向量写入，建 HNSW 索引，写一条「先按 category 过滤、再按余弦距离排序取 Top-5」的查询。
• 验收标准：EXPLAIN 输出显示查询走了 hnsw 索引扫描而非全表顺扫；同一条查询在建索引前后各跑一次，延迟有可观测的下降；过滤条件生效（结果里 category 全部符合）。

练习三：混合检索对比实验

• 任务：在练习一的语料上，挑 10 条包含专有名词或编号的查询（如产品型号、人名），分别用「纯向量」「纯 BM25」（可用 rank_bm25 库）与「RRF 融合」三种方式检索 Top-5，人工标注每条查询的正确文档。
• 验收标准：产出一张 10 行 3 列的命中对照表；RRF 的总命中数不低于两个单路中的任何一个；能用一句话总结哪类查询向量弱、哪类查询 BM25 弱。

练习四：向量库选型决策

读以下三个场景，结合本模块学到的方案对比知识，判断应该选哪个向量库、用什么索引，并写出至少一条理由：

• 场景一「个人知识库」：约 5000 条笔记文本，在个人笔记本上运行，需要本地检索，没有运维团队。期望零部署、开箱即用。
• 场景二「企业级搜索引擎」：约 500 万条商品描述，日均查询量 50 万次，需要高并发低延迟检索，后续可能扩展到千万级别，团队有 K8s 运维经验。
• 场景三「移动端离线词典」：约 1000 条词条，在手机上离线运行，需要极低延迟（< 10ms），不能联网，不能依赖外部服务。