1准备开发环境
2      ↓
3编写并测试 Python 算子
4      ↓
5打包为 whl 文件
6      ↓
7在百度胜算中创建算子和算子版本
8      ↓
9将自定义算子加入工作流
10      ↓
11运行工作流并查看结果

1# 1. 下载模型包
2wget -O db_operator_dev.tar "https://offical-images.bj.bcebos.com/ray/ray-2.1.7.tar?authorization=bce-auth-v1/33bda7b5e8ea4d669c2c7944449665c6/2026-08-27T11:13:55Z/-1/host/9e2b6bc9d3ab41c2d4b140ea9f2509dac30d44d23a8029967190593a52fb3589"
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 \
6  iregistry.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            'text': '特殊的康熙部首或者扩展部首会被去除，⼏几⺇'
5        },
6        {
7            'text': '请问你是谁dasoidhao@1264fg.45om'
8        },
9        {
10            'text': '匹配汉字包括繁體字'
11        }
12    ]
13
14    input_dataset = RayDataset.from_list(samples)
15
16    # 2. create operator
17    op = PunctuationReplacer()
18
19    # 3. run
20    output_dataset = input_dataset.run(op)
21
22    # 4. get & check result
23    text_data = output_dataset.get_column(column=op.text_key)
24    
25    print(f'---result: {text_data}')

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}')

