import logging class ImageSearch: __logger = logging.getLogger(__name__) def __init__(self, feature_extractor, milvus): self.feature_extractor = feature_extractor self.milvus = milvus # id = product_id def image_to_image_search(self, image, key_name, top_k = 100): try: # 提取查询图像的特征 query_embedding = self.feature_extractor.extract_features(image) results = self.milvus.search(query_embedding, limit=top_k) # 处理结果 if not results or len(results) == 0: return [] # 返回结果 keys = [hit.entity.get(key_name) for hit in results[0]] scores = [hit.score for hit in results[0]] return list(zip(keys, scores)) except Exception as e: self.__logger.error(f"Image search failed: {e}") raise ValueError(f"Image search failed: {e}")