ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

从零搭建多模态本地搜索:基于CLIP与FAISS实现以文搜图与视频检索

从零搭建多模态本地搜索:基于CLIP与FAISS实现以文搜图与视频检索 1. 多模态本地搜索为什么文件多了之后传统搜索就不够用了相信不少开发者和内容创作者都有这样的经历电脑里存了上万张图片、几十个视频素材某天想找一张“去年活动海报里带蓝色背景的那张图”或者想定位某个视频里“主角出现在第几分钟”的画面传统文件管理器只能按文件名、修改日期、类型过滤完全帮不上忙。即使你把文件命名得再规范也挡不住素材量增长到一定程度后的检索困难。更麻烦的是图片和视频本身是“非结构化数据”没有干净的字段可以索引。想搜图片中的物体、场景、文字想搜视频中的某一帧画面传统基于文件名的搜索根本无法理解内容语义。这时候就需要引入多模态本地搜索让机器像人一样理解图像、视频、文本之间的关系在本地环境完成“以文搜图”“以图搜图”“以文搜视频片段”等操作。腾讯开源生态中已经有相关方向的多模态检索项目出现这里不讨论某一个特定仓库的名字和版本而是围绕多模态本地搜索这一技术方向拆解从原理到部署落地的完整方案。整篇文章会覆盖技术概念、环境准备、索引构建、代码示例、常见报错和工程最佳实践无论你是想给个人素材库做检索工具还是想在业务系统里实现多模态搜索能力都可以直接参考。2. 多模态搜索的核心概念向量化、嵌入与相似度检索在动手写代码之前先理解多模态检索背后最重要的三个概念。只有理解了向量化为什么是“让机器理解图片和视频”的基础后面调参和排错才不会一头雾水。2.1 什么是多模态嵌入“模态”在机器学习里指的是数据的存在形式比如文本、图像、音频、视频。单模态模型只能处理一种输入多模态模型则能把不同类型的输入映射到同一个数学空间。多模态嵌入的核心思想是用一个神经网络模型把一张图片压缩成一个固定长度的向量例如 512 维或 1024 维浮点数数组把一段文本也压缩成同样长度的向量。然后通过训练让模型学会语义相近的图文在向量空间里的距离更近语义不相关的图文距离更远。经典的 CLIP 模型就是这种思路的代表它把图片和文本分别编码再通过对比学习拉近匹配图文对的距离。这种“图文对齐”能力正是多模态本地搜索的技术基石。2.2 相似度检索向量之间怎么“比大小”当图片、视频帧、文本都被转换成向量之后搜索问题就变成了向量检索问题。给定一个查询文本系统先把它编码成查询向量然后在海量向量库中查找与查询向量最相似的若干条记录。常用指标有余弦相似度Cosine Similarity和内积Dot Product。余弦相似度关注方向一致性对向量长度不敏感因此在文本和图文检索中非常常用。计算公式也很简单similarity (A · B) / (||A|| * ||B||)实际工程中很少暴力遍历所有向量而是使用近似最近邻ANNApproximate Nearest Neighbor索引来加速。常见开源方案有 Meta 开源的 FAISS、Zilliz 的 Milvus、以及各种 HNSW 图索引实现。2.3 为什么“本地搜索”有独特价值把搜索工具部署在本地环境个人 PC、内网服务器、边缘设备而不是调用云端 API主要有三大优势数据隐私图片、视频往往包含敏感信息本地索引可以做到数据不出域。离线可用内网环境或者无外网环境下也可以完成索引和检索。成本可控数据量大时云 API 的调用费用会快速上升本地推理一次性投入显卡或 CPU 算力即可。代价是需要自己管理模型、索引和运维。下面我们就来完整走一遍这套流程。3. 环境准备与版本说明多模态模型对运行环境有一定要求但并不是说没有 3090 或者 A100 就跑不起来。个人素材库规模的检索CPU 推理虽然慢一些但也能完成任务如果数据量达到几十万张图或者需要实时索引视频就建议准备一块 NVIDIA 显卡。3.1 操作系统与 Python 环境本文示例以 Linux 或 Windows 10/11 为演示环境Python 版本建议使用 3.9 或 3.10。如果你是 Windows 用户注意部分深度学习和向量检索库在安装时对 Python 版本比较敏感建议直接用 Anaconda 创建独立虚拟环境避免污染系统 Python。# 创建并激活虚拟环境 conda create -n multimodal-search python3.10 conda activate multimodal-search版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你的环境是 ARM 架构或者 CUDA 版本比较特殊需要额外留意依赖是否提供了对应预编译包。3.2 核心依赖库多模态搜索项目通常由四部分依赖组成模型推理框架负责加载多模态模型并生成向量常见如 PyTorch。多模态模型库提供预训练模型权重常见如 Hugging Face Transformers、OpenCLIP。向量检索库负责向量的索引和相似度检索常见如 FAISS。图像/视频处理库负责读取图片、抽视频帧常见如 Pillow、OpenCV。安装示例pip install torch torchvision pip install transformers pip install faiss-cpu pip install opencv-python pillow如果你有 NVIDIA GPU并且 CUDA 环境已经准备好可以把 FAISS 替换成 GPU 版本pip install faiss-gpu这里需要特别说明不同库的版本迭代速度很快本文不写死具体版本号。你在安装时如果遇到依赖冲突建议优先查看各库官方文档中列出的兼容版本组合或者使用pip check检查依赖关系。3.3 模型选择思路多模态模型的选择范围很广。个人学习和本地小规模数据可以选择参数量较小的开源 CLIP 系列模型如果追求更高的图文匹配精度可以选用更大的开源多模态模型。模型选型时重点考虑三点向量维度不同模型输出的向量维度不同例如 512 维、768 维、1024 维维度越高通常表达力越强但存储和检索开销也越大。中文支持如果你的素材以中文场景为主优先选择在中文图文对上训练过的模型否则中文检索效果会明显下降。显存占用个人本地部署时优先选 2GB 显存以内能跑起来的模型视频抽帧索引才不至于卡顿。4. 从零实现一个多模态本地搜索工具这一节是全文的重点。我们将实现一个最小可运行的多模态本地搜索工具支持图片素材批量索引以文搜图以图搜图视频关键帧提取、索引与搜索为了便于理解我们把整个项目拆成数据准备、索引构建、搜索服务三个模块。4.1 创建项目结构建议目录结构如下multimodal_search/ ├── config.py # 配置文件 ├── index_images.py # 图片索引脚本 ├── index_videos.py # 视频索引脚本 ├── search_engine.py # 搜索服务 ├── data/ │ ├── images/ # 待索引图片 │ └── videos/ # 待索引视频 └── storage/ # 向量索引保存位置4.2 配置文件所有可调整参数集中放在config.py中方便后续切换模型和调整检索参数。# 文件路径config.py # 模型名称这里用 Hugging Face 上常见的 CLIP 模型为例 # 实际名称需要根据你可获取的模型调整 MODEL_NAME openai/clip-vit-base-patch32 # 向量维度不同模型不一样需要和模型输出保持一致 VECTOR_DIM 512 # 图片/视频帧输入尺寸 IMAGE_SIZE 224 # 向量索引保存路径 FAISS_INDEX_PATH storage/faiss_index.bin META_PATH storage/meta.jsonl # 视频抽帧间隔秒可根据视频时长和素材类型调整 FRAME_INTERVAL 2 # 检索返回数量 TOPK 5把模型名和向量维度拆出来是为了避免在代码里到处写魔法值。换模型时只需要改配置文件不需要动业务代码。4.3 编写向量编码模块核心功能是加载多模态模型把图片和文本编码成向量。这里我们封装一个Encoder类# 文件路径encoder.py import torch import torch.nn.functional as F from PIL import Image from transformers import CLIPProcessor, CLIPModel class Encoder: def __init__(self, model_name): self.device cuda if torch.cuda.is_available() else cpu self.model CLIPModel.from_pretrained(model_name).to(self.device) self.processor CLIPProcessor.from_pretrained(model_name) self.model.eval() def encode_image(self, image_path: str): image Image.open(image_path).convert(RGB) inputs self.processor(imagesimage, return_tensorspt).to(self.device) with torch.no_grad(): image_features self.model.get_image_features(**inputs) # 归一化便于后续余弦相似度计算 image_features F.normalize(image_features, p2, dim-1) return image_features.cpu().numpy()[0] def encode_text(self, text: str): inputs self.processor(texttext, return_tensorspt).to(self.device) with torch.no_grad(): text_features self.model.get_text_features(**inputs) text_features F.normalize(text_features, p2, dim-1) return text_features.cpu().numpy()[0]这里需要注意几点get_image_features和get_text_features是 CLIP 模型提供的独立编码接口不要直接走forward否则得到的是图文匹配 logits不是用于检索的特征向量。归一化是很有必要的。后续使用 FAISS 的内积检索时归一化后的向量内积就等价于余弦相似度效果更稳定。with torch.no_grad()能显著减少推理时的显存占用索引阶段不需要反向传播。4.4 图片索引构建有了编码器之后遍历图片目录逐张编码并写入 FAISS 索引# 文件路径index_images.py import os import json import faiss import numpy as np from tqdm import tqdm from encoder import Encoder from config import * def build_image_index(image_dir: str): encoder Encoder(MODEL_NAME) if os.path.exists(FAISS_INDEX_PATH): os.remove(FAISS_INDEX_PATH) # FAISS 索引IndexFlatIP 是精确内积检索数据量小时最准确 index faiss.IndexFlatIP(VECTOR_DIM) # 保存元数据每个向量对应的文件路径 meta_list [] image_paths [] for root, _, files in os.walk(image_dir): for name in files: if name.lower().endswith((.jpg, .jpeg, .png, .webp, .bmp)): image_paths.append(os.path.join(root, name)) print(f发现 {len(image_paths)} 张图片开始编码...) for img_path in tqdm(image_paths): try: vec encoder.encode_image(img_path) index.add(np.array([vec], dtypenp.float32)) meta_list.append({path: img_path, type: image}) except Exception as e: print(f处理失败: {img_path}错误: {e}) continue # 保存索引和元数据 os.makedirs(storage, exist_okTrue) faiss.write_index(index, FAISS_INDEX_PATH) with open(META_PATH, w, encodingutf-8) as f: for meta in meta_list: f.write(json.dumps(meta, ensure_asciiFalse) \n) print(f索引构建完成共 {len(meta_list)} 条记录) if __name__ __main__: build_image_index(data/images)说明几点tqdm用来显示进度图片量大的时候非常有用。编码失败时不要直接中断整个流程单张损坏图片跳过即可。IndexFlatIP是暴力精确检索准确度最高。数据量超过 50 万条时再考虑IndexIVFFlat或IndexHNSWFlat等近似索引。4.5 视频索引构建抽帧与索引视频本身不是静态内容直接给整个视频编码并不现实。常规做法是按固定间隔抽帧把每一帧作为一张“图片”来索引同时保留帧对应的时间点。# 文件路径index_videos.py import os import json import faiss import numpy as np import cv2 from tqdm import tqdm from encoder import Encoder from config import * def extract_frames(video_path: str, interval: float): 按间隔抽取视频帧返回帧列表每帧为 (时间戳, numpy图像) cap cv2.VideoCapture(video_path) if not cap.isOpened(): return [] fps cap.get(cv2.CAP_PROP_FPS) total_frames int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) frame_step max(1, int(fps * interval)) frames [] frame_idx 0 while True: ret, frame cap.read() if not ret: break if frame_idx % frame_step 0: timestamp frame_idx / fps # BGR 转 RGB方便和图片编码流程保持一致 frame_rgb cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frames.append((timestamp, frame_rgb)) frame_idx 1 cap.release() return frames def build_video_index(video_dir: str): encoder Encoder(MODEL_NAME) video_paths [] for root, _, files in os.walk(video_dir): for name in files: if name.lower().endswith((.mp4, .avi, .mov, .mkv)): video_paths.append(os.path.join(root, name)) print(f发现 {len(video_paths)} 个视频开始抽帧并编码...) index faiss.IndexFlatIP(VECTOR_DIM) meta_list [] for video_path in tqdm(video_paths): frames extract_frames(video_path, FRAME_INTERVAL) print(f{video_path}: 抽取 {len(frames)} 帧) for timestamp, frame in frames: # 将 numpy 图像转换为 PIL Image复用图文编码逻辑 from PIL import Image pil_image Image.fromarray(frame) vec encoder.encode_image_from_pil(pil_image) index.add(np.array([vec], dtypenp.float32)) meta_list.append({ path: video_path, type: video, timestamp: round(timestamp, 2) }) faiss.write_index(index, FAISS_INDEX_PATH if not os.path.exists(FAISS_INDEX_PATH) else FAISS_INDEX_PATH.replace(.bin, _video.bin)) meta_path META_PATH.replace(.jsonl, _video.jsonl) with open(meta_path, w, encodingutf-8) as f: for meta in meta_list: f.write(json.dumps(meta, ensure_asciiFalse) \n) print(f视频索引构建完成共 {len(meta_list)} 帧记录)这里需要同步修改encoder.py增加一个接收 PIL Image 对象的方法# 在 encoder.py 中新增方法 def encode_image_from_pil(self, image: Image.Image): inputs self.processor(imagesimage, return_tensorspt).to(self.device) with torch.no_grad(): image_features self.model.get_image_features(**inputs) image_features F.normalize(image_features, p2, dim-1) return image_features.cpu().numpy()[0]视频抽帧间隔是一个需要权衡的参数间隔太短帧数爆炸索引时间和存储空间都会大幅上升。间隔太长可能漏掉关键画面。一般建议先按 2 到 5 秒抽帧观察检索效果后再调整。对于有字幕或语音讲解较多的视频可以进一步结合 OCR 或语音识别把文本信息也纳入索引形成“视觉 文本”的混合检索但那是进阶方案本文先不展开。4.6 搜索服务实现索引构建完成后就可以编写检索函数了。给定一句查询文本编码成向量后从 FAISS 索引中搜索最相近的 K 条记录# 文件路径search_engine.py import json import faiss import numpy as np from encoder import Encoder from config import * class SearchEngine: def __init__(self): self.encoder Encoder(MODEL_NAME) self.index faiss.read_index(FAISS_INDEX_PATH) self.meta_list self._load_meta(META_PATH) def _load_meta(self, path): metas [] with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if line: metas.append(json.loads(line)) return metas def search_by_text(self, query_text: str, topk: int TOPK): vec self.encoder.encode_text(query_text) scores, indices self.index.search( np.array([vec], dtypenp.float32), topk ) results [] for score, idx in zip(scores[0], indices[0]): if idx 0: continue results.append({ score: float(score), **self.meta_list[idx] }) return results def search_by_image(self, image_path: str, topk: int TOPK): vec self.encoder.encode_image(image_path) scores, indices self.index.search( np.array([vec], dtypenp.float32), topk ) results [] for score, idx in zip(scores[0], indices[0]): if idx 0: continue results.append({ score: float(score), **self.meta_list[idx] }) return results if __name__ __main__: engine SearchEngine() print( 以文搜图示例 ) query 一只在草地上奔跑的狗 results engine.search_by_text(query) for r in results: print(f相似度: {r[score]:.4f}, 文件: {r[path]}) print(\n 以图搜图示例 ) results engine.search_by_image(query_images/example.jpg) for r in results: print(f相似度: {r[score]:.4f}, 文件: {r[path]})搜索接口返回的结果中不仅包含相似度分数还保留了文件的路径和类型。对于视频帧元数据里还有timestamp可以精确告诉用户“在第几分第几秒出现该画面”。4.7 运行与验证依次执行下面的命令# 1. 把测试图片放到 data/images 目录 # 2. 构建图片索引 python index_images.py # 3. 搜索测试 python search_engine.py预期输出类似发现 120 张图片开始编码... 索引构建完成共 120 条记录 以文搜图示例 相似度: 0.8632, 文件: data/images/dog_running_01.jpg 相似度: 0.8120, 文件: data/images/park_dog_03.jpg分数越高表示相似度越高。注意不同模型的分数分布区间不一样有些模型的相似度普遍在 0.6 以上才认为相关有些则 0.3 就可能语义相近建议在实际数据上多测试找到合适的阈值。5. 多模态搜索的进阶优化从能用走向好用最小可运行版本跑通之后可以按需投入以下优化点。5.1 使用近似最近邻索引提升检索速度IndexFlatIP是精确检索数据量很大时内存和耗时都会线性增长。如果索引条目超过几十万推荐换成IndexHNSWFlatimport faiss # HNSW 图索引检索速度和精度之间可以调参 index faiss.IndexHNSWFlat(VECTOR_DIM, 32) # 32 表示邻接数量 index.hnsw.efConstruction 200 # 构建索引时的搜索范围越大越精准但构建越慢 index.hnsw.efSearch 64 # 检索时的搜索范围越大越精准但耗时越久HNSW 是当前工业界非常流行的 ANN 算法兼顾了高召回率和低查询延迟。如果你的数据本身就有明显的类别结构也可以考虑 IVF 系列索引。5.2 重排序向量检索不是终点向量检索找到的 top-K 只是“候选集”并不一定完全精确。更稳妥的做法是第一轮用 ANN 索引快速召回 200 条候选。第二轮用更精确的跨模态模型或更大的模型对候选做精细排序取前 5 条返回。这种“粗排 精排”的架构在搜索系统中非常常见也是多模态搜索工具达到生产可用水平的关键一步。5.3 混合检索叠加 OCR 和音频信息视频搜索的体验瓶颈往往在于“画面里有文字但模型没有识别出文字”。例如想搜“视频里出现 Hello World 文字的所有画面”纯视觉特征很难命中。解决方案是在抽帧后加一步 OCRimport pytesseract from PIL import Image text pytesseract.image_to_string(pil_image, langeng)OCR 结果可以作为文本字段保存到元数据中构建一个额外的文本倒排索引例如使用 Whoosh、Elasticsearch。查询时先做常规多模态向量检索再做关键字过滤两者取交集或者加权融合。5.4 增量索引与数据更新本地素材库是动态增长的每次全量重建索引会非常耗时。更合理的做法是对新加入的文件只对新文件编码追加到已有 FAISS 索引。对删除的文件在元数据中标记为不可用定期全量重建清理。对同一目录设置监听比如watchdog文件变化时自动增量更新。增量追加 FAISS 索引很简单index.add(new_vectors)但 FAISS 本身不支持删除指定向量所以生产环境通常需要维护“失效列表”或者定期重建索引。6. 常见问题与排查思路多模态本地搜索在部署和使用阶段有一些高频问题整理成表格方便快速定位。问题现象常见原因解决思路安装 faiss 时报错Python 版本或平台不支持预编译包使用 conda 安装conda install -c conda-forge faiss-cpu编码速度极慢使用了 CPU 推理模型又偏大换小模型开启fp16升级到 GPU中文查询效果差模型以英文图文训练为主换用中文多模态模型在元数据中补充中文关键词视频抽帧数量过多抽帧间隔设置太短提高FRAME_INTERVAL对画面相似度做去重搜索结果的相似度都偏高查询词太笼统或模型区分度不足细化查询描述换更大参数量模型索引构建时内存溢出图片太多且一次性加载分批编码每批处理后释放变量改用磁盘映射索引换模型后向量维度对不上配置文件中的VECTOR_DIM未更新统一修改VECTOR_DIM并删除旧索引重建另一个值得注意的点是不要在所有素材上盲目追求高精度模型。图片和视频检索任务里速度往往比一点点精度提升更重要。建立索引时多花一小时可能拖垮整个更新流程。7. 工程落地建议与最佳实践7.1 索引与元数据分离向量索引只负责“找相似向量”元数据负责“告诉用户这是什么”。永远不要在向量文件中塞太多业务字段否则后续更新、迁移都会非常痛苦。建议元数据使用 JSON Lines 或者 SQLite按行读写方便也支持按路径去重。7.2 使用独立的模型服务如果搜索服务会提供给多个模块使用强烈建议把模型推理改造成独立服务例如通过 FastAPI 暴露 HTTP 接口而不是在业务进程中直接加载模型。原因有两点模型加载会占用较多内存多次加载浪费资源。模型推理是 CPU/GPU 密集操作独立服务更方便做并发控制和水平扩展。下面是一个简单的 FastAPI 检索服务片段仅作参考思路from fastapi import FastAPI from search_engine import SearchEngine app FastAPI() engine SearchEngine() app.get(/search) def search(q: str, topk: int 5): results engine.search_by_text(q, topktopk) return {results: results}7.3 安全与权限边界涉及多模态内容检索时需要注意合规与安全边界本地索引的元数据中可能包含文件路径等敏感信息跨机器传输时要脱敏。如果搜索服务部署在服务器并对外提供接口必须加认证鉴权避免未授权访问导致数据泄露。内部数据处理要遵守最小权限原则模型文件、索引文件尽量不放在公网可访问目录下。上线生产环境前先在测试环境完整跑通索引构建、检索、索引重建流程确认没有副作用再切换。7.4 显存与性能优化多模态模型的显存占用主要来自模型权重和中间激活值。如果显存比较紧张可以尝试加载模型时使用半精度from_pretrained(model_name, torch_dtypetorch.float16)推理不用的变量及时del并调用torch.cuda.empty_cache()批量编码而不是单张循环利用 batch 提升吞吐批量编码示例思路# 假设 batch_size 为 32 images [] paths [] batch_size 32 for img_path in all_paths: img load_image(img_path) images.append(img) paths.append(img_path) if len(images) batch_size: inputs processor(imagesimages, return_tensorspt).to(device) with torch.no_grad(): feats model.get_image_features(**inputs) # 归一化、保存 feats 和 paths images [] paths []7.5 日志与可观测性索引任务通常耗时较长一定要记录足够多的日志每批处理了多少文件、耗时多少、失败原因是什么。检索服务也要记录查询语句、响应耗时和命中结果方便后续分析用户搜索意图、优化模型和阈值。8. 总结与下一步学习方向基于开源多模态模型和向量检索库完全可以搭建一套可用的本地搜索工具让图片、视频素材不再“只可存放不可检索”。本文实现的核心链路可以概括为用多模态编码模型把图片、视频帧转化为高维向量。用 FAISS 构建向量索引并保存文件路径、视频时间戳等元数据。查询时把文本或图片编码为向量在索引中做相似度检索。通过近似索引、重排序、OCR 混合检索等机制提升效果。下一步的方向可以从这几个点深入尝试不同规模的多模态模型测量同一批数据上的检索精度和耗时差异。引入更完整的视频理解能力不单单抽帧还可以做镜头切换检测、关键人物识别。构建一个 Web 界面用拖拽上传和实时检索替代命令行。如果数据量达到百万级考虑迁移到 Milvus 等分布式向量数据库并接入 Kubernetes 部署。不要一开始就追求完美先用最小的数据量跑通闭环再逐步迭代优化。毕竟搜索工具的核心价值是“帮助人在海量素材里快速找到目标”如果你的素材库现在还只有几百张图直接上大模型和分布式架构反而是过度设计。先动手试试第一批索引跑起来你对多模态检索的理解就已经超过大多数停留在概念阶段的人了。如果这篇文章对你有帮助建议收藏备用后续需要给个人素材库或业务系统搭建多模态本地搜索时可以直接参照这份完整流程操作。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进