AI 数据湖服务提供高效的多模态数据处理能力。其图像检索 Pipeline 解决方案，基于 Daft 分布式计算框架和 LAS 图像处理算子，旨在帮助用户从海量原始图像数据中高效地提取高质量的特征向量。该方案能够将格式各异的图像数据，转化为符合现代检索应用标准的、统一的图像向量数据集，为构建高级图像检索和多模态应用奠定坚实基础。

背景信息 #

随着数字内容的爆炸式增长，图像已成为信息传递的核心媒介之一。传统的基于标签（Tags）或元数据（Metadata）的图像搜索方法，严重依赖人工标注，不仅成本高昂，而且无法理解图像深层次的语义内容，难以实现精准的“相似性”查找。
以图搜图、相似推荐等现代应用的核心，是语义检索，即通过理解图像内容本身来进行匹配。这依赖于将图像转化为能够表征其语义的数学表示——特征向量（Embeddings）。然而，构建一个高效的图像向量化流程面临诸多挑战：

• 输入标准化：海量图像数据在尺寸、分辨率、长宽比和格式上存在巨大差异，直接处理会影响模型的效率和特征提取的一致性。
• 高质量特征提取：需要借助先进的视觉模型（如 Vision Transformer）才能从图像中提取出能够精准捕捉其内容、风格、纹理等丰富语义的高维向量。
• 规模化处理效率：图像处理是计算密集型任务，面对百万、千万甚至亿级的图像数据，必须具备高性能的分布式处理和 GPU 加速能力。

LAS 的图像检索 Pipeline 解决方案，正是为了应对上述挑战而设计，提供了一套从图像预处理到向量化输出的端到端自动化流程。

适用场景 #

• 图像检索系统：适用于构建大规模图像检索（以图搜图）系统的场景，通过图像向量化实现快速、精准的相似图像搜索和商品推荐。
• 跨模态检索：适用于图文匹配、以图搜文、以文搜图等跨模态检索应用。生成的图像向量可以与文本向量在同一语义空间中进行相似度计算。
• 内容审核与聚类：适用于图像去重、相似内容聚类、版权图片识别等场景，通过向量相似度对海量图片进行高效分析。

优势特性 #

• 端到端处理流程：支持从原始图像到最终向量特征的完整链路，覆盖图像标准化重采样和特征向量化两大核心步骤。
• 高效图像预处理：集成专业的图像重采样算子，支持多种插值方法，可批量统一图像尺寸和分辨率，为特征提取提供标准化输入。
• 先进图像向量化：基于 DINOv2-Large 等先进视觉 Transformer 模型，提取高质量、高区分度的图像特征向量。
• 灵活存储格式：支持多种图像输入格式（URL、Base64、二进制），输出标准化的向量数据，便于下游应用直接使用。
• 分布式处理能力：依托 Daft 提供的并行调度机制，可实现大规模图像数据的高性能处理与向量化。
• GPU 加速支持：图像特征提取过程支持 GPU 加速，可数十倍提升大规模图像数据的处理效率。
• 多模态兼容：生成的图像向量与主流文本模型的向量空间兼容，可直接用于图文匹配等多模态应用。

前提条件 #

• 已注册火山引擎账号。如果您是首次登录火山引擎控制台，请先完成实名认证。
• 已开通AI数据湖服务产品。
• 已创建通用队列、计算队列。
• 已创建开发机。
• 已将原始图像上传至对象存储（ TOS） 的指定桶（Bucket）中。

工作原理 #

本解决方案采用两阶段流水线架构，包含图像预处理和图像向量化两个核心环节。

核心处理流程 #

案例展示 #

流水线执行后，会生成一个结构化数据集，其中包含原始图像的位置、其唯一标识符、标准化/重采样图像的路径以及生成的高维特征向量：

操作流程 #

步骤一：开发调试 #

1. 在 AI 数据湖服务的开发机上，编写 Python 脚本来定义和执行图像处理 Pipeline。该脚本将使用 Daft 框架串联 LAS 图像处理算子，实现从加载图像到输出向量的全过程。

