2025年物体检测技术全面解析：从传统到零样本的完整入门指南

# 基于Transformer的高级物体检测器
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import DetrForObjectDetection, DetrImageProcessor
from PIL import Image
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as patches
from typing import List, Dict, Tuple, Optional, Union

class TransformerObjectDetector:
    def __init__(self, model_name: str = "facebook/detr-resnet-50"):
        # 加载预训练的DETR模型
        self.model = DetrForObjectDetection.from_pretrained(model_name)
        self.processor = DetrImageProcessor.from_pretrained(model_name)
        self.model.eval()  # 设置为评估模式
        
        # 默认参数设置
        self.default_params = {
            "confidence_threshold": 0.5,
            "max_detections": 100,
            "device": "cuda" if torch.cuda.is_available() else "cpu"
        }
        
        # 将模型移至指定设备
        self.model.to(self.default_params["device"])
        
        # 获取COCO数据集的类别名称
        self.coco_classes = self.processor.image_processor.id2label

    def detect_objects(self, image: Union[Image.Image, np.ndarray], **kwargs) -> Dict:
        """执行物体检测"""
        # 合并默认参数和用户提供的参数
        params = {**self.default_params, **kwargs}
        
        # 预处理图像
        inputs = self.processor(images=image, return_tensors="pt")
        inputs = {k: v.to(params["device"]) for k, v in inputs.items()}
        
        # 模型推理
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        # 后处理检测结果
        target_sizes = torch.tensor([image.size[::-1]]).to(params["device"])
        results = self.processor.post_process_object_detection(
            outputs,
            threshold=params["confidence_threshold"],
            target_sizes=target_sizes
        )[0]
        
        # 整理结果
        detections = {
            "boxes": results["boxes"].cpu().numpy(),
            "scores": results["scores"].cpu().numpy(),
            "labels": results["labels"].cpu().numpy(),
            "label_names": [self.coco_classes[label.item()] for label in results["labels"]]
        }
        
        # 限制最大检测数量
        if len(detections["boxes"]) > params["max_detections"]:
            # 按置信度排序并取前N个
            top_indices = np.argsort(detections["scores"])[-params["max_detections"]:][::-1]
            for key in detections:
                detections[key] = detections[key][top_indices]
        
        return detections

    def visualize_detections(self, image: Union[Image.Image, np.ndarray], detections: Dict, 
                            show_labels: bool = True, show_confidence: bool = True) -> np.ndarray:
        """可视化检测结果"""
        # 确保图像是numpy数组格式
        if isinstance(image, Image.Image):
            image_np = np.array(image).copy()
        else:
            image_np = image.copy()
        
        # 创建绘图对象
        fig, ax = plt.subplots(1, figsize=(12, 9))
        ax.imshow(image_np)
        
        # 为每个检测结果绘制边界框
        for i, (box, score, label_name) in enumerate(zip(
                detections["boxes"], detections["scores"], detections["label_names"])):
            # 将边界框坐标转换为整数
            x_min, y_min, x_max, y_max = box.astype(int)
            
            # 生成唯一颜色
            color = plt.cm.hsv(i / len(detections["boxes"]))[:3]
            
            # 绘制边界框
            rect = patches.Rectangle((x_min, y_min), x_max - x_min, y_max - y_min, 
                                    linewidth=2, edgecolor=color, facecolor='none')
            ax.add_patch(rect)
            
            # 绘制标签和置信度
            if show_labels:
                label_text = label_name
                if show_confidence:
                    label_text += f" ({score:.2f})"
                
                # 放置标签文本
                plt.text(x_min, y_min - 10, label_text, color='white', 
                        fontsize=10, bbox=dict(facecolor=color, alpha=0.7))
        
        # 隐藏坐标轴
        ax.axis('off')
        
        # 将绘制结果转换为numpy数组
        plt.tight_layout()
        fig.canvas.draw()
        result_image = np.frombuffer(fig.canvas.tostring_rgb(), dtype=np.uint8)
        result_image = result_image.reshape(fig.canvas.get_width_height()[::-1] + (3,))
        plt.close(fig)
        
        return result_image

    def batch_detect_objects(self, images: List[Union[Image.Image, np.ndarray]], 
                           batch_size: int = 8, **kwargs) -> List[Dict]:
        """批量检测物体"""
        # 合并默认参数和用户提供的参数
        params = {**self.default_params, **kwargs}
        
        all_detections = []
        
        # 分批处理图像
        for i in range(0, len(images), batch_size):
            batch_images = images[i:i+batch_size]
            
            # 预处理批量图像
            inputs = self.processor(images=batch_images, return_tensors="pt")
            inputs = {k: v.to(params["device"]) for k, v in inputs.items()}
            
            # 模型推理
            with torch.no_grad():
                outputs = self.model(**inputs)
            
            # 后处理每个图像的检测结果
            target_sizes = torch.tensor([img.size[::-1] for img in batch_images]).to(params["device"])
            results = self.processor.post_process_object_detection(
                outputs,
                threshold=params["confidence_threshold"],
                target_sizes=target_sizes
            )
            
            # 整理每个图像的结果
            for j, result in enumerate(results):
                detections = {
                    "boxes": result["boxes"].cpu().numpy(),
                    "scores": result["scores"].cpu().numpy(),
                    "labels": result["labels"].cpu().numpy(),
                    "label_names": [self.coco_classes[label.item()] for label in result["labels"]]
                }
                
                # 限制最大检测数量
                if len(detections["boxes"]) > params["max_detections"]:
                    top_indices = np.argsort(detections["scores"])[-params["max_detections"]:][::-1]
                    for key in detections:
                        detections[key] = detections[key][top_indices]
                
                all_detections.append(detections)
        
        return all_detections

    def evaluate_detections(self, detections: Dict, ground_truth: Dict, 
                          iou_threshold: float = 0.5) -> Dict:
        """评估检测结果"""
        true_positives = 0
        false_positives = 0
        false_negatives = 0
        
        # 创建已匹配的真值标志
        matched_gt = [False] * len(ground_truth["boxes"])
        
        # 按置信度排序检测结果
        sorted_indices = np.argsort(-detections["scores"])
        
        # 遍历检测结果
        for i in sorted_indices:
            box = detections["boxes"][i]
            label = detections["labels"][i]
            matched = False
            
            # 寻找匹配的真值框
            for j, gt_box in enumerate(ground_truth["boxes"]):
                if not matched_gt[j] and ground_truth["labels"][j] == label:
                    iou = self._calculate_iou(box, gt_box)
                    if iou >= iou_threshold:
                        true_positives += 1
                        matched_gt[j] = True
                        matched = True
                        break
            
            if not matched:
                false_positives += 1
        
        # 计算假阴性
        false_negatives = sum(not matched for matched in matched_gt)
        
        # 计算精度、召回率和F1分数
        precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
        recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
        f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        return {
            "precision": precision,
            "recall": recall,
            "f1_score": f1_score,
            "true_positives": true_positives,
            "false_positives": false_positives,
            "false_negatives": false_negatives
        }

    def _calculate_iou(self, box1: np.ndarray, box2: np.ndarray) -> float:
        """计算两个边界框的IOU"""
        x1, y1, x2, y2 = box1
        x1g, y1g, x2g, y2g = box2
        
        # 计算交集区域
        x_left = max(x1, x1g)
        y_top = max(y1, y1g)
        x_right = min(x2, x2g)
        y_bottom = min(y2, y2g)
        
        if x_right < x_left or y_bottom < y_top:
            return 0.0
        
        # 计算交集面积
        intersection_area = (x_right - x_left) * (y_bottom - y_top)
        
        # 计算两个边界框的面积
        box1_area = (x2 - x1) * (y2 - y1)
        box2_area = (x2g - x1g) * (y2g - y1g)
        
        # 计算并集面积
        union_area = box1_area + box2_area - intersection_area
        
        # 计算IOU
        iou = intersection_area / union_area
        
        return iou

