目录
- 知识点补充
- 前言
- 生成 ahash
- 向量数据库
- 创建集合
- 插入 ahash 到 milvus
- 查询 ahash from milvus
知识点补充
aHash算法
Hash算法进行图片相似度识别的本质，就是将图片进行Hash转化，生成一组二进制数字，然后通过比较不同图片的Hash值距离找出相似图片。aHash中文叫平均哈希算法，顾名思义，在进行转化过程中将用到像素均值。
基本原理：
1、缩小尺寸。这样做会去除图片的细节，只保留结构、明暗等基本信息，目的是统一图片大小，保证后续图片都有相同长度的哈希值，方便距离计算。网上看到的案例基本都将尺寸缩小为8*8，64个像素点，暂时不清楚缩小为这个尺寸的原因，但如果觉得损失的信息太多，个人认为可以将尺寸适当调大，当然像素点多了后续计算就会稍慢一些。
2、灰度化处理。将图片全部转换为统一的灰度图。
3、计算像素均值。计算像素的灰度平均值（此处均值出现）。
4、哈希值计算。将每个像素的灰度，与平均值进行比较。大于或等于平均值，记为1，小于平均值，记为0，由此生成二进制数组。
5、图片配对，计算汉明距离。距离越近，越相似。当图片缩小为8*8时，通常认为汉明距离小于10的一组图片为相似图片。
前言
有一个需求：计算图片的相似度
需要解决两个问题：
- 生成 ahash
- 存储和计算 ahash 之间的距离
生成 ahash
『生成 ahash』 选用 python 下面的一个 imagehash 库。（github：https://github.com/JohannesBuchner/imagehash）
from io import BytesIO
import numpy
import imagehash
from PIL import Image
def create_vector(file: BytesIO) -> bytes:
image = Image.open(file)
hash = imagehash.average_hash(image)
_vector = []
for h in hash.hash:
_vector.extend(h)
vector = bytes(
numpy.packbits(
[
int(v)
for v in _vector
],
axis=-1
).tolist()
)
return vectorcreate_vector 函数输出的类型是 bytes，就是二进制序列
imagehash.average_hash(image) 输出的 hash 对象，hash 对象有一个 hash 属性，这个属性的类型是 list[list[bool]]打印出来就是长下面这样子，其实就是一个 8x8=64 bit 的序列
[[False False False False False False False False]
[ True False False False True False False False]
[False False True True True True False False]
[False False False True True False True True]
[False False True True True False False False]
[False True True True True False False False]
[False True True True True False True True]
[False False False True True False True True]]
向量数据库
『存储和计算 ahash 之间的距离』选用 milvus
创建集合
定义集合：
import settings
from pymilvus import (
connections,
Collection,
FieldSchema,
CollectionSchema,
DataType,
)
from loggers import logger
connections.connect(
host=settings.MILVUS_CONFIG.host,
port=settings.MILVUS_CONFIG.port,
)
schema = CollectionSchema([
FieldSchema("id", DataType.INT64, is_primary=True, auto_id=True),
FieldSchema("meta_id", DataType.INT64),
FieldSchema("company_id", DataType.INT64),
FieldSchema("image_vector", dtype=DataType.BINARY_VECTOR, dim=64)
])
# 集合不存在，则会自动创建集合；已存在，不会重复创建
collection = Collection(settings.MILVUS_CONFIG.collection.name, schema)使用的向量类型是 dtype=DataType.BINARY_VECTOR,
为什么不选 float 是因为我不知道怎么把 ahash 转成 float
插入 ahash 到 milvus
class TestVector(unittest.TestCase):
def test_insert_vector(self):
"""
插入 ahash 到 milvus
python -m unittest testing.test_milvus.TestVector.test_insert_vector
"""
oss_file_path = 'image_hash/testing/WechatIMG193.jpeg'
file = BytesIO(bucket.get_object(oss_file_path).read())
vector = create_vector(file)
m_pk = insert_vector(vector, meta_id=2, company_id=1)
logger.debug(f'milvus pk: {m_pk}')查询 ahash from milvus
def test_search(self):
"""
批量调用后端接口入库
python -m unittest testing.test_milvus.TestVector.test_search
"""
oss_file_path = 'image_hash/testing/WechatIMG193.jpeg'
file = BytesIO(open(BASE_DIR/'testing'/'resource'/'WechatIMG193.jpeg','rb').read())
vector = create_vector(file)
logger.debug(vector)
rows: list[dict[str, Any]] = collection.search(
data=[vector],
param={"metric_type": 'L2', "params": {"nprobe": 32}},
anns_field='image_vector',
output_fields=['id', 'meta_id', 'company_id'],
limit=10,
)
logger.debug(rows)
logger.debug(type(rows))
提供CDN加速