1import pandas as pd
2import ray
3from typing import Dict
4import numpy as np
5import torch
6
7from palette.ops.base import Transformer
8from palette.core.ray_dataset import RayDataset
9
10# 1.写自定义算子
11class TorchPredictor(Transformer):
12    ##算子设置为transform类型
13    _op_type = "transform"
14    ##算子设置为批处理，即__call__中接收一批数据，返回一批数据的处理结果
15    _batched_op = True
16    ## 算子的名称
17    _name = "operator_transform"
18    ## 设置算子执行模式为PIPELINE_ACTOR，这样init只会执行一次
19    _ray_execute_mode = "PIPELINE_ACTOR"
20    ## 设置算子输入数据格式为numpy
21    _ray_batch_format = "numpy"
22    ## 若算子需要使用GPU，则设置算子处理器为cuda
23    _processor = "cuda"
24
25
26    def __init__(self, *args, **kwargs):
27        ##调用父类的初始化函数
28        super().__init__(*args, **kwargs)
29
30        ##初始化模型，也可以使用模型权重文件加载
31        self.model = torch.nn.Identity().cuda()
32        self.model.eval()
33
34    ## 实现算子的__call__方法，接收一批数据，返回一批数据的处理结果，在__call__中调用初始化函数中的模型进行推理
35    def __call__(self, batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
36        inputs = torch.as_tensor(batch["data"], dtype=torch.float32).cuda()
37        with torch.inference_mode():
38            ## 调用模型进行推理
39            batch["output"] = self.model(inputs).detach().cpu().numpy()
40        return batch

1import pandas as pd
2import ray
3from typing import Dict
4import numpy as np
5import torch
6
7from palette.ops.base import Transformer
8from palette.core.ray_dataset import RayDataset
9
10# 1.写自定义算子
11class TorchPredictor(Transformer):
12    ##算子设置为transform类型
13    _op_type = "transform"
14    ##算子设置为批处理，即__call__中接收一批数据，返回一批数据的处理结果
15    _batched_op = True
16    ## 算子的名称
17    _name = "operator_transform"
18    ## 设置算子执行模式为PIPELINE_ACTOR，这样init只会执行一次
19    _ray_execute_mode = "PIPELINE_ACTOR"
20    ## 设置算子输入数据格式为numpy
21    _ray_batch_format = "numpy"
22    ## 若算子需要使用GPU，则设置算子处理器为GPU
23    _processor = "cuda"
24
25
26    def __init__(self, *args, **kwargs):
27        ##调用父类的初始化函数
28        super().__init__(*args, **kwargs)
29
30        ##初始化模型，也可以使用模型权重文件加载
31        self.model = torch.nn.Identity().cuda()
32        self.model.eval()
33
34    ## 实现算子的__call__方法，接收一批数据，返回一批数据的处理结果，在__call__中调用初始化函数中的模型进行推理
35    def __call__(self, batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
36        inputs = torch.as_tensor(batch["data"], dtype=torch.float32).cuda()
37        with torch.inference_mode():
38            ## 调用模型进行推理
39            batch["output"] = self.model(inputs).detach().cpu().numpy()
40        return batch
41
42# 2.构造数据测试自定义算子
43# 2.1. create dataset
44samples = ray.data.from_numpy(np.ones((32, 100)))
45input_dataset = RayDataset(ray_dataset=samples)
46
47# 2.2. create operator
48op = TorchPredictor()
49
50# 2.3. run
51output_dataset = input_dataset.run(op)
52
53# 2.4. show result
54output_dataset.data.show()

1* 测试结果：在本地执行test.py

1# 1. input data
2    samples = [
3        {
4            'text': '特殊的康熙部首或者扩展部首会被去除，⼏几⺇'
5        },
6        {
7            'text': '请问你是谁dasoidhao@1264fg.45om'
8        },
9        {
10            'text': '匹配汉字包括繁體字'
11        }
12    ]
13
14    input_dataset = RayDataset.from_list(samples)
15
16    # 2. create operator
17    op = PunctuationFilter()
18
19    # 3. run
20    output_dataset = input_dataset.run(op)
21
22    # 4. get & check result
23    text_data = output_dataset.get_column(column=op.text_key)
24    
25    print(f'---result: {text_data}')

1* 保持与输入相同的记录数（3条）
2* 每个输出包含处理后的图像路径
3* 路径格式：`./ret/原始文件名.jpg`

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

1/root/databuilder_dev/output/dist

1cd /root/databuilder_dev/output/dist
2pip install databuilder_vendor_operators-<版本>-py3-none-any.whl

1import base64
2import json
3import time
4import os
5import re
6
7from palette.ops.base import Transformer
8from palette.util.file_utils import SmartFile
9from databuilder.model.util import get_model_path
10
11system_prompt = """
12You are an expert driving-scene annotator. You classify what is visible in a single image and produce a JSON object with fixed keys.xxx
13"""
14user_prompt = """You will receive a single forward-driving scene image (e.g. base64 or URL). xxxxx"""
15
16def run_inference(image_path, system_prompt_text, user_prompt_text, model, sampling_params):
17    inference_start_time = time.perf_counter()
18    print(f"Start to annotation input image {image_path}. Start timestamp: {inference_start_time}#############")
19    user_content = [
20        {"type": "text", "text": user_prompt_text},
21    ]
22
23    with open(image_path, "rb") as image_file:
24        image_data = base64.b64encode(image_file.read()).decode("utf-8")
25        new_image = {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}}
26        user_content.append(new_image)
27
28    messages = [{"role": "system", "content": [{"type": "text", "text": system_prompt_text}]},
29                {"role": "user", "content": user_content}]
30
31    outputs = model.chat([messages], sampling_params)
32    inference_end_time = time.perf_counter()
33    inference_elapsed_time = inference_end_time - inference_start_time
34    print(f"Finish annotation input image {image_path}. End timestamp: {inference_end_time}, use {inference_elapsed_time:.4f}s#############")
35    return outputs[0].outputs[0].text
36
37def parse_llm_json(llm_output):
38    try:
39        # 1. 尝试使用正则表达式提取 ```json ... ``` 或 ``` ... ``` 中的内容
40        # re.DOTALL 让 . 也能匹配换行符
41        pattern = r"```(?:json)?\s*(.*?)```"
42        match = re.search(pattern, llm_output, re.DOTALL)
43
44        if match:
45            # 如果找到了 Markdown 代码块，提取其中的内容
46            json_str = match.group(1)
47        else:
48            # 2. 如果没找到 Markdown 标记，尝试直接寻找最外层的 {} 或 []
49            # 这一步是为了处理 "好的，这是你的 JSON：{...}" 这种情况
50            # 寻找第一个 { 或 [ 开始，到最后一个 } 或 ] 结束
51
52            # 简单的启发式搜索：找到第一个 { 或 [
53            start_index = -1
54            end_index = -1
55
56            # 寻找最早出现的 { 或 [
57            first_curly = llm_output.find('{')
58            first_square = llm_output.find('[')
59
60            if first_curly == -1 and first_square == -1:
61                # 既没有 { 也没有 [，可能不是 JSON
62                raise ValueError("未找到 JSON 开始符号")
63
64            # 确定开始位置
65            if first_curly != -1 and (first_square == -1 or first_curly < first_square):
66                start_index = first_curly
67                end_index = llm_output.rfind('}') + 1
68            else:
69                start_index = first_square
70                end_index = llm_output.rfind(']') + 1
71
72            if start_index != -1 and end_index != -1:
73                json_str = llm_output[start_index:end_index]
74            else
75                json_str = llm_output
76
77        # 3. 解析 JSON
78        return json.loads(json_str)
79
80    except json.JSONDecodeError as e:
81        print(f"JSON 解析失败: {e}")
82        return None
83    except Exception as e:
84        print(f"发生错误: {e}")
85        return None
86
87#定义算子类ImageAnnotationVLLM，继承于Transformer
88class ImageAnnotationVLLM(Transformer):
89    _op_type = "transform"  # 算子类型：转换类
90    _batched_op = True  # 支持批量处理
91    _processor = 'cuda'  # 默认使用CPU加速，设置为cuda表示算子需要使用GPU资源运行
92    _name = "image_annotation_vllm"  # 算子名称
93    _ray_execute_mode = "PIPELINE_ACTOR"  # Ray的Actor执行模式
94    _ray_batch_format = "numpy"  # 批处理格式
95
96    def __init__(self,
97                 max_tokens: int = 4096, 
98                 max_model_len: int = 8192,
99                 temperature: float = 0.7,
100                 top_p: float = 0.9,
101                 model_name: str = None,
102                 preload: bool = False, #若算子有模型加载等一系列数据处理过程中只运行一次的操作，可以在构造器中加入preload参数，当preload为True完成模型加载等一系列操作。
103                 *args, **kwargs):
104        super().__init__(*args, **kwargs)
105        self.model = None
106        self.sampling_params = None
107        self.smart_file = SmartFile()
108        self.download_path = os.path.join(os.getcwd(), os.path.splitext(os.path.basename(__file__))[0])
109        self.model_name = model_name
110        self.max_tokens = max_tokens
111        self.max_model_len = max_model_len
112        self.temperature = temperature
113        self.top_p = top_p
114
115        #自定义算子创建时，可设置依赖的模型版本，Databuilder会自动下载和复用模型文件。此处可以通过get_model_path来获取依赖模型的下载地址
116        self.model_path = get_model_path(self.model_name)
117
118        # 当preload为True时，完成模型加载和下载目录的初始化，此处可以保证整个数据处理过程中，模型只会加载一次。
119        if preload:
120            os.makedirs(self.download_path, exist_ok=True)
121
122            print("preload model")
123            self.init_model()
124
125    def init_model(self):
126        from vllm import LLM, SamplingParams
127        process_start_time = time.perf_counter()
128        
129        # 使用LLM加载本地的模型文件
130        self.model = LLM(
131            model=self.model_path,
132            trust_remote_code=True,
133            max_model_len=self.max_model_len,
134        )
135
136        self.sampling_params = SamplingParams(
137            max_tokens=self.max_tokens,
138            temperature=self.temperature,
139            top_p=self.top_p,
140        )
141
142        process_elapsed_time = time.perf_counter() - process_start_time
143        print(f"Finish loading {self.model_name} model by vllm. Use {process_elapsed_time:.4f}s#############")
144
145    def download_image(self, image_path):
146        local_path = self.smart_file.volume_2_local(image_path, self.download_path)
147        return local_path
148
149    #实现transform算子的transform_batched函数，因为_ray_batch_format设置为numpy，所以sample的类型为Dict[str, numpy.ndarray]
150    def transform_batched(self, samples):
151        #取出samples中的images列
152        src_images = samples[self.image_key]
153        #最后返回图片名及标注信息，由此创建两个list。
154        file_name = []
155        annotations = []
156        #对于每张图片，先从volume下载到本地，随后传入图片下载路径、模型等参数完成模型推理。
157        for image_path in src_images:
158            local_path = self.download_image(image_path)
159            annotation = run_inference(local_path, system_prompt, user_prompt, self.model, self.sampling_params)
160            ret = parse_llm_json(annotation)
161            #将图片名加入file_name
162            file_name.append(image_path)
163            #将标注信息加入annotations
164            annotations.append(ret)
165        
166        #最后返回一个Dict[str, list]也可返回Dict[str, numpy.ndarray]，包含两个元素分别是file_name和annotations。
167        return {"file_name": file_name, "annotations": annotations}

1import base64
2import json
3import re
4import time
5import os
6
7from databuilder.model.util import get_model_path
8from palette.ops.base import Transformer
9from palette.util.file_utils import SmartFile, make_unique_name_by_dict
10
11system_prompt = """
12You are an expert driving-scene annotator. You classify what is visible in a single image and produce a JSON object with fixed keys.xxx
13"""
14user_prompt = """You will receive a single forward-driving scene image (e.g. base64 or URL). Classify all attributes below.xxx
15"""
16
17def run_inference(image_path, system_prompt_text, user_prompt_text, model, processor):
18    inference_start_time = time.perf_counter()
19    print(f"Start to annotation input image {image_path}. Start timestamp: {inference_start_time}#############")
20    with open(image_path, "rb") as image_file:
21        image_data = base64.b64encode(image_file.read()).decode("utf-8")
22    # prepare Qwen-VL message template
23    user_content = [
24            {"type": "text", "text": user_prompt_text},
25            {"type": "image", "image": image_data}
26        ]
27    messages = [{"role": "system", "content": [{"type": "text", "text": system_prompt_text}]}, {"role": "user", "content": user_content}]
28
29    inputs = processor.apply_chat_template(
30        messages, tokenize=True, add_generation_prompt=True, return_dict=True, return_tensors="pt"
31    )
32
33    inputs = inputs.to(model.device)
34
35    generated_ids = model.generate(**inputs, max_new_tokens=4096)
36    generated_ids_trimmed = [
37        out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
38    ]
39    output_text = processor.batch_decode(
40        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
41    )
42
43    inference_end_time = time.perf_counter()
44    inference_elapsed_time = inference_end_time - inference_start_time
45    print(f"Finish annotation input image {image_path}. End timestamp: {inference_end_time}, use {inference_elapsed_time:.4f}s#############")
46    return output_text[0]
47
48def parse_llm_json(llm_output):
49    try:
50        # 1. 尝试使用正则表达式提取 ```json ... ``` 或 ``` ... ``` 中的内容
51        # re.DOTALL 让 . 也能匹配换行符
52        pattern = r"```(?:json)?\s*(.*?)```"
53        match = re.search(pattern, llm_output, re.DOTALL)
54
55        if match:
56            # 如果找到了 Markdown 代码块，提取其中的内容
57            json_str = match.group(1)
58        else:
59            # 2. 如果没找到 Markdown 标记，尝试直接寻找最外层的 {} 或 []
60            # 这一步是为了处理 "好的，这是你的 JSON：{...}" 这种情况
61            # 寻找第一个 { 或 [ 开始，到最后一个 } 或 ] 结束
62
63            # 简单的启发式搜索：找到第一个 { 或 [
64            start_index = -1
65            end_index = -1
66
67            # 寻找最早出现的 { 或 [
68            first_curly = llm_output.find('{')
69            first_square = llm_output.find('[')
70
71            if first_curly == -1 and first_square == -1:
72                # 既没有 { 也没有 [，可能不是 JSON
73                raise ValueError("未找到 JSON 开始符号")
74
75            # 确定开始位置
76            if first_curly != -1 and (first_square == -1 or first_curly < first_square):
77                start_index = first_curly
78                end_index = llm_output.rfind('}') + 1
79            else:
80                start_index = first_square
81                end_index = llm_output.rfind(']') + 1
82
83            if start_index != -1 and end_index != -1:
84                json_str = llm_output[start_index:end_index]
85            else:
86                json_str = llm_output
87
88        # 3. 解析 JSON
89        return json.loads(json_str)
90
91    except json.JSONDecodeError as e:
92        print(f"JSON 解析失败: {e}")
93        return None
94    except Exception as e:
95        print(f"发生错误: {e}")
96        return None
97
98#定义算子类ImageAnnotationVLLM，继承于Transformer
99class ImageAnnotationTransform(Transformer):
100    _op_type = "transform"  # 算子类型：转换类
101    _batched_op = True  # 支持批量处理
102    _processor = 'cuda'  # 默认使用CPU加速，设置为cuda表示算子需要使用GPU资源运行
103    _name = "image_annotation_transform"  # 算子名称
104    _ray_execute_mode = "PIPELINE_ACTOR"  # Ray的Actor执行模式
105    _ray_batch_format = "numpy"  # 批处理格式
106
107    def __init__(self,
108                 model_name: str = None,
109                 preload: bool = False, #若算子有模型加载等一系列数据处理过程中只运行一次的操作，可以在构造器中加入preload参数，当preload为True完成模型加载等一系列操作。
110                 *args, **kwargs):
111        super().__init__(*args, **kwargs)
112        self.processor = None
113        self.model = None
114        self.smart_file = SmartFile()
115        self.download_path = os.path.join(os.getcwd(), os.path.splitext(os.path.basename(__file__))[0])
116        self.model_name = model_name
117        
118        #自定义算子创建时，可设置依赖的模型版本，Databuilder会自动下载和复用模型文件。此处可以通过get_model_path来获取依赖模型的下载地址
119        self.model_path = get_model_path(self.model_name)
120
121        # 当preload为True时，完成模型加载和下载目录的初始化，此处可以保证整个数据处理过程中，模型只会加载一次。
122        if preload:
123            print("preload model")
124            self.init_model()
125
126    def init_model(self):
127        # 使用transformers加载本地的模型文件
128        from transformers import AutoModelForImageTextToText, AutoProcessor
129        process_start_time = time.perf_counter()
130        self.model = AutoModelForImageTextToText.from_pretrained(
131            self.model_path,
132            dtype="auto",
133            device_map="auto"
134        )
135        self.processor = AutoProcessor.from_pretrained(self.model_path)
136        process_elapsed_time = time.perf_counter() - process_start_time
137        print(f"Finish loading {self.model_name} model. Use {process_elapsed_time:.4f}s#############")
138
139    def download_image(self, image_path):
140        local_path = self.smart_file.volume_2_local(image_path, self.download_path)
141        return local_path
142
143    #实现transform算子的transform_batched函数，因为_ray_batch_format设置为numpy，所以sample的类型为Dict[str, numpy.ndarray]
144    def transform_batched(self, samples):
145        #取出samples中的images列
146        src_images = samples[self.image_key]
147        #最后返回图片名及标注信息，由此创建两个list。
148        file_name = []
149        annotations = []
150        #对于每张图片，先从volume下载到本地，随后传入图片下载路径、模型等参数完成模型推理。
151        for image_path in src_images:
152            os.makedirs(self.download_path, exist_ok=True)
153            local_path = self.download_image(image_path)
154            annotation = run_inference(local_path, system_prompt, user_prompt, self.model, self.processor)
155            ret = parse_llm_json(annotation)
156            #将图片名加入file_name
157            file_name.append(image_path)
158            #将标注信息加入annotations
159            annotations.append(ret)
160            
161        #最后返回一个Dict[str, list]也可返回Dict[str, numpy.ndarray]，包含两个元素分别是file_name和annotations。
162        return {"file_name": file_name, "annotations": annotations}

1* 推荐版本以4段式表示，每个版本段数字以递增的方式修改，正式版本更新时更新前3个版本段数字，非正式版本仅更新第4个版本段数字
2    * 例如 0.0.1.12
3
4* 不管正式版本还是非正式版本，上线到DataBuilder平台时，都需要更新版本号，更新前后相同的版本号whl包DataBuilder平台不会拉取更新