# 高级零样本物体检测器
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoProcessor, AutoModelForZeroShotObjectDetection
from PIL import Image, ImageDraw, ImageFont
import numpy as np
import matplotlib.pyplot as plt
from typing import List, Dict, Tuple, Optional, Union

class AdvancedZeroShotDetector:
    def __init__(self, model_name: str = "google/owl-vit-base-patch32"):
        # 加载预训练的零样本物体检测模型
        self.model = AutoModelForZeroShotObjectDetection.from_pretrained(model_name)
        self.processor = AutoProcessor.from_pretrained(model_name)
        self.model.eval()  # 设置为评估模式
        
        # 默认参数设置
        self.default_params = {
            "num_detections": 10,
            "detection_threshold": 0.5,
            "text_threshold": 0.25,
            "nms_threshold": 0.45,
            "device": "cuda" if torch.cuda.is_available() else "cpu"
        }
        
        # 将模型移至指定设备
        self.model.to(self.default_params["device"])

    def detect_objects(self, image: Union[Image.Image, np.ndarray], 
                      candidate_labels: List[str], **kwargs) -> Dict:
        """执行零样本物体检测"""
        # 合并默认参数和用户提供的参数
        params = {**self.default_params, **kwargs}
        
        # 预处理输入
        if isinstance(image, np.ndarray):
            image = Image.fromarray(image)
        
        # 处理输入
        inputs = self.processor(
            images=image,
            text=candidate_labels,
            return_tensors="pt"
        )
        inputs = {k: v.to(params["device"]) for k, v in inputs.items()}
        
        # 执行检测
        with torch.no_grad():
            outputs = self.model.detect_objects(
                **inputs,
                num_detections=params["num_detections"],
                detection_threshold=params["detection_threshold"],
                text_threshold=params["text_threshold"],
                nms_threshold=params["nms_threshold"]
            )
        
        # 处理输出
        detections = {
            "boxes": outputs.boxes.cpu().numpy(),
            "scores": outputs.scores.cpu().numpy(),
            "labels": outputs.labels.cpu().numpy(),
            "label_names": [candidate_labels[i] for i in outputs.labels.cpu().numpy()]
        }
        
        return detections

    def detect_objects_with_descriptions(self, image: Union[Image.Image, np.ndarray], 
                                       category_descriptions: Dict[str, str], **kwargs) -> Dict:
        """使用详细的类别描述进行零样本物体检测"""
        # 合并默认参数和用户提供的参数
        params = {**self.default_params, **kwargs}
        
        # 提取类别名称和描述
        categories = list(category_descriptions.keys())
        descriptions = list(category_descriptions.values())
        
        # 预处理输入
        if isinstance(image, np.ndarray):
            image = Image.fromarray(image)
        
        # 处理输入
        inputs = self.processor(
            images=image,
            text=descriptions,
            return_tensors="pt"
        )
        inputs = {k: v.to(params["device"]) for k, v in inputs.items()}
        
        # 执行检测
        with torch.no_grad():
            outputs = self.model.detect_objects(
                **inputs,
                num_detections=params["num_detections"],
                detection_threshold=params["detection_threshold"],
                text_threshold=params["text_threshold"],
                nms_threshold=params["nms_threshold"]
            )
        
        # 处理输出，将标签索引映射到原始类别名称
        detections = {
            "boxes": outputs.boxes.cpu().numpy(),
            "scores": outputs.scores.cpu().numpy(),
            "labels": outputs.labels.cpu().numpy(),
            "label_names": [categories[i] for i in outputs.labels.cpu().numpy()]
        }
        
        return detections

    def visualize_detections(self, image: Union[Image.Image, np.ndarray], 
                           detections: Dict, show_confidence: bool = True) -> Image.Image:
        """可视化检测结果"""
        # 确保图像是PIL格式
        if isinstance(image, np.ndarray):
            image = Image.fromarray(image)
        
        # 创建可绘制的图像副本
        draw_image = image.copy()
        draw = ImageDraw.Draw(draw_image)
        
        # 尝试加载字体
        try:
            font = ImageFont.truetype("arial.ttf", 16)
        except:
            # 如果无法加载字体，使用默认字体
            font = ImageFont.load_default()
        
        # 为每个检测结果绘制边界框和标签
        for i, (box, score, label) in enumerate(zip(
                detections["boxes"], detections["scores"], detections["label_names"])):
            # 将边界框坐标转换为整数
            x1, y1, x2, y2 = box.astype(int)
            
            # 生成唯一颜色
            color = self._get_unique_color(i)
            
            # 绘制边界框
            draw.rectangle([(x1, y1), (x2, y2)], outline=color, width=3)
            
            # 准备标签文本
            if show_confidence:
                label_text = f"{label} ({score:.2f})"
            else:
                label_text = label
            
            # 测量文本大小
            text_width, text_height = draw.textsize(label_text, font=font)
            
            # 绘制标签背景
            draw.rectangle(
                [(x1, y1 - text_height - 5), (x1 + text_width + 5, y1)],
                fill=color
            )
            
            # 绘制标签文本
            draw.text((x1 + 2, y1 - text_height - 3), label_text, fill=(255, 255, 255), font=font)
        
        return draw_image

    def _get_unique_color(self, index: int) -> Tuple[int, int, int]:
        """为每个类别生成唯一的颜色"""
        # 使用HSV颜色空间生成唯一颜色
        h = index * 0.618033988749895  # 黄金分割比例，生成均匀分布的颜色
        s = 0.7
        v = 0.9
        
        return self._hsv_to_rgb(h, s, v)

    def _hsv_to_rgb(self, h: float, s: float, v: float) -> Tuple[int, int, int]:
        """将HSV颜色转换为RGB颜色"""
        h = h % 1.0
        i = int(h * 6.0)
        f = h * 6.0 - i
        p = v * (1.0 - s)
        q = v * (1.0 - f * s)
        t = v * (1.0 - (1.0 - f) * s)
        
        if i == 0:
            r, g, b = v, t, p
        elif i == 1:
            r, g, b = q, v, p
        elif i == 2:
            r, g, b = p, v, t
        elif i == 3:
            r, g, b = p, q, v
        elif i == 4:
            r, g, b = t, p, v
        else:
            r, g, b = v, p, q
        
        return (int(r * 255), int(g * 255), int(b * 255))

    def evaluate_detections(self, detections: Dict, ground_truth: Dict, 
                          iou_threshold: float = 0.5) -> Dict:
        """评估检测结果"""
        true_positives = 0
        false_positives = 0
        false_negatives = 0
        
        # 创建已匹配的真值标志
        matched_gt = [False] * len(ground_truth["boxes"])
        
        # 按置信度排序检测结果
        sorted_indices = np.argsort(-detections["scores"])
        
        # 遍历检测结果
        for i in sorted_indices:
            box = detections["boxes"][i]
            label = detections["label_names"][i]
            matched = False
            
            # 寻找匹配的真值框
            for j, gt_box in enumerate(ground_truth["boxes"]):
                if not matched_gt[j] and ground_truth["label_names"][j] == label:
                    iou = self._calculate_iou(box, gt_box)
                    if iou >= iou_threshold:
                        true_positives += 1
                        matched_gt[j] = True
                        matched = True
                        break
            
            if not matched:
                false_positives += 1
        
        # 计算假阴性
        false_negatives = sum(not matched for matched in matched_gt)
        
        # 计算精度、召回率和F1分数
        precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
        recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
        f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        return {
            "precision": precision,
            "recall": recall,
            "f1_score": f1_score,
            "true_positives": true_positives,
            "false_positives": false_positives,
            "false_negatives": false_negatives
        }

    def _calculate_iou(self, box1: np.ndarray, box2: np.ndarray) -> float:
        """计算两个边界框的IOU"""
        x1, y1, x2, y2 = box1
        x1g, y1g, x2g, y2g = box2
        
        # 计算交集区域
        x_left = max(x1, x1g)
        y_top = max(y1, y1g)
        x_right = min(x2, x2g)
        y_bottom = min(y2, y2g)
        
        if x_right < x_left or y_bottom < y_top:
            return 0.0
        
        # 计算交集面积
        intersection_area = (x_right - x_left) * (y_bottom - y_top)
        
        # 计算两个边界框的面积
        box1_area = (x2 - x1) * (y2 - y1)
        box2_area = (x2g - x1g) * (y2g - y1g)
        
        # 计算并集面积
        union_area = box1_area + box2_area - intersection_area
        
        # 计算IOU
        iou = intersection_area / union_area
        
        return iou

