使用 Milvus 實現高效圖片查重功能

本文將介紹如何利用 Milvus 向量數據庫構建一個高效的圖片查重系統，通過傳入圖片就能快速從已有數據中找出匹配度高的相似圖片。

一.什么是圖片查重？

圖片查重指的是通過算法識別出內容相同或高度相似的圖片，即使它們可能存在尺寸、格式、輕微編輯等差異。傳統的基于文件名或元數據的查重方法效果極差，而基于內容的圖片查重則能真正識別視覺上相似的圖片。

二. 技術原理

基于 Milvus 的圖片查重系統主要依賴以下關鍵技術：

1. 圖片特征提取：使用深度學習模型將圖片轉換為固定維度的特征向量，捕捉圖片的視覺特征
2. 向量相似度搜索：通過計算向量之間的距離（相似度）來判斷圖片的相似程度
3. 高效向量數據庫：Milvus 提供的高性能向量索引和搜索能力，支持億級數據的毫秒級檢索

核心流程：

• 預處理：將所有圖片轉換為特征向量并存儲到 Milvus
• 查重階段：對輸入圖片提取特征向量，在 Milvus 中搜索相似度高于閾值的向量，找到對應的圖片

三 .實現步驟

1. 核心代碼實現

下面是完整的圖片查重系統實現，包含特征提取和 Milvus 操作：

2. 代碼解析

核心組件

1. 圖片特征提取器（ImageFeatureExtractor）：

class ImageFeatureExtractor:"""圖片特征提取器，將圖片轉換為特征向量"""def __init__(self):# 使用預訓練的ResNet50模型self.model = models.resnet50(pretrained=True)# 移除最后一層全連接層，保留特征提取部分self.model = torch.nn.Sequential(*list(self.model.children())[:-1])self.model.eval()  # 切換到評估模式# 確保使用適當的設備（GPU如果可用）self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")self.model.to(self.device)# 圖片預處理流程，與模型訓練時保持一致self.transform = transforms.Compose([transforms.Resize(256),transforms.CenterCrop(224),transforms.ToTensor(),transforms.Normalize(mean=[0.485, 0.456, 0.406],std=[0.229, 0.224, 0.225])])logger.info(f"特征提取器初始化完成，使用設備: {self.device}")def extract(self, image_path):"""提取單張圖片的特征向量"""try:# 打開圖片并轉換為RGB格式img = Image.open(image_path).convert('RGB')# 預處理img_tensor = self.transform(img).unsqueeze(0)  # 添加批次維度img_tensor = img_tensor.to(self.device)# 提取特征with torch.no_grad():  # 不計算梯度，加速推理features = self.model(img_tensor)# 展平為一維向量并歸一化feature_vector = features.squeeze().cpu().numpy()normalized_vector = feature_vector / np.linalg.norm(feature_vector)return normalized_vectorexcept Exception as e:logger.error(f"提取圖片 {image_path} 特征失敗: {str(e)}")return Nonedef batch_extract(self, image_paths):"""批量提取圖片特征"""features = []valid_paths = []for path in image_paths:feat = self.extract(path)if feat is not None:features.append(feat)valid_paths.append(path)return valid_paths, features

• 使用預訓練的 ResNet50 模型提取圖片特征
• 對圖片進行標準化處理（Resize、裁剪、歸一化）
• 支持單張和批量圖片特征提取
• 自動選擇 GPU/CPU 設備加速處理

