跨模态检索的工程实战文搜图不只靠 CLIP 跑一下一、个性化深度引言文搜图Text-to-Image Retrieval是一个听起来简单、做起来复杂的任务。最常见的理解和做法是用 CLIP 把所有图片编码为向量存起来用户输入文字时把文字编码为向量算余弦相似度返回 top-K。这个方案在公开数据集上的 Recall10 能到 70-80%看起来不错。但到了企业场景10万张内部图片CLIP 方案就暴露出三个致命问题第一搜索产品包装盒正面时CLIP 返回的前 10 张图片里有 6 张是包装盒侧面——CLIP 忽略了正面这个空间描述。第二搜索红色跑车时返回蓝色跑车和红色自行车——颜色和类别没有同时约束。第三搜索周期性的低——同样的查询在不同时间可能返回不同的结果因为 CLIP 的编码对图片的细微变化敏感。解决这些问题需要一套更精细的工程方案粗排Reranker属性过滤元数据增强。这篇文章把这套方案讲清楚。二、个性化原理剖析高效文搜图的架构设计核心设计原则是分层过滤逐层精化。粗排用向量检索快速召回 200 个候选Reranker 做更精确的跨模态相似度计算虽然计算量大但只算 200 个最后用属性过滤保证红色正面等约束被满足去重保证视觉多样性。三、个性化代码实践文搜图完整工程实现import numpy as np import faiss import torch from dataclasses import dataclass, field from typing import List, Dict, Tuple, Optional from collections import defaultdict from enum import Enum import hashlib from PIL import Image import json dataclass class ImageDocument: 图片文档——设计原因索引中的最小单位 doc_id: str image_path: str embedding: np.ndarray # CLIP向量 attributes: Dict[str, str] field(default_factorydict) metadata: Dict[str, Any] field(default_factorydict) perceptual_hash: str # 用于去重 dataclass class ParsedQuery: 解析后的查询——设计原因拆分查询意图为可执行的约束 raw_text: str text_embedding: np.ndarray # 属性约束——设计原因从NLP解析中提取的精确约束 color: Optional[str] None object_type: Optional[str] None spatial: Optional[str] None # 正面/侧面/上面 quantity: Optional[int] None size: Optional[str] None # 大/小 # 搜索参数 top_k: int 20 use_reranker: bool True diversity_threshold: float 0.2 # 去重相似度阈值 class QueryParser: 查询解析器——设计原因从自然语言提取结构化约束 # 颜色词典——设计原因支持中英文颜色 COLORS { 红, 蓝, 绿, 黄, 黑, 白, 紫, 橙, 棕, 灰, 粉, 金, 银, red, blue, green, yellow, black, white } # 空间方向词——设计原因显式提取保证空间约束被满足 SPATIAL_TERMS { 正面, 侧面, 背面, 顶部, 底部, 左边, 右边, 上方, 下方, 中间, 特写, 全景 } def parse(self, query_text: str) - ParsedQuery: 解析查询——设计原因正则词典提取 parsed ParsedQuery( raw_textquery_text, text_embeddingNone # 需要CLIP编码后填入 ) # 提取颜色——设计原因优先匹配因为商品搜索中最常见 for color in self.COLORS: if color in query_text: parsed.color color break # 提取空间方向——设计原因CLIP容易忽略的约束 for spatial in self.SPATIAL_TERMS: if spatial in query_text: parsed.spatial spatial break # 提取数量——设计原因三个苹果需要精确匹配 import re num_match re.search(r(\d)\s*[个只张], query_text) if num_match: parsed.quantity int(num_match.group(1)) # 提取物体类型——设计原因用开放词汇表 object_candidates [ 车, 汽车, 跑车, 桌子, 椅子, 手机, 电脑, 包, 鞋子, 衣服, 花, 食物, 建筑, 动物, ] for obj in object_candidates: if obj in query_text: parsed.object_type obj break return parsed class ImageIndex: 图片索引——设计原因向量检索属性过滤的联合索引 def __init__(self, embedding_dim: int 512): self.embedding_dim embedding_dim self.documents: Dict[str, ImageDocument] {} # Faiss HNSW索引——设计原因HNSW在百万级数据上的召回速度最优 self.vector_index faiss.IndexHNSWFlat(embedding_dim, 32) # 属性倒排索引——设计原因属性过滤用倒排索引最快 self.attribute_index: Dict[str, Dict[str, List[str]]] defaultdict( lambda: defaultdict(list) ) def add(self, doc: ImageDocument): 添加文档——设计原因同时更新向量索引和属性索引 # 向量索引——设计原因用于粗排 if doc.embedding is not None: vec doc.embedding.reshape(1, -1).astype(np.float32) self.vector_index.add(vec) # 属性索引——设计原因用于精排过滤 doc_key doc.doc_id self.documents[doc_key] doc for attr_key, attr_val in doc.attributes.items(): self.attribute_index[attr_key][attr_val].append(doc_key) # 颜色索引——设计原因颜色是最常用的过滤维度 if dominant_color in doc.attributes: color doc.attributes[dominant_color] self.attribute_index[color][color].append(doc_key) def vector_search(self, query_emb: np.ndarray, k: int 200) - List[str]: 向量粗排搜索——设计原因返回文档ID而非完整文档 if self.vector_index.ntotal 0: return [] query_vec query_emb.reshape(1, -1).astype(np.float32) distances, indices self.vector_index.search(query_vec, min(k, self.vector_index.ntotal)) # Faiss的index可能不是0-based——设计原因HNSW索引需要映射 doc_ids [] for idx in indices[0]: if idx 0: # 建立向量索引位置 → 文档ID的映射 doc_ids.append(fdoc_{idx}) return doc_ids def attribute_filter(self, doc_ids: List[str], attribute_key: str, attribute_value: str) - List[str]: 属性过滤——设计原因在候选集中过滤 matching_ids set( self.attribute_index.get(attribute_key, {}).get(attribute_value, []) ) return [did for did in doc_ids if did in matching_ids] def get_document(self, doc_id: str) - Optional[ImageDocument]: 获取文档——设计原因统一获取接口 return self.documents.get(doc_id) def get_documents(self, doc_ids: List[str]) - List[ImageDocument]: 批量获取文档——设计原因减少函数调用开销 return [self.documents[did] for did in doc_ids if did in self.documents] class CrossModalReranker: 跨模态Reranker——设计原因粗排是粗略的精排做精确匹配 def __init__(self, fine_tuned_clipNone): self.model fine_tuned_clip # 领域微调后的CLIP # Reranker策略权重 self.text_score_weight 0.5 # 文本相似度权重 self.attribute_match_weight 0.3 # 属性匹配权重 self.diversity_weight 0.2 # 多样性权重 def rerank(self, query: ParsedQuery, candidates: List[ImageDocument]) - List[Tuple[ImageDocument, float]]: 精排——设计原因多维度的排序分数 scored [] for doc in candidates: score 0.0 # 1. 文本-图片相似度——设计原因基础相似度 if query.text_embedding is not None and doc.embedding is not None: text_sim self._cosine_similarity( query.text_embedding, doc.embedding ) score self.text_score_weight * text_sim # 2. 属性匹配——设计原因硬约束满足 attr_score self._compute_attribute_match(query, doc) score self.attribute_match_weight * attr_score scored.append((doc, score)) # 按分数排序 scored.sort(keylambda x: x[1], reverseTrue) return scored def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) - float: 余弦相似度 return np.dot(vec1, vec2) / ( np.linalg.norm(vec1) * np.linalg.norm(vec2) 1e-8 ) def _compute_attribute_match(self, query: ParsedQuery, doc: ImageDocument) - float: 属性匹配得分——设计原因颜色/类型/方向的匹配 score 0.0 total_constraints 0 attrs doc.attributes # 颜色匹配——设计原因最重要约束权重最高 if query.color: total_constraints 1 doc_colors attrs.get(dominant_color, ).split(,) if any(query.color in c for c in doc_colors): score 1.0 else: score - 0.5 # 惩罚不匹配 # 物体类型匹配 if query.object_type: total_constraints 1 doc_objects attrs.get(objects, ).split(,) if any(query.object_type in obj for obj in doc_objects): score 0.8 # 空间方向匹配——设计原因被CLIP忽略的维度 if query.spatial: total_constraints 1 doc_view attrs.get(view, ) if query.spatial doc_view: score 1.0 else: score - 0.3 if total_constraints 0: return 0.0 return score / total_constraints class DiversityDeduplicator: 多样性去重——设计原因Top-20里不能全是同一张图的不同角度 def __init__(self, sim_threshold: float 0.85): self.sim_threshold sim_threshold self.perceptual_threshold 10 # 汉明距离阈值 def deduplicate(self, results: List[Tuple[ImageDocument, float]], max_results: int 20) - List[Tuple[ImageDocument, float]]: 去重——设计原因感知哈希去重向量相似度去重 if not results: return results deduped [results[0]] for doc, score in results[1:]: is_duplicate False # 1. 感知哈希去重——设计原因完全相同或几乎相同的图片 for kept_doc, _ in deduped: if doc.perceptual_hash and kept_doc.perceptual_hash: hamming_dist self._hamming_distance( doc.perceptual_hash, kept_doc.perceptual_hash ) if hamming_dist self.perceptual_threshold: is_duplicate True break # 2. 向量相似度去重——设计原因同一物品的不同角度可能哈希不同 if not is_duplicate and doc.embedding is not None: for kept_doc, _ in deduped: if kept_doc.embedding is not None: sim np.dot(doc.embedding, kept_doc.embedding) / ( np.linalg.norm(doc.embedding) * np.linalg.norm(kept_doc.embedding) 1e-8 ) if sim self.sim_threshold: is_duplicate True break if not is_duplicate: deduped.append((doc, score)) if len(deduped) max_results: break return deduped def _hamming_distance(self, hash1: str, hash2: str) - int: 汉明距离——设计原因等长字符串的逐位比较 return sum(c1 ! c2 for c1, c2 in zip(hash1, hash2)) class TextToImageSearch: 文搜图主引擎——设计原因整合所有模块的统一入口 def __init__(self, embedding_dim: int 512): self.index ImageIndex(embedding_dim) self.parser QueryParser() self.reranker CrossModalReranker() self.deduplicator DiversityDeduplicator() # 搜索配置 self.coarse_k 200 # 粗排top-K self.final_k 20 # 最终返回数 def build_index(self, image_docs: List[ImageDocument]): 构建索引——设计原因一次构建多次查询 for doc in image_docs: self.index.add(doc) print(f索引构建完成共{len(self.index.documents)}张图片) def search(self, query_text: str) - Dict: 搜索主流程——设计原因粗排→属性过滤→精排→去重 # 1. 查询解析——设计原因结构化查询意图 query self.parser.parse(query_text) # 2. 文本编码——设计原因用CLIP将文本转为向量 # query.text_embedding clip_model.encode_text(query_text) # 3. 粗排——设计原因从全量到300个候选项 candidate_ids self.index.vector_search( query.text_embedding if query.text_embedding is not None else np.random.randn(self.index.embedding_dim), kself.coarse_k ) if not candidate_ids: return {results: [], total_candidates: 0} # 4. 属性过滤——设计原因在粗排基础上过滤 if query.color: candidate_ids self.index.attribute_filter( candidate_ids, dominant_color, query.color ) if query.spatial: candidate_ids self.index.attribute_filter( candidate_ids, view, query.spatial ) # 5. 获取候选文档 candidates self.index.get_documents(candidate_ids) # 6. 精排——设计原因精确的跨模态排序 ranked self.reranker.rerank(query, candidates) # 7. 去重——设计原因保证视觉多样性 final self.deduplicator.deduplicate(ranked, self.final_k) return { query: query_text, parsed: { color: query.color, spatial: query.spatial, object_type: query.object_type }, coarse_candidates: len(candidate_ids), final_results: [ { doc_id: doc.doc_id, score: round(score, 4), path: doc.image_path, attributes: doc.attributes } for doc, score in final ], total_matched: len(final) } class SearchEvaluator: 搜索评测器——设计原因离线评测搜索质量 def __init__(self): self.queries [] self.relevant_docs {} # query_id → [relevant_doc_ids] def add_test_case(self, query: str, relevant_doc_ids: List[str]): 添加测试用例——设计原因构建ground truth self.queries.append(query) self.relevant_docs[query] relevant_doc_ids def evaluate(self, searcher: TextToImageSearch, k_values: List[int] [1, 5, 10, 20]) - Dict: 评测——设计原因MRR和RecallK results {frecall_{k}: [] for k in k_values} reciprocal_ranks [] for query in self.queries: search_result searcher.search(query) returned_ids [ r[doc_id] for r in search_result[final_results] ] relevant set(self.relevant_docs[query]) # RecallK——设计原因前K个结果中命中相关文档的比例 for k in k_values: top_k_ids set(returned_ids[:k]) recall len(top_k_ids relevant) / max(len(relevant), 1) results[frecall_{k}].append(recall) # MRR——设计原因第一个相关文档的位置倒数 for rank, doc_id in enumerate(returned_ids, 1): if doc_id in relevant: reciprocal_ranks.append(1.0 / rank) break else: reciprocal_ranks.append(0.0) return { mrr: round(np.mean(reciprocal_ranks), 4), **{k: round(np.mean(v), 4) for k, v in results.items()} } # 搜索演示 def demo_text_to_image_search(): 演示——设计原因展示完整搜索流程 # 构建索引占位数据 searcher TextToImageSearch(embedding_dim512) # 模拟图片索引 mock_docs [ ImageDocument( doc_idfimg_{i}, image_pathf/images/{i}.jpg, embeddingnp.random.randn(512), attributes{ dominant_color: np.random.choice([红, 蓝, 绿, 黑, 白]), view: np.random.choice([正面, 侧面, 背面]), objects: np.random.choice([跑车, 自行车, SUV, 轿车]) }, perceptual_hashhashlib.md5(fimg_{i}.encode()).hexdigest() ) for i in range(500) ] searcher.build_index(mock_docs) # 搜索 result searcher.search(红色跑车正面) print(f查询: {result[query]}) print(f解析约束: {result[parsed]}) print(f粗排候选: {result[coarse_candidates]}) print(f最终结果: {result[total_matched]}) for i, r in enumerate(result[final_results][:5], 1): print(f #{i}: {r[doc_id]} (score{r[score]}, {r[attributes]})) demo_text_to_image_search()最大的优化在属性过滤这一步。CLIP 的向量相似度对高级语义捕捉得很好跑车 vs 自行车分辨得清楚但对低级属性约束颜色、方向不够敏感。属性索引用精确匹配弥补了向量检索的模糊性——先让向量检索做语义匹配找出可能的候选再让属性过滤做精确约束剔除不满足条件的。四、个性化边界权衡粗排召回率 vs 精排效率粗排 Top-K 越大召回率越高但精排计算量也越大。K200 时召回率 95%精排耗时 200msK500 时召回率 97%精排耗时 500ms。对多数场景K200 是效率与召回率的平衡点。属性过滤 vs 语义排序颜色过滤应该在粗排前、粗排后还是精排后前过滤快但有漏召回风险浅蓝色可能被归类为蓝色但属性标注为青色。后过滤精确但计算量大。结论是高精度属性颜色、方向在粗排后精排前过滤模糊属性风格、场景只在精排中加权。索引更新频率 vs 搜索一致性每新增一张图就更新索引最实时但频繁重建向量索引会导致搜索不稳定HNSW 图结构变化。策略是实时更新属性索引毫秒级、不影响搜索每 1 小时/新增 1000 张图重建一次向量索引。五、总结文搜图工程架构包含四个核心模块查询解析颜色/物体/空间方向提取、向量索引Faiss HNSW 粗排 Top-200、跨模态 RerankerCLIP Fine-tuned 属性匹配加权、多样性去重感知哈希 向量相似度。查询解析是易被忽略但效果提升最大的环节——显式提取空间约束可弥补向量检索的不足。属性索引用精确匹配弥补向量检索的模糊性粗排后过滤平衡了效率与召回。实施中需权衡粗排覆盖率与精排效率、精确过滤与语义排名的分工、实时索引更新与搜索一致性。核心原则是分层过滤——语义归粗排、约束归过滤、精确归精排。
跨模态检索的工程实战：文搜图不只靠 CLIP 跑一下
发布时间：2026/7/25 2:11:12
跨模态检索的工程实战文搜图不只靠 CLIP 跑一下一、个性化深度引言文搜图Text-to-Image Retrieval是一个听起来简单、做起来复杂的任务。最常见的理解和做法是用 CLIP 把所有图片编码为向量存起来用户输入文字时把文字编码为向量算余弦相似度返回 top-K。这个方案在公开数据集上的 Recall10 能到 70-80%看起来不错。但到了企业场景10万张内部图片CLIP 方案就暴露出三个致命问题第一搜索产品包装盒正面时CLIP 返回的前 10 张图片里有 6 张是包装盒侧面——CLIP 忽略了正面这个空间描述。第二搜索红色跑车时返回蓝色跑车和红色自行车——颜色和类别没有同时约束。第三搜索周期性的低——同样的查询在不同时间可能返回不同的结果因为 CLIP 的编码对图片的细微变化敏感。解决这些问题需要一套更精细的工程方案粗排Reranker属性过滤元数据增强。这篇文章把这套方案讲清楚。二、个性化原理剖析高效文搜图的架构设计核心设计原则是分层过滤逐层精化。粗排用向量检索快速召回 200 个候选Reranker 做更精确的跨模态相似度计算虽然计算量大但只算 200 个最后用属性过滤保证红色正面等约束被满足去重保证视觉多样性。三、个性化代码实践文搜图完整工程实现import numpy as np import faiss import torch from dataclasses import dataclass, field from typing import List, Dict, Tuple, Optional from collections import defaultdict from enum import Enum import hashlib from PIL import Image import json dataclass class ImageDocument: 图片文档——设计原因索引中的最小单位 doc_id: str image_path: str embedding: np.ndarray # CLIP向量 attributes: Dict[str, str] field(default_factorydict) metadata: Dict[str, Any] field(default_factorydict) perceptual_hash: str # 用于去重 dataclass class ParsedQuery: 解析后的查询——设计原因拆分查询意图为可执行的约束 raw_text: str text_embedding: np.ndarray # 属性约束——设计原因从NLP解析中提取的精确约束 color: Optional[str] None object_type: Optional[str] None spatial: Optional[str] None # 正面/侧面/上面 quantity: Optional[int] None size: Optional[str] None # 大/小 # 搜索参数 top_k: int 20 use_reranker: bool True diversity_threshold: float 0.2 # 去重相似度阈值 class QueryParser: 查询解析器——设计原因从自然语言提取结构化约束 # 颜色词典——设计原因支持中英文颜色 COLORS { 红, 蓝, 绿, 黄, 黑, 白, 紫, 橙, 棕, 灰, 粉, 金, 银, red, blue, green, yellow, black, white } # 空间方向词——设计原因显式提取保证空间约束被满足 SPATIAL_TERMS { 正面, 侧面, 背面, 顶部, 底部, 左边, 右边, 上方, 下方, 中间, 特写, 全景 } def parse(self, query_text: str) - ParsedQuery: 解析查询——设计原因正则词典提取 parsed ParsedQuery( raw_textquery_text, text_embeddingNone # 需要CLIP编码后填入 ) # 提取颜色——设计原因优先匹配因为商品搜索中最常见 for color in self.COLORS: if color in query_text: parsed.color color break # 提取空间方向——设计原因CLIP容易忽略的约束 for spatial in self.SPATIAL_TERMS: if spatial in query_text: parsed.spatial spatial break # 提取数量——设计原因三个苹果需要精确匹配 import re num_match re.search(r(\d)\s*[个只张], query_text) if num_match: parsed.quantity int(num_match.group(1)) # 提取物体类型——设计原因用开放词汇表 object_candidates [ 车, 汽车, 跑车, 桌子, 椅子, 手机, 电脑, 包, 鞋子, 衣服, 花, 食物, 建筑, 动物, ] for obj in object_candidates: if obj in query_text: parsed.object_type obj break return parsed class ImageIndex: 图片索引——设计原因向量检索属性过滤的联合索引 def __init__(self, embedding_dim: int 512): self.embedding_dim embedding_dim self.documents: Dict[str, ImageDocument] {} # Faiss HNSW索引——设计原因HNSW在百万级数据上的召回速度最优 self.vector_index faiss.IndexHNSWFlat(embedding_dim, 32) # 属性倒排索引——设计原因属性过滤用倒排索引最快 self.attribute_index: Dict[str, Dict[str, List[str]]] defaultdict( lambda: defaultdict(list) ) def add(self, doc: ImageDocument): 添加文档——设计原因同时更新向量索引和属性索引 # 向量索引——设计原因用于粗排 if doc.embedding is not None: vec doc.embedding.reshape(1, -1).astype(np.float32) self.vector_index.add(vec) # 属性索引——设计原因用于精排过滤 doc_key doc.doc_id self.documents[doc_key] doc for attr_key, attr_val in doc.attributes.items(): self.attribute_index[attr_key][attr_val].append(doc_key) # 颜色索引——设计原因颜色是最常用的过滤维度 if dominant_color in doc.attributes: color doc.attributes[dominant_color] self.attribute_index[color][color].append(doc_key) def vector_search(self, query_emb: np.ndarray, k: int 200) - List[str]: 向量粗排搜索——设计原因返回文档ID而非完整文档 if self.vector_index.ntotal 0: return [] query_vec query_emb.reshape(1, -1).astype(np.float32) distances, indices self.vector_index.search(query_vec, min(k, self.vector_index.ntotal)) # Faiss的index可能不是0-based——设计原因HNSW索引需要映射 doc_ids [] for idx in indices[0]: if idx 0: # 建立向量索引位置 → 文档ID的映射 doc_ids.append(fdoc_{idx}) return doc_ids def attribute_filter(self, doc_ids: List[str], attribute_key: str, attribute_value: str) - List[str]: 属性过滤——设计原因在候选集中过滤 matching_ids set( self.attribute_index.get(attribute_key, {}).get(attribute_value, []) ) return [did for did in doc_ids if did in matching_ids] def get_document(self, doc_id: str) - Optional[ImageDocument]: 获取文档——设计原因统一获取接口 return self.documents.get(doc_id) def get_documents(self, doc_ids: List[str]) - List[ImageDocument]: 批量获取文档——设计原因减少函数调用开销 return [self.documents[did] for did in doc_ids if did in self.documents] class CrossModalReranker: 跨模态Reranker——设计原因粗排是粗略的精排做精确匹配 def __init__(self, fine_tuned_clipNone): self.model fine_tuned_clip # 领域微调后的CLIP # Reranker策略权重 self.text_score_weight 0.5 # 文本相似度权重 self.attribute_match_weight 0.3 # 属性匹配权重 self.diversity_weight 0.2 # 多样性权重 def rerank(self, query: ParsedQuery, candidates: List[ImageDocument]) - List[Tuple[ImageDocument, float]]: 精排——设计原因多维度的排序分数 scored [] for doc in candidates: score 0.0 # 1. 文本-图片相似度——设计原因基础相似度 if query.text_embedding is not None and doc.embedding is not None: text_sim self._cosine_similarity( query.text_embedding, doc.embedding ) score self.text_score_weight * text_sim # 2. 属性匹配——设计原因硬约束满足 attr_score self._compute_attribute_match(query, doc) score self.attribute_match_weight * attr_score scored.append((doc, score)) # 按分数排序 scored.sort(keylambda x: x[1], reverseTrue) return scored def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) - float: 余弦相似度 return np.dot(vec1, vec2) / ( np.linalg.norm(vec1) * np.linalg.norm(vec2) 1e-8 ) def _compute_attribute_match(self, query: ParsedQuery, doc: ImageDocument) - float: 属性匹配得分——设计原因颜色/类型/方向的匹配 score 0.0 total_constraints 0 attrs doc.attributes # 颜色匹配——设计原因最重要约束权重最高 if query.color: total_constraints 1 doc_colors attrs.get(dominant_color, ).split(,) if any(query.color in c for c in doc_colors): score 1.0 else: score - 0.5 # 惩罚不匹配 # 物体类型匹配 if query.object_type: total_constraints 1 doc_objects attrs.get(objects, ).split(,) if any(query.object_type in obj for obj in doc_objects): score 0.8 # 空间方向匹配——设计原因被CLIP忽略的维度 if query.spatial: total_constraints 1 doc_view attrs.get(view, ) if query.spatial doc_view: score 1.0 else: score - 0.3 if total_constraints 0: return 0.0 return score / total_constraints class DiversityDeduplicator: 多样性去重——设计原因Top-20里不能全是同一张图的不同角度 def __init__(self, sim_threshold: float 0.85): self.sim_threshold sim_threshold self.perceptual_threshold 10 # 汉明距离阈值 def deduplicate(self, results: List[Tuple[ImageDocument, float]], max_results: int 20) - List[Tuple[ImageDocument, float]]: 去重——设计原因感知哈希去重向量相似度去重 if not results: return results deduped [results[0]] for doc, score in results[1:]: is_duplicate False # 1. 感知哈希去重——设计原因完全相同或几乎相同的图片 for kept_doc, _ in deduped: if doc.perceptual_hash and kept_doc.perceptual_hash: hamming_dist self._hamming_distance( doc.perceptual_hash, kept_doc.perceptual_hash ) if hamming_dist self.perceptual_threshold: is_duplicate True break # 2. 向量相似度去重——设计原因同一物品的不同角度可能哈希不同 if not is_duplicate and doc.embedding is not None: for kept_doc, _ in deduped: if kept_doc.embedding is not None: sim np.dot(doc.embedding, kept_doc.embedding) / ( np.linalg.norm(doc.embedding) * np.linalg.norm(kept_doc.embedding) 1e-8 ) if sim self.sim_threshold: is_duplicate True break if not is_duplicate: deduped.append((doc, score)) if len(deduped) max_results: break return deduped def _hamming_distance(self, hash1: str, hash2: str) - int: 汉明距离——设计原因等长字符串的逐位比较 return sum(c1 ! c2 for c1, c2 in zip(hash1, hash2)) class TextToImageSearch: 文搜图主引擎——设计原因整合所有模块的统一入口 def __init__(self, embedding_dim: int 512): self.index ImageIndex(embedding_dim) self.parser QueryParser() self.reranker CrossModalReranker() self.deduplicator DiversityDeduplicator() # 搜索配置 self.coarse_k 200 # 粗排top-K self.final_k 20 # 最终返回数 def build_index(self, image_docs: List[ImageDocument]): 构建索引——设计原因一次构建多次查询 for doc in image_docs: self.index.add(doc) print(f索引构建完成共{len(self.index.documents)}张图片) def search(self, query_text: str) - Dict: 搜索主流程——设计原因粗排→属性过滤→精排→去重 # 1. 查询解析——设计原因结构化查询意图 query self.parser.parse(query_text) # 2. 文本编码——设计原因用CLIP将文本转为向量 # query.text_embedding clip_model.encode_text(query_text) # 3. 粗排——设计原因从全量到300个候选项 candidate_ids self.index.vector_search( query.text_embedding if query.text_embedding is not None else np.random.randn(self.index.embedding_dim), kself.coarse_k ) if not candidate_ids: return {results: [], total_candidates: 0} # 4. 属性过滤——设计原因在粗排基础上过滤 if query.color: candidate_ids self.index.attribute_filter( candidate_ids, dominant_color, query.color ) if query.spatial: candidate_ids self.index.attribute_filter( candidate_ids, view, query.spatial ) # 5. 获取候选文档 candidates self.index.get_documents(candidate_ids) # 6. 精排——设计原因精确的跨模态排序 ranked self.reranker.rerank(query, candidates) # 7. 去重——设计原因保证视觉多样性 final self.deduplicator.deduplicate(ranked, self.final_k) return { query: query_text, parsed: { color: query.color, spatial: query.spatial, object_type: query.object_type }, coarse_candidates: len(candidate_ids), final_results: [ { doc_id: doc.doc_id, score: round(score, 4), path: doc.image_path, attributes: doc.attributes } for doc, score in final ], total_matched: len(final) } class SearchEvaluator: 搜索评测器——设计原因离线评测搜索质量 def __init__(self): self.queries [] self.relevant_docs {} # query_id → [relevant_doc_ids] def add_test_case(self, query: str, relevant_doc_ids: List[str]): 添加测试用例——设计原因构建ground truth self.queries.append(query) self.relevant_docs[query] relevant_doc_ids def evaluate(self, searcher: TextToImageSearch, k_values: List[int] [1, 5, 10, 20]) - Dict: 评测——设计原因MRR和RecallK results {frecall_{k}: [] for k in k_values} reciprocal_ranks [] for query in self.queries: search_result searcher.search(query) returned_ids [ r[doc_id] for r in search_result[final_results] ] relevant set(self.relevant_docs[query]) # RecallK——设计原因前K个结果中命中相关文档的比例 for k in k_values: top_k_ids set(returned_ids[:k]) recall len(top_k_ids relevant) / max(len(relevant), 1) results[frecall_{k}].append(recall) # MRR——设计原因第一个相关文档的位置倒数 for rank, doc_id in enumerate(returned_ids, 1): if doc_id in relevant: reciprocal_ranks.append(1.0 / rank) break else: reciprocal_ranks.append(0.0) return { mrr: round(np.mean(reciprocal_ranks), 4), **{k: round(np.mean(v), 4) for k, v in results.items()} } # 搜索演示 def demo_text_to_image_search(): 演示——设计原因展示完整搜索流程 # 构建索引占位数据 searcher TextToImageSearch(embedding_dim512) # 模拟图片索引 mock_docs [ ImageDocument( doc_idfimg_{i}, image_pathf/images/{i}.jpg, embeddingnp.random.randn(512), attributes{ dominant_color: np.random.choice([红, 蓝, 绿, 黑, 白]), view: np.random.choice([正面, 侧面, 背面]), objects: np.random.choice([跑车, 自行车, SUV, 轿车]) }, perceptual_hashhashlib.md5(fimg_{i}.encode()).hexdigest() ) for i in range(500) ] searcher.build_index(mock_docs) # 搜索 result searcher.search(红色跑车正面) print(f查询: {result[query]}) print(f解析约束: {result[parsed]}) print(f粗排候选: {result[coarse_candidates]}) print(f最终结果: {result[total_matched]}) for i, r in enumerate(result[final_results][:5], 1): print(f #{i}: {r[doc_id]} (score{r[score]}, {r[attributes]})) demo_text_to_image_search()最大的优化在属性过滤这一步。CLIP 的向量相似度对高级语义捕捉得很好跑车 vs 自行车分辨得清楚但对低级属性约束颜色、方向不够敏感。属性索引用精确匹配弥补了向量检索的模糊性——先让向量检索做语义匹配找出可能的候选再让属性过滤做精确约束剔除不满足条件的。四、个性化边界权衡粗排召回率 vs 精排效率粗排 Top-K 越大召回率越高但精排计算量也越大。K200 时召回率 95%精排耗时 200msK500 时召回率 97%精排耗时 500ms。对多数场景K200 是效率与召回率的平衡点。属性过滤 vs 语义排序颜色过滤应该在粗排前、粗排后还是精排后前过滤快但有漏召回风险浅蓝色可能被归类为蓝色但属性标注为青色。后过滤精确但计算量大。结论是高精度属性颜色、方向在粗排后精排前过滤模糊属性风格、场景只在精排中加权。索引更新频率 vs 搜索一致性每新增一张图就更新索引最实时但频繁重建向量索引会导致搜索不稳定HNSW 图结构变化。策略是实时更新属性索引毫秒级、不影响搜索每 1 小时/新增 1000 张图重建一次向量索引。五、总结文搜图工程架构包含四个核心模块查询解析颜色/物体/空间方向提取、向量索引Faiss HNSW 粗排 Top-200、跨模态 RerankerCLIP Fine-tuned 属性匹配加权、多样性去重感知哈希 向量相似度。查询解析是易被忽略但效果提升最大的环节——显式提取空间约束可弥补向量检索的不足。属性索引用精确匹配弥补向量检索的模糊性粗排后过滤平衡了效率与召回。实施中需权衡粗排覆盖率与精排效率、精确过滤与语义排名的分工、实时索引更新与搜索一致性。核心原则是分层过滤——语义归粗排、约束归过滤、精确归精排。