# 物体检测模型知识蒸馏示例
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import DetrForObjectDetection, YolosForObjectDetection

class KnowledgeDistillationObjectDetector:
    def __init__(self, teacher_model_name, student_model_name, num_classes=91):
        # 加载教师模型（大型高精度模型）
        self.teacher_model = DetrForObjectDetection.from_pretrained(teacher_model_name)
        for param in self.teacher_model.parameters():
            param.requires_grad = False  # 冻结教师模型参数
        
        # 加载学生模型（小型高效模型）
        self.student_model = YolosForObjectDetection.from_pretrained(
            student_model_name,
            num_labels=num_classes
        )
        
        # 定义优化器
        self.optimizer = torch.optim.AdamW(
            self.student_model.parameters(),
            lr=1e-4,
            weight_decay=1e-4
        )
        
        # 定义温度参数（控制软标签的平滑程度）
        self.temperature = 2.0
        
        # 定义损失权重
        self.cls_weight = 0.5  # 分类损失权重
        self.box_weight = 0.3  # 边界框回归损失权重
        self.distill_weight = 0.2  # 知识蒸馏损失权重
    
    def compute_distillation_loss(self, teacher_logits, student_logits, temperature):
        # 计算知识蒸馏损失
        teacher_probs = F.softmax(teacher_logits / temperature, dim=-1)
        student_log_probs = F.log_softmax(student_logits / temperature, dim=-1)
        
        # 使用KL散度计算蒸馏损失
        distillation_loss = F.kl_div(
            student_log_probs,
            teacher_probs,
            reduction='batchmean'
        ) * (temperature ** 2)  # 缩放损失以保持梯度大小
        
        return distillation_loss
    
    def train_step(self, pixel_values, pixel_mask, labels):
        # 启用教师模型的推理模式
        with torch.no_grad():
            teacher_outputs = self.teacher_model(
                pixel_values=pixel_values,
                pixel_mask=pixel_mask
            )
        
        # 学生模型前向传播
        student_outputs = self.student_model(
            pixel_values=pixel_values,
            pixel_mask=pixel_mask,
            labels=labels
        )
        
        # 计算原始损失
        original_loss = student_outputs.loss
        
        # 计算知识蒸馏损失
        distillation_loss = self.compute_distillation_loss(
            teacher_logits=teacher_outputs.logits,
            student_logits=student_outputs.logits,
            temperature=self.temperature
        )
        
        # 计算总损失
        total_loss = (
            self.cls_weight * student_outputs.loss_dict['loss_ce'] +
            self.box_weight * student_outputs.loss_dict['loss_bbox'] +
            self.distill_weight * distillation_loss
        )
        
        # 反向传播和参数更新
        self.optimizer.zero_grad()
        total_loss.backward()
        self.optimizer.step()
        
        return {
            'total_loss': total_loss.item(),
            'original_loss': original_loss.item(),
            'distillation_loss': distillation_loss.item()
        }

