层级分类器
更新时间：2026-07-29
简介
HierarchicalClassifier 是基于 CLIP zero-shot 的层级分类算子，按照 label_tree 指定的 prompt 树对图像逐层分类，输出层级路径字符串（如 general/live_action/human）及沿路径的最小置信度。
功能描述
• 使用本地 CLIP 权重（默认 openai/clip-vit-base-patch16）做 zero-shot 分类，无需重新训练，适合冷启动。
• 通过 label_tree 自定义层级 prompt 树；预先为每一层 label 缓存 text embedding，避免重复编码。
• 逐层 softmax 取 argmax 向下深入，直至叶节点；当某层置信度低于 min_confidence 时输出 unknown。
• 支持 URL / base64 / binary 三种图像输入，视频可先抽首帧再喂入；GPU/CPU 均可运行。
算子参数
输入
| 输入 | 含义 |
|---|---|
| images | 待分类的图像列，取值由 image_src_type 决定（图像 URL / base64 字符串 / 二进制）；视频请先抽取首帧 |
输出
| 输出 | 含义 |
|---|---|
| result | struct{path: string, confidence: float32}，path 为层级路径（如 general/live_action/human，低置信度为 unknown），confidence 为路径上各层的最小置信度 |
参数
| 参数名称 | 类型 | 默认值 | 描述 | |
|---|---|---|---|---|
| image_src_type | str | "image_url" | 图像输入类型，可选 image_url / image_binary / image_base64 |
|
| model_path | str | "/opt/aihc/models" | CLIP 权重根目录，与 model_name 拼接为实际模型目录 |
|
| model_name | str | "openai/clip-vit-base-patch16" | CLIP 模型名（相对 model_path 的子目录） |
|
| label_tree | dict[str, Any] |None | None | 层级 prompt 树；为 None 时使用内置默认树 _DEFAULT_LABEL_TREE（general / physical_ai 两大顶层） |
|
| prompt_template | str | "a photo of {label}" | 每个 label 的 prompt 模板，{label} 会替换为叶节点名 |
|
| batch_size | int | 16 | 图像推理批大小 | |
| dtype | str | "float32" | 模型精度，可选 float32 / float16 / bfloat16（zero-shot 下 float32 更稳） |
|
| rank | int | 0 | 多卡 rank，用于选择 cuda:{rank % cuda_device_count} |
|
| min_confidence | float | 0.0 | 置信度低于该阈值时输出 unknown；默认 0.0 表示不做阈值过滤 |
|
| fail_on_error | bool | False | 批处理出错时是否抛异常；False 时对应行返回 None |
调用示例
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.hierarchical_classifier import HierarchicalClassifier
10
11IMAGE = "bos://your-bucket/sample.jpg" # MOCK
12MODEL_PATH = "/opt/aihc/models" # MOCK
13
14os.environ.setdefault("BOS_ENDPOINT", "http://bj.bcebos.com")
15os.environ.setdefault("BOS_REGION", "bj")
16
17if __name__ == "__main__":
18 if os.getenv("DAFT_RUNNER", "native") == "ray":
19 import ray
20
21 ray.init(ignore_reinit_error=True)
22 daft.set_runner_ray()
23 daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
24
25 samples = {"image": [IMAGE]}
26 ds = daft.from_pydict(samples)
27 ds = ds.with_column(
28 "label",
29 aihc_udf(
30 HierarchicalClassifier,
31 construct_args={
32 "image_src_type": "image_url",
33 "model_path": MODEL_PATH,
34 "model_name": "openai/clip-vit-base-patch16",
35 "min_confidence": 0.0,
36 },
37 num_cpus=1,
38 num_gpus=1,
39 concurrency=1,
40 batch_size=1,
41 )(col("image")),
42 )
43 ds.show()评价此篇文章