向量数据实战：斗图总是慢人一步？谁还没学会AI表情包检索

pip install towhee gradio opencv-pythonpip install pymilvus==2.2.11

curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh
bash standalone_embed.sh start

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
def create_milvus_collection(collection_name, dim):# 连接milvus服务，其中的ip地址为milvus服务所在地址connections.connect(host='192.168.2.134', port='19530')# 如果collection存在就销毁，类似mysql中的if table exists，drop xxxif utility.has_collection(collection_name):utility.drop_collection(collection_name)# 设定字段/列fields = [FieldSchema(name='id', dtype=DataType.INT64, descrition='ids', is_primary=True, auto_id=False),FieldSchema(name='embedding', dtype=DataType.FLOAT_VECTOR, descrition='embedding vectors', dim=dim)]schema = CollectionSchema(fields=fields, description='text image search')collection = Collection(name=collection_name, schema=schema)
# 对embedding建立索引# create IVF_FLAT index for collection.index_params = {'metric_type':'L2','index_type':"IVF_FLAT",'params':{"nlist":512}}collection.create_index(field_name="embedding", index_params=index_params)return collection
collection = create_milvus_collection('meme_search', 512)

meme_images.csv：import csvimport os
def generate_csv(folder_path, save_path):index = 0with open(save_path, 'w', encoding='utf-8', newline='') as f:csv_writer = csv.writer(f)# 写入csv标题csv_writer.writerow(['id', 'path', 'label'])# 遍历目录，将id和图片路径添加到csv文件中for filepath, _, files in os.walk(folder_path):for file_name in files:file_all_path = os.path.join(filepath, file_name)csv_writer.writerow([index, file_all_path,''])index += 1if __name__ == '__main__':# 表情包存放路径img_folder = './meme'# csv文件生成路径save_csv = './meme_images.csv'generate_csv(img_folder, save_csv)

# 定义csv文件的数据处理方式def read_csv(csv_path, encoding='utf-8-sig'):import csvwith open(csv_path, 'r', encoding=encoding) as f:data = csv.DictReader(f)for line in data:yield int(line['id']), line['path']
# 定义数据处理流水线：# 1. 输入csv文件# 2. 通过上述定义的read_csv函数处理csv数据，拿到其中的id和path# 3. 通过ops.image_decode.cv2读取路径中的image图像# 4. 通过ops.image_text_embedding.clip指定clip_vit_base_patch16作为处理模型，将图像转换为向量数据# 5. 通过lambda x: x / np.linalg.norm(x)对向量数据进行线性处理# 6. 通过ops.ann_insert.milvus_client将处理后的向量数据及id存储到milvus数据库中，其中的collection_name参数需和上述建立的collection名称一致p_csv_load = (pipe.input('csv_file').flat_map('csv_file', ('id', 'path'), read_csv).map('path', 'img', ops.image_decode.cv2('rgb')).map('img', 'vec', ops.image_text_embedding.clip(model_name='clip_vit_base_patch16', modality='image', device=0)).map('vec', 'vec', lambda x: x / np.linalg.norm(x)).map(('id', 'vec'), (), ops.ann_insert.milvus_client(host='192.168.2.134', port='19530', collection_name='meme_search')).output())
ret = p_csv_load('meme_images.csv')
# 载入数据collection.load()

RuntimeError: Node-image-decode/cv2-1 runs failed, error msg: Read image ./meme\img-1721552589754b27f5c97bcec58d8e51c7652f5bfd006.jpg failed, Traceback (most recent call last):File "C:\Users\Dubito\AppData\Local\Programs\Python\Python310\lib\site-packages\towhee\runtime\nodes\node.py", line 158, in _callreturn True, self._op(*inputs), NoneFile "C:\Users\Dubito\.towhee\operators\image-decode\cv2\versions\main\image_decode_cv2.py", line 69, in __call__raise RuntimeError(err)RuntimeError: Read image ./meme\img-1721552589754b27f5c97bcec58d8e51c7652f5bfd006.jpg failed, Traceback (most recent call last):File "C:\Users\Dubito\AppData\Local\Programs\Python\Python310\lib\site-packages\towhee\runtime\nodes\node.py", line 171, in processself.process_step()File "C:\Users\Dubito\AppData\Local\Programs\Python\Python310\lib\site-packages\towhee\runtime\nodes\_map.py", line 63, in process_stepassert succ, msgAssertionError: Read image ./meme\img-1721552589754b27f5c97bcec58d8e51c7652f5bfd006.jpg failed, Traceback (most recent call last):File "C:\Users\Dubito\AppData\Local\Programs\Python\Python310\lib\site-packages\towhee\runtime\nodes\node.py", line 158, in _callreturn True, self._op(*inputs), NoneFile "C:\Users\Dubito\.towhee\operators\image-decode\cv2\versions\main\image_decode_cv2.py", line 69, in __call__raise RuntimeError(err)RuntimeError: Read image ./meme\img-1721552589754b27f5c97bcec58d8e51c7652f5bfd006.jpg failed

