1688 图片搜索逆向工程与多模态搜索融合实践 —— 基于 CLIP 模型的特征向量落地

{  "imageUrl": "aHR0cHM6Ly9...",  // 图片URL或base64编码  "similarityThreshold": 0.75,   // 相似度阈值，低于此值的结果会过滤  "searchScene": "reverseImageSearch",  // 固定场景值  "clientVersion": "5.12.0"      // 客户端版本，低版本可能被拒绝}

// 原算法逆向还原（已做格式调整）function generateSign(timestamp, deviceId) {  // 拼接密钥前缀+时间戳+设备ID前8位  const rawStr = `Alibaba_${timestamp}_${deviceId.slice(0,8)}`;  // MD5加密后取中间16位  return md5(rawStr).slice(8,24);}

import hashlibimport timeimport uuiddef generate_sign():    timestamp = str(int(time.time()))    # 模拟设备ID，实际应从合法渠道获取    device_id = str(uuid.uuid4()).replace("-", "")[:8]    raw_str = f"Alibaba_{timestamp}_{device_id}"    return hashlib.md5(raw_str.encode()).hexdigest()[8:24], timestamp, device_id

from PIL import Imageimport clipimport torch# 加载模型（首次运行会自动下载）model, preprocess = clip.load("ViT-B/32", device="cuda" if torch.cuda.is_available() else "cpu")def extract_image_features(img_path):    """提取图片特征向量"""    image = preprocess(Image.open(img_path)).unsqueeze(0).to(device)    with torch.no_grad():        # 生成512维特征向量并归一化        features = model.encode_image(image).numpy()[0]        return features / (features ** 2).sum() ** 0.5def extract_text_features(text):    """提取文本特征向量（用于后续图文融合）"""    tokens = clip.tokenize([text]).to(device)    with torch.no_grad():        features = model.encode_text(tokens).numpy()[0]        return features / (features ** 2).sum() ** 0.5

import faissclass FeatureIndexer:    def __init__(self, dimension=512):        # 构建内积索引（归一化后等价于余弦相似度）        self.index = faiss.IndexFlatIP(dimension)        # 预加载商品特征库（实际项目中可持久化到磁盘）        self.product_ids = []        self.load_product_features()    def load_product_features(self):        """加载商品特征（图片+标题融合特征）"""        # 实际项目中从数据库读取商品数据        products = get_1688_products()  # 自定义函数：获取商品列表        for product in products:            img_feat = extract_image_features(product["img_url"])            text_feat = extract_text_features(product["title"])            # 图文特征融合（权重可根据场景调整）            fused_feat = 0.7 * img_feat + 0.3 * text_feat            self.index.add(fused_feat.reshape(1, -1))            self.product_ids.append(product["id"])    def search(self, query_feat, top_k=10):        """搜索最相似的商品"""        # D：相似度距离，I：索引位置        D, I = self.index.search(query_feat.reshape(1, -1), top_k)        # 映射回商品ID并过滤低相似度结果        return [            {"product_id": self.product_ids[i], "similarity": float(D[0][j])}            for j, i in enumerate(I[0]) if float(D[0][j]) >= 0.75        ]

import requestsdef clip_based_image_search(img_path):    # 1. 提取图片特征    query_feat = extract_image_features(img_path)        # 2. 生成签名与请求头    x_sign, timestamp, device_id = generate_sign()    headers = {        "x-sign": x_sign,        "x-version": "5.12.0",        "x-device-id": device_id,        "Content-Type": "application/json"    }        # 3. 构造请求体（融合CLIP特征与原生参数）    payload = {        "embedding": query_feat.tolist(),        "searchType": "vector",  # 自定义参数：启用向量搜索        "similarityThreshold": 0.75,        "clientVersion": "5.12.0"    }        # 4. 发送请求并处理结果    response = requests.post(        "https://api.1688.com/image-search/v1/search",        json=payload,        headers=headers,        timeout=10    )        # 5. 用本地索引二次筛选（提升准确率）    raw_items = response.json()["result"]["items"]    indexed_results = feature_indexer.search(query_feat)    indexed_ids = {item["product_id"] for item in indexed_results}        # 返回交集结果（兼顾原生准确性与自定义需求）    return [item for item in raw_items if item["productId"] in indexed_ids]