1# 1. 下载模型包
2wget -O  db_operator_dev.tar  "上述物料中镜像链接"
3# 2. 进入算子镜像tar包所在目录
4cd /xxx
5# 3. 展开镜像
6docker load < ./db_operator_dev.tar

1docker image ls | grep iregistry.baidu-int.com/doris-rdw/algorithm/operator

1docker run -itd --gpus=all \
2--privileged=true  \
3--shm-size=64g \
4--net=host \
5--name db_op_dev \
6iregistry.baidu-int.com/doris-rdw/algorithm/operator:0.7 \
7/bin/bash

1docker exec -it db_op_dev /bin/bash

1# 1. 进入阈值自定义算子测试目录
2cd /root/databuilder_dev/tests
3# 2. 运行标点替换算子测试脚本
4sh run_punctuation_replace.sh

1# 1. input data
2    samples = [
3        {
4            'images': './images/cat.jpg'
5        },
6        {
7            'images': './images/cat2.jpg'
8        },
9        {
10            'images': './images/lena.jpg'
11        }
12    ]
13
14    input_dataset = RayDataset.from_list(samples)
15
16    # 2. create operator
17    dst_path = './ret'
18    os.makedirs(dst_path, exist_ok=True)
19    op = ImageResizer(width=1024, height=1024, dst_path=dst_path, need_hash_name=False)
20
21    # 3. run
22    output_dataset = input_dataset.run(op)
23
24    # 4. get & check result
25    image_path = output_dataset.get_column(column=op.image_key)
26    
27    for name in image_path:
28        image = Image.open(name)
29        print(f'---result image {name} shape: {image.size}')