代码参考
以下是完整的 Pipeline 实现代码，展示了如何加载 TOS 上的图像数据，并依次进行重采样和向量化。

from __future__ import annotations

import logging

import daft
from daft import col
from daft.las.functions.image.embedding.image_vit_embedding import ImageViTEmbedding
from daft.las.functions.image.image_resample import ImageResample
from daft.las.functions.udf import las_udf
from daft.las.io.tos import TOSConfig

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.StreamHandler(),
    ],
)

logger = logging.getLogger(__name__)

def run_pipeline(input_tos_dir: str):
    input_s3_dir = input_tos_dir.replace("tos://", "s3://", 1)
    tos_config = TOSConfig.from_env()
    IO_CONFIG = daft.io.IOConfig(s3=tos_config.to_s3_config())
    model_path = "/opt/las/models"
    logger.info("图像检索pipeline开始运行...")
    
    df = daft.from_glob_path(
        f"{input_s3_dir}/*.png",
        io_config=IO_CONFIG,
    )
    
    df = df.with_column(
        "image_tos_path",
        col("path").str.replace("s3://", "tos://")
    )
    df = df.with_column(
        "image_id",
        col("path").str.split("/").list.get(-1).str.replace(".png", "")
    )
    df = df.select("image_tos_path", "image_id")
    
    logger.info("图像文件加载完成: %d 条记录", df.count_rows())
    
    logger.info("开始图像重采样...")
    df = df.with_column(
        "resampled_image",
        las_udf(
            ImageResample,
            construct_args={
                "image_suffix": ".png",
                "tos_dir": "tos://<YOUR_BUCKET_NAME>/image_retrieval/resample_image/",
                "local_dir": "",
                "image_src_type": "image_url",
                "target_size": [224, 224],
                "target_dpi": [72, 72],
                "method": "lanczos",
            },
            num_gpus=0,
            batch_size=32,
            concurrency=1,
        )(col("image_tos_path")),
    )
    df.collect()
    logger.info("图像重采样完成: %d 条记录", df.count_rows())
    
    logger.info("开始图像向量化...")
    df = df.with_column(
        "image_embedding",
        las_udf(
            ImageViTEmbedding,
            construct_args={
                "image_src_type": "image_base64",
                "dtype": "float32",
                "batch_size": 32,
                "model_path": model_path,
                "model_name": "facebook/dinov2-large",
                "use_cls_token_embedding": True,
                "rank": 0,
            },
            num_gpus=1,
            batch_size=32,
            concurrency=1,
        )(col("resampled_image")["base64"]),
    )
    df.collect()
    logger.info("图像向量化完成: %d 条记录", df.count_rows())
    
    df = df.with_column(
        "resampled_image_tos_path",
        col("resampled_image")["image_path"]
    )
    df = df.exclude("resampled_image")
    df.show()

if __name__ == "__main__":
    input_tos_dir = "tos://<YOUR_BUCKET_NAME>/image_retrieval/test_image"
    run_pipeline(input_tos_dir)

2. 查看处理结果，根据处理结果调整 Pipeline 脚本，以实现图像处理需求。

执行 Pipeline 后，您将得到一个包含图像标识、原始路径、重采样后路径以及图像特征向量的 Daft DataFrame。
示例日志输出：

2025-08-21 07:30:10,INFO - 图像检索 Pipeline 开始运行...
2025-08-21 07:30:12,INFO - 图像文件加载完成: 150 条记录
2025-08-21 07:30:13,INFO - 开始图像重采样...
2025-08-21 07:31:05,INFO - 图像重采样完成: 150 条记录
2025-08-21 07:31:06,INFO - 开始图像向量化...
2025-08-21 07:32:30,INFO - 图像向量化完成: 150 条记录
2025-08-21 07:32:31,INFO - Pipeline 执行完毕，最终结果如下：

示例输出数据 (df.show()):

步骤二：全量任务部署 #

当处理管道在开发机上调试无误后，您可以将其保存为自定义镜像，并在 AI 数据湖服务 -> 任务管理 中创建批量任务。通过该任务，可以对存储在 TOS 上的数百万张图片进行规模化、自动化的向量化处理，为上层检索应用生产数据。