使用DataWorks Notebook实现智能图片标注，给你的图片加个“注释”

DataWorks作为一站式智能数据开发治理平台，积累和沉淀了阿里巴巴十余年大数据建设方法论和最佳实践，为数仓、数据湖、OpenLake湖仓一体数据架构提供智能化ETL开发、数据分析与主动式数据资产治理服务，助力“Data+AI”全生命周期的数据管理。面向大数据&AI协同开发场景，DataWorks提供了交互式开发和分析工具Notebook。DataWorks Notebook具备灵活且可复用的数据处理和分析环境，增强了数据开发和分析体验的直观性、模块化和交互性，能够帮助用户轻松高效地进行数据处理、数据探索、数据可视化和模型构建。

2. 点击“载入案例”，选择“工作空间”与“个人开发环境”。

3. 如何创建工作空间？

• 进入 DataWorks控制台 > 工作空间 列表，点击创建工作空间按钮。（链接：https://x.sm.cn/4GtdPhX）

4. 如何创建个人开发环境实例？

• GPU 推荐使用 24GB 显存的 A10（ecs.gn7i-c8g1.2xlarge）或更高配置

• 镜像选择 DSW 官方镜像 modelscope:1.18.0-pytorch2.3.0-gpu-py310-cu121-ubuntu22.04

! pip install git+https://github.com/xinyu1205/recognize-anything.git

[  {    "conversations": [      {        "from": "system",        "value": "你是一个导游，请生动有趣地回答游客提出的问题"      },      {        "from": "human",        "value": "给我讲讲这个东西"      },      {        "from": "gpt",        "value": "各位游客朋友们，大家好！\n\n现在映入大家眼帘的，是一件极为珍贵的文物——**神面纹玉戚**。它可不是一般的宝贝，而是来自遥远的新石器时代的一件精美玉器哦！\n\n这件宝贝现藏于山西博物院，它的尺寸相当精致：高20.6厘米，宽13.1厘米，厚仅0.4厘米，真是薄如蝉翼，工艺精湛！\n\n1964年，它在黎城县后庄村广志山出土，仿佛穿越时空，向我们诉说着古老的故事。\n\n大家仔细看，这块玉戚采用了线刻工艺，一侧雕刻着神人的半侧面头像，头戴华美的冠饰，披拂着飘逸的长发，眼眉清晰可辨，仿佛在凝视着我们；而另一侧则是一个方形台座上置神人冠饰的形象，显得神秘而威严。\n\n要知道，玉戚在古代可是象征军事统帅权力的神器，握有它，就如同掌握了千军万马！\n\n大家不妨多停留片刻，感受一下这件文物散发出的古老而神秘的气息吧！\n\n谢谢大家的聆听，接下来我们将继续参观其他精彩展品，敬请期待！"      }    ],    "images": [      "images/instance_1579398113581395972.jpg"    ]  }]

import osregion = os.getenv('DATAWORKS_REGION')!wget https://dataworks-notebook-{region}.oss-{region}-internal.aliyuncs.com/public-datasets/Image_Tagging_GPU/Qwen2-VL-History.zip!mkdir data && unzip -q Qwen2-VL-History.zip -d ./data

04
下载模型

1. 下载 ram_plus_swin_large_14m 模型

!mkdir models!wget -P ./models https://dataworks-notebook-{region}.oss-{region}-internal.aliyuncs.com/public-datasets/Image_Tagging_GPU/ram_plus_swin_large_14m.pth

2. 下载 BERT-base-uncased 模型




    

!wget https://dataworks-notebook-{region}.oss-{region}-internal.aliyuncs.com/public-datasets/Image_Tagging_GPU/bert-base-uncased.zip!unzip -q bert-base-uncased.zip -d ./models

├─models│  │─ ram_plus_swin_large_14m.pth│  │─ bert-base-uncased│  │   │─coreml│  │   │  │─...│  │   │─config.json│  │   │─model.safetensors│  │   │─pytorch_model.bin│  │   │─...

from collections import Counterfrom datasets import Imageimport osimport numpy as np
import torchfrom ram.models import ram_plusfrom ram.transform import get_transform
import logginglogging.basicConfig(level=logging.INFO,                    format='%(asctime)s:%(name)s:%(levelname)s:%(message)s')
IMAGE_BASE_DIR = './data'

def load_image(path):    img_feature = Image()    img = img_feature.decode_example(img_feature.encode_example(path))    img = img.convert('RGB')    return img

