使用 MyScale 构建多模式图像搜索应用程序

# 导入 pandas
将 pandas 导入为 pd
# 从目录加载照片文件
df_photos = pd.read_csv("tmp/photos.tsv", sep='\t', header=0)
df_photos

# 导入 pytorch
进口火炬
# 导入变压器以从 Hugging Face 加载模型和处理器
从 Transformer 导入 CLIPProcessor、CLIPModel

# 从 Hugging Face 加载 CLIP 模型
模型 = CLIPModel.from_pretrained('openai/clip-vit-base-patch32')
# 加载用于预处理图像的处理器并使它们与模型兼容
处理器 = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# 定义方法
def create_embeddings（图像=无，文本=无）：
    # 初始化嵌入
    图像嵌入=无
    文本嵌入=无

    # 处理图像（如果提供）
    如果图像不是无：
        image_embeddings = extract_image_features(图像)
        image_embeddings = torch.tensor(image_embeddings)
        image_embeddings = image_embeddings / image_embeddings.norm(dim=-1, keepdim=True)

    # 处理文本（如果提供）
    如果文本不是“无”：
        text_inputs = 处理器(text=[text], return_tensors="pt", padding=True)
        使用 torch.no_grad()：
            text_outputs = model.get_text_features(**text_inputs)
            text_embeddings = text_outputs / text_outputs.norm(dim=-1, keepdim=True)
            text_embeddings = text_embeddings.squeeze(0).tolist()

    # 如果同时提供了图像和文本，则合并嵌入，并标准化
    如果 image_embeddings 不是 None 并且 text_embeddings 不是 None：
        组合嵌入 = (图像嵌入 + torch.tensor(文本嵌入)) / 2
        组合嵌入 = 组合嵌入 / 组合嵌入.norm(dim=-1, keepdim=True)
        返回combined_embeddings.tolist()

    # 仅返回图像或文本嵌入（如果提供了其中之一）
如果 text_embeddings 是 None，则返回 image_embeddings.tolist() text_embeddings

# 导入Image moduke进行图像处理
从 PIL 导入图像
# 导入requests模块用于发出HTTP请求
导入请求
# 导入tqdm用于处理条形图可视化
从 tqdm.auto 导入 tqdm

# 获取前1000张图像
photo_ids = df_photos['photo_id'][:1000]
# 过滤DataFrame以获取所需的列
df_photos = df_photos.loc[photo_ids.index, ['photo_id', 'photo_image_url']]
# 创建一个会话来发出HTTP请求
会话 = requests.Session()

# 定义用于下载和获取嵌入的 Python 函数
def process_image(url):
    尝试：
        # 发出 GET 请求来下载图像
        响应 = session.get(url, 流=True)
        响应.raise_for_status()
        图像 = Image.open(response.raw)
        # 获取嵌入并返回
        返回create_embeddings（图像）
    除了 requests.RequestException：
        返回无
# 构造一个URL来下载较小尺寸的图片
df_photos['photo_image_url'] = df_photos['photo_image_url'].apply(lambda x: x + "?q=75&fm=jpg&w=200&fit=max")
# 将图像一张一张地传递到“process_image”并将嵌入保存到新创建的列“photo_embed”
df_photos['photo_embed'] = [tqdm 中 url 的 process_image(url)(df_photos['photo_image_url'],total=len(df_photos))]

# 删除图像处理失败的行
df_photos.dropna(子集=['photo_embed'], inplace=True)
# 重置索引并将“id”列重命名为“index”
df_photos = df_photos[df_photos['photo_id'].isin(photo_ids)].reset_index().rename(columns={'index': 'id'})

# 关闭会话
session.close()

导入clickhouse_connect

客户端 = clickhouse_connect.get_client(
    主机='您的主机地址',
    端口=443，
    username='您的用户名',
    密码='您的密码'
）

df_photos

# 检查是否存在同名表。如果存在，则删除该表
client.command("如果存在则删除表default.myscale_photos")
# 创建一个照片表
客户端.命令("""
创建表default.myscale_photos
（
    id UInt64,
    照片_id 字符串，
    photo_image_url 字符串，
    photo_embed 数组(Float32),
    约束向量_len 检查长度（photo_embed）= 512
）
按 ID 排序
""")

# 从数据集中上传数据
client.insert("default.myscale_photos", df_photos.to_records(index=False).tolist(),
              column_names=df_photos.columns.tolist())

# 检查插入数据的数量
print(f"照片数量：{client.command('SELECT count(*) FROM default.myscale_photos')}")

# 用余弦创建向量索引
客户端.命令("""
更改表default.myscale_photos
添加矢量索引 photo_embed_index photo_embed
类型 MSTG
('metric_type=余弦')
“””）

# 检查向量索引的状态，确保向量索引已准备好并处于“已构建”状态
get_index_status="从 system.vector_indices 中选择状态，其中 name='photo_embed_index'"
print(f"索引构建状态：{client.command(get_index_status)}")

导入请求
将 matplotlib.pyplot 导入为 plt
从 PIL 导入图像
从 io 导入 BytesIO

# 下载图片及其 url
默认下载（网址）：
    响应 = requests.get(url)
    返回 Image.open(BytesIO(response.content))
def show_image(url, 标题=无):
    img = 下载（网址）
    图 = plt.figure(figsize=(4, 4))
    plt.imshow(img)
    plt.show()
random_image = client.query("SELECT * FROM default.myscale_photos ORDER BY rand() LIMIT 1")
target_image_url = random_image.first_item["photo_image_url"]
print("正在加载目标图像...")
显示图像（目标图像网址）