class MilvusImageChecker:"""基于Milvus的圖片查重工具"""def __init__(self, host=Config.MILVUS_HOST, port=Config.MILVUS_PORT, collection_name=Config.COLLECTION_NAME):self.host = hostself.port = portself.collection_name = collection_nameself.collection = None# 連接Milvus并初始化集合self.connect()self.init_collection()def connect(self):"""連接到Milvus服務器"""try:connections.connect(alias="default",host=self.host,port=self.port)logger.info(f"成功連接到Milvus服務器: {self.host}:{self.port}")except Exception as e:logger.error(f"連接Milvus服務器失敗: {str(e)}")raisedef init_collection(self):"""初始化集合，如不存在則創建"""if utility.has_collection(self.collection_name):self.collection = Collection(self.collection_name)logger.info(f"已加載集合: {self.collection_name}")return# 定義集合字段fields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),FieldSchema(name="image_path", dtype=DataType.VARCHAR, max_length=512),FieldSchema(name="feature_vector", dtype=DataType.FLOAT_VECTOR, dim=Config.VECTOR_DIM),FieldSchema(name="upload_time", dtype=DataType.INT64, description="圖片上傳時間戳，用于去重時保留最早/最新版本")]# 創建集合schemaschema = CollectionSchema(fields, "用于圖片查重的集合，存儲圖片路徑和特征向量",enable_dynamic_field=False)# 創建集合self.collection = Collection(self.collection_name, schema)logger.info(f"成功創建集合: {self.collection_name}")# 創建索引，優化查詢性能index_params = {"index_type": "HNSW",  # 適用于高維向量的高效索引"metric_type": "IP",   # 內積，適用于歸一化向量的相似度計算"params": {"M": 16,           # HNSW參數，影響索引質量和查詢速度"efConstruction": 200  # 構建索引時的參數}}self.collection.create_index(field_name="feature_vector",index_params=index_params)logger.info("集合索引創建完成")def insert_images(self, image_paths, upload_timestamps=None):"""插入圖片特征到Milvus參數:image_paths: 圖片路徑列表upload_timestamps: 上傳時間戳列表，用于去重時判斷保留哪個版本"""if not image_paths:logger.warning("沒有圖片路徑可插入")return []# 處理時間戳（默認使用當前時間）if upload_timestamps is None:current_ts = int(os.path.getmtime(image_paths[0])) if image_paths else 0upload_timestamps = [current_ts] * len(image_paths)# 批量提取特征extractor = ImageFeatureExtractor()valid_paths, features = extractor.batch_extract(image_paths)if not valid_paths:logger.warning("沒有有效圖片可插入")return []# 準備插入數據data = [valid_paths,  # image_path字段[f.tolist() for f in features],  # feature_vector字段upload_timestamps[:len(valid_paths)]  # upload_time字段]# 執行插入try:insert_result = self.collection.insert(data)self.collection.flush()  # 刷新到磁盤logger.info(f"成功插入 {len(valid_paths)} 張圖片，ID范圍: {insert_result.primary_keys}")return insert_result.primary_keysexcept Exception as e:logger.error(f"插入圖片失敗: {str(e)}")return []def check_duplicates(self, image_path, threshold=Config.DEFAULT_THRESHOLD, top_k=Config.DEFAULT_TOP_K):"""檢查指定圖片是否存在重復或高度相似的圖片參數:image_path: 待檢查的圖片路徑threshold: 相似度閾值，高于此值認為是相似圖片top_k: 返回的最大相似圖片數量返回:相似圖片列表，按相似度降序排列"""# 提取查詢圖片特征extractor = ImageFeatureExtractor()query_vector = extractor.extract(image_path)if query_vector is None:logger.error("無法提取查詢圖片特征，查重失敗")return []# 加載集合到內存（如果尚未加載）if not self.collection.is_loaded:self.collection.load()logger.info(f"集合 {self.collection_name} 已加載到內存")# 配置搜索參數search_params = {"metric_type": "IP","params": {"ef": 64}  # 搜索時的參數，影響查詢精度和速度}# 執行相似度搜索try:results = self.collection.search(data=[query_vector.tolist()],anns_field="feature_vector",param=search_params,limit=top_k,output_fields=["image_path", "upload_time"])# 處理搜索結果，過濾低于閾值的結果duplicates = []for hits in results:for hit in hits:similarity = hit.distanceif similarity >= threshold:duplicates.append({"image_path": hit.entity.get("image_path"),"similarity": float(similarity),"milvus_id": hit.id,"upload_time": hit.entity.get("upload_time")})# 按相似度降序排序duplicates.sort(key=lambda x: x["similarity"], reverse=True)logger.info(f"找到 {len(duplicates)} 張相似圖片 (閾值: {threshold})")return duplicatesexcept Exception as e:logger.error(f"查重搜索失敗: {str(e)}")return []def delete_duplicates(self, duplicate_ids):"""刪除指定ID的重復圖片記錄"""if not duplicate_ids:return Truetry:self.collection.delete(f"id in {duplicate_ids}")self.collection.flush()logger.info(f"成功刪除 {len(duplicate_ids)} 條重復記錄")return Trueexcept Exception as e:logger.error(f"刪除重復記錄失敗: {str(e)}")return False

3. Milvus 查重工具（MilvusImageChecker）：
   • 負責與 Milvus 服務器的連接和交互
   • 初始化集合并創建高效索引（使用 HNSW 索引）
   • 提供圖片特征插入、重復圖片查詢和刪除功能
   • 支持批量操作，提高處理效率

關鍵技術點

• 特征向量歸一化：確保內積（IP）可以直接作為余弦相似度使用
• 合適的索引選擇：使用 HNSW 索引平衡查詢速度和精度
• 相似度閾值：可根據業務需求調整，值越高表示要求越相似
• 時間戳管理：記錄圖片上傳時間，便于去重時選擇保留最早或最新版本

五 .重點

1. 模型選擇：

   • 追求精度：可使用更復雜的模型如 ResNet101 或 Vision Transformer
   • 追求速度：可使用輕量級模型如 MobileNet 或 EfficientNet

2. 索引優化：

   • 對于大規模數據，可調整 HNSW 索引的 M 和 efConstruction 參數
   • 可嘗試不同索引類型（如 IVF_FLAT、IVF_SQ8）找到性能平衡點

3. 閾值調整：

   • 對于嚴格查重（完全相同的圖片），可將閾值設為 0.98 以上
   • 對于相似圖片檢索，可將閾值設為 0.85-0.95 之間

4. 分布式部署：

   • 對于超大規模圖片庫，可使用 Milvus 集群提高吞吐量和可靠性

六 應用場景

• 內容管理系統：自動檢測并去重上傳的圖片
• 電商平臺：識別盜圖和相似商品圖片
• 版權保護：追蹤未經授權使用的圖片
• 相冊管理：自動整理相似照片，減少冗余

總結

基于 Milvus 的圖片查重系統能夠高效處理海量圖片數據，通過特征向量和相似度搜索技術，實現了精準的重復圖片識別。相比傳統方法，它具有以下優勢：

• 識別真正視覺相似的圖片，不受文件名或格式影響
• 支持億級圖片的快速檢索，毫秒級響應
• 可靈活調整相似度閾值，適應不同業務需求
• 易于擴展和集成到現有系統中