1#!/usr/bin/env python
2# -*- coding: UTF-8 -*-
3# Copyright (c) 2025 Baidu, Inc. All Rights Reserved.
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9#     http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16
17"""File load_localmodel_vllm.py"""
18
19import base64
20import json
21import os
22import re
23import time
24
25from palette.ops.base import Transformer
26from databuilder.model.util import get_model_path
27from palette.util.file_utils import SmartFile
28
29os.environ["TRANSFORMERS_OFFLINE"] = "1"
30os.environ["HF_DATASETS_OFFLINE"] = "1"
31
32def _parse_llm_json(llm_output):
33    """从模型输出中提取并解析 JSON，失败时返回 None"""
34    try:
35        match = re.search(r"```(?:json)?\s*(.*?)```", llm_output, re.DOTALL)
36        if match:
37            json_str = match.group(1)
38        else:
39            first_curly  = llm_output.find('{')
40            first_square = llm_output.find('[')
41            if first_curly == -1 and first_square == -1:
42                raise ValueError("未找到 JSON 开始符号")
43            if first_curly != -1 and (first_square == -1 or first_curly < first_square):
44                json_str = llm_output[first_curly: llm_output.rfind('}') + 1]
45            else:
46                json_str = llm_output[first_square: llm_output.rfind(']') + 1]
47        return json.loads(json_str)
48    except Exception as e:
49        print(f"[LoadLocalModelVLLM] JSON 解析失败: {e}")
50        return None
51
52
53class LoadLocalModelVLLM(Transformer):
54    """
55    本地 vLLM 模型推理算子
56    功能：
57    1. 从 volume 路径下载图片到本地
58    2. 使用本地 vLLM 模型对图片进行推理标注
59    3. 解析模型输出的 JSON，写入 annotations 字段
60    """
61    _op_type = "transform"
62    _batched_op = True
63    _processor = 'cuda'
64    _name = "load_localmodel_vllm"
65    _ray_execute_mode = "PIPELINE_ACTOR"
66    _ray_batch_format = "numpy"
67
68    def __init__(self,
69                 model_path: str = '',
70                 system_prompt: str = '',
71                 user_prompt: str = '',
72                 max_tokens: int = 4096,
73                 max_model_len: int = 8192,
74                 temperature: float = 0.7,
75                 top_p: float = 0.9,
76                 preload: bool = False,
77                 *args, **kwargs):
78        """
79        :param model_path: 模型路径，由平台注册的完整路径
80        :param system_prompt: 系统提示词，为空时使用内置默认值
81        :param user_prompt: 用户提示词，为空时使用内置默认值
82        :param max_tokens: 最大输出 token 数
83        :param max_model_len: 模型最大上下文长度
84        :param temperature: 生成随机性
85        :param top_p: 核采样阈值
86        :param preload: 为 True 时在构造阶段完成模型加载
87        """
88        super().__init__(*args, **kwargs)
89        self.model_path = model_path
90        self.system_prompt = system_prompt or (
91            '你是一个专业的图像分析专家。'
92            '请仔细分析图片内容，以 JSON 格式返回结果。'
93        )
94        self.user_prompt = user_prompt or (
95            '请分析这张图片，提取能描述图片内容的标签列表。'
96            '标签应涵盖主体、场景、风格、颜色等维度。'
97            '以 JSON 格式返回，格式为：{"tags": ["标签1", "标签2", ...]}'
98        )
99        self.max_tokens = max_tokens
100        self.max_model_len = max_model_len
101        self.temperature = temperature
102        self.top_p = top_p
103        self.model = None
104        self.sampling_params = None
105        self.smart_file = SmartFile()
106        self.download_path = os.path.join(
107            os.getcwd(),
108            os.path.splitext(os.path.basename(__file__))[0]
109        )
110        print(f"[LoadLocalModelVLLM] init, model_path={model_path}, preload={preload}")
111        if preload:
112            from vllm import LLM, SamplingParams
113            import torch
114
115            os.makedirs(self.download_path, exist_ok=True)
116            full_model_path = get_model_path(model_path)
117            print(f"[LoadLocalModelVLLM] preload=True, full_model_path={full_model_path}")
118
119            t0 = time.perf_counter()
120            device = 'cuda' if torch.cuda.is_available() else 'cpu'
121            print(f"[LoadLocalModelVLLM] 开始加载模型, device={device}")
122            self.model = LLM(
123                model=full_model_path,
124                tokenizer=full_model_path,
125                trust_remote_code=True,
126                max_model_len=self.max_model_len,
127                device=device,
128            )
129            self.sampling_params = SamplingParams(
130                max_tokens=self.max_tokens,
131                temperature=self.temperature,
132                top_p=self.top_p,
133            )
134            print(f"[LoadLocalModelVLLM] 模型加载完成，耗时 {time.perf_counter() - t0:.4f}s")
135
136    def __call__(self, samples, *args, **kwargs):
137        src_images = samples[self.image_key].tolist()
138        print(f"[LoadLocalModelVLLM] __call__ 开始，batch size={len(src_images)}")
139        tags_list = []
140
141        for i, image_path in enumerate(src_images):
142            print(f"[LoadLocalModelVLLM] 处理第 {i+1}/{len(src_images)} 张: {image_path}")
143            try:
144                local_path = self.smart_file.volume_2_local(image_path, self.download_path)
145                with open(local_path, "rb") as f:
146                    image_data = base64.b64encode(f.read()).decode("utf-8")
147
148                messages = [
149                    {"role": "system", "content": [{"type": "text", "text": self.system_prompt}]},
150                    {"role": "user", "content": [
151                        {"type": "text", "text": self.user_prompt},
152                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}},
153                    ]},
154                ]
155                outputs = self.model.chat([messages], self.sampling_params)
156                raw_output = outputs[0].outputs[0].text
157                print(f"[LoadLocalModelVLLM] 推理完成，raw_output length={len(raw_output)}")
158                parsed = _parse_llm_json(raw_output)
159                tags = parsed.get('tags', parsed) if isinstance(parsed, dict) else parsed
160            except Exception as e:
161                print(f"[LoadLocalModelVLLM] 推理失败 {image_path}: {e}")
162                tags = None
163
164            tags_list.append(tags)
165
166        print(f"[LoadLocalModelVLLM] __call__ 完成，processed={len(tags_list)}")
167        samples['tags'] = tags_list
168        return samples

1cd /root/databuilder_dev
2sh ./build.sh

1cd /root/databuilder_dev/output/dist 
2pip install databuilder_vendor_operators-0.3.0-py3-none-any.whl