影刀RPA 训练数据标注：AI模型数据集自动构建

影刀RPA 训练数据标注AI模型数据集自动构建作者林焱什么情况用这个做AI应用开发的同学应该深有体会模型效果好不好数据质量占七成。但高质量标注数据从哪来纯人工标注不仅慢还贵——外包团队每条数据几毛到几块不等一个中型项目动辄上万条数据需求。还有一个场景你已经在用某个AI模型比如OCR识别发票、NLP分类工单但通用模型的准确率在你们业务场景下就是差那么一点。你需要用自己的业务数据做微调Fine-tuning但首先得有标注好的数据集。用影刀RPA配合现有的AI服务来做半自动标注——机器先标一轮人工复核修正效率提升5-10倍。这就是本文要讲的内容。怎么做第一步设计标注任务的结构先明确你要标注什么数据、标注格式是什么。以最常见的三个场景为例场景A文本分类标注输入客服工单文本 输出分类标签退款/换货/咨询/投诉 格式JSONL每行 {text: ..., label: 退款}场景B图像分类标注输入商品图片 输出类目标签服装/电子/食品/家居 格式按文件夹分类存放或CSV记录路径标签场景C实体识别NER标注输入合同文本 输出标注实体甲方、乙方、金额、日期 格式BIO标注序列 [video(video-sMSRuKNa-1784927583575)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)]第二步用现有AI服务做预标注先用现成的AI服务做第一轮自动标注。比如用大模型API做文本分类预标注importrequestsimportjsonimporttimedefai_pre_label_text(text,categories,api_key,api_url): 用大模型API对文本做预分类 promptf请将以下文本分类到以下类别之一{, .join(categories)}只返回类别名称不要任何解释。 文本{text}类别headers{Authorization:fBearer{api_key},Content-Type:application/json}data{model:gpt-3.5-turbo,# 或其他模型messages:[{role:system,content:你是一个文本分类助手只输出类别名称。},{role:user,content:prompt}],temperature:0,max_tokens:20}try:resprequests.post(api_url,headersheaders,jsondata,timeout30)resultresp.json()labelresult[choices][0][message][content].strip()# 验证标签是否在类别列表中iflabelincategories:returnlabelelse:# 模糊匹配forcatincategories:ifcatinlabel:returncatreturnf待确认:{label}exceptExceptionase:returnf预标注失败:{str(e)}第三步影刀流程编排批量预标注用影刀批量处理待标注数据1. 【读取Excel】→ 读取待标注文本列表 2. 【ForEach循环】→ 遍历每条文本 3. 【Python节点】→ 调用ai_pre_label_text()预分类 4. 【写入Excel】→ 将预标注结果写入新列 5. 【等待】→ 0.5秒API速率限制 6. 【保存Excel】→ 导出预标注结果对于图像分类可以对接现有的图像识别APIimportbase64defai_pre_label_image(image_path,categories,api_key): 用多模态大模型对图片做预分类 withopen(image_path,rb)asf:image_base64base64.b64encode(f.read()).decode()headers{Authorization:fBearer{api_key},Content-Type:application/json}data{model:gpt-4o,# 支持视觉的模型messages:[{role:user,content:[{type:text,text:f这张图片最可能属于以下哪个类别{, .join(categories)}。只输出类别名。},{type:image_url,image_url:{url:fdata:image/jpeg;base64,{image_base64}}}]}],max_tokens:20}resprequests.post(https://api.openai.com/v1/chat/completions,headersheaders,jsondata,timeout30)returnresp.json()[choices][0][message][content].strip()第四步用规则引擎补充标注不是所有标注都需要AI。很多场景下规则比AI更准。用Python写规则引擎与AI标注互补importredefrule_based_label(text,rules_config): 基于规则的自动标注 优先用规则100%准确规则覆盖不了的才用AI forruleinrules_config:labelrule[label]keywordsrule.get(keywords,[])patternsrule.get(patterns,[])excluderule.get(exclude,[])# 先检查排除词ifany(exintextforexinexclude):continue# 关键词匹配ifkeywordsandany(kwintextforkwinkeywords):returnlabel,规则-关键词# 正则匹配ifpatterns:forpatinpatterns:ifre.search(pat,text):returnlabel,规则-正则returnNone,规则未命中# 规则配置示例客服工单分类RULES[{label:退款,keywords:[退款,退钱,退货退款,返还款项],exclude:[不退款,无法退款]},{label:换货,keywords:[换货,换一个,更换,调换],exclude:[]},{label:投诉,keywords:[投诉,举报,差评,曝光],patterns:[r客服.*[差垃圾烂],r服务.*太差]},{label:咨询,keywords:[请问,怎么,如何,咨询,问一下],exclude:[]}]第五步标注一致性检查预标注完成后做一致性校验——标完后找AI再审一遍把置信度低的挑出来人工复核defconfidence_check(text,label,categories,api_key): 让AI对预标注结果打分 返回置信度低于阈值的人工复核 promptf文本{text}预标注类别{label}可选类别{, .join(categories)}请判断预标注是否正确只返回数字分值1-5 5完全正确 4基本正确 3不确定 2可能错误 1肯定错误 分值# 调用API获取打分...代码类似上面# 返回分值# 简化返回return4# 占位第六步导出标准数据集格式标注完成后导出为模型训练所需的标准格式importjsonimportcsvimportosimportshutildefexport_dataset(labeled_data,output_dir,format_typejsonl): 导出标注数据集 支持的格式 - jsonl: 每行一个JSON对象 - csv: CSV表格 - folder: 按类别分文件夹图片分类 os.makedirs(output_dir,exist_okTrue)ifformat_typejsonl:output_pathos.path.join(output_dir,dataset.jsonl)withopen(output_path,w,encodingutf-8)asf:foriteminlabeled_data:f.write(json.dumps(item,ensure_asciiFalse)\n)# 统计信息labels{}foriteminlabeled_data:labelitem.get(label,unknown)labels[label]labels.get(label,0)1# 生成统计报告stats_pathos.path.join(output_dir,dataset_stats.json)withopen(stats_path,w,encodingutf-8)asf:json.dump({总数:len(labeled_data),类别分布:labels,导出格式:format_type},f,ensure_asciiFalse,indent2)returnoutput_path,stats_pathelifformat_typefolder:# 按类别分文件夹foriteminlabeled_data:labelitem[label]src_pathitem.get(image_path,)ifnotsrc_pathornotos.path.exists(src_path):continuelabel_diros.path.join(output_dir,label)os.makedirs(label_dir,exist_okTrue)dest_pathos.path.join(label_dir,os.path.basename(src_path))shutil.copy2(src_path,dest_path)有什么坑坑一API调用费用失控现象月底看账单标注5000条数据花了300块API费用完全没想到。原因每次API调用都消耗token大模型的token计费是累计的。5000条数据每条200 tokens就是100万tokens。解决先规则后AI能用规则匹配的先过滤掉只把规则覆盖不了的送AI批量处理一次API调用同时标注多条数据batch模式减少请求次数设置每日限额代码中加计数器超过预算自动停止classAPIBudget:API调用预算控制def__init__(self,daily_limit1000):self.daily_limitdaily_limit self.today_count0self.today_datedatetime.date.today()defcan_call(self):todaydatetime.date.today()iftoday!self.today_date:self.today_count0self.today_datetodayifself.today_countself.daily_limit:returnFalseself.today_count1returnTruedefremaining(self):returnmax(0,self.daily_limit-self.today_count)坑二预标注错误引入系统性偏差现象训练出来的模型表现很奇怪对某些类型的数据总是分类错误。原因AI预标注本身有偏见比如模型把价格太贵了总是标为投诉而非咨询你没纠正就直接用来训练了等于用错误答案训练模型。解决必须有人工复核环节。影刀可以把置信度低的标出来生成一个待人工确认列表。关键指标预标注准确率低于90%的数据集人工复核比例不能低于20%defsampling_for_review(labeled_data,confidence_threshold3,sample_rate0.2): 抽样供人工复核 - 所有置信度低于阈值的全部复核 - 置信度高的随机抽取 sample_rate 比例复核 importrandom review_list[]foriteminlabeled_data:confidenceitem.get(confidence,5)# 默认满分ifconfidenceconfidence_threshold:review_list.append({**item,复核原因:低置信度})# 从高置信度数据中随机抽取high_conf[itemforiteminlabeled_dataifitem.get(confidence,5)confidence_threshold]sample_countmax(int(len(high_conf)*sample_rate),10)sampledrandom.sample(high_conf,min(sample_count,len(high_conf)))foriteminsampled:review_list.append({**item,复核原因:随机抽检})returnreview_list坑三数据格式不统一TEMU店群如何管理运营现象导出的数据集训练时各种报错编码问题、缺少必填字段、JSON格式错误。原因不同来源的数据格式不一样合并导出时没有做统一归一化。解决在标注开始前先定义一个严格的Schema所有数据在进入标注流程前先做格式转换# 定义标准数据格式STANDARD_SCHEMA{id:str,# 唯一IDtext:str,# 文本内容必填label:str,# 标注标签必填confidence:int,# 置信度1-5source:str,# 数据来源labeled_by:str,# 标注方式ai/rule/humanlabeled_at:str,# 标注时间ISO格式reviewed:bool,# 是否已人工复核}defnormalize_data(raw_data,schemaSTANDARD_SCHEMA):数据格式归一化normalized{}forfield,field_typeinschema.items():valueraw_data.get(field)iffieldin[text,label]andnotvalue:raiseValueError(f必填字段{field}缺失)normalized[field]field_type(value)ifvalueisnotNoneelsereturnnormalized坑四图片数据集文件路径混乱现象数据集导出后图片文件和标注记录对不上。原因标注过程中文件被移动/重命名或者路径使用了相对路径。解决全程使用绝对路径标注记录中同时保存文件路径和MD5哈希importhashlibdefget_file_hash(file_path):计算文件MD5用于唯一标识hasherhashlib.md5()withopen(file_path,rb)asf:forchunkiniter(lambda:f.read(4096),b):hasher.update(chunk)returnhasher.hexdigest()# 标注记录示例image_record{image_path:os.path.abspath(D:/data/images/img001.jpg),md5:get_file_hash(D:/data/images/img001.jpg),label:电子,confidence:5}坑五标注工具选择不当现象标注效率低、容易出错、多人协作困难。原因用Excel直接编辑标注数据没有分类筛选、没有快捷键、没有一致性检查。解决如果标注量超过1000条建议对接专业标注工具如Label Studio、Doccano的API。影刀负责数据预处理和导出标注工具的API负责提供高效的标注界面defupload_to_labelstudio(data_list,labelstudio_url,api_token,project_id):将预标注数据导入Label Studioimportrequests headers{Authorization:fToken{api_token}}tasks[]foritemindata_list:task{data:{text:item[text]},predictions:[{result:[{from_name:label,to_name:text,type:choices,value:{choices:[item[label]]}}]}]ifitem.get(label)else[]}tasks.append(task)resprequests.post(f{labelstudio_url}/api/projects/{project_id}/import,headersheaders,jsontasks)returnresp.json()总结用影刀RPA做AI数据标注不是要替代人工标注而是做预标注辅助校验。机器标得快但不一定准人标得准但慢——两者结合才是最优解。核心流程是规则过滤→AI预标→置信度打分→低分送人工复核→导出标准格式。这套流程跑顺了标注效率提升5-10倍同时质量还有保障。