跨模态检索的工程挑战文本搜图、图搜文本和图搜图的统一架构一、深度引言与场景痛点大家好我是赵咕咕。电商设计团队的一个日常需求是这样的运营给一张竞品图说帮我找一下我们库里跟这个风格最接近的 banner 素材。设计师可能再补一句色调尽量类似上周那个蓝色渐变方案的。这个需求看起来简单拆开来看包含三种不同的检索模式文本搜图用文字描述图片内容找到匹配的图片蓝色渐变 banner。图搜文本给一张图找到描述这张图的文字找到对应的设计说明、prompt 或标签。图搜图给一张图找到视觉上相似的图竞品图 → 库内相似素材。三种模式对应三个不同的检索方向。传统做法是三套独立系统但本质上它们应该共享同一个理解模型——因为无论是文字描述一张图还是判断两张图是否相似底层都是对视觉语义的理解。这篇文章我聊聊如何在一个统一架构里支持这三种跨模态检索核心依靠 CLIP 模型将图文映射到同一向量空间。二、底层机制与原理深度剖析2.1 CLIP 的核心思想CLIPContrastive Language-Image Pre-training的核心思路很优雅用 4 亿对图文配对数据训练让模型学会把匹配的图文对拉近不匹配的推远。训练完成后CLIP 有两个输出分支文本编码器任意文本 → 固定维度向量如 512 维图像编码器任意图片 → 同一维度向量因为它们在同一个向量空间里所以文本搜图 在图片向量库中找跟查询文本向量最近的图片图搜文本 在文本向量库中找跟查询图片向量最近的文本图搜图 在图片向量库中找跟查询图片向量最近的图片这就是统一架构的数学基础——不需要三套模型一个 CLIP 模型就够了。2.2 统一架构设计架构的四个关键层共享编码层文本和图像经过各自的编码器进入同一个 L2 归一化的向量空间。这是统一架构的基础。多模态索引图片特征和文本特征存在同一个向量数据库中使用相同的距离度量只是 collection 分开。查询路由层根据输入类型路由到不同的检索路径。text→搜图image→搜图image→搜文本。后处理层元数据过滤、去重、可解释性。特别是为什么这张图跟查询匹配的解释——对于创意工作者来说知道因为色调相似所以推荐了这张比知道相似度 0.87有用得多。2.3 混合查询的加权融合最实用的场景是模糊文本描述 一张参考图的混合查询帮我找跟这张图上传竞品截图风格类似的素材或者大概长这样文字描述色调参考这张上传参考图处理方式是文本编码向量 × α 图片编码向量 × (1-α) → 融合向量 → 搜索图片索引。α 是一个可调节的权重。α0.7 表示更重视文字描述的语义α0.3 表示更接近上传图片的风格。业务上可以让用户在 UI 中拖动滑块调整 α。三、生产级代码实现import asyncio import base64 import hashlib import logging from dataclasses import dataclass, field from enum import Enum from typing import Any import numpy as np import torch from PIL import Image from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue, ) # 通过 sentence-transformers 或 OpenCLIP 加载 CLIP from sentence_transformers import SentenceTransformer logger logging.getLogger(__name__) # ── 数据模型 ─────────────────────────────────────────── class SearchMode(Enum): TEXT_TO_IMAGE text_to_image IMAGE_TO_IMAGE image_to_image IMAGE_TO_TEXT image_to_text HYBRID hybrid class Modality(Enum): IMAGE image TEXT text dataclass class MediaAsset: 媒体资产模型。 asset_id: str modality: Modality content: str # 文本内容 或 图片路径 tags: list[str] field(default_factorylist) metadata: dict[str, Any] field(default_factorydict) embedding: np.ndarray | None None dataclass class CrossModalResult: 跨模态检索结果。 asset: MediaAsset score: float modality: Modality source: str # 来自哪个索引 # ── CLIP 编码器 ──────────────────────────────────────── class CLIPEncoder: CLIP 模型封装统一文本和图像编码。 def __init__(self, model_name: str clip-ViT-B-32): self._model_name model_name self._model: SentenceTransformer | None None self._dim 512 async def initialize(self) - None: if self._model is not None: return try: self._model await asyncio.to_thread( SentenceTransformer, self._model_name ) # 获取模型输出维度 self._dim self._model.get_sentence_embedding_dimension() logger.info(CLIP 模型 %s 加载完成, 维度%d, self._model_name, self._dim) except Exception as e: logger.error(CLIP 模型加载失败: %s, e) raise async def encode_text(self, text: str) - np.ndarray: 文本编码。 assert self._model is not None, 模型未初始化 embedding await asyncio.to_thread( self._model.encode, [text], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] async def encode_image(self, image_path: str) - np.ndarray: 图像编码。 assert self._model is not None, 模型未初始化 try: img Image.open(image_path).convert(RGB) embedding await asyncio.to_thread( self._model.encode, [img], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] except Exception as e: logger.error(图像编码失败 %s: %s, image_path, e) raise async def encode_image_base64(self, image_b64: str) - np.ndarray: 从 base64 编码图像。 import io img_data base64.b64decode(image_b64) img Image.open(io.BytesIO(img_data)).convert(RGB) embedding await asyncio.to_thread( self._model.encode, [img], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] property def dimension(self) - int: return self._dim # ── 统一的多模态检索引擎 ─────────────────────────────── class CrossModalSearchEngine: 跨模态检索引擎支持 text→image, image→image, image→text 三种模式。 IMAGE_COLLECTION media_images TEXT_COLLECTION media_texts def __init__(self, encoder: CLIPEncoder | None None): self._encoder encoder or CLIPEncoder() self._client QdrantClient(path./qdrant_multimodal) async def initialize(self) - None: 初始化编码器和向量集合。 await self._encoder.initialize() for coll in [self.IMAGE_COLLECTION, self.TEXT_COLLECTION]: if not self._client.collection_exists(coll): self._client.create_collection( collection_namecoll, vectors_configVectorParams( sizeself._encoder.dimension, distanceDistance.COSINE, ), ) # 创建标签索引 self._client.create_payload_index( collection_namecoll, field_nametags, field_schemakeyword, ) logger.info(多模态检索引擎初始化完成) async def index_assets(self, assets: list[MediaAsset]) - None: 批量索引媒体资产图片和文本混合。 image_points [] text_points [] for asset in assets: try: if asset.modality Modality.IMAGE: embedding await self._encoder.encode_image(asset.content) point_id hashlib.md5(asset.asset_id.encode()).hexdigest()[:16] image_points.append(PointStruct( idpoint_id, vectorembedding.tolist(), payload{ asset_id: asset.asset_id, path: asset.content, tags: asset.tags, **asset.metadata, }, )) else: embedding await self._encoder.encode_text(asset.content) point_id hashlib.md5(asset.asset_id.encode()).hexdigest()[:16] text_points.append(PointStruct( idpoint_id, vectorembedding.tolist(), payload{ asset_id: asset.asset_id, content: asset.content, tags: asset.tags, **asset.metadata, }, )) except Exception as e: logger.error(索引资产 %s 失败: %s, asset.asset_id, e) if image_points: self._client.upsert(collection_nameself.IMAGE_COLLECTION, pointsimage_points) logger.info(已索引 %d 张图片, len(image_points)) if text_points: self._client.upsert(collection_nameself.TEXT_COLLECTION, pointstext_points) logger.info(已索引 %d 条文本, len(text_points)) async def search( self, query_text: str | None None, query_image_path: str | None None, query_image_b64: str | None None, mode: SearchMode SearchMode.TEXT_TO_IMAGE, hybrid_alpha: float 0.5, filter_tags: list[str] | None None, top_k: int 10, ) - list[CrossModalResult]: 统一的跨模态搜索入口。 results [] # 构建查询过滤器 query_filter None if filter_tags: conditions [ FieldCondition(keytags, matchMatchValue(valuetag)) for tag in filter_tags ] query_filter Filter(mustconditions) try: if mode SearchMode.TEXT_TO_IMAGE and query_text: # 文本 → 图像 vec await self._encoder.encode_text(query_text) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourceimage_index, )) elif mode SearchMode.IMAGE_TO_IMAGE and (query_image_path or query_image_b64): # 图像 → 图像 vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourceimage_index, )) elif mode SearchMode.IMAGE_TO_TEXT and (query_image_path or query_image_b64): # 图像 → 文本 vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) hits self._client.search( collection_nameself.TEXT_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.TEXT, contentp.get(content, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.TEXT, sourcetext_index, )) elif mode SearchMode.HYBRID and query_text and (query_image_path or query_image_b64): # 混合查询文本向量 × α 图片向量 × (1-α) text_vec await self._encoder.encode_text(query_text) img_vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) # 加权融合确保归一化 fused text_vec * hybrid_alpha img_vec * (1 - hybrid_alpha) fused fused / np.linalg.norm(fused) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorfused.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourcefhybrid(α{hybrid_alpha}), )) except Exception as e: logger.exception(跨模态搜索失败: mode%s, mode) return [] # 最大边缘相关度MMR去重保证结果多样性 results self._mmr_dedup(results, lambda_coef0.5, final_kmin(top_k, len(results))) return results def _mmr_dedup( self, results: list[CrossModalResult], lambda_coef: float 0.5, final_k: int 5, ) - list[CrossModalResult]: MMR 去重平衡相关性和多样性。 if len(results) final_k: return results selected: list[CrossModalResult] [] remaining list(results) # 第一个选最高分的 remaining.sort(keylambda x: x.score, reverseTrue) selected.append(remaining.pop(0)) while len(selected) final_k and remaining: best_score -float(inf) best_idx 0 for i, r in enumerate(remaining): # 相关性分数 relevance r.score # 与已选择的最大相似度多样性惩罚 max_sim max( abs(s.score * r.score) # 近似相似度 for s in selected ) if selected else 0 mmr lambda_coef * relevance - (1 - lambda_coef) * max_sim if mmr best_score: best_score mmr best_idx i selected.append(remaining.pop(best_idx)) return selected async def find_similar_pairs(self, top_k: int 20) - list[dict]: 发现图库中高相似度的图片对用于去重和聚类。 # 获取所有图片 scroll_result self._client.scroll( collection_nameself.IMAGE_COLLECTION, limit1000, with_vectorsTrue, )[0] if len(scroll_result) 2: return [] vectors [r.vector for r in scroll_result] ids [r.id for r in scroll_result] # 批量计算余弦相似度矩阵 import torch mat torch.tensor(np.array(vectors)) sim_matrix torch.mm(mat, mat.T) similar_pairs [] for i in range(len(vectors)): for j in range(i 1, len(vectors)): sim float(sim_matrix[i][j]) if sim 0.95: # 高相似度阈值 similar_pairs.append({ asset_a: ids[i], asset_b: ids[j], similarity: sim, }) similar_pairs.sort(keylambda x: x[similarity], reverseTrue) return similar_pairs[:top_k] # ── 使用示例 ──────────────────────────────────────────── async def main(): encoder CLIPEncoder() engine CrossModalSearchEngine(encoder) await engine.initialize() # 索引图片和文本资产 assets [ MediaAsset( asset_idimg_001, modalityModality.IMAGE, content/path/to/summer_banner_blue.png, tags[banner, summer, blue], metadata{season: summer, campaign: 七月大促}, ), MediaAsset( asset_idtxt_001, modalityModality.TEXT, content夏日促销活动 banner蓝色海洋渐变背景清爽简约风格产品居中展示, tags[prompt, summer], metadata{author: designer_a}, ), ] await engine.index_assets(assets) # 文本搜图 results await engine.search( query_text蓝色渐变科技感 banner, modeSearchMode.TEXT_TO_IMAGE, filter_tags[banner], ) print(f文本搜图: {len(results)} 条结果) for r in results: print(f [{r.score:.3f}] {r.asset.content}) # 图搜图 results await engine.search( query_image_path/path/to/competitor_banner.png, modeSearchMode.IMAGE_TO_IMAGE, ) print(f图搜图: {len(results)} 条结果) # 图搜文本 results await engine.search( query_image_path/path/to/reference.png, modeSearchMode.IMAGE_TO_TEXT, ) print(f图搜文本: 找到匹配的描述/标签) for r in results: print(f [{r.score:.3f}] {r.asset.content[:100]}) if __name__ __main__: asyncio.run(main())核心设计决策图片和文本分集合存储虽然它们在同一个向量空间但分开存有两个好处一是查询时不需要过滤 modality提高效率二是可以做不同的索引参数调优图片集合可能需要更大的ef_construct。单一搜索入口模式参数化search()函数的mode参数路由到不同检索路径。使用者不需要知道图片索引叫media_images文本索引叫media_texts这些细节。混合查询的向量融合文本向量和图片向量直接做加权平均前提是它们已经 L2 归一化。归一化后才能保证加权融合后不偏离单位球。MMR 去重检索可能返回多张视觉上几乎一样的图同一张素材的不同版本、不同尺寸。MMR 保证结果多样性让用户看到不同的选择。四、边界分析与架构权衡4.1 CLIP 的局限性——它不懂文字CLIP 擅长的是理解图片的整体视觉风格和语义概念。但它有个显著的弱点不擅长识别图片中的文字。如果你搜图片里有全场5折这几个字的 bannerCLIP 可能找不到——因为它的训练目标是匹配图文对的整体语义不是 OCR。对于需要识别图中文字的场景需要额外的 OCR 处理层提取图中文字存到元数据走文本搜索。4.2 检索延迟CLIP 编码一张图片约 10-50ms取决于硬件Qdrant 检索 100 万向量约 1-5ms。整体检索延迟 15-55ms对交互式场景完全够用。如果图片库达到千万级别需要用 FAISS 的 IVF PQ 索引来加速。但绝大多数企业素材库不会超过百万Qdrant 单机足够。4.3 可解释性相似度 0.87对于非技术用户完全没意义。你需要解释为什么这两张图相似。一个实用的方案是用 CLIP 的注意力权重做可视化高亮查询图片和结果图片中共通的关键区域色调、构图、主体。或者用大语言模型GPT-4 Vision来生成自然语言解释这两张 banner 的相似之处在于都使用了蓝橙对比色调、居中构图、以及圆角卡片式产品展示。4.4 中文 CLIP 的选择标准的 OpenAI CLIP 对中文支持有限。中文场景推荐使用Chinese-CLIPOFA-Sys 的中文 CLIP 变体支持中英文双语。AltCLIP智源研究院的多语言 CLIP支持中文和英文。M-CLIP通过多语言蒸馏增强的 CLIP。替换只需要改CLIPEncoder的model_name参数架构其余部分不需要变动。五、总结跨模态检索的工程本质是用 CLIP 把文本和图片映射到同一个向量空间然后检索就变成了简单的最近邻搜索。三个关键决策分集合存储但共享编码器图片和文本用不同的 Qdrant collection 存储但使用同一个 CLIP 编码器。这样 query 时可以根据模式直接路由到目标集合。混合查询是杀手级特性文本 参考图的混合查询通过加权融合两个向量来实现。α 参数让用户可以在更像文字描述和更像参考图之间调节。不要忽视可解释性对于创意工作者来说知道推荐原因比知道分数更重要。预留注意力可视化和 LLM 解释的接口。这个架构的可扩展性很好。新做视频检索——用 VideoCLIP 替换编码器。新做 3D 模型检索——用 PointCLIP。架构不变只换编码器。这就是抽象的价值。下一篇预告技术博客怎么写才能在保证质量的同时高效交付聊聊我全职写 10 篇技术文章的流程复盘。
跨模态检索的工程挑战文本搜图、图搜文本和图搜图的统一架构一、深度引言与场景痛点大家好我是赵咕咕。电商设计团队的一个日常需求是这样的运营给一张竞品图说帮我找一下我们库里跟这个风格最接近的 banner 素材。设计师可能再补一句色调尽量类似上周那个蓝色渐变方案的。这个需求看起来简单拆开来看包含三种不同的检索模式文本搜图用文字描述图片内容找到匹配的图片蓝色渐变 banner。图搜文本给一张图找到描述这张图的文字找到对应的设计说明、prompt 或标签。图搜图给一张图找到视觉上相似的图竞品图 → 库内相似素材。三种模式对应三个不同的检索方向。传统做法是三套独立系统但本质上它们应该共享同一个理解模型——因为无论是文字描述一张图还是判断两张图是否相似底层都是对视觉语义的理解。这篇文章我聊聊如何在一个统一架构里支持这三种跨模态检索核心依靠 CLIP 模型将图文映射到同一向量空间。二、底层机制与原理深度剖析2.1 CLIP 的核心思想CLIPContrastive Language-Image Pre-training的核心思路很优雅用 4 亿对图文配对数据训练让模型学会把匹配的图文对拉近不匹配的推远。训练完成后CLIP 有两个输出分支文本编码器任意文本 → 固定维度向量如 512 维图像编码器任意图片 → 同一维度向量因为它们在同一个向量空间里所以文本搜图 在图片向量库中找跟查询文本向量最近的图片图搜文本 在文本向量库中找跟查询图片向量最近的文本图搜图 在图片向量库中找跟查询图片向量最近的图片这就是统一架构的数学基础——不需要三套模型一个 CLIP 模型就够了。2.2 统一架构设计架构的四个关键层共享编码层文本和图像经过各自的编码器进入同一个 L2 归一化的向量空间。这是统一架构的基础。多模态索引图片特征和文本特征存在同一个向量数据库中使用相同的距离度量只是 collection 分开。查询路由层根据输入类型路由到不同的检索路径。text→搜图image→搜图image→搜文本。后处理层元数据过滤、去重、可解释性。特别是为什么这张图跟查询匹配的解释——对于创意工作者来说知道因为色调相似所以推荐了这张比知道相似度 0.87有用得多。2.3 混合查询的加权融合最实用的场景是模糊文本描述 一张参考图的混合查询帮我找跟这张图上传竞品截图风格类似的素材或者大概长这样文字描述色调参考这张上传参考图处理方式是文本编码向量 × α 图片编码向量 × (1-α) → 融合向量 → 搜索图片索引。α 是一个可调节的权重。α0.7 表示更重视文字描述的语义α0.3 表示更接近上传图片的风格。业务上可以让用户在 UI 中拖动滑块调整 α。三、生产级代码实现import asyncio import base64 import hashlib import logging from dataclasses import dataclass, field from enum import Enum from typing import Any import numpy as np import torch from PIL import Image from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue, ) # 通过 sentence-transformers 或 OpenCLIP 加载 CLIP from sentence_transformers import SentenceTransformer logger logging.getLogger(__name__) # ── 数据模型 ─────────────────────────────────────────── class SearchMode(Enum): TEXT_TO_IMAGE text_to_image IMAGE_TO_IMAGE image_to_image IMAGE_TO_TEXT image_to_text HYBRID hybrid class Modality(Enum): IMAGE image TEXT text dataclass class MediaAsset: 媒体资产模型。 asset_id: str modality: Modality content: str # 文本内容 或 图片路径 tags: list[str] field(default_factorylist) metadata: dict[str, Any] field(default_factorydict) embedding: np.ndarray | None None dataclass class CrossModalResult: 跨模态检索结果。 asset: MediaAsset score: float modality: Modality source: str # 来自哪个索引 # ── CLIP 编码器 ──────────────────────────────────────── class CLIPEncoder: CLIP 模型封装统一文本和图像编码。 def __init__(self, model_name: str clip-ViT-B-32): self._model_name model_name self._model: SentenceTransformer | None None self._dim 512 async def initialize(self) - None: if self._model is not None: return try: self._model await asyncio.to_thread( SentenceTransformer, self._model_name ) # 获取模型输出维度 self._dim self._model.get_sentence_embedding_dimension() logger.info(CLIP 模型 %s 加载完成, 维度%d, self._model_name, self._dim) except Exception as e: logger.error(CLIP 模型加载失败: %s, e) raise async def encode_text(self, text: str) - np.ndarray: 文本编码。 assert self._model is not None, 模型未初始化 embedding await asyncio.to_thread( self._model.encode, [text], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] async def encode_image(self, image_path: str) - np.ndarray: 图像编码。 assert self._model is not None, 模型未初始化 try: img Image.open(image_path).convert(RGB) embedding await asyncio.to_thread( self._model.encode, [img], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] except Exception as e: logger.error(图像编码失败 %s: %s, image_path, e) raise async def encode_image_base64(self, image_b64: str) - np.ndarray: 从 base64 编码图像。 import io img_data base64.b64decode(image_b64) img Image.open(io.BytesIO(img_data)).convert(RGB) embedding await asyncio.to_thread( self._model.encode, [img], normalize_embeddingsTrue, show_progress_barFalse ) return embedding[0] property def dimension(self) - int: return self._dim # ── 统一的多模态检索引擎 ─────────────────────────────── class CrossModalSearchEngine: 跨模态检索引擎支持 text→image, image→image, image→text 三种模式。 IMAGE_COLLECTION media_images TEXT_COLLECTION media_texts def __init__(self, encoder: CLIPEncoder | None None): self._encoder encoder or CLIPEncoder() self._client QdrantClient(path./qdrant_multimodal) async def initialize(self) - None: 初始化编码器和向量集合。 await self._encoder.initialize() for coll in [self.IMAGE_COLLECTION, self.TEXT_COLLECTION]: if not self._client.collection_exists(coll): self._client.create_collection( collection_namecoll, vectors_configVectorParams( sizeself._encoder.dimension, distanceDistance.COSINE, ), ) # 创建标签索引 self._client.create_payload_index( collection_namecoll, field_nametags, field_schemakeyword, ) logger.info(多模态检索引擎初始化完成) async def index_assets(self, assets: list[MediaAsset]) - None: 批量索引媒体资产图片和文本混合。 image_points [] text_points [] for asset in assets: try: if asset.modality Modality.IMAGE: embedding await self._encoder.encode_image(asset.content) point_id hashlib.md5(asset.asset_id.encode()).hexdigest()[:16] image_points.append(PointStruct( idpoint_id, vectorembedding.tolist(), payload{ asset_id: asset.asset_id, path: asset.content, tags: asset.tags, **asset.metadata, }, )) else: embedding await self._encoder.encode_text(asset.content) point_id hashlib.md5(asset.asset_id.encode()).hexdigest()[:16] text_points.append(PointStruct( idpoint_id, vectorembedding.tolist(), payload{ asset_id: asset.asset_id, content: asset.content, tags: asset.tags, **asset.metadata, }, )) except Exception as e: logger.error(索引资产 %s 失败: %s, asset.asset_id, e) if image_points: self._client.upsert(collection_nameself.IMAGE_COLLECTION, pointsimage_points) logger.info(已索引 %d 张图片, len(image_points)) if text_points: self._client.upsert(collection_nameself.TEXT_COLLECTION, pointstext_points) logger.info(已索引 %d 条文本, len(text_points)) async def search( self, query_text: str | None None, query_image_path: str | None None, query_image_b64: str | None None, mode: SearchMode SearchMode.TEXT_TO_IMAGE, hybrid_alpha: float 0.5, filter_tags: list[str] | None None, top_k: int 10, ) - list[CrossModalResult]: 统一的跨模态搜索入口。 results [] # 构建查询过滤器 query_filter None if filter_tags: conditions [ FieldCondition(keytags, matchMatchValue(valuetag)) for tag in filter_tags ] query_filter Filter(mustconditions) try: if mode SearchMode.TEXT_TO_IMAGE and query_text: # 文本 → 图像 vec await self._encoder.encode_text(query_text) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourceimage_index, )) elif mode SearchMode.IMAGE_TO_IMAGE and (query_image_path or query_image_b64): # 图像 → 图像 vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourceimage_index, )) elif mode SearchMode.IMAGE_TO_TEXT and (query_image_path or query_image_b64): # 图像 → 文本 vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) hits self._client.search( collection_nameself.TEXT_COLLECTION, query_vectorvec.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.TEXT, contentp.get(content, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.TEXT, sourcetext_index, )) elif mode SearchMode.HYBRID and query_text and (query_image_path or query_image_b64): # 混合查询文本向量 × α 图片向量 × (1-α) text_vec await self._encoder.encode_text(query_text) img_vec ( await self._encoder.encode_image(query_image_path) if query_image_path else await self._encoder.encode_image_base64(query_image_b64) ) # 加权融合确保归一化 fused text_vec * hybrid_alpha img_vec * (1 - hybrid_alpha) fused fused / np.linalg.norm(fused) hits self._client.search( collection_nameself.IMAGE_COLLECTION, query_vectorfused.tolist(), query_filterquery_filter, limittop_k, ) for hit in hits: p hit.payload or {} results.append(CrossModalResult( assetMediaAsset( asset_idp.get(asset_id, ), modalityModality.IMAGE, contentp.get(path, ), tagsp.get(tags, []), metadatap, ), scorehit.score, modalityModality.IMAGE, sourcefhybrid(α{hybrid_alpha}), )) except Exception as e: logger.exception(跨模态搜索失败: mode%s, mode) return [] # 最大边缘相关度MMR去重保证结果多样性 results self._mmr_dedup(results, lambda_coef0.5, final_kmin(top_k, len(results))) return results def _mmr_dedup( self, results: list[CrossModalResult], lambda_coef: float 0.5, final_k: int 5, ) - list[CrossModalResult]: MMR 去重平衡相关性和多样性。 if len(results) final_k: return results selected: list[CrossModalResult] [] remaining list(results) # 第一个选最高分的 remaining.sort(keylambda x: x.score, reverseTrue) selected.append(remaining.pop(0)) while len(selected) final_k and remaining: best_score -float(inf) best_idx 0 for i, r in enumerate(remaining): # 相关性分数 relevance r.score # 与已选择的最大相似度多样性惩罚 max_sim max( abs(s.score * r.score) # 近似相似度 for s in selected ) if selected else 0 mmr lambda_coef * relevance - (1 - lambda_coef) * max_sim if mmr best_score: best_score mmr best_idx i selected.append(remaining.pop(best_idx)) return selected async def find_similar_pairs(self, top_k: int 20) - list[dict]: 发现图库中高相似度的图片对用于去重和聚类。 # 获取所有图片 scroll_result self._client.scroll( collection_nameself.IMAGE_COLLECTION, limit1000, with_vectorsTrue, )[0] if len(scroll_result) 2: return [] vectors [r.vector for r in scroll_result] ids [r.id for r in scroll_result] # 批量计算余弦相似度矩阵 import torch mat torch.tensor(np.array(vectors)) sim_matrix torch.mm(mat, mat.T) similar_pairs [] for i in range(len(vectors)): for j in range(i 1, len(vectors)): sim float(sim_matrix[i][j]) if sim 0.95: # 高相似度阈值 similar_pairs.append({ asset_a: ids[i], asset_b: ids[j], similarity: sim, }) similar_pairs.sort(keylambda x: x[similarity], reverseTrue) return similar_pairs[:top_k] # ── 使用示例 ──────────────────────────────────────────── async def main(): encoder CLIPEncoder() engine CrossModalSearchEngine(encoder) await engine.initialize() # 索引图片和文本资产 assets [ MediaAsset( asset_idimg_001, modalityModality.IMAGE, content/path/to/summer_banner_blue.png, tags[banner, summer, blue], metadata{season: summer, campaign: 七月大促}, ), MediaAsset( asset_idtxt_001, modalityModality.TEXT, content夏日促销活动 banner蓝色海洋渐变背景清爽简约风格产品居中展示, tags[prompt, summer], metadata{author: designer_a}, ), ] await engine.index_assets(assets) # 文本搜图 results await engine.search( query_text蓝色渐变科技感 banner, modeSearchMode.TEXT_TO_IMAGE, filter_tags[banner], ) print(f文本搜图: {len(results)} 条结果) for r in results: print(f [{r.score:.3f}] {r.asset.content}) # 图搜图 results await engine.search( query_image_path/path/to/competitor_banner.png, modeSearchMode.IMAGE_TO_IMAGE, ) print(f图搜图: {len(results)} 条结果) # 图搜文本 results await engine.search( query_image_path/path/to/reference.png, modeSearchMode.IMAGE_TO_TEXT, ) print(f图搜文本: 找到匹配的描述/标签) for r in results: print(f [{r.score:.3f}] {r.asset.content[:100]}) if __name__ __main__: asyncio.run(main())核心设计决策图片和文本分集合存储虽然它们在同一个向量空间但分开存有两个好处一是查询时不需要过滤 modality提高效率二是可以做不同的索引参数调优图片集合可能需要更大的ef_construct。单一搜索入口模式参数化search()函数的mode参数路由到不同检索路径。使用者不需要知道图片索引叫media_images文本索引叫media_texts这些细节。混合查询的向量融合文本向量和图片向量直接做加权平均前提是它们已经 L2 归一化。归一化后才能保证加权融合后不偏离单位球。MMR 去重检索可能返回多张视觉上几乎一样的图同一张素材的不同版本、不同尺寸。MMR 保证结果多样性让用户看到不同的选择。四、边界分析与架构权衡4.1 CLIP 的局限性——它不懂文字CLIP 擅长的是理解图片的整体视觉风格和语义概念。但它有个显著的弱点不擅长识别图片中的文字。如果你搜图片里有全场5折这几个字的 bannerCLIP 可能找不到——因为它的训练目标是匹配图文对的整体语义不是 OCR。对于需要识别图中文字的场景需要额外的 OCR 处理层提取图中文字存到元数据走文本搜索。4.2 检索延迟CLIP 编码一张图片约 10-50ms取决于硬件Qdrant 检索 100 万向量约 1-5ms。整体检索延迟 15-55ms对交互式场景完全够用。如果图片库达到千万级别需要用 FAISS 的 IVF PQ 索引来加速。但绝大多数企业素材库不会超过百万Qdrant 单机足够。4.3 可解释性相似度 0.87对于非技术用户完全没意义。你需要解释为什么这两张图相似。一个实用的方案是用 CLIP 的注意力权重做可视化高亮查询图片和结果图片中共通的关键区域色调、构图、主体。或者用大语言模型GPT-4 Vision来生成自然语言解释这两张 banner 的相似之处在于都使用了蓝橙对比色调、居中构图、以及圆角卡片式产品展示。4.4 中文 CLIP 的选择标准的 OpenAI CLIP 对中文支持有限。中文场景推荐使用Chinese-CLIPOFA-Sys 的中文 CLIP 变体支持中英文双语。AltCLIP智源研究院的多语言 CLIP支持中文和英文。M-CLIP通过多语言蒸馏增强的 CLIP。替换只需要改CLIPEncoder的model_name参数架构其余部分不需要变动。五、总结跨模态检索的工程本质是用 CLIP 把文本和图片映射到同一个向量空间然后检索就变成了简单的最近邻搜索。三个关键决策分集合存储但共享编码器图片和文本用不同的 Qdrant collection 存储但使用同一个 CLIP 编码器。这样 query 时可以根据模式直接路由到目标集合。混合查询是杀手级特性文本 参考图的混合查询通过加权融合两个向量来实现。α 参数让用户可以在更像文字描述和更像参考图之间调节。不要忽视可解释性对于创意工作者来说知道推荐原因比知道分数更重要。预留注意力可视化和 LLM 解释的接口。这个架构的可扩展性很好。新做视频检索——用 VideoCLIP 替换编码器。新做 3D 模型检索——用 PointCLIP。架构不变只换编码器。这就是抽象的价值。下一篇预告技术博客怎么写才能在保证质量的同时高效交付聊聊我全职写 10 篇技术文章的流程复盘。
)
