趋近智
近似最近邻(ANN)算法由多种参数 (parameter)控制。通过实践,将直接观察到调整这些参数如何影响搜索性能。本次实践旨在培养对索引时间、搜索速度和准确率(召回率)之间权衡的直观感受。
我们将主要关注HNSW,因为它是一种广泛使用且有效的算法,在许多向量 (vector)数据库中均有提供。我们将使用qdrant-client库,该库允许我们在内存中运行Qdrant实例,方便试验,并对HNSW参数提供精细控制。
首先,确保你已安装所需的库。你需要qdrant-client用于向量 (vector)数据库交互,numpy用于数值运算,sentence-transformers用于生成示例嵌入 (embedding)(或者你可以使用预计算好的),time用于测量延迟,以及scikit-learn用于根据真实数据计算召回率。
pip install qdrant-client numpy sentence-transformers scikit-learn
接下来,我们准备一些数据并设置在内存中运行的Qdrant客户端。为简单起见,我们将生成一小组随机向量,但你可以轻松地用使用all-MiniLM-L6-v2等模型从真实文本数据生成的嵌入来替代它们。
import numpy as np
from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer # Or use your own embeddings
import time
from sklearn.neighbors import NearestNeighbors
# 配置
NUM_VECTORS = 1000
DIMENSION = 384 # all-MiniLM-L6-v2 的示例维度
COLLECTION_NAME = "hnsw_experiment_collection"
DISTANCE_METRIC = models.Distance.COSINE
# 生成示例数据(如果需要,请替换为真实嵌入)
print(f"正在生成 {NUM_VECTORS} 个维度为 {DIMENSION} 的随机向量...")
# 使用固定随机种子以确保结果可重现
np.random.seed(42)
data_vectors = np.random.rand(NUM_VECTORS, DIMENSION).astype(np.float32)
query_vector = np.random.rand(1, DIMENSION).astype(np.float32)
# --- 可选:使用 Sentence Transformers 生成更真实的数据 ---
# model = SentenceTransformer('all-MiniLM-L6-v2')
# sample_texts = ["这是第一句话。", "这是另一句话。", "用于向量搜索的示例文本。"] # 添加更多文本
# data_vectors = model.encode(sample_texts)
# query_text = "关于句子的查询。"
# query_vector = model.encode([query_text])
# DIMENSION = data_vectors.shape[1]
# NUM_VECTORS = data_vectors.shape[0]
# ------------------------------------------------------------------
print("示例数据已生成。")
print(f"Data vectors shape: {data_vectors.shape}")
print(f"Query vector shape: {query_vector.shape}")
# Initialize Qdrant client in memory
client = QdrantClient(":memory:")
print("Qdrant 客户端已在内存中初始化。")
# 使用 scikit-learn 计算真实结果(精确最近邻)
print("正在计算真实结果(精确KNN)...")
nn_exact = NearestNeighbors(n_neighbors=10, metric='cosine', algorithm='brute')
nn_exact.fit(data_vectors)
ground_truth_indices = nn_exact.kneighbors(query_vector, return_distance=False)[0]
ground_truth_set = set(ground_truth_indices)
print("真实结果已计算。")
# 计算 recall@10 的辅助函数
def calculate_recall(retrieved_indices, ground_truth):
retrieved_set = set(idx for idx in retrieved_indices)
true_positives = len(retrieved_set.intersection(ground_truth))
recall = true_positives / len(ground_truth) if len(ground_truth) > 0 else 0
return recall
此设置创建了NUM_VECTORS个随机向量,定义了一个query_vector,初始化了一个内存中的Qdrant客户端,并且重要的一点是,使用scikit-learn的精确最近邻搜索计算了ground_truth_indices。这个真实结果将作为我们计算召回率的参照。
HNSW有几个重要参数。我们将关注其中三个:
m:图中每层每个节点的最大连接数。更高的m通常会带来更好的召回率,但会增加索引大小和索引时间。ef_construct:索引构建期间使用的动态列表的大小。更大的ef_construct在构建图时会考虑更多的潜在邻居,从而可能获得更高质量的索引(更好的召回率),但代价是更长的索引时间。ef_search(在Qdrant API中查询时常称为ef,与ef_construct不同):搜索期间使用的动态列表的大小。更大的ef_search在查询时会查看更多的潜在邻居,通常会提高召回率但会增加查询延迟。我们来设计一个函数,用于创建集合、索引数据和执行搜索,从而允许我们改变这些参数。
def run_hnsw_experiment(m_value, ef_construct_value, ef_search_value, k=10):
"""
创建具有特定HNSW参数的Qdrant集合,索引数据,
执行搜索,并返回指标。
"""
# 如果集合在之前运行中存在,则删除它
try:
client.delete_collection(collection_name=COLLECTION_NAME)
# print(f"集合 '{COLLECTION_NAME}' 已删除。")
except Exception:
# print(f"集合 '{COLLECTION_NAME}' 不存在,跳过删除。")
pass
time.sleep(0.1) # 短暂暂停,确保删除完成
# Create collection with specified HNSW parameters
start_index_time = time.time()
client.create_collection(
collection_name=COLLECTION_NAME,
vectors_config=models.VectorParams(
size=DIMENSION,
distance=DISTANCE_METRIC
),
hnsw_config=models.HnswConfigDiff(
m=m_value,
ef_construct=ef_construct_value
)
)
# Index the data (using IDs 0 to NUM_VECTORS-1)
client.upsert(
collection_name=COLLECTION_NAME,
points=models.Batch(
ids=list(range(NUM_VECTORS)),
vectors=data_vectors.tolist()
),
wait=True # 等待索引完成
)
index_time = time.time() - start_index_time
# Perform the search
start_search_time = time.time()
search_result = client.search(
collection_name=COLLECTION_NAME,
query_vector=query_vector[0].tolist(),
search_params=models.SearchParams(
hnsw_ef=ef_search_value, # 这是搜索时的 ef 参数
exact=False # 确保我们使用 ANN 索引
),
limit=k # 获取前 k 个结果
)
search_latency = time.time() - start_search_time
# Calculate recall
retrieved_ids = [hit.id for hit in search_result]
recall = calculate_recall(retrieved_ids, ground_truth_set)
# print(f"参数: m={m_value}, ef_construct={ef_construct_value}, ef_search={ef_search_value}")
# print(f" 索引时间: {index_time:.4f}秒")
# print(f" 搜索延迟: {search_latency:.4f}秒")
# print(f" 召回率@{k}: {recall:.4f}")
# print("-" * 20)
return {
"m": m_value,
"ef_construct": ef_construct_value,
"ef_search": ef_search_value,
"index_time": index_time,
"search_latency": search_latency,
"recall": recall
}
现在我们可以通过调用此函数并使用不同的参数组合来运行试验。
ef_search让我们保持m和ef_construct不变,看看改变ef_search如何影响召回率和延迟。更高的ef_search通常会同时增加这两者。
results_ef_search = []
m_fixed = 16
ef_construct_fixed = 100
ef_search_values = [10, 20, 50, 100, 200] # 尝试不同的搜索复杂度
print(f"\n--- 试验:改变 ef_search (m={m_fixed}, ef_construct={ef_construct_fixed}) ---")
for ef_s in ef_search_values:
result = run_hnsw_experiment(m_fixed, ef_construct_fixed, ef_s)
results_ef_search.append(result)
print(f"ef_search={ef_s}: 召回率={result['recall']:.3f}, 延迟={result['search_latency']:.4f}秒")
# 准备绘图数据
latencies = [r['search_latency'] for r in results_ef_search]
recalls = [r['recall'] for r in results_ef_search]
ef_s_labels = [str(r['ef_search']) for r in results_ef_search]
我们可以可视化这种权衡:
随着
ef_search的增加,召回率明显提升,但搜索延迟也随之增加。最佳值取决于你的应用是优先考虑更高的准确率还是更低的延迟。注意:具体数值很大程度上取决于数据、硬件和库的实现。
ef_construct现在,我们固定m和ef_search,看看ef_construct如何影响索引构建时间和最终的搜索质量。更高的ef_construct应该会增加索引时间,但即使ef_search较低,也可能带来更好的召回率。
results_ef_construct = []
m_fixed = 16
ef_search_fixed = 50 # 保持搜索工作量适中
ef_construct_values = [50, 100, 200, 400] # 尝试不同的构建复杂度
print(f"\n--- 试验:改变 ef_construct (m={m_fixed}, ef_search={ef_search_fixed}) ---")
for ef_c in ef_construct_values:
result = run_hnsw_experiment(m_fixed, ef_c, ef_search_fixed)
results_ef_construct.append(result)
print(f"ef_construct={ef_c}: 索引时间={result['index_time']:.3f}秒, 召回率={result['recall']:.3f}, 延迟={result['search_latency']:.4f}秒")
# 准备绘图数据
index_times_efc = [r['index_time'] for r in results_ef_construct]
recalls_efc = [r['recall'] for r in results_ef_construct]
ef_c_labels = [str(r['ef_construct']) for r in results_ef_construct]
增加
ef_construct会明显增加构建索引所需的时间。虽然它可以在固定的ef_search下提高召回率,但收益可能会在某个点后递减。更高的ef_construct构建了一个可能“更好”的图,这可能允许之后在略低的ef_search下也能获得良好的召回率。
m最后,我们来改变m,同时保持ef参数不变。请记住,m控制着图的连接性。
results_m = []
ef_construct_fixed = 100
ef_search_fixed = 50
m_values = [8, 16, 32, 64] # 尝试不同的连接级别
print(f"\n--- 试验:改变 m (ef_construct={ef_construct_fixed}, ef_search={ef_search_fixed}) ---")
for m_val in m_values:
result = run_hnsw_experiment(m_val, ef_construct_fixed, ef_search_fixed)
results_m.append(result)
print(f"m={m_val}: 索引时间={result['index_time']:.3f}秒, 召回率={result['recall']:.3f}, 延迟={result['search_latency']:.4f}秒")
# 准备绘图数据
index_times_m = [r['index_time'] for r in results_m]
recalls_m = [r['recall'] for r in results_m]
m_labels = [str(r['m']) for r in results_m]
增加
m也会增加索引构建时间(以及索引大小,尽管此处未测量),同时通常会提高召回率。与ef_construct类似,对召回率的影响可能会显示递减的收益。m的常见值通常在16到64之间。
这个动手练习表明了调整近似最近邻索引参数 (parameter)的实际效果。你观察到:
ef_search直接以更高的搜索延迟换取更好的召回率。ef_construct和m会在索引期间投入更多时间(和潜在的内存)来构建更高质量的图结构,这可以带来更好的召回率,或者在查询期间允许使用更低的ef_search值。最佳参数并非通用。它们很大程度上取决于:
参数调整通常涉及更系统的试验,可能使用更大的数据集和自动化评估框架来考察参数空间,并为你的特定需求找到最佳平衡。然而,这种实践经验提供了一种基本理解,即这些参数如何影响系统的行为,从而帮助你在构建自己的语义搜索系统时做出明智的决定。
这部分内容有帮助吗?
© 2026 ApX Machine LearningAI伦理与透明度•