目录
三、VLM多模态方案
四、AI质检工作台搭建
1.构建Gradio界面
1)初始化代码
2)创建页签Tabs
3)布局
4)相关的控件
2.页面逻辑
1)缺陷检测
yolo检测
VLM检测
2)检测记录
3)统计分析
4)数据导出
五、Bad Cases持续改善
三、VLM多模态方案
VLM是用提示词的方式,描述了6种缺陷的样式,并让大模型根据样式的描述将缺陷按照规则框起来。主要逻辑如下:
response = self.client.chat.completions.create( model=self.model_name, messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{b64_image}" }, }, ], } ], max_tokens=1024, temperature=0.1, ) reply = response.choices[0].message.content.strip()先将图片转化成Base64。用提示词的方式,描述缺陷的类型,并规定了传出的类型放入多模态大模型中进行推理。解析推理内容reply,将Box框绘制在图片上,并输出出来。
大家看下效果。对比下来感觉在这个问题上yolo的速度更快,效果更好。VLM是偏通用的,而yolo偏专业,二者可以互补。如在这个案例中,实时的检测用yolo,让VLM来做辅助分析。
四、AI质检工作台搭建
如上所示,是工作台的完整设计。包含了界面的搭建,两种缺陷的检测,数据库的记录,检测记录,统计记录,数据导出。
1.构建Gradio界面
1)初始化代码
gr.Blocks创建了一个 Web 应用根容器。
with gr.Blocks( # 1.浏览器标签页显示的标题(类似 HTML 的`<title>` )。 title="AI质检工作台", # 2.使用 Gradio 内置主题 :Soft 配色柔和、有圆角阴影。Gradio 还提供`Default` 、`Base` 、`Monochrome` 等预设主题,也可以自定义颜色。 theme=gr.themes.Soft(), # 凡是带`elem_classes="result-text"` 的组件,文字都用等宽字体 css=""" .result-text { font-family: 'Consolas', 'Courier New', monospace; } """ ) as app: gr.Markdown("# AI质检工作台 - 钢铁表面缺陷检测") gr.Markdown("支持 **YOLO目标检测** 和 **Qwen-VL多模态大模型** 两种检测引擎, 可对比分析")2)创建页签Tabs
使用此方法,创建上面设计图给出的四个标签:缺陷检测,检测记录,统计分析和数据导出四个页签
with gr.Tabs(): with gr.Tab("缺陷检测"): #.... with gr.Tab("检测记录"): #....3)布局
gr.Row()为水平布局的容器,有三列,宽度比例为2:1:2
# 水平布局 with gr.Row(): with gr.Column(scale=2): # 左列 ...... with gr.Column(scale=1): # 中列 ...... with gr.Column(scale=2): # 右列 ......4)相关的控件
# 图片组件 上传,显示图片 input_image = gr.Image( label="上传钢铁表面图片", type="numpy", height=300, ) # 单选按钮,多选一 model_type = gr.Radio( choices=["YOLO", "VLM (Qwen-VL)"], value="YOLO", label="选择检测引擎", ) # 滑动条,拖动选择数值 conf_slider = gr.Slider( minimum=0.05, maximum=0.95, value=0.25, step=0.05, label="YOLO置信度阈值", visible=True, ) # gr.Accordion为折叠框 # gr.Textbox 文本框,输入/显示文本 with gr.Accordion("VLM API配置", open=False): vlm_api_key = gr.Textbox( label="API Key", value=os.environ.get("DASHSCOPE_API_KEY", ""), placeholder="未设置DASHSCOPE_API_KEY环境变量, 请手动填入", type="password", # interactive=True, 只读 ) ...... # 按钮 review_btn = gr.Button("提交审核") review_btn.click( fn=save_review, #绑定的函数 inputs=[current_log_id, review_status, review_note], outputs=[review_result], ) # 下拉框 rec_model_filter = gr.Dropdown( choices=["全部", "yolo", "vlm"], value="全部", label="模型筛选", )2.页面逻辑
1)缺陷检测
yolo检测
从文件夹里拖动一张照片到左上角位置 或者 在左下角选择测试图片并点击【加载选中图片】按钮,点击【开始检测】按钮,右上角显示检测结果和检测详情
主要逻辑:
# Step 1:加载model,model_path是之前训练好的best.pt的路径 model = YOLO(model_path) # Step 2: 推理 results = self.model.predict( source=image_path, # 图片路径 conf=conf, # 置信度 iou=iou, verbose=False, # augment=True, # 开启TTA ) # Step 3: 解析推理结果 boxes = [] if len(results) > 0 and results[0].boxes is not None: det_boxes = results[0].boxes xyxy = det_boxes.xyxy.cpu().numpy() confs = det_boxes.conf.cpu().numpy() classes = det_boxes.cls.cpu().numpy() for box, c, cls_id in zip(xyxy, confs, classes): cls_id = int(cls_id) boxes.append(DetectionBox( class_id=cls_id, #缺陷Id class_name=CLASS_NAMES.get(cls_id, f"class_{cls_id}"), #缺陷名称 confidence=float(c), #置信度 x1=float(box[0]), y1=float(box[1]), #坐标 x2=float(box[2]), y2=float(box[3]), )) # Step 4: 将推理内容整理成DetectionResult的格式 result = DetectionResult( model_type="yolo", image_path=image_path, boxes=boxes, inference_time=elapsed, ) # Step 5:将框绘制到图片上,并保存图片 result.annotated_image = self._draw_boxes(image_path, boxes)VLM检测
图片不变,【选择检测引擎】切换为VLM再看一下检测结果
主要逻辑:推理过程和上面VLM多模态检测方案一样。这里主要讲下如何处理数据的。
boxes = [] defect_types = [] description = reply try: boxes, defect_types, description = self._parse_bbox_response( reply, img_width, img_height ) except (json.JSONDecodeError, KeyError, IndexError, TypeError): description = f"VLM原始回复 (JSON解析失败):\n{reply}" # 把数据整理成统一格式 result = DetectionResult( model_type="vlm", image_path=image_path, boxes=boxes, vlm_text=description, vlm_defect_types=defect_types, inference_time=elapsed, ) # 在图片上绘制框 result.annotated_image = self._draw_boxes(image_path, boxes)# 解析Json json_str = self._extract_json(reply) parsed = json.loads(json_str) ....... for item in parsed: bbox_2d = item.get("bbox_2d") label = item.get("label", "unknown") description = item.get("description", "") if not bbox_2d or len(bbox_2d) != 4: continue x1, y1, x2, y2 = self._convert_bbox(bbox_2d, img_width, img_height) class_id = CLASS_NAME_TO_ID.get(label, -1)--------------------------------------------------------------------
在人工审核处,提交审核结果,收集bad Case。
2)检测记录
在第一行,筛选模型和审核状态,点击【刷新记录】,从数据库加载检测记录。
将检测记录列表的ID再进行详细审核,输入ID,点击【查看详情】从数据库中加载出某次记录的结果。填入审核需要的内容【提交审核】。
--------------------------------------------------------------------
之所以能加载出检测记录,是因为在上一步缺陷检测的时候将检测记录记载到sqlite3这个轻量级数据库里了。
# 打开数据库 def _get_conn(self): conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() # 存储数据 with self._get_conn() as conn: conn.execute(""" INSERT INTO detection_log (image_path, image_name, model_type, num_detections, detections_json, vlm_text, vlm_defect_types, inference_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( data.get("image_path", ""), image_name, data.get("model_type", "unknown"), len(data.get("boxes", [])), json.dumps(data.get("boxes", []), ensure_ascii=False), data.get("vlm_text", ""), json.dumps(data.get("vlm_defect_types", []), ensure_ascii=False), data.get("inference_time", 0), )) log_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] # 后面省略...sql语句大同小异3)统计分析
点击【生成统计报告】,查看统计报告。
读取数据库内保存的数据,将数据查出来,然后用特定的格式展示出来。
4)数据导出
导出误检和漏检的图片
# 导出图片和记录,bad_cases是从数据库读出的badCases exported = [] for record in bad_cases: # 拷贝图片 img_path = record["image_path"] if os.path.exists(img_path): dst = os.path.join(output_dir, "images", record["image_name"]) shutil.copy2(img_path, dst) exported.append({ "id": record["id"], "image_name": record["image_name"], "model_type": record["model_type"], "review_status": record["review_status"], "review_note": record["review_note"], "num_detections": record["num_detections"], "detections": record["detections_json"], "vlm_text": record["vlm_text"], "created_at": record["created_at"], }) # 写入CSV csv_path = os.path.join(output_dir, "bad_cases.csv") if exported: with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=exported[0].keys()) writer.writeheader() writer.writerows(exported) # 写入导出信息 export_info = { "export_time": datetime.now().isoformat(), "total_bad_cases": len(exported), "wrong_count": sum(1 for r in exported if r["review_status"] == "wrong"), "missed_count": sum(1 for r in exported if r["review_status"] == "missed"), "model_type_filter": model_type or "all", } with open(os.path.join(output_dir, "export_info.json"), "w", encoding="utf-8") as f: json.dump(export_info, f, ensure_ascii=False, indent=2) print(f"导出完成:") print(f" Bad Case 总数: {len(exported)}") print(f" 误检(wrong): {export_info['wrong_count']}") print(f" 漏检(missed): {export_info['missed_count']}") print(f" 导出目录: {output_dir}")五、Bad Cases持续改善
模型上线后性能会逐渐衰减:
• 产线环境变化(光照、材料批次、设备老化)
• 新缺陷类型出现
• 边缘case 积累
我们需要把这些Bad Case 定期的导出,用于后续的训练。完整流程是:
模型检测-->人工审核(标记误检/漏检)-->导出Bad Case-->人工重新标注-->合并到训练集-->重训练