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，倒排文件）索引為例：

1. 構建索引時：Milvus會將所有向量分成nlist個“簇”（Cluster），每個簇有一個中心點。
2. 查詢時：Milvus會先計算您的查詢向量與所有nlist個簇中心的距離，然后只選擇最接近的nprobe個簇。
3. 最終搜索：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）

這是一種更高級的策略，通過結合多種信息來提升最終結果的準確性。

• 兩階段搜索：
  1. 召回階段：使用ANN索引快速召回一個較大的候選集（例如，您需要Top 10的結果，可以先召回Top 100或Top 200）。
  2. 重排階段：對這個小規模的候選集（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 resultsexcept Exception as e:logger.error(f"模擬Query解析失敗: {e}")return []# --- 步驟 3: 實現高級搜索邏輯 ---
class AdvancedSearcher:def __init__(self, mock_db: MockMilvus):self.db = mock_dbasync def hybrid_search_with_rerank(self, query_vec, final_top_k=10, recall_multiplier=10):recall_top_k = final_top_k * recall_multiplierlogger.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.9317ID: ...a9e0f13e1a3e, ANN分數: 0.9998ID: ...7a2b97c234a4, ANN分數: 0.9421ID: ...a59a729e8c4e, ANN分數: 0.9997ID: ...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 | 是否為已知相似項: True2. 名稱: similar_image_4.jpg  | 精確分數: 0.9998 | 是否為已知相似項: True3. 名稱: similar_image_1.jpg  | 精確分數: 0.9998 | 是否為已知相似項: True4. 名稱: similar_image_5.jpg  | 精確分數: 0.9997 | 是否為已知相似項: True5. 名稱: 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 | 是否為已知相似項: True2. 名稱: similar_image_4.jpg  | 精確分數: 0.9998 | 是否為已知相似項: True3. 名稱: similar_image_1.jpg  | 精確分數: 0.9998 | 是否為已知相似項: True4. 名稱: similar_image_5.jpg  | 精確分數: 0.9997 | 是否為已知相似項: True5. 名稱: similar_image_3.jpg  | 精確分數: 0.9997 | 是否為已知相似項: True驗證: 在范圍搜索中找到了 5 / 5 個已知的相似項。

4.總結與建議

• 理解權衡：首先要明確，在生產環境中，使用ANN搜索是在速度和100%準確率之間做出的必要權衡。
• 從搜索參數入手：對于您當前的問題，最簡單的嘗試就是逐步增大您搜索時的nprobe（如果用IVF索引）或ef（如果用HNSW索引）參數，觀察召回率是否能滿足您的要求，同時監控查詢延遲是否在可接受范圍內。
• 評估索引：如果調整搜索參數效果不佳，可以考慮重新構建索引，選擇更合適的索引類型和構建參數。
• 考慮重排：如果對結果的精度要求非常高，引入一個基于ANN的粗召回+精確重排的兩階段策略是業界的常用最佳實踐。