# 零样本物体检测模型量化优化示例
import torch
import torch.quantization
from transformers import AutoProcessor, AutoModelForZeroShotObjectDetection
from PIL import Image
import numpy as np
import time
import psutil
import os

class QuantizedZeroShotDetector:
    def __init__(self, model_name="google/owl-vit-base-patch32"):
        # 加载预训练的零样本物体检测模型
        self.model = AutoModelForZeroShotObjectDetection.from_pretrained(model_name)
        self.processor = AutoProcessor.from_pretrained(model_name)
        
        # 保存原始模型用于比较
        self.original_model = self.model
        
        # 移至CPU（量化通常在CPU上进行）
        self.model.to("cpu")
        self.original_model.to("cpu")
        
        # 设置默认参数
        self.default_params = {
            "num_detections": 10,
            "detection_threshold": 0.5,
            "text_threshold": 0.25,
            "nms_threshold": 0.45
        }
    
    def quantize_dynamic(self):
        # 执行动态量化
        print("开始执行动态量化...")
        start_time = time.time()
        
        # 对模型进行动态量化
        self.model = torch.quantization.quantize_dynamic(
            self.model,
            {torch.nn.Linear, torch.nn.Conv2d},
            dtype=torch.qint8
        )
        
        end_time = time.time()
        print(f"动态量化完成，耗时: {end_time - start_time:.2f}秒")
        
        return self.model
    
    def detect_objects(self, image, candidate_labels, use_quantized=True, **kwargs):
        # 合并默认参数和用户提供的参数
        params = {**self.default_params, **kwargs}
        
        # 选择使用量化模型还是原始模型
        model = self.model if use_quantized else self.original_model
        
        # 处理输入
        inputs = self.processor(
            images=image,
            text=candidate_labels,
            return_tensors="pt"
        )
        
        # 执行检测
        with torch.no_grad():
            outputs = model.detect_objects(
                **inputs,
                num_detections=params["num_detections"],
                detection_threshold=params["detection_threshold"],
                text_threshold=params["text_threshold"],
                nms_threshold=params["nms_threshold"]
            )
        
        # 处理输出
        detections = {
            "boxes": outputs.boxes.cpu().numpy(),
            "scores": outputs.scores.cpu().numpy(),
            "labels": outputs.labels.cpu().numpy(),
            "label_names": [candidate_labels[i] for i in outputs.labels.cpu().numpy()]
        }
        
        return detections
    
    def compare_performance(self, image, candidate_labels):
        # 比较量化模型和原始模型的性能
        process = psutil.Process(os.getpid())
        
        # 测量原始模型性能
        print("测量原始模型性能...")
        start_time = time.time()
        start_memory = process.memory_info().rss / 1024 / 1024  # MB
        
        original_detections = self.detect_objects(image, candidate_labels, use_quantized=False)
        
        end_time = time.time()
        end_memory = process.memory_info().rss / 1024 / 1024  # MB
        
        original_time = end_time - start_time
        original_memory = end_memory - start_memory
        
        # 测量量化模型性能
        print("测量量化模型性能...")
        start_time = time.time()
        start_memory = process.memory_info().rss / 1024 / 1024  # MB
        
        quantized_detections = self.detect_objects(image, candidate_labels, use_quantized=True)
        
        end_time = time.time()
        end_memory = process.memory_info().rss / 1024 / 1024  # MB
        
        quantized_time = end_time - start_time
        quantized_memory = end_memory - start_memory
        
        # 计算加速比和内存减少率
        speedup = original_time / quantized_time if quantized_time > 0 else float('inf')
        memory_reduction = (1 - quantized_memory / original_memory) * 100 if original_memory > 0 else 0
        
        return {
            "original_time": original_time,
            "original_memory": original_memory,
            "quantized_time": quantized_time,
            "quantized_memory": quantized_memory,
            "speedup": speedup,
            "memory_reduction": memory_reduction
        }