动手实践：调整索引参数的试验

近似最近邻（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参数 (parameter)试验

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值。

最佳参数并非通用。它们很大程度上取决于：

你的具体数据集： 向量 (vector)的分布和维度很重要。
你的硬件： CPU性能和内存带宽会影响索引和搜索速度。
你的应用需求： 你是否需要尽可能高的召回率，即使花费更多时间？还是将最小化延迟作为主要目标，接受略低的召回率？

参数调整通常涉及更系统的试验，可能使用更大的数据集和自动化评估框架来考察参数空间，并为你的特定需求找到最佳平衡。然而，这种实践经验提供了一种基本理解，即这些参数如何影响系统的行为，从而帮助你在构建自己的语义搜索系统时做出明智的决定。

这部分内容有帮助吗？