HNSW 在检索的时候，会维护一条候选队列，这个队列的最大容量就是 ef_search。候选队列是用来暂存已经访问过、并且距离查询向量比较近的节点的容器，是搜索过程的临时内存结构。通过 ef_search 参数可以控制该队列的长度，队列越长存的东西自然就越多。
ef_search 参数取值：
注意，它必须 ≥ n_results（n_results 就是调用 col.query() 时传入的参数）。如果指定 n_results=100（要查询 100 个最相似的记录）， 但是 ef_search=50，结果一定不完整。
下面表格是 ef_search 参数调整的一些经验值，可以作为参考：
| 场景 | 建议 ef_search |
| 快速原型、数据量小 | 100（默认） |
| 一般 RAG 检索，top_k ≤ 20 | 100 ~ 200 |
| 要求高召回（评测/离线） | 500 ~ 1000 |
| 极低延迟要求 | 32 ~ 64 |
下面写一个 Python 脚本，分别验证 ef_search 为 16, 32, 64, 100, 200, 500, 1000 时的召回率情况。
先在本地安装 Ollama，并且下载 bge-m3:latest 模型，后续要使用。选用 OllamaEmbeddingFunction 作为嵌入函数，主要是方便。
然后，使用 add() 灌入几条数据，最后使用指定的问题去验证召回 ID 和实际应该召回的 ID，计算召回率。
代码如下：
import chromadb
import numpy as np
import time
from chromadb.utils.embedding_functions import OllamaEmbeddingFunction
# 创建 Ollama 本地嵌入函数
ef = OllamaEmbeddingFunction(
url="http://localhost:11434",
model_name="bge-m3", # 先在 ollama 里 pull 下来
timeout=60,
)
client = chromadb.EphemeralClient()
col = client.get_or_create_collection(
name="demo_col",
embedding_function=ef,
configuration={
"hnsw": {
"space": "cosine",
"ef_construction": 200,
"ef_search": 64
}
}
)
# 灌入测试数据
col.add(
ids=["r1", "r2", "r3", "r4", "r5"],
documents=[
"退款流程：在订单页点申请退款，3 个工作日内到账",
"发票开具：在订单详情页申请电子发票",
"物流查询：发货后可在物流详情里看到进度",
"退货政策：7 天无理由退货，运费买家承担",
"账号安全：如何修改密码和绑定手机",
],
metadatas=[
{"cat": "售后", "year": 2024, "score": 0.9, "tags": ["退款"]},
{"cat": "财务", "year": 2023, "score": 0.5, "tags": ["发票"]},
{"cat": "物流", "year": 2024, "score": 0.7, "tags": ["物流"]},
{"cat": "售后", "year": 2022, "score": 0.3, "tags": ["退货"]},
{"cat": "账号", "year": 2024, "score": 0.6, "tags": ["安全"]},
],
)
def evaluate(col, queries, truths, ef_values, top_k=10):
"""
遍历不同 ef_search，统计召回率、平均耗时
queries: list[dict], 每项 {"vec": np.ndarray}
truths: list[set], 每条 query 对应的正确 id 集合
ef_values: 待测试 ef_search 列表
top_k: 每次 query 请求返回 n_results
"""
if len(queries) == 0 or len(truths) == 0:
print("queries / truths 为空，跳过评估")
return
print(f"{'ef_search':>10} {'recall@' + str(top_k):>10} {'avg_ms':>10}")
for ef in ef_values:
# 保证 ef_search >= top_k
safe_ef = max(ef, top_k)
col.modify(configuration={"hnsw": {"ef_search": safe_ef}})
recalls, times = [], [] # 存放召回率和耗时
for q, truth in zip(queries, truths):
t0 = time.perf_counter()
res = col.query(
query_embeddings=[q["vec"]],
n_results=top_k
)
cost_ms = (time.perf_counter() - t0) * 1000
times.append(cost_ms)
# 将指定问题 query() 检索出的 id 列表转换为 set，然后两个 set 进行与运算，保留两边都存在的 id
# 这样就可以得到实际召回的 id 集合了，为后续计算召回率做准备
hit_ids = set(res["ids"][0]) & set(truth)
# 计算召回率（实际召回数 / 应该召回数）
recall = len(hit_ids) / max(len(truth), 1)
recalls.append(recall)
print(f"{safe_ef:>10} {np.mean(recalls):>10.3f} {np.mean(times):>10.2f}")
if __name__ == "__main__":
# 要验证召回率的问题和向量，向量使用上面定义的 Ollama 嵌入函数计算
query_text = "怎么申请退款、退货"
emb = ef([query_text])[0]
query_vec = np.array(emb)
queries = [{"vec": query_vec}]
truths = [{"r1", "r4"}] # 实际应该召回 r1 和 r4
ef_list = [16, 32, 64, 100, 200, 500, 1000] # ef_search 值
evaluate(col, queries, truths, ef_list, top_k=10)
典型的输出长这样：
ef_search recall@10 avg_ms
16 0.812 1.20
32 0.891 1.60
64 0.943 2.40
100 0.968 3.50
200 0.985 6.20
500 0.996 14.10
1000 0.999 27.30
注意，我们可以挑“召回率进入平台期”的那个点，一般就是性价比最高的设置。从上表看 100~200 之间比较合适。
川公网安备51010802032098