Milvus中两种核心查询方式：暴力搜索（Brute-force Search） 和 近似最近邻搜索（Approximate Nearest Neighbor, ANN）。
- 逐一计算相似度：这是暴力搜索，能保证100%找到最相似的向量，但速度极慢，不适用于大规模数据。
- 使用Milvus的召回工具：这是近似最近邻（ANN）搜索，它通过构建索引来极大地提升查询速度，但牺牲了绝对的准确性，因此“有可能”找不到理论上最相似的那个向量。
1 Milvus 的两种核心查询模式
1.1 暴力搜索 (Brute-force / Exact Search)
- 工作原理：将您的查询向量与集合中的每一个向量进行距离计算，然后排序，返回距离最近的结果。
- 优点：
- 100%准确：能保证找到理论上最精确的结果，召回率永远是100%。
- 缺点：
- 性能极差：计算量与数据总量成正比（O(n)复杂度）。当数据量达到百万、千万甚至亿级时，查询会变得极其缓慢，甚至无法在可接受的时间内完成。
- 适用场景：
- 数据集非常小（例如几万到几十万级别）。
- 对准确率有绝对要求，不容许任何误差的场景，如金融或医疗领域的某些关键匹配任务。
- 在Milvus中的实现：使用
FLAT类型的索引。当您不对向量字段建立任何其他索引时，Milvus默认就会采用这种方式。
1.2 近似最近邻搜索 (ANN Search)
这是Milvus等向量数据库的核心与精髓。
- 工作原理：它不对整个数据集进行详尽搜索。相反，它在数据入库时会预先通过特定算法（如聚类、图、树等）将相似的向量组织在一起，构建出一个“索引”数据结构。查询时，它利用这个索引快速定位到可能包含最近邻的一个或多个“区域”，然后只在这些小区域内进行精确搜索。
- 优点：
- 速度极快：通过牺牲少量精度，查询速度可以比暴力搜索快几个数量级，能够轻松应对海量数据的实时查询需求。
- 缺点：
- 结果是近似的：由于只搜索了部分数据，存在一定的概率会错过真正的最近邻，导致召回率（Recall）低于100%。
- 需要调优：索引的类型和参数选择会直接影响查询的速度和召回率，需要根据具体场景进行权衡和调整。
- 适用场景：
- 绝大多数大规模向量检索应用，如以图搜图、推荐系统、语义搜索等，这些场景下用户对微小的精度损失不敏感，但对查询速度要求很高。
- 在Milvus中的实现：通过创建如
IVF_FLAT,IVF_PQ,HNSW,DiskANN等索引类型来实现。
2.为什么ANN搜索会“找不到”最相似的向量？
这正是 速度与准确率的权衡（Trade-off） 的体现。
以常用的IVF（Inverted File，倒排文件）索引为例：
- 构建索引时：Milvus会将所有向量分成
nlist个“簇”（Cluster），每个簇有一个中心点。 - 查询时：Milvus会先计算您的查询向量与所有
nlist个簇中心的距离，然后只选择最接近的nprobe个簇。 - 最终搜索：Milvus只在这
nprobe个簇包含的向量中进行暴力搜索，找出最终结果。
如果最相似的那个目标向量，不幸地被分到了一个距离您的查询向量稍远的簇中，而这个簇又没有被选入nprobe个搜索范围内，那么这个目标向量就永远不会被找到。
3.如何提升ANN搜索的召回率（Recall）？
既然您遇到了召回率问题，以下是几种在Milvus中提升召回率的有效方法：
3.1. 调整搜索参数
这是最直接、最常用的方法。 在执行search()时，可以调整search_params来扩大搜索范围。
- 对于
IVF系列索引（如IVF_FLAT,IVF_PQ）：- 提高
nprobe值：这个参数决定了要搜索多少个“簇”。nprobe值越高，搜索的范围越大，找到真正最近邻的概率就越高，召回率也随之提升。但同时，查询时间也会增加。 建议您可以从一个较小的值（如16, 32）开始，逐步增加并测试召回率和延迟，找到最佳平衡点。
- 提高
- 对于
HNSW索引：- 提高
ef值：这个参数控制了搜索时维护的动态候选列表的大小。ef越大，搜索的路径越多，越不容易错过最近邻，召回率越高，但查询也越慢。
- 提高
3.2. 调整索引构建参数
在创建索引时，合理的参数也能提升后续的查询效果。
- 对于
IVF系列索引：- 选择合适的
nlist：nlist（簇的数量）需要根据您的数据量来定。一个常用的经验法则是设置为4 * sqrt(n)（n是向量总数）。nlist过小或过大都可能影响效果。
- 选择合适的
- 对于
HNSW索引：- 提高
M和efConstruction：M定义了图中每个节点的最大出度，efConstruction控制了建图时的搜索深度。增加这两个值可以构建出质量更高、连通性更好的图，从而提升召回率，但会增加索引构建时间和内存占用。
- 提高
3.3. 使用混合搜索与重排（Hybrid Search & Reranking）
这是一种更高级的策略，通过结合多种信息来提升最终结果的准确性。
- 两阶段搜索：
- 召回阶段：使用ANN索引快速召回一个较大的候选集（例如，您需要Top 10的结果，可以先召回Top 100或Top 200）。
- 重排阶段：对这个小规模的候选集（100或200个向量）进行精确的暴力计算，或者使用更复杂的重排模型（Reranker Model）进行排序，最后返回最精准的Top 10结果。
- 多向量混合搜索：如果您的数据可以使用不同模型生成多种向量（例如，一个向量代表颜色，一个向量代表纹理），Milvus 2.4版本以上支持多向量搜索，可以综合多个向量字段的结果进行重排，显著提升召回率。
代码示例：
3.4. 范围搜索（Range Search）
如果您关心的不是返回“前K个最相似的”，而是返回“所有相似度高于某个阈值的”，可以使用范围搜索。这种方式可以根据您设定的距离范围来确保所有符合条件的结果都被召回。
3.5 代码实践
仅针对3和4撰写测试代码，供参考，可能存在错误：
import numpy as np
import asyncio
from typing import List, Dict
import logging
import uuid
import time
# --- 配置 ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
DIMENSION = 128 # 为方便演示，降低维度
TOTAL_VECTORS = 2000 # 模拟数据总量
SIMILAR_COUNT = 5 # 制造5个与基准相似的向量
# --- 向量工具函数 ---
def normalize_l2(vectors: np.ndarray) -> np.ndarray:
"""对Numpy向量或向量数组进行L2归一化"""
if vectors.ndim == 1:
norm = np.linalg.norm(vectors)
return vectors / max(norm, 1e-10)
else:
norms = np.linalg.norm(vectors, axis=1, keepdims=True)
norms = np.maximum(norms, 1e-10)
return vectors / norms
# --- 步骤 1: 构造模拟数据 ---
def generate_mock_data() -> List[Dict]:
"""
生成包含已知相似簇的模拟数据
"""
logger.info("开始构造模拟数据...")
# a. 创建一个归一化的基准向量
base_vector = normalize_l2(np.random.random(DIMENSION).astype(np.float32))
mock_data = []
# b. 制造 SIMILAR_COUNT 个与基准相似的向量
for i in range(SIMILAR_COUNT):
# 添加微小噪声
noise = np.random.random(DIMENSION).astype(np.float32) * 0.05
similar_vector = normalize_l2(base_vector + noise)
mock_data.append({
"id": str(uuid.uuid4()),
"image_name": f"similar_image_{i+1}.jpg",
"image_path": f"/path/to/similar_{i+1}.jpg",
"vector": similar_vector,
"label": "similar_group"
})
# c. 制造大量随机的背景噪声向量
for i in range(TOTAL_VECTORS - SIMILAR_COUNT):
random_vector = normalize_l2(np.random.random(DIMENSION).astype(np.float32))
mock_data.append({
"id": str(uuid.uuid4()),
"image_name": f"random_image_{i+1}.jpg",
"image_path": f"/path/to/random_{i+1}.jpg",
"vector": random_vector,
"label": f"random_group_{i % 10}"
})
logger.info(f"数据构造完成，共 {len(mock_data)} 条记录。")
return mock_data, base_vector
# --- 步骤 2: 模拟 Milvus 环境 ---
class MockMilvus:
"""一个简单的内存数据库，用于模拟Milvus的行为"""
def __init__(self, data: List[Dict]):
self._data = {item['id']: item for item in data}
self._vectors = np.array([item['vector'] for item in data])
self._ids = [item['id'] for item in data]
logger.info("模拟Milvus环境已就绪，数据已加载到内存。")
def search(self, query_vector: np.ndarray, limit: int, params: Dict) -> List[Dict]:
"""模拟 ANN search，返回的结果顺序可能不精确"""
logger.info(f"[模拟Search] 参数: limit={limit}, params={params}")
# 精确计算所有向量的相似度
scores = np.dot(self._vectors, query_vector)
# 模拟 ANN 的不确定性：打乱前 30% 的结果顺序
top_30_percent_idx = int(len(scores) * 0.3)
top_indices = np.argsort(-scores) # 获取降序排序的索引
# 打乱前30%的索引
shuffled_top_part = top_indices[:top_30_percent_idx]
np.random.shuffle(shuffled_top_part)
top_indices[:top_30_percent_idx] = shuffled_top_part
# 范围搜索逻辑
if 'range_filter' in params:
radius = params.get('radius', 1.0)
range_filter = params.get('range_filter', 0.0)
# 筛选出在范围内的
indices_in_range = [i for i in top_indices if range_filter <= scores[i] <= radius]
result_indices = indices_in_range[:limit]
else:
result_indices = top_indices[:limit]
# 返回模拟的 Milvus Hit 对象
results = [{"id": self._ids[i], "distance": scores[i]} for i in result_indices]
return [results]
def query(self, expr: str, output_fields: List[str]) -> List[Dict]:
"""模拟 query by id"""
# 这是一个简化的解析，只处理 "id in [...]" 的情况
try:
id_list_str = expr.split(' in ')[1].strip('[]')
id_list = [s.strip().strip("'\"") for s in id_list_str.split(',')]
results = [self._data[id_] for id_ in id_list if id_ in self._data]
logger.info(f"[模拟Query] 表达式 '{expr[:50]}...' 命中 {len(results)} 条记录。")
return results
except Exception as e:
logger.error(f"模拟Query解析失败: {e}")
return []
# --- 步骤 3: 实现高级搜索逻辑 ---
class AdvancedSearcher:
def __init__(self, mock_db: MockMilvus):
self.db = mock_db
async def hybrid_search_with_rerank(self, query_vec, final_top_k=10, recall_multiplier=10):
recall_top_k = final_top_k * recall_multiplier
logger.info(f"\n--- [混合搜索] 开始 ---")
logger.info(f"阶段 1: ANN 召回 Top {recall_top_k}...")
# 1. 召回
recall_results = self.db.search(query_vec, limit=recall_top_k, params={"nprobe": 64})
candidate_hits = recall_results[0]
logger.info(f"召回了 {len(candidate_hits)} 个候选者。")
# 打印召回结果的前几个，展示其顺序可能不精确
print("召回结果（前5，顺序可能不精确）:")
for hit in candidate_hits[:5]:
print(f" ID: ...{hit['id'][-12:]}, ANN分数: {hit['distance']:.4f}")
# 2. 重排
logger.info("阶段 2: 精确重排...")
candidate_ids = [hit['id'] for hit in candidate_hits]
candidate_entities = self.db.query(f"id in {candidate_ids}", output_fields=['*'])
reranked_candidates = []
for entity in candidate_entities:
exact_score = np.dot(np.array(entity['vector']), query_vec)
entity['score'] = float(exact_score)
reranked_candidates.append(entity)
reranked_candidates.sort(key=lambda x: x['score'], reverse=True)
logger.info("重排完成。")
return reranked_candidates[:final_top_k]
async def range_search_with_rerank(self, query_vec, threshold=0.9, limit=100):
logger.info(f"\n--- [范围搜索] 开始 (阈值 > {threshold}) ---")
logger.info(f"阶段 1: 范围召回...")
# 1. 范围召回
recall_results = self.db.search(query_vec, limit=limit, params={"radius": 1.0, "range_filter": threshold})
candidate_hits = recall_results[0]
logger.info(f"范围召回了 {len(candidate_hits)} 个候选者。")
# 2. 重排
logger.info("阶段 2: 精确重排...")
candidate_ids = [hit['id'] for hit in candidate_hits]
candidate_entities = self.db.query(f"id in {candidate_ids}", output_fields=['*'])
reranked_candidates = []
for entity in candidate_entities:
exact_score = np.dot(np.array(entity['vector']), query_vec)
if exact_score >= threshold:
entity['score'] = float(exact_score)
reranked_candidates.append(entity)
reranked_candidates.sort(key=lambda x: x['score'], reverse=True)
logger.info("重排完成。")
return reranked_candidates
# --- 步骤 4: 执行与结果展示 ---
async def main():
# 1. 构造数据和查询向量
mock_data, base_vector = generate_mock_data()
# 构造一个与基准向量极其相似的查询向量
query_vector = normalize_l2(base_vector + np.random.random(DIMENSION).astype(np.float32) * 0.01)
# 2. 初始化模拟环境和搜索器
mock_db = MockMilvus(mock_data)
searcher = AdvancedSearcher(mock_db)
# 3. 执行混合搜索
final_hybrid_results = await searcher.hybrid_search_with_rerank(query_vector, final_top_k=5, recall_multiplier=10)
print("\n✅ 混合搜索与重排的最终结果 (Top 5):")
for i, result in enumerate(final_hybrid_results):
is_truly_similar = "similar_image" in result['image_name']
print(f" {i+1}. 名称: {result['image_name']:<20} | 精确分数: {result['score']:.4f} | 是否为已知相似项: {is_truly_similar}")
# 4. 执行范围搜索
similarity_threshold = 0.98 # 设置一个较高的阈值，理论上只有我们制造的相似向量能满足
final_range_results = await searcher.range_search_with_rerank(query_vector, threshold=similarity_threshold)
print(f"\n✅ 范围搜索与重排的最终结果 (相似度 > {similarity_threshold}):")
for i, result in enumerate(final_range_results):
is_truly_similar = "similar_image" in result['image_name']
print(f" {i+1}. 名称: {result['image_name']:<20} | 精确分数: {result['score']:.4f} | 是否为已知相似项: {is_truly_similar}")
# 验证：检查是否所有已知的相似项都被找到了
found_similar_count = sum(1 for r in final_range_results if "similar_image" in r['image_name'])
print(f"\n验证: 在范围搜索中找到了 {found_similar_count} / {SIMILAR_COUNT} 个已知的相似项。")
if __name__ == "__main__":
asyncio.run(main())
你会看到类似的输出：
INFO:__main__:开始构造模拟数据...
INFO:__main__:数据构造完成，共 2000 条记录。
INFO:__main__:模拟Milvus环境已就绪，数据已加载到内存。
--- [混合搜索] 开始 ---
INFO:__main__:阶段 1: ANN 召回 Top 50...
INFO:__main__:[模拟Search] 参数: limit=50, params={'nprobe': 64}
INFO:__main__:召回了 50 个候选者。
召回结果（前5，顺序可能不精确）:
ID: ...e4d8c83a0b4c, ANN分数: 0.9317
ID: ...a9e0f13e1a3e, ANN分数: 0.9998
ID: ...7a2b97c234a4, ANN分数: 0.9421
ID: ...a59a729e8c4e, ANN分数: 0.9997
ID: ...3f4b5a2d1e9f, ANN分数: 0.9288
INFO:__main__:阶段 2: 精确重排...
INFO:__main__:[模拟Query] 表达式 'id in ['e4d8c83a-0b4c-4c...' 命中 50 条记录。
INFO:__main__:重排完成。
✅ 混合搜索与重排的最终结果 (Top 5):
1. 名称: similar_image_2.jpg | 精确分数: 0.9999 | 是否为已知相似项: True
2. 名称: similar_image_4.jpg | 精确分数: 0.9998 | 是否为已知相似项: True
3. 名称: similar_image_1.jpg | 精确分数: 0.9998 | 是否为已知相似项: True
4. 名称: similar_image_5.jpg | 精确分数: 0.9997 | 是否为已知相似项: True
5. 名称: similar_image_3.jpg | 精确分数: 0.9997 | 是否为已知相似项: True
--- [范围搜索] 开始 (阈值 > 0.98) ---
INFO:__main__:阶段 1: 范围召回...
INFO:__main__:[模拟Search] 参数: limit=100, params={'radius': 1.0, 'range_filter': 0.98}
INFO:__main__:范围召回了 5 个候选者。
INFO:__main__:阶段 2: 精确重排...
INFO:__main__:[模拟Query] 表达式 'id in ['a9e0f13e-1a3e-4d...' 命中 5 条记录。
INFO:__main__:重排完成。
✅ 范围搜索与重排的最终结果 (相似度 > 0.98):
1. 名称: similar_image_2.jpg | 精确分数: 0.9999 | 是否为已知相似项: True
2. 名称: similar_image_4.jpg | 精确分数: 0.9998 | 是否为已知相似项: True
3. 名称: similar_image_1.jpg | 精确分数: 0.9998 | 是否为已知相似项: True
4. 名称: similar_image_5.jpg | 精确分数: 0.9997 | 是否为已知相似项: True
5. 名称: similar_image_3.jpg | 精确分数: 0.9997 | 是否为已知相似项: True
验证: 在范围搜索中找到了 5 / 5 个已知的相似项。
4.总结与建议
- 理解权衡：首先要明确，在生产环境中，使用ANN搜索是在速度和100%准确率之间做出的必要权衡。
- 从搜索参数入手：对于您当前的问题，最简单的尝试就是逐步增大您搜索时的
nprobe（如果用IVF索引）或ef（如果用HNSW索引）参数，观察召回率是否能满足您的要求，同时监控查询延迟是否在可接受范围内。 - 评估索引：如果调整搜索参数效果不佳，可以考虑重新构建索引，选择更合适的索引类型和构建参数。
- 考虑重排：如果对结果的精度要求非常高，引入一个基于ANN的粗召回+精确重排的两阶段策略是业界的常用最佳实践。