向量数据库并非万能：何时该用，何时不该用？

向量数据库确实解决了一个实际问题，并且在许多情况下，它们是RAG系统的正确选择。但关键在于：仅仅因为使用了嵌入向量，并不意味着就一定需要一个向量数据库。

目前存在一种日益明显的趋势，即每个RAG实现都从接入一个向量数据库开始。对于大规模、持久化的知识库而言，这或许合理，但这并非总是最高效的路径，尤其是在应用场景更具动态性或对时间更敏感的情况下。

在Planck，嵌入向量被用于增强基于大语言模型的系统。然而，在一个实际应用中，选择避免使用向量数据库，转而采用一个简单的键值存储，结果证明后者是更合适的选择。

在深入探讨该案例之前，可以通过一个简化的通用场景来解释其原因。

示例场景

设想一个简单的RAG风格系统。用户上传几个文本文件，可能是报告或会议记录。这些文件被分割成块，为每个块生成嵌入向量，并利用这些嵌入向量来回答问题。用户在接下来的几分钟内提出少量问题，然后离开。此时，文件及其嵌入向量都已无用，可以安全丢弃。

换言之，数据是临时性的，用户只会提出少量问题，并且希望尽可能快地得到答案。

现在请稍作停顿并自问：

这些嵌入向量应该存储在哪里？

大多数人的直觉反应是：“我有嵌入向量，所以我需要一个向量数据库”。但请稍作思考，看看这个抽象概念背后实际发生了什么。当向向量数据库发送嵌入向量时，它并非仅仅是“存储”它们。它会构建一个索引以加速相似性搜索。索引构建工作正是许多神奇效果的来源，也是大量成本所在。

对于一个长期存在、大规模的知识库，这种权衡完全合理：只需支付一次索引构建成本（或在数据变更时增量支付），然后将其分摊到数百万次查询中。但在上述示例场景中，情况并非如此。恰恰相反：系统需要不断添加小型、一次性的嵌入向量批次，每个批次只回答极少量的查询，然后丢弃所有内容。

因此，真正的问题不是“我应该使用向量数据库吗？”，而是“索引构建工作是否值得？”要回答这个问题，可以参考一个简单的基准测试。

基准测试：无索引检索 vs. 索引检索

汽车与向量数据库概念图

图片由 Julia Fiander 发布于 Unsplash

本节内容技术性较强。将涉及Python代码并解释底层算法。如果具体的实现细节与您无关，可以跳过直接阅读 结果 部分。

需要比较两种系统：

无索引：完全不构建索引，仅将嵌入向量保存在内存中并直接扫描。
向量数据库：预先支付索引构建成本，以使每次查询更快。

首先，考虑“无向量数据库”的方法。当查询到来时，计算查询嵌入向量与所有存储嵌入向量之间的相似度，然后选择top-k个最相似的。这本质上就是没有任何索引的K-最近邻算法。

import numpy as np
def run_knn(embeddings: np.ndarray, query_embedding: np.ndarray, top_k: int) -> np.ndarray:

sims = embeddings @ query_embedding

return sims.argsort()[-top_k:][::-1]

该代码使用点积作为余弦相似度的代理（假设向量已归一化），并对分数进行排序以找到最佳匹配。它实际上就是扫描所有向量并挑选最接近的那些。

现在，看看向量数据库通常做什么。在底层，大多数向量数据库依赖于一个近似最近邻索引。ANN方法牺牲一点精度以大幅提升搜索速度，而HNSW是其中最广泛使用的算法之一。将使用hnswlib库来模拟索引行为。

import numpy as np
import hnswlib
def create_hnsw_index(embeddings: np.ndarray, num_dims: int) -> hnswlib.Index:

index = hnswlib.Index(space='cosine', dim=num_dims)

index.init_index(max_elements=embeddings.shape[0])

index.add_items(embeddings)

return index
def query_hnsw(index: hnswlib.Index, query_embedding: np.ndarray, top_k: int) -> np.ndarray:

labels, distances = index.knn_query(query_embedding, k=top_k)

return labels[0]

为了看清权衡点，可以生成一些随机嵌入向量，将其归一化，并测量每个步骤所需的时间：

import time
import numpy as np
import hnswlib
from tqdm import tqdm
def run_benchmark(num_embeddings: int, num_dims: int, top_k: int, num_iterations: int) -> None:

print(f"Benchmarking with {num_embeddings} embeddings of dimension {num_dims}, retrieving top-{top_k} nearest neighbors.")
knn_times: list[float] = []
index_times: list[float] = []
hnsw_query_times: list[float] = []

for _ in tqdm(range(num_iterations), desc="Running benchmark"):
    embeddings = np.random.rand(num_embeddings, num_dims).astype('float32')
    embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
    query_embedding = np.random.rand(num_dims).astype('float32')
    query_embedding = query_embedding / np.linalg.norm(query_embedding)

    start_time = time.time()
    run_knn(embeddings, query_embedding, top_k)
    knn_times.append((time.time() - start_time) * 1e3)

    start_time = time.time()
    vector_db_index = create_hnsw_index(embeddings, num_dims)
    index_times.append((time.time() - start_time) * 1e3)

    start_time = time.time()
    query_hnsw(vector_db_index, query_embedding, top_k)
    hnsw_query_times.append((time.time() - start_time) * 1e3)

print(f"BENCHMARK RESULTS (averaged over {num_iterations} iterations)")
print(f"[Naive KNN] Average search time without indexing: {np.mean(knn_times):.2f} ms")
print(f"[HNSW Index] Average index construction time: {np.mean(index_times):.2f} ms")
print(f"[HNSW Index] Average query time with indexing: {np.mean(hnsw_query_times):.2f} ms")

run_benchmark(num_embeddings=50000, num_dims=1536, top_k=5, num_iterations=20)