p_image_insert = (pipe.input("id", "path").map("path", "img", ops.image_decode.cv2("rgb")).map("img", "vec", ops.image_text_embedding.clip(model_name='clip_vit_base_patch16', modality="image")).map("vec", "vec", lambda x: x / np.linalg.norm(x)).map(("id", "vec"), "result", ops.ann_insert.milvus_client(host='192.168.2.134', port='19530', collection_name='meme_search')).output("result"))

import pandas as pdimport cv2
# 定义读取图像的函数，根据图像id从csv文件中获取其对应路径再读取def read_image(image_ids):df = pd.read_csv('meme_images.csv')id_img = df.set_index('id')['path'].to_dict()imgs = []decode = ops.image_decode.cv2('rgb')for image_id in image_ids:path = id_img[image_id]imgs.append(decode(path))return imgs
# 定义数据处理流水线：# 1. 输入text文本# 2. 通过ops.image_text_embedding.clip指定clip_vit_base_patch16作为模型处理文本，得到向量数据# 3. 通过lambda x: x / np.linalg.norm(x)将向量数据进行线性转换# 4. 通过ops.ann_search.milvus_client检索数据库中的相似向量，其中的collection_name参数需和上述建立的collection名称一致# 5. 对检索结果进行处理，获取图像id# 6. 通过上述定义的read_image函数，读取图像内容p_search = (pipe.input('text').map('text', 'vec', ops.image_text_embedding.clip(model_name='clip_vit_base_patch16', modality='text')).map('vec', 'vec', lambda x: x / np.linalg.norm(x)).map('vec', 'result', ops.ann_search.milvus_client(host='192.168.2.134', port='19530', collection_name='meme_search', limit=5)).map('result', 'image_ids', lambda x: [item[0] for item in x]).map('image_ids', 'images', read_image).output('text', 'images'))
# 针对文本，执行流水线检索图像DataCollection(p_search("white cat")).show()DataCollection(p_search("black dog")).show()DataCollection(p_search("smile")).show()DataCollection(p_search("angry")).show()

# -*- coding: utf-8 -*-import hashlibimport hmacimport jsonimport sysimport timefrom datetime import datetimefrom http.client import HTTPSConnection
def sign(key, msg):return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
global_config = {"tx_sid": "xxx","tx_skey": "xxx"}
def tx_translate(msg, s="zh", t="en"):secret_id = global_config["tx_sid"]secret_key = global_config["tx_skey"]token = ""service = "tmt"host = "tmt.tencentcloudapi.com"region = "ap-beijing"version = "2018-03-21"action = "TextTranslate"
params = {"SourceText": msg,"Source": s,"Target": t,"ProjectId": 0}payload = json.dumps(params)endpoint = "https://tmt.tencentcloudapi.com"algorithm = "TC3-HMAC-SHA256"timestamp = int(time.time())date = datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")
# ************* 步骤 1：拼接规范请求串 *************http_request_method = "POST"canonical_uri = "/"canonical_querystring = ""ct = "application/json; charset=utf-8"canonical_headers = "content-type:%s\nhost:%s\nx-tc-action:%s\n" % (ct, host, action.lower())signed_headers = "content-type;host;x-tc-action"hashed_request_payload = hashlib.sha256(payload.encode("utf-8")).hexdigest()canonical_request = (http_request_method + "\n" + canonical_uri + "\n" + canonical_querystring + "\n" + canonical_headers + "\n" + signed_headers + "\n" + hashed_request_payload)
# ************* 步骤 2：拼接待签名字符串 *************credential_scope = date + "/" + service + "/" + "tc3_request"hashed_canonical_request = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()string_to_sign = (algorithm + "\n" +str(timestamp) + "\n" +credential_scope + "\n" +hashed_canonical_request)
# ************* 步骤 3：计算签名 *************secret_date = sign(("TC3" + secret_key).encode("utf-8"), date)secret_service = sign(secret_date, service)secret_signing = sign(secret_service, "tc3_request")signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
# ************* 步骤 4：拼接 Authorization *************authorization = (algorithm + " " + "Credential=" + secret_id + "/" + credential_scope + ", " + "SignedHeaders=" + signed_headers + ", " + "Signature=" + signature)
# ************* 步骤 5：构造并发起请求 *************headers = {"Authorization": authorization,"Content-Type": "application/json; charset=utf-8","Host": host,"X-TC-Action": action,"X-TC-Timestamp": timestamp,"X-TC-Version": version}if region:headers["X-TC-Region"] = regionif token:headers["X-TC-Token"] = token
try:req = HTTPSConnection(host)req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))resp = req.getresponse()return json.loads(resp.read())['Response']['TargetText']except Exception as err:print(f"[tx_translate] translate [{msg}] from {s} to {t} failed: {err}")return msg

