CLIP-GmP-ViT-L-14实操手册自定义Embedding缓存加速重复查询你是不是也遇到过这种情况每次用CLIP模型计算图片和文字的相似度都得从头到尾跑一遍模型哪怕图片和文字之前已经算过了。特别是当你需要批量处理大量数据或者反复查询同一组素材时这种重复计算简直是在浪费时间和算力。今天我们就来解决这个问题。我将带你深入CLIP-GmP-ViT-L-14模型手把手教你如何构建一个自定义的Embedding缓存系统。这个系统能让你把计算过的图片和文本特征也就是Embedding存起来下次再用时直接读取速度能提升几十倍甚至上百倍。简单来说就是让模型“记住”它算过的东西别再傻乎乎地重复劳动了。1. 为什么你需要Embedding缓存在开始动手之前我们先搞清楚为什么要做这件事。理解了动机后面的步骤才会更有方向。1.1 重复计算的代价CLIP模型的核心工作是把图片和文字都转换成一组数字向量/Embedding然后计算它们之间的相似度。这个过程我们称之为“推理”。每次推理无论输入是全新的还是之前出现过的模型都需要对图片进行预处理裁剪、归一化等。让庞大的神经网络ViT-L-14进行前向传播生成图片特征。对文本进行分词、编码生成文本特征。计算两个特征之间的余弦相似度。对于CLIP-GmP-ViT-L-14这样的模型单次推理在CPU上可能需要几百毫秒到几秒在GPU上快一些但依然有开销。想象一下你有1000张商品图片需要和10个不同的文案描述进行匹配比如“夏日清新风”、“高端商务感”。如果每次匹配都重新计算那就是1000 * 10 10000次推理。这不仅是时间的浪费在云服务环境下更是实实在在的金钱浪费。1.2 缓存带来的效率革命缓存的核心思想是“空间换时间”。我们把第一次计算得到的Embedding保存到硬盘或内存里并建立一个快速的查找表比如用文件路径或文本内容作为键。当同样的查询再次出现时系统会检查缓存中是否存在对应的Embedding。如果存在直接读取。如果不存在才调用模型进行计算并将结果存入缓存。这样做的好处立竿见影首次查询速度不变因为需要计算并存入缓存。后续重复查询速度极快可能只需要几毫秒来读取数据避免了模型前向传播的巨大开销。批量处理对于固定的图片库或文本库可以预先计算并缓存所有Embedding后续的匹配操作就变成了快速的向量检索效率极高。接下来我们就基于你提供的CLIP-GmP-ViT-L-14项目来构建这样一个系统。2. 项目环境与快速回顾在开始改造之前我们先确保环境是一致的并快速回顾一下原始项目的结构。2.1 项目启动与访问按照你提供的文档项目位于/root/CLIP-GmP-ViT-L-14/。我们用最推荐的方式启动它cd /root/CLIP-GmP-ViT-L-14 ./start.sh启动后在浏览器中打开http://localhost:7860你应该能看到Gradio的Web界面。这个界面提供了两个核心功能单图单文相似度计算上传一张图输入一段文字得到匹配分数。批量检索上传一张图输入多个文本提示得到按相关性排序的结果。我们的目标就是在不改变这个友好界面的前提下为它背后的计算引擎装上“缓存”这个加速器。2.2 理解原始计算流程为了添加缓存我们需要先找到模型计算Embedding的代码位置。通常这类项目的核心逻辑会封装在一个函数里比如叫get_image_embedding和get_text_embedding。我们可以先简单浏览一下项目目录寻找主逻辑文件比如app.py,clip_model.py或utils.pycd /root/CLIP-GmP-ViT-L-14 ls -la假设我们找到了核心模型加载和推理的文件。原始的计算流程抽象来看是这样的# 伪代码展示原始流程 def compute_image_embedding_original(image_path): # 1. 加载图片 image load_image(image_path) # 2. 图片预处理 processed_image preprocess(image) # 3. 模型推理耗时操作 with torch.no_grad(): embedding model.encode_image(processed_image) return embedding def compute_text_embedding_original(text): # 1. 文本分词 tokens tokenize(text) # 2. 模型推理耗时操作 with torch.no_grad(): embedding model.encode_text(tokens) return embedding我们的任务就是给这两个函数穿上“缓存”的外衣。3. 构建自定义Embedding缓存系统现在进入核心环节。我们将设计一个简单、高效且可靠的缓存系统。3.1 缓存系统设计思路一个好的缓存系统需要考虑以下几点键Key如何生成用什么来唯一标识一张图片或一段文本图片可以用文件路径的哈希值MD5, SHA256文本可以直接用字符串本身或其哈希值。值Value如何存储Embedding是PyTorch Tensor或NumPy数组。我们需要将其序列化后存储。常用的格式有.pt(PyTorch),.npy(NumPy) 或者通用的.pkl。缓存存储在哪里内存快但重启丢失还是硬盘慢但持久化对于Embedding这种一旦计算基本不变的数据硬盘缓存是更好的选择。我们可以设置一个缓存目录如./embedding_cache。缓存失效策略虽然Embedding通常不变但如果模型更新了怎么办我们可以在缓存键中加入模型版本或配置的哈希当模型改变时自动失效旧缓存。为了保持教程的简洁和实用性我们先实现一个基于硬盘的、以哈希值为键的基础版本。3.2 实现缓存工具类我们创建一个新的Python文件比如叫embedding_cache.py来实现缓存逻辑。# embedding_cache.py import os import pickle import hashlib import torch import numpy as np from pathlib import Path import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class EmbeddingCache: 一个简单的Embedding硬盘缓存系统。 def __init__(self, cache_dir./embedding_cache, model_signatureclip_gmp_vit_l_14): 初始化缓存。 Args: cache_dir: 缓存文件存储的目录。 model_signature: 模型标识符用于区分不同模型生成的缓存。 self.cache_dir Path(cache_dir) self.model_signature model_signature # 创建缓存目录 self.cache_dir.mkdir(parentsTrue, exist_okTrue) # 子目录分别存放图片和文本缓存 self.image_cache_dir self.cache_dir / image self.text_cache_dir self.cache_dir / text self.image_cache_dir.mkdir(exist_okTrue) self.text_cache_dir.mkdir(exist_okTrue) logger.info(fEmbedding缓存初始化完成。缓存根目录: {self.cache_dir.absolute()}) def _get_file_hash(self, file_path): 计算文件的MD5哈希值作为缓存的键。 hash_md5 hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(4096), b): hash_md5.update(chunk) return hash_md5.hexdigest() def _get_text_hash(self, text): 计算文本的MD5哈希值作为缓存的键。 return hashlib.md5(text.encode(utf-8)).hexdigest() def _get_cache_path(self, cache_dir, key): 根据键生成缓存文件路径。 # 将模型标识符作为文件名前缀防止不同模型缓存冲突 filename f{self.model_signature}_{key}.pkl return cache_dir / filename def get_image_embedding(self, image_path, compute_func): 获取图片的Embedding优先从缓存读取。 Args: image_path: 图片文件路径。 compute_func: 一个函数当缓存不存在时它接受image_path并返回embedding(torch.Tensor或np.ndarray)。 Returns: 图片的Embedding。 # 1. 生成缓存键 if not os.path.exists(image_path): raise FileNotFoundError(f图片文件不存在: {image_path}) cache_key self._get_file_hash(image_path) cache_file self._get_cache_path(self.image_cache_dir, cache_key) # 2. 检查缓存是否存在 if cache_file.exists(): logger.debug(f图片缓存命中: {image_path}) try: with open(cache_file, rb) as f: cached_data pickle.load(f) # 假设缓存的数据是 (embedding, config_hash) embedding cached_data[embedding] # 这里可以添加配置校验例如检查model_signature是否一致 return embedding except Exception as e: logger.warning(f读取图片缓存失败 {cache_file}将重新计算。错误: {e}) # 3. 缓存不存在或读取失败调用计算函数 logger.info(f计算图片Embedding: {image_path}) embedding compute_func(image_path) # 4. 将结果存入缓存 try: # 将embedding转换为CPU上的numpy数组存储节省空间且兼容性好 if torch.is_tensor(embedding): embedding_to_save embedding.cpu().numpy() else: embedding_to_save np.array(embedding) cache_data { embedding: embedding_to_save, model_signature: self.model_signature, source: image_path } with open(cache_file, wb) as f: pickle.dump(cache_data, f) logger.debug(f图片缓存已保存: {cache_file}) except Exception as e: logger.error(f保存图片缓存失败: {e}) return embedding def get_text_embedding(self, text, compute_func): 获取文本的Embedding优先从缓存读取。 Args: text: 文本字符串。 compute_func: 一个函数当缓存不存在时它接受text并返回embedding。 Returns: 文本的Embedding。 # 1. 生成缓存键 cache_key self._get_text_hash(text) cache_file self._get_cache_path(self.text_cache_dir, cache_key) # 2. 检查缓存是否存在 if cache_file.exists(): logger.debug(f文本缓存命中: {text[:50]}...) try: with open(cache_file, rb) as f: cached_data pickle.load(f) embedding cached_data[embedding] return embedding except Exception as e: logger.warning(f读取文本缓存失败 {cache_file}将重新计算。错误: {e}) # 3. 缓存不存在或读取失败调用计算函数 logger.info(f计算文本Embedding: {text[:50]}...) embedding compute_func(text) # 4. 将结果存入缓存 try: if torch.is_tensor(embedding): embedding_to_save embedding.cpu().numpy() else: embedding_to_save np.array(embedding) cache_data { embedding: embedding_to_save, model_signature: self.model_signature, source_text: text } with open(cache_file, wb) as f: pickle.dump(cache_data, f) logger.debug(f文本缓存已保存: {cache_file}) except Exception as e: logger.error(f保存文本缓存失败: {e}) return embedding def clear_cache(self, cache_typeall): 清空缓存。cache_type可以是 image, text, 或 all。 dirs_to_clear [] if cache_type in [image, all]: dirs_to_clear.append(self.image_cache_dir) if cache_type in [text, all]: dirs_to_clear.append(self.text_cache_dir) for dir_path in dirs_to_clear: if dir_path.exists(): for file in dir_path.iterdir(): file.unlink() logger.info(f已清空缓存目录: {dir_path})这个EmbeddingCache类提供了核心的缓存功能。它通过哈希值唯一标识输入将计算结果以pickle格式存储在硬盘上。get_image_embedding和get_text_embedding方法接受一个计算函数作为参数实现了“缓存-计算”的逻辑分离非常灵活。3.3 集成缓存到CLIP项目现在我们需要修改项目原有的代码将缓存系统集成进去。我们需要找到原本负责计算Embedding的地方可能在app.py或某个模型工具文件中。假设我们找到了一个核心函数它加载模型并提供编码方法。修改后的代码可能如下所示# 假设在原项目的 model_utils.py 或类似文件中 import torch from PIL import Image from transformers import CLIPProcessor, CLIPModel import os import sys # 导入我们刚写的缓存类 sys.path.append(os.path.dirname(os.path.abspath(__file__))) from embedding_cache import EmbeddingCache class CachedCLIPModel: def __init__(self, model_nameopenai/clip-vit-large-patch14): logger.info(f正在加载模型: {model_name}) self.model CLIPModel.from_pretrained(model_name) self.processor CLIPProcessor.from_pretrained(model_name) self.device cuda if torch.cuda.is_available() else cpu self.model.to(self.device) self.model.eval() # 初始化缓存使用模型名称作为签名 self.cache EmbeddingCache(model_signaturemodel_name.replace(/, _)) logger.info(模型与缓存初始化完成。) def _compute_image_embedding_uncached(self, image_path): 不经过缓存的计算函数原始逻辑。 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 image_features / image_features.norm(dim-1, keepdimTrue) return image_features def _compute_text_embedding_uncached(self, text): 不经过缓存的计算函数原始逻辑。 inputs self.processor(texttext, return_tensorspt, paddingTrue).to(self.device) with torch.no_grad(): text_features self.model.get_text_features(**inputs) text_features text_features / text_features.norm(dim-1, keepdimTrue) return text_features def get_image_embedding(self, image_path): 对外提供的接口获取图片Embedding带缓存。 return self.cache.get_image_embedding( image_path, self._compute_image_embedding_uncached ) def get_text_embedding(self, text): 对外提供的接口获取文本Embedding带缓存。 return self.cache.get_text_embedding( text, self._compute_text_embedding_uncached ) def compute_similarity(self, image_path, text): 计算图片和文本的相似度。 image_emb self.get_image_embedding(image_path) text_emb self.get_text_embedding(text) # 计算余弦相似度 similarity (image_emb text_emb.T).item() return similarity关键点在于我们创建了一个CachedCLIPModel类它内部封装了原始模型和一个缓存实例。对外提供的get_image_embedding和get_text_embedding方法已经是带缓存功能的了。原有的业务逻辑比如Gradio界面只需要调用这些方法无需关心底层是否有缓存。3.4 在Gradio应用中调用最后我们需要修改Gradio应用app.py的接口函数使其使用我们新的带缓存的模型类。# 在 app.py 中 import gradio as gr # ... 其他导入 ... from model_utils import CachedCLIPModel # 假设我们的类放在 model_utils.py # 全局初始化一次模型和缓存 cached_model CachedCLIPModel() def calculate_similarity(image, text): Gradio接口函数计算单图单文相似度 if image is None or not text: return 请上传图片并输入文本 # 临时保存上传的图片 temp_path /tmp/gradio_temp_image.png image.save(temp_path) try: similarity cached_model.compute_similarity(temp_path, text) return f相似度得分: {similarity:.4f} except Exception as e: return f计算出错: {str(e)} finally: # 清理临时文件可选 if os.path.exists(temp_path): os.remove(temp_path) def batch_retrieve(image, text_list_str): Gradio接口函数批量检索 if image is None or not text_list_str: return 请上传图片并输入文本列表每行一个 temp_path /tmp/gradio_temp_image_batch.png image.save(temp_path) text_list [t.strip() for t in text_list_str.split(\n) if t.strip()] try: # 获取图片特征缓存生效 image_emb cached_model.get_image_embedding(temp_path) results [] for text in text_list: # 获取每个文本特征缓存生效 text_emb cached_model.get_text_embedding(text) similarity (image_emb text_emb.T).item() results.append((text, similarity)) # 按相似度排序 results.sort(keylambda x: x[1], reverseTrue) # 格式化输出 output 匹配结果从高到低\n for i, (text, score) in enumerate(results): output f{i1}. {text}: {score:.4f}\n return output except Exception as e: return f批量检索出错: {str(e)} finally: if os.path.exists(temp_path): os.remove(temp_path) # ... 后续Gradio界面构建代码保持不变 ...至此缓存系统就集成完毕了。重启Gradio应用后你会发现首次计算某张图片或某个文本时速度正常但第二次及以后的计算速度会飞快因为直接从缓存读取了。4. 缓存效果验证与进阶技巧系统搭好了我们来验证一下效果并探讨一些可以让它更好的进阶技巧。4.1 如何验证缓存是否生效查看日志我们在缓存类中设置了logging.debug。你可以将日志级别调为DEBUG在控制台会看到“缓存命中”或“计算...”的信息。观察硬盘查看项目目录下是否生成了embedding_cache文件夹里面是否有.pkl缓存文件。性能对比最直观的方法是用代码计时。你可以写一个简单的测试脚本import time # 假设 cached_model 已经初始化 test_image_path 你的测试图片路径 test_text 你的测试文本 # 第一次计算冷启动 start time.time() emb1 cached_model.get_image_embedding(test_image_path) 第一次耗时 time.time() - start print(f第一次计算图片Embedding耗时: {第一次耗时:.3f}秒) # 第二次计算应命中缓存 start time.time() emb2 cached_model.get_image_embedding(test_image_path) 第二次耗时 time.time() - start print(f第二次计算图片Embedding耗时: {第二次耗时:.3f}秒) print(f速度提升: {第一次耗时/第二次耗时:.1f}倍) # 检查两次结果是否一致 print(f两次结果是否一致: {torch.allclose(emb1, emb2)})对于文本Embedding也可以做同样的测试。你会看到第二次及以后的耗时主要是磁盘I/O的时间相比模型推理可以忽略不计。4.2 进阶优化技巧我们的基础版本已经能工作得很好但如果你面对更复杂的需求可以考虑以下优化内存二级缓存LRU Cache硬盘缓存解决了持久化问题但读取速度仍受限于磁盘。可以在EmbeddingCache类内部添加一个基于functools.lru_cache的内存缓存将最近使用过的Embedding保存在内存中实现毫秒级读取。批量预计算与加载如果你的应用场景是有一个固定的图片库比如商品图库和固定的文本标签库可以在服务启动时或者用一个单独的脚本预先计算所有Embedding并缓存。这样线上服务启动后所有的查询都将是缓存命中响应速度极快。分布式缓存当服务部署在多台机器上时可以使用Redis或Memcached作为集中式缓存。这样在一台机器上计算的Embedding其他机器也能共享避免重复计算。缓存版本管理与失效如前所述在缓存键中加入模型配置、预处理参数的哈希。当这些发生变更时新计算的缓存会自动存入新文件旧缓存自然失效。还可以定期清理过期的缓存文件。更高效的序列化格式对于非常大的Embedding集合可以考虑使用更高效的序列化格式如numpy.savez_compressed或torch.save或者使用专门的向量数据库如FAISS, Milvus进行存储和检索它们针对向量操作做了大量优化。5. 总结通过这篇教程我们完成了一项非常有价值的工程优化为CLIP-GmP-ViT-L-14模型添加了自定义的Embedding缓存系统。我们来回顾一下核心收获理解了缓存的价值它通过“空间换时间”彻底避免了模型对相同输入的重复计算在批量处理和重复查询场景下能带来数量级的性能提升。实现了缓存系统我们设计并实现了一个基于硬盘的、键值对结构的EmbeddingCache类。它使用哈希值作为键pickle格式存储Embedding值逻辑清晰易于集成。完成了项目集成我们将缓存系统无缝集成到了原有的CLIP项目中修改了模型封装类和Gradio接口使得上层应用无感知地享受到了缓存带来的加速。掌握了验证与优化方法学会了如何验证缓存生效并了解了一系列进阶优化方向如内存缓存、预计算、分布式缓存等可以根据实际需求进行扩展。这个方案不仅适用于这个特定的CLIP模型其设计思路可以轻松迁移到任何需要缓存模型中间结果Embedding、特征图等的场景。下次当你遇到需要反复调用同一个昂贵模型的时候不妨试试给它加上一个“记忆”系统。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
CLIP-GmP-ViT-L-14实操手册：自定义Embedding缓存加速重复查询
CLIP-GmP-ViT-L-14实操手册自定义Embedding缓存加速重复查询你是不是也遇到过这种情况每次用CLIP模型计算图片和文字的相似度都得从头到尾跑一遍模型哪怕图片和文字之前已经算过了。特别是当你需要批量处理大量数据或者反复查询同一组素材时这种重复计算简直是在浪费时间和算力。今天我们就来解决这个问题。我将带你深入CLIP-GmP-ViT-L-14模型手把手教你如何构建一个自定义的Embedding缓存系统。这个系统能让你把计算过的图片和文本特征也就是Embedding存起来下次再用时直接读取速度能提升几十倍甚至上百倍。简单来说就是让模型“记住”它算过的东西别再傻乎乎地重复劳动了。1. 为什么你需要Embedding缓存在开始动手之前我们先搞清楚为什么要做这件事。理解了动机后面的步骤才会更有方向。1.1 重复计算的代价CLIP模型的核心工作是把图片和文字都转换成一组数字向量/Embedding然后计算它们之间的相似度。这个过程我们称之为“推理”。每次推理无论输入是全新的还是之前出现过的模型都需要对图片进行预处理裁剪、归一化等。让庞大的神经网络ViT-L-14进行前向传播生成图片特征。对文本进行分词、编码生成文本特征。计算两个特征之间的余弦相似度。对于CLIP-GmP-ViT-L-14这样的模型单次推理在CPU上可能需要几百毫秒到几秒在GPU上快一些但依然有开销。想象一下你有1000张商品图片需要和10个不同的文案描述进行匹配比如“夏日清新风”、“高端商务感”。如果每次匹配都重新计算那就是1000 * 10 10000次推理。这不仅是时间的浪费在云服务环境下更是实实在在的金钱浪费。1.2 缓存带来的效率革命缓存的核心思想是“空间换时间”。我们把第一次计算得到的Embedding保存到硬盘或内存里并建立一个快速的查找表比如用文件路径或文本内容作为键。当同样的查询再次出现时系统会检查缓存中是否存在对应的Embedding。如果存在直接读取。如果不存在才调用模型进行计算并将结果存入缓存。这样做的好处立竿见影首次查询速度不变因为需要计算并存入缓存。后续重复查询速度极快可能只需要几毫秒来读取数据避免了模型前向传播的巨大开销。批量处理对于固定的图片库或文本库可以预先计算并缓存所有Embedding后续的匹配操作就变成了快速的向量检索效率极高。接下来我们就基于你提供的CLIP-GmP-ViT-L-14项目来构建这样一个系统。2. 项目环境与快速回顾在开始改造之前我们先确保环境是一致的并快速回顾一下原始项目的结构。2.1 项目启动与访问按照你提供的文档项目位于/root/CLIP-GmP-ViT-L-14/。我们用最推荐的方式启动它cd /root/CLIP-GmP-ViT-L-14 ./start.sh启动后在浏览器中打开http://localhost:7860你应该能看到Gradio的Web界面。这个界面提供了两个核心功能单图单文相似度计算上传一张图输入一段文字得到匹配分数。批量检索上传一张图输入多个文本提示得到按相关性排序的结果。我们的目标就是在不改变这个友好界面的前提下为它背后的计算引擎装上“缓存”这个加速器。2.2 理解原始计算流程为了添加缓存我们需要先找到模型计算Embedding的代码位置。通常这类项目的核心逻辑会封装在一个函数里比如叫get_image_embedding和get_text_embedding。我们可以先简单浏览一下项目目录寻找主逻辑文件比如app.py,clip_model.py或utils.pycd /root/CLIP-GmP-ViT-L-14 ls -la假设我们找到了核心模型加载和推理的文件。原始的计算流程抽象来看是这样的# 伪代码展示原始流程 def compute_image_embedding_original(image_path): # 1. 加载图片 image load_image(image_path) # 2. 图片预处理 processed_image preprocess(image) # 3. 模型推理耗时操作 with torch.no_grad(): embedding model.encode_image(processed_image) return embedding def compute_text_embedding_original(text): # 1. 文本分词 tokens tokenize(text) # 2. 模型推理耗时操作 with torch.no_grad(): embedding model.encode_text(tokens) return embedding我们的任务就是给这两个函数穿上“缓存”的外衣。3. 构建自定义Embedding缓存系统现在进入核心环节。我们将设计一个简单、高效且可靠的缓存系统。3.1 缓存系统设计思路一个好的缓存系统需要考虑以下几点键Key如何生成用什么来唯一标识一张图片或一段文本图片可以用文件路径的哈希值MD5, SHA256文本可以直接用字符串本身或其哈希值。值Value如何存储Embedding是PyTorch Tensor或NumPy数组。我们需要将其序列化后存储。常用的格式有.pt(PyTorch),.npy(NumPy) 或者通用的.pkl。缓存存储在哪里内存快但重启丢失还是硬盘慢但持久化对于Embedding这种一旦计算基本不变的数据硬盘缓存是更好的选择。我们可以设置一个缓存目录如./embedding_cache。缓存失效策略虽然Embedding通常不变但如果模型更新了怎么办我们可以在缓存键中加入模型版本或配置的哈希当模型改变时自动失效旧缓存。为了保持教程的简洁和实用性我们先实现一个基于硬盘的、以哈希值为键的基础版本。3.2 实现缓存工具类我们创建一个新的Python文件比如叫embedding_cache.py来实现缓存逻辑。# embedding_cache.py import os import pickle import hashlib import torch import numpy as np from pathlib import Path import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class EmbeddingCache: 一个简单的Embedding硬盘缓存系统。 def __init__(self, cache_dir./embedding_cache, model_signatureclip_gmp_vit_l_14): 初始化缓存。 Args: cache_dir: 缓存文件存储的目录。 model_signature: 模型标识符用于区分不同模型生成的缓存。 self.cache_dir Path(cache_dir) self.model_signature model_signature # 创建缓存目录 self.cache_dir.mkdir(parentsTrue, exist_okTrue) # 子目录分别存放图片和文本缓存 self.image_cache_dir self.cache_dir / image self.text_cache_dir self.cache_dir / text self.image_cache_dir.mkdir(exist_okTrue) self.text_cache_dir.mkdir(exist_okTrue) logger.info(fEmbedding缓存初始化完成。缓存根目录: {self.cache_dir.absolute()}) def _get_file_hash(self, file_path): 计算文件的MD5哈希值作为缓存的键。 hash_md5 hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(4096), b): hash_md5.update(chunk) return hash_md5.hexdigest() def _get_text_hash(self, text): 计算文本的MD5哈希值作为缓存的键。 return hashlib.md5(text.encode(utf-8)).hexdigest() def _get_cache_path(self, cache_dir, key): 根据键生成缓存文件路径。 # 将模型标识符作为文件名前缀防止不同模型缓存冲突 filename f{self.model_signature}_{key}.pkl return cache_dir / filename def get_image_embedding(self, image_path, compute_func): 获取图片的Embedding优先从缓存读取。 Args: image_path: 图片文件路径。 compute_func: 一个函数当缓存不存在时它接受image_path并返回embedding(torch.Tensor或np.ndarray)。 Returns: 图片的Embedding。 # 1. 生成缓存键 if not os.path.exists(image_path): raise FileNotFoundError(f图片文件不存在: {image_path}) cache_key self._get_file_hash(image_path) cache_file self._get_cache_path(self.image_cache_dir, cache_key) # 2. 检查缓存是否存在 if cache_file.exists(): logger.debug(f图片缓存命中: {image_path}) try: with open(cache_file, rb) as f: cached_data pickle.load(f) # 假设缓存的数据是 (embedding, config_hash) embedding cached_data[embedding] # 这里可以添加配置校验例如检查model_signature是否一致 return embedding except Exception as e: logger.warning(f读取图片缓存失败 {cache_file}将重新计算。错误: {e}) # 3. 缓存不存在或读取失败调用计算函数 logger.info(f计算图片Embedding: {image_path}) embedding compute_func(image_path) # 4. 将结果存入缓存 try: # 将embedding转换为CPU上的numpy数组存储节省空间且兼容性好 if torch.is_tensor(embedding): embedding_to_save embedding.cpu().numpy() else: embedding_to_save np.array(embedding) cache_data { embedding: embedding_to_save, model_signature: self.model_signature, source: image_path } with open(cache_file, wb) as f: pickle.dump(cache_data, f) logger.debug(f图片缓存已保存: {cache_file}) except Exception as e: logger.error(f保存图片缓存失败: {e}) return embedding def get_text_embedding(self, text, compute_func): 获取文本的Embedding优先从缓存读取。 Args: text: 文本字符串。 compute_func: 一个函数当缓存不存在时它接受text并返回embedding。 Returns: 文本的Embedding。 # 1. 生成缓存键 cache_key self._get_text_hash(text) cache_file self._get_cache_path(self.text_cache_dir, cache_key) # 2. 检查缓存是否存在 if cache_file.exists(): logger.debug(f文本缓存命中: {text[:50]}...) try: with open(cache_file, rb) as f: cached_data pickle.load(f) embedding cached_data[embedding] return embedding except Exception as e: logger.warning(f读取文本缓存失败 {cache_file}将重新计算。错误: {e}) # 3. 缓存不存在或读取失败调用计算函数 logger.info(f计算文本Embedding: {text[:50]}...) embedding compute_func(text) # 4. 将结果存入缓存 try: if torch.is_tensor(embedding): embedding_to_save embedding.cpu().numpy() else: embedding_to_save np.array(embedding) cache_data { embedding: embedding_to_save, model_signature: self.model_signature, source_text: text } with open(cache_file, wb) as f: pickle.dump(cache_data, f) logger.debug(f文本缓存已保存: {cache_file}) except Exception as e: logger.error(f保存文本缓存失败: {e}) return embedding def clear_cache(self, cache_typeall): 清空缓存。cache_type可以是 image, text, 或 all。 dirs_to_clear [] if cache_type in [image, all]: dirs_to_clear.append(self.image_cache_dir) if cache_type in [text, all]: dirs_to_clear.append(self.text_cache_dir) for dir_path in dirs_to_clear: if dir_path.exists(): for file in dir_path.iterdir(): file.unlink() logger.info(f已清空缓存目录: {dir_path})这个EmbeddingCache类提供了核心的缓存功能。它通过哈希值唯一标识输入将计算结果以pickle格式存储在硬盘上。get_image_embedding和get_text_embedding方法接受一个计算函数作为参数实现了“缓存-计算”的逻辑分离非常灵活。3.3 集成缓存到CLIP项目现在我们需要修改项目原有的代码将缓存系统集成进去。我们需要找到原本负责计算Embedding的地方可能在app.py或某个模型工具文件中。假设我们找到了一个核心函数它加载模型并提供编码方法。修改后的代码可能如下所示# 假设在原项目的 model_utils.py 或类似文件中 import torch from PIL import Image from transformers import CLIPProcessor, CLIPModel import os import sys # 导入我们刚写的缓存类 sys.path.append(os.path.dirname(os.path.abspath(__file__))) from embedding_cache import EmbeddingCache class CachedCLIPModel: def __init__(self, model_nameopenai/clip-vit-large-patch14): logger.info(f正在加载模型: {model_name}) self.model CLIPModel.from_pretrained(model_name) self.processor CLIPProcessor.from_pretrained(model_name) self.device cuda if torch.cuda.is_available() else cpu self.model.to(self.device) self.model.eval() # 初始化缓存使用模型名称作为签名 self.cache EmbeddingCache(model_signaturemodel_name.replace(/, _)) logger.info(模型与缓存初始化完成。) def _compute_image_embedding_uncached(self, image_path): 不经过缓存的计算函数原始逻辑。 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 image_features / image_features.norm(dim-1, keepdimTrue) return image_features def _compute_text_embedding_uncached(self, text): 不经过缓存的计算函数原始逻辑。 inputs self.processor(texttext, return_tensorspt, paddingTrue).to(self.device) with torch.no_grad(): text_features self.model.get_text_features(**inputs) text_features text_features / text_features.norm(dim-1, keepdimTrue) return text_features def get_image_embedding(self, image_path): 对外提供的接口获取图片Embedding带缓存。 return self.cache.get_image_embedding( image_path, self._compute_image_embedding_uncached ) def get_text_embedding(self, text): 对外提供的接口获取文本Embedding带缓存。 return self.cache.get_text_embedding( text, self._compute_text_embedding_uncached ) def compute_similarity(self, image_path, text): 计算图片和文本的相似度。 image_emb self.get_image_embedding(image_path) text_emb self.get_text_embedding(text) # 计算余弦相似度 similarity (image_emb text_emb.T).item() return similarity关键点在于我们创建了一个CachedCLIPModel类它内部封装了原始模型和一个缓存实例。对外提供的get_image_embedding和get_text_embedding方法已经是带缓存功能的了。原有的业务逻辑比如Gradio界面只需要调用这些方法无需关心底层是否有缓存。3.4 在Gradio应用中调用最后我们需要修改Gradio应用app.py的接口函数使其使用我们新的带缓存的模型类。# 在 app.py 中 import gradio as gr # ... 其他导入 ... from model_utils import CachedCLIPModel # 假设我们的类放在 model_utils.py # 全局初始化一次模型和缓存 cached_model CachedCLIPModel() def calculate_similarity(image, text): Gradio接口函数计算单图单文相似度 if image is None or not text: return 请上传图片并输入文本 # 临时保存上传的图片 temp_path /tmp/gradio_temp_image.png image.save(temp_path) try: similarity cached_model.compute_similarity(temp_path, text) return f相似度得分: {similarity:.4f} except Exception as e: return f计算出错: {str(e)} finally: # 清理临时文件可选 if os.path.exists(temp_path): os.remove(temp_path) def batch_retrieve(image, text_list_str): Gradio接口函数批量检索 if image is None or not text_list_str: return 请上传图片并输入文本列表每行一个 temp_path /tmp/gradio_temp_image_batch.png image.save(temp_path) text_list [t.strip() for t in text_list_str.split(\n) if t.strip()] try: # 获取图片特征缓存生效 image_emb cached_model.get_image_embedding(temp_path) results [] for text in text_list: # 获取每个文本特征缓存生效 text_emb cached_model.get_text_embedding(text) similarity (image_emb text_emb.T).item() results.append((text, similarity)) # 按相似度排序 results.sort(keylambda x: x[1], reverseTrue) # 格式化输出 output 匹配结果从高到低\n for i, (text, score) in enumerate(results): output f{i1}. {text}: {score:.4f}\n return output except Exception as e: return f批量检索出错: {str(e)} finally: if os.path.exists(temp_path): os.remove(temp_path) # ... 后续Gradio界面构建代码保持不变 ...至此缓存系统就集成完毕了。重启Gradio应用后你会发现首次计算某张图片或某个文本时速度正常但第二次及以后的计算速度会飞快因为直接从缓存读取了。4. 缓存效果验证与进阶技巧系统搭好了我们来验证一下效果并探讨一些可以让它更好的进阶技巧。4.1 如何验证缓存是否生效查看日志我们在缓存类中设置了logging.debug。你可以将日志级别调为DEBUG在控制台会看到“缓存命中”或“计算...”的信息。观察硬盘查看项目目录下是否生成了embedding_cache文件夹里面是否有.pkl缓存文件。性能对比最直观的方法是用代码计时。你可以写一个简单的测试脚本import time # 假设 cached_model 已经初始化 test_image_path 你的测试图片路径 test_text 你的测试文本 # 第一次计算冷启动 start time.time() emb1 cached_model.get_image_embedding(test_image_path) 第一次耗时 time.time() - start print(f第一次计算图片Embedding耗时: {第一次耗时:.3f}秒) # 第二次计算应命中缓存 start time.time() emb2 cached_model.get_image_embedding(test_image_path) 第二次耗时 time.time() - start print(f第二次计算图片Embedding耗时: {第二次耗时:.3f}秒) print(f速度提升: {第一次耗时/第二次耗时:.1f}倍) # 检查两次结果是否一致 print(f两次结果是否一致: {torch.allclose(emb1, emb2)})对于文本Embedding也可以做同样的测试。你会看到第二次及以后的耗时主要是磁盘I/O的时间相比模型推理可以忽略不计。4.2 进阶优化技巧我们的基础版本已经能工作得很好但如果你面对更复杂的需求可以考虑以下优化内存二级缓存LRU Cache硬盘缓存解决了持久化问题但读取速度仍受限于磁盘。可以在EmbeddingCache类内部添加一个基于functools.lru_cache的内存缓存将最近使用过的Embedding保存在内存中实现毫秒级读取。批量预计算与加载如果你的应用场景是有一个固定的图片库比如商品图库和固定的文本标签库可以在服务启动时或者用一个单独的脚本预先计算所有Embedding并缓存。这样线上服务启动后所有的查询都将是缓存命中响应速度极快。分布式缓存当服务部署在多台机器上时可以使用Redis或Memcached作为集中式缓存。这样在一台机器上计算的Embedding其他机器也能共享避免重复计算。缓存版本管理与失效如前所述在缓存键中加入模型配置、预处理参数的哈希。当这些发生变更时新计算的缓存会自动存入新文件旧缓存自然失效。还可以定期清理过期的缓存文件。更高效的序列化格式对于非常大的Embedding集合可以考虑使用更高效的序列化格式如numpy.savez_compressed或torch.save或者使用专门的向量数据库如FAISS, Milvus进行存储和检索它们针对向量操作做了大量优化。5. 总结通过这篇教程我们完成了一项非常有价值的工程优化为CLIP-GmP-ViT-L-14模型添加了自定义的Embedding缓存系统。我们来回顾一下核心收获理解了缓存的价值它通过“空间换时间”彻底避免了模型对相同输入的重复计算在批量处理和重复查询场景下能带来数量级的性能提升。实现了缓存系统我们设计并实现了一个基于硬盘的、键值对结构的EmbeddingCache类。它使用哈希值作为键pickle格式存储Embedding值逻辑清晰易于集成。完成了项目集成我们将缓存系统无缝集成到了原有的CLIP项目中修改了模型封装类和Gradio接口使得上层应用无感知地享受到了缓存带来的加速。掌握了验证与优化方法学会了如何验证缓存生效并了解了一系列进阶优化方向如内存缓存、预计算、分布式缓存等可以根据实际需求进行扩展。这个方案不仅适用于这个特定的CLIP模型其设计思路可以轻松迁移到任何需要缓存模型中间结果Embedding、特征图等的场景。下次当你遇到需要反复调用同一个昂贵模型的时候不妨试试给它加上一个“记忆”系统。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