class ImageTaggingMapper(object):    """Generate image tags.    """
    def __init__(self,                 image_filed: str = 'images',                 tag_field_name: str = 'image_tags',                 *args,                 **kwargs):        """        Initialization method.        :param image_filed: the field name of images.        :param tag_field_name: the field name to store the tags.         :param args: extra args        :param kwargs: extra args        """        super().__init__(*args, **kwargs)
        self.image_filed = image_filed
        logging.info('Loading recognizeAnything model...')        self.model = ram_plus(            pretrained='./models/ram_plus_swin_large_14m.pth',            text_encoder_type='./models/bert-base-uncased',            image_size=384,            vit='swin_l',            threshold=0.68)        self.device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')        self.model.to(self.device).eval()                self.transform = get_transform(image_size=384)        self.tag_field_name = tag_field_name
    def process_single(self, sample):        # 检查标签是否已生成        if self.tag_field_name in sample:            return sample
        # 如果图像不存在，则返回空标签        if self.image_filed not in sample or not sample[self.image_filed]:            sample[self.tag_field_name] = np.array([[]], dtype=np.str_)            return sample
        # 加载图像        image_paths = sample[self.image_filed]        




    
        image_tags = []        for img_path in image_paths:            img_path = os.path.join(IMAGE_BASE_DIR, img_path)  # 获取图像的路径            image = load_image(img_path)
            image_tensor = torch.unsqueeze(self.transform(image), dim=0).to(self.device)            with torch.no_grad():                tags, _ = self.model.generate_tag(image_tensor)
            words = [word.strip() for word in tags[0].split('|')]            word_count = Counter(words)            sorted_word_list = [item for item, _ in word_count.most_common()]            image_tags.append(np.array(sorted_word_list, dtype=np.str_))
        sample[self.tag_field_name] = image_tags        return sample

from datasets import load_dataset
data_path = './data/train.json'tag_data_path = './out_tag_data.json'
# 加载数据dataset = load_dataset('json', data_files=data_path)
# 初始化打标算子image_tagging_op = ImageTaggingMapper()
# 数据集打标dataset = dataset.map(function=image_tagging_op.process_single)
# 保存打标结果，在原数据文件中新增一个tag field保存打标结果dataset['train'].to_json(tag_data_path, force_ascii=False)

!head -n 1 ./out_tag_data.json

import matplotlib.pyplot as pltfrom PIL import Image, ImageDraw, ImageFontimport osimport jsonimport random

TAG_FONT_SIZE = 30FONT_PATH = "./DingTalk JinBuTi.ttf"  # 使用字体文件TAG_COLOR = "red"

# 可视化函数def visualize_with_tags(image_path, tags):    try:        # 加载图片        img = Image.open(image_path)        draw = ImageDraw.Draw(img)                # 加载字体并设置大小        try:            font = ImageFont.truetype(FONT_PATH, TAG_FONT_SIZE)        except Exception as e:            print('load default font')            font = ImageFont.load_default()  # 默认字体可能无法缩放                # 在图片左上角添加标签        tag_text = "标签：\n" + ", \n".join(tags)        draw.text((10, 10), tag_text, fill=TAG_COLOR, font=font)                # 显示图片        plt.figure(figsize=(10, 8))        plt.imshow(img)        plt.axis("off")        plt.title(os.path.basename(image_path))        plt.show()            except FileNotFoundError:        print(f"图片文件不存在：{image_path}")    except Exception as e:        print(f"处理图片时发生错误：{str(e)}")

# 加载数据with open(tag_data_path, 'r') as f:    data_list = f.readlines()    random.shuffle(data_list)  # 有一些连续相同的图像，shuffle 显示不同的图像        # 批量可视化    for item in data_list[:3]:  # 显示3个样本        item = json.loads(item)        images = item['images']        tags = item['image_tags']        for i in range(len(images)):            image_path = os.path.join(IMAGE_BASE_DIR, images[i])            tag = item['image_tags'][i]            print(f"图片路径：{images[i]}")            print(f"标注标签：{tags[i]}")                    visualize_with_tags(image_path, tag)

图片路径：images/instance_1579398113589784578.jpg标注标签：['calligraphy', 'ink', 'manuscript', 'mark', 'pen', 'scroll', 'text', 'write', 'writing']

图片路径：images/instance_1579398113581395972.jpg标注标签：['artifact', 'hole', 'metal', 'rust', 'writing']

图片路径：images/instance_1586990758474346497.jpg标注标签：['architecture', 'building', 'city', 'pillar', 'entrance', 'palace', 'place', 'plaza', 'red', 'shrine', 'sky', 'structure', 'temple', 'worship']

立即行动：免费额度，即刻体验！

如您对文中所提到的内容及产品有疑问，或者有明确的业务需求，可 点击 「阅读原文」 填写表单联系我们，有专业的人员来联系您响应您的需求。