def insert_image(image_id, image_path):p_image_insert(image_id, image_path)collection.load()

# 检查图片后缀是否合法def check_image_suffix(filename):valid_suffixes = {"png", "jpg", "jpeg"}return "." in filename and filename.rsplit(".", 1)[1].lower() in valid_suffixes
# 检查图片位深度是否为32def check_bit_depth(image):return image.mode == "RGBA" and image.info.get("bitdepth") == 32
# 检查图片分辨率是否为72dpidef check_resolution(image):return image.info.get("dpi") == (72, 72)
@bp.route("/image_upload", methods=["POST"])def image_upload():if "image" not in request.files:return jsonify({"error": "No image part"}), 400
file = request.files["image"]logger.debug("start upload image <{}>".format(file.filename))if file.filename == "":return jsonify({"error": "No selected image"}), 400
if not check_image_suffix(file.filename):return jsonify({"error": "Invalid image suffix"}), 400
try:file_data = file.read()file_io_data = io.BytesIO(file_data)image = Image.open(file_io_data)except IOError:return jsonify({"error": "Invalid image file"}), 400
if check_bit_depth(image):return jsonify({"error": "32-bit images are not supported"}), 400
if check_resolution(image):return jsonify({"error": "72dpi resolution is not supported"}), 400
file_path = os.path.join("meme", file.filename)image_id = get_imgs_id_max() + 1set_imgs_id_max(image_id)save_image(image, file_path)insert_image(image_id, file_path)return jsonify({"message": "Image uploaded successfully"})

from pymilvus import MilvusClient
client = MilvusClient(uri=f"http://192.168.2.134:19530",)
def delete_image(image_ids):res = client.delete(collection_name='meme_search', ids=image_ids)collection.load()return res@bp.route("/image_delete", methods=["POST"])def image_delete():image_ids = request.get_json().get("image_id")if not image_ids:return jsonify({"error": "Image ID is required"}), 400
# 调用函数删除图片res = delete_image(image_ids)
return jsonify({"message": "Image deleted successfully: {}".format(res)})

p_search = (pipe.input('text').map('text', 'vec', ops.image_text_embedding.clip(model_name='clip_vit_base_patch16', modality='text')).map('vec', 'vec', lambda x: x / np.linalg.norm(x)).map('vec', 'result', ops.ann_search.milvus_client(host='192.168.2.134', port='19530', collection_name='meme_search', limit=5)).map("result", "image_ids", lambda x: [item[0] for item in x]).output("image_ids"))
def search_image(text):return p_search(text).get()@bp.route("/image_search", methods=["POST"])def image_search():text = request.get_json().get("text")if not text:return jsonify({"error": "text is required"}), 400
# 判断是否含有中文字符，是的话调用翻译接口if any('\u4e00' <= char <= '\u9fff' for char in text):text = tx_translate(text)
# 调用函数进行搜索img_ids = search_image(text)img_paths = get_image_paths(img_ids)
return jsonify({"results": img_paths})