图像对相似度（CLIP）
更新时间：2026-09-14
简介
图图相似度算子：两张图各过一遍 CLIP 图像塔，取余弦相似度，逐行产出分数，过滤交给 pipeline。CLIP 封装复用 daft/aihc/functions/multimodal/_clip_dual_tower.py。
功能描述
- 双输入列：图像 A、图像 B 按行配对，两列必须用同一种
image_src_type；两列行数不一致直接抛 ValueError - 分数 = 两侧 L2 归一化向量的点积，等价于余弦相似度，结果
round(..., 6)，取值 [-1, 1] - 权重从本地目录加载（
{model_path}/{model_name}），不联网下载；目录不存在时初始化阶段抛 FileNotFoundError - 设备由
aihc_udf(num_gpus=...)决定：num_gpus > 0且torch.cuda.is_available()时用cuda:{rank % 卡数}，否则 CPU dtype="float16"落到 CPU 上会被强制回退 float32（CPU 的 fp16 矩阵乘法不完整），并打 warning- 图像先
convert("RGB")再交给CLIPProcessor做 resize/归一化 - 任一侧为 None 或空白字符串的行不进推理，直接输出 None
- 按
batch_size微批推理；微批内任一张图解码或推理失败，整个微批记 exception 日志并留 None，其余微批照常 - 两路输入按列组织，不要求 A、B 是不同的图；A/B 传同一张图会正常返回约 1.0
算子参数
输入
| 输入 | 含义 |
|---|---|
| images_a | 第一路图像列，内容形式由 image_src_type 决定 |
| images_b | 第二路图像列，与 images_a 按行配对，形式相同 |
输出
| 输出 | 含义 |
|---|---|
| sim | float64，两张图的 CLIP 余弦相似度，取值 [-1, 1]；任一侧为空或该微批失败时为 None |
参数
| 参数名称 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| image_src_type | str | "image_url" | 两列图像共用的编码形式：image_url / image_base64 / image_binary |
| model_path | str | "/opt/aihc/model" | 权重根目录 |
| model_name | str | "openai/clip-vit-base-patch32" | 相对 model_path 的 CLIP 权重子目录 |
| dtype | str | "float32" | 权重精度：float16 / float32 / bfloat16，其他值抛 ValueError |
| batch_size | int | 16 | 推理微批大小（算子内部分批，与 aihc_udf(batch_size=...) 的行批是两回事） |
| rank | int | 0 | 多卡场景 worker 序号，用来选 cuda:{rank % 卡数} |
注意事项
- 本算子只产出统计字段，不做过滤。例如「只保留相似度落在 [0.1, 1.0] 的样本」这类条件要自己在 pipeline 里写。
- 加载的是完整
CLIPModel（含文本塔），只用到图像塔；显存占用按整模型算。 - 微批粒度的异常兜底意味着一张坏图会连带同微批的其它行变成 None。数据脏的时候把
batch_size调小，能缩小连带范围。
调用示例
Python
1from __future__ import annotations
2
3import os
4
5import daft
6from daft import col
7
8from daft.aihc.common.udf import aihc_udf
9from daft.aihc.functions.image.image_pair_similarity import ImagePairSimilarity
10
11if __name__ == "__main__":
12 if os.getenv("DAFT_RUNNER", "native") == "ray":
13 import ray
14 ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
15 daft.set_runner_ray()
16 daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
17
18samples = {
19 "a": ["bos://your-bucket/sample.jpg"],
20 "b": ["bos://your-bucket/sample.jpg"],
21 }
22 ds = daft.from_pydict(samples)
23 ds = ds.with_column(
24 "sim",
25 aihc_udf(
26 ImagePairSimilarity,
27 construct_args={
28 "image_src_type": "image_url",
29 "model_path": "/path/to/models",
30 "model_name": "openai/clip-vit-base-patch32",
31 "dtype": "float32",
32 "batch_size": 16,
33 },
34 num_cpus=1,
35 num_gpus=1,
36 concurrency=1,
37 batch_size=16,
38 )(col("a"), col("b")),
39 )
40 ds.show()评价此篇文章