YOLO+VLM：AOI的下一站，从“检出一个缺陷”到“维护一个产品状态”

class DefectState:
    """AOI缺陷状态条目——多模型融合后的统一输出"""
    def __init__(self):
        self.defects = {}  # defect_id -> DefectEntry
        
    class DefectEntry:
        def __init__(self, defect_id, category, bbox, confidence):
            self.id = defect_id
            self.category = category        # YOLO检测结果
            self.bbox_2d = bbox              # 二维框
            self.position_3d = None          # Depth/Stereo提供
            self.area_pixel = None           # Segmentation提供
            self.depth_mm = None             # Depth估计提供
            self.timestamp = now()           # 时间戳
            self.track_id = None             # Tracker提供
            self.semantic_judgment = None    # VLM提供
            self.confidence = confidence     # 多模型交叉验证后的置信度

class AOIMultiModelPipeline:
    """AOI多模型协同流水线——让不同模型回答不同问题"""
    
    def __init__(self):
        self.detector = YOLO("yolo26n.pt")      # 检测：是什么
        self.segmenter = SegmentationModel()     # 分割：边界怎样
        self.depth_estimator = YOLO("yolo26n-depth.pt")  # 深度：离多远/多深
        self.tracker = SORT()                    # 跟踪：还是不是它
        self.vlm = VLMClient()                   # VLM：意味着什么
        
    def process_board(self, image, board_id):
        # Step 1: 检测——YOLO26快速检出候选缺陷
        det_results = self.detector(image, conf=0.25)
        defects = []
        
        for det in det_results:
            # Step 2: 分割——精确定义缺陷边界
            mask = self.segmenter(image, det.bbox)
            
            # Step 3: 深度——获取缺陷深度信息（3D AOI场景）
            depth_map = self.depth_estimator(image)
            depth_value = depth_map[det.bbox]
            
            # Step 4: VLM——模糊/新型缺陷语义判断（仅低置信度触发）
            if det.conf < 0.7 or det.category == "unknown":
                semantic = self.vlm.analyze(image, det.bbox, 
                    prompt="判断该缺陷是否达到报废标准，并说明原因")
            else:
                semantic = None
                
            defects.append({
                "category": det.category,
                "bbox": det.bbox,
                "mask": mask,
                "depth": depth_value,
                "area_px": mask.area(),
                "confidence": det.conf,
                "vlm_judgment": semantic
            })
        
        # Step 5: 跟踪——与历史批次同一位置关联
        for defect in defects:
            defect.track_id = self.tracker.update(defect.bbox, board_id)
            
        return self._build_board_state(board_id, defects)

class AOIScheduler:
    """AOI模型调度器——不同模型不同频率，按需触发"""
    
    def __init__(self):
        self.detector = YOLO("yolo26n.pt")
        self.segmenter = SegmentationModel()
        self.depth_estimator = YOLO("yolo26n-depth.pt") if USE_3D else None
        self.vlm = VLMClient()
        self.tracker = SORT()
        
        # 调度策略
        self.detect_every_n_frames = 1      # 每帧检测
        self.segment_on_defect = True       # 检出缺陷时才分割
        self.depth_on_3d_required = True    # 需要三维信息时才开启
        self.vlm_on_low_conf = True         # 低置信度时才调用VLM
        
    def run_pipeline(self, frame, frame_id):
        # 1. 检测——每帧都跑
        detections = self.detector(frame, conf=0.25)
        
        # 2. 跟踪——每帧更新
        self.tracker.update(detections)
        
        # 3. 分割——只在检出缺陷时触发
        if detections and self.segment_on_defect:
            for det in detections:
                det.mask = self.segmenter(frame, det.bbox)
                
        # 4. 深度——只在需要3D信息时触发（如靠近目标或配置开启）
        depth_map = None
        if self.depth_estimator and self.depth_on_3d_required:
            depth_map = self.depth_estimator(frame)
            
        # 5. VLM——只在低置信度、新型缺陷或语义歧义时触发
        for det in detections:
            if det.conf < 0.7 or det.category == "unknown":
                det.vlm_result = self.vlm.analyze(
                    frame, det.bbox,
                    prompt="分析该缺陷类型、严重程度和是否符合IPC标准"
                )
                
        return self._merge_results(frame_id, detections, depth_map)

class AOISharedBuffer:
    """共享帧Buffer——避免多模型间的数据重复拷贝"""
    
    def __init__(self):
        self.frame_buffer = None          # 统一的图像Tensor
        self.buffer_owner = None          # 当前持有者
        self.timestamp = None
        self.lock = threading.Lock()
        
    def acquire(self, model_name):
        """模型获取帧数据（共享读取，不拷贝）"""
        with self.lock:
            if self.frame_buffer is None:
                return None
            # 返回引用，而非拷贝
            return self.frame_buffer
        
    def release(self, model_name):
        """模型释放帧数据"""
        with self.lock:
            # 减少引用计数，不释放原始Buffer
            pass

class AOIPipeline:
    def process_frame(self, raw_frame):
        # 写入共享Buffer（零拷贝）
        shared = AOISharedBuffer()
        shared.frame_buffer = raw_frame  # 仅保存引用
        
        # 多模型并行读取同一份数据
        det_thread = Thread(target=self.detector.predict, args=(shared,))
        depth_thread = Thread(target=self.depth_estimator.predict, args=(shared,))
        
        det_thread.start()
        depth_thread.start()
        # ...
        
        # 所有模型读完后再释放
        # 避免在模型读取期间修改Buffer

class AOIFusionEngine:
    """多模型交叉验证融合——不是取平均，而是验证一致性"""
    
    def fuse(self, det_result, seg_result, depth_result, history, vlm_result):
        """融合多模型结果，输出最终可信度"""
        base_conf = det_result.confidence
        
        # 条件1：分割一致性——缺陷边界清晰且面积合理
        if seg_result and seg_result.area > 20:
            base_conf *= 1.1
        
        # 条件2：深度一致性——深度值在合理范围内
        if depth_result and 0.1 < depth_result < 10:
            base_conf *= 1.1
        
        # 条件3：历史一致性——同一位置历史缺陷记录
        historical = history.get(det_result.bbox, None)
        if historical:
            if historical.category == det_result.category:
                base_conf *= 1.2  # 历史一致，信任度提升
            else:
                base_conf *= 0.8  # 历史不一致，怀疑
            if historical.count > 5:
                # 连续多块板同一位置出现，预警工艺问题
                self.alert_工艺偏差(det_result.bbox)
        
        # 条件4：VLM语义验证
        if vlm_result:
            if vlm_result.severity == "critical":
                base_conf *= 1.3
            elif vlm_result.severity == "acceptable":
                base_conf *= 0.7
        
        # 时间一致性检查——确保所有结果来自同一帧
        assert det_result.timestamp == seg_result.timestamp == depth_result.timestamp
        
        return min(base_conf, 1.0)

class AOISceneState:
    """AOI最终输出：统一的产品质量状态"""
    
    def __init__(self):
        self.board_id = None
        self.defects = []           # 缺陷列表
        self.pass_fail = "PASS"     # 整板判定
        self.timestamp = now()
        self.metadata = {}          # 工艺参数、批次信息等

class AOIAgent:
    """AOI智能调度器——按需调用模型，维护统一状态"""
    
    def process(self, board_image, board_id, task):
        # 判断任务类型
        if task == "quick_scan":
            # 只需快速筛查
            results = self.lightweight_model(board_image)
            return self._build_state(results)
            
        elif task == "defect_analysis":
            # 检出缺陷后详细分析
            detections = self.yolo26(board_image)
            for det in detections:
                # 按需调用
                if det.conf < 0.7:
                    det.vlm_result = self.vlm(board_image, det.bbox)
                if self.need_3d:
                    det.depth = self.depth_estimator(board_image, det.bbox)
                if self.need_precision:
                    det.mask = self.segmenter(board_image, det.bbox)
            return self._build_state(detections)
            
        elif task == "conflict_resolution":
            # 结果冲突时调用VLM重新判断
            return self.vlm(board_image, task.context)