AI计算机视觉应用开发完全教程

1. 计算机视觉概述与技术栈

计算机视觉（Computer Vision）旨在让计算机从图像和视频中获取高层次的理解能力。从技术栈角度来看，一个典型的视觉系统包含以下层次：

基础设施层：GPU/TPU计算资源、CUDA/cuDNN加速库、容器化部署 框架层：PyTorch、TensorFlow、ONNX Runtime、OpenCV 模型层：预训练模型（ImageNet、COCO）、自定义微调模型 应用层：图像分类、目标检测、分割、OCR、视频分析

现代计算机视觉的技术演进可以概括为：手工特征（SIFT/HOG）→ CNN（AlexNet/VGG）→ 深度CNN（ResNet/EfficientNet）→ Transformer（ViT/Swin）→ 多模态大模型（CLIP/GPT-4V）。每一轮演进都显著提升了视觉系统的感知能力。

2. 图像分类与CNN架构演进

2.1 CNN基础架构

卷积神经网络通过卷积层提取局部特征、池化层降低空间维度、全连接层进行分类决策。

import torch
import torch.nn as nn
import torch.nn.functional as F

class BasicCNN(nn.Module):
    """基础CNN分类网络"""
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            # 第一个卷积块
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),

            # 第二个卷积块
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),

            # 第三个卷积块
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.AdaptiveAvgPool2d((1, 1))
        )
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(128, num_classes)
        )

    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)

# 模型实例化与推理
model = BasicCNN(num_classes=10)
x = torch.randn(1, 3, 32, 32)  # 批次1，3通道，32x32图像
output = model(x)
print(f"输出形状: {output.shape}")  # (1, 10)

2.2 经典架构演进

import torchvision.models as models

# 各阶段经典模型加载
models_dict = {
    "AlexNet": models.alexnet(pretrained=True),           # 2012, 开启深度学习时代
    "VGG16": models.vgg16(pretrained=True),               # 2014, 更深的网络
    "ResNet50": models.resnet50(pretrained=True),          # 2015, 残差连接
    "DenseNet121": models.densenet121(pretrained=True),    # 2017, 密集连接
    "EfficientNet-B0": models.efficientnet_b0(pretrained=True),  # 2019, 复合缩放
}

# 使用预训练ResNet进行推理
from torchvision import transforms
from PIL import Image

model = models.resnet50(pretrained=True)
model.eval()

preprocess = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                         std=[0.229, 0.224, 0.225]),
])

# image = Image.open("cat.jpg")
# input_tensor = preprocess(image).unsqueeze(0)
# with torch.no_grad():
#     output = model(input_tensor)
#     probabilities = torch.softmax(output[0], dim=0)
#     top5 = torch.topk(probabilities, 5)

2.3 迁移学习实战

import torch
import torch.nn as nn
from torchvision import models

def create_classifier(num_classes, freeze_backbone=True):
    """基于预训练ResNet50的迁移学习分类器"""
    model = models.resnet50(pretrained=True)

    if freeze_backbone:
        # 冻结骨干网络参数
        for param in model.parameters():
            param.requires_grad = False

    # 替换最后的全连接层
    in_features = model.fc.in_features
    model.fc = nn.Sequential(
        nn.Dropout(0.3),
        nn.Linear(in_features, 512),
        nn.ReLU(),
        nn.Dropout(0.2),
        nn.Linear(512, num_classes)
    )

    return model

# 创建5分类模型
model = create_classifier(num_classes=5)

# 只训练最后的分类层
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in model.parameters())
print(f"可训练参数: {trainable_params:,} / 总参数: {total_params:,}")
print(f"训练比例: {trainable_params/total_params*100:.2f}%")

3. 目标检测：YOLO系列与DETR

3.1 YOLO系列

YOLO（You Only Look Once）将目标检测视为单次回归问题，实现了速度与精度的良好平衡。

# 使用Ultralytics YOLOv8
# pip install ultralytics
from ultralytics import YOLO

# 加载预训练模型
model = YOLO("yolov8n.pt")  # nano版本，速度最快

# 推理
results = model("street_scene.jpg", conf=0.5, iou=0.45)

# 解析检测结果
for result in results:
    boxes = result.boxes
    for box in boxes:
        cls_id = int(box.cls[0])
        conf = float(box.conf[0])
        xyxy = box.xyxy[0].tolist()  # [x1, y1, x2, y2]
        class_name = model.names[cls_id]
        print(f"检测到: {class_name}, 置信度: {conf:.3f}, 位置: {xyxy}")

# 自定义训练
def train_yolo_custom():
    """训练自定义YOLO检测模型"""
    model = YOLO("yolov8s.pt")  # 使用small版本作为基础

    # 训练（需要准备YOLO格式的数据集）
    results = model.train(
        data="custom_dataset.yaml",
        epochs=100,
        imgsz=640,
        batch=16,
        lr0=0.01,
        lrf=0.01,
        momentum=0.937,
        weight_decay=0.0005,
        augment=True,
        device="0"  # GPU编号
    )
    return results

3.2 DETR：基于Transformer的端到端检测

DETR（Detection Transformer）摒弃了锚框和NMS等手工组件，实现了真正的端到端检测。

import torch
from transformers import DetrImageProcessor, DetrForObjectDetection
from PIL import Image

# 加载DETR模型
processor = DetrImageProcessor.from_pretrained("facebook/detr-resnet-50")
model = DetrForObjectDetection.from_pretrained("facebook/detr-resnet-50")

# 推理
image = Image.open("street_scene.jpg")
inputs = processor(images=image, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)

# 解析结果
results = processor.post_process_object_detection(
    outputs, threshold=0.9, target_sizes=[(image.height, image.width)]
)[0]

for score, label, box in zip(results["scores"], results["labels"], results["boxes"]):
    box = [round(i, 2) for i in box.tolist()]
    print(f"检测到 {model.config.id2label[label.item()]}: "
          f"置信度={round(score.item(), 3)}, 位置={box}")

3.3 YOLO与DETR对比

4. 实例分割与语义分割

4.1 Mask R-CNN实例分割

import torch
import torchvision
from torchvision.models.detection import maskrcnn_resnet50_fpn
from torchvision import transforms
from PIL import Image
import numpy as np

# 加载预训练Mask R-CNN
model = maskrcnn_resnet50_fpn(pretrained=True)
model.eval()

# 图像预处理
image = Image.open("scene.jpg").convert("RGB")
transform = transforms.Compose([transforms.ToTensor()])
input_tensor = transform(image).unsqueeze(0)

# 推理
with torch.no_grad():
    predictions = model(input_tensor)

# 解析分割结果
pred = predictions[0]
masks = pred["masks"]           # (N, 1, H, W) 分割掩码
boxes = pred["boxes"]           # (N, 4) 边界框
labels = pred["labels"]         # (N,) 类别标签
scores = pred["scores"]         # (N,) 置信度

# 筛选高置信度结果
threshold = 0.7
high_conf = scores > threshold
masks = masks[high_conf].squeeze(1)  # (N, H, W)
boxes = boxes[high_conf]
labels = labels[high_conf]

print(f"检测到 {len(masks)} 个实例")
for i in range(len(masks)):
    print(f"实例 {i+1}: 类别={labels[i].item()}, "
          f"面积={masks[i].sum().item():.0f} 像素")

4.2 SAM：Segment Anything Model

SAM是Meta推出的通用分割模型，能够根据点击、框选或文本提示对任意物体进行分割。

# pip install segment-anything
from segment_anything import sam_model_registry, SamPredictor
import cv2
import numpy as np

# 加载SAM模型
sam_checkpoint = "sam_vit_h_4b8939.pth"
model_type = "vit_h"
sam = sam_model_registry[model_type](checkpoint=sam_checkpoint)
sam.to("cuda")
predictor = SamPredictor(sam)

# 加载图像并设置
image = cv2.imread("scene.jpg")
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
predictor.set_image(image_rgb)

# 点击式分割：指定一个点进行分割
input_point = np.array([[500, 375]])  # 点击位置 (x, y)
input_label = np.array([1])           # 1=前景, 0=背景

masks, scores, logits = predictor.predict(
    point_coords=input_point,
    point_labels=input_label,
    multimask_output=True,  # 输出多个候选掩码
)

# 选择最佳掩码
best_mask = masks[np.argmax(scores)]
print(f"掩码形状: {best_mask.shape}")
print(f"分割面积: {best_mask.sum()} 像素")

# 框选式分割
box = np.array([100, 100, 400, 400])  # [x1, y1, x2, y2]
masks, scores, logits = predictor.predict(
    box=box,
    multimask_output=False
)

4.3 语义分割

语义分割为图像中的每个像素分配类别标签，不区分同类实例。

from transformers import SegformerForSemanticSegmentation, SegformerFeatureExtractor
from PIL import Image
import torch
import numpy as np

# 加载SegFormer语义分割模型
model = SegformerForSemanticSegmentation.from_pretrained(
    "nvidia/segformer-b0-finetuned-ade-512-512"
)
feature_extractor = SegformerFeatureExtractor.from_pretrained(
    "nvidia/segformer-b0-finetuned-ade-512-512"
)

# 推理
image = Image.open("street_scene.jpg")
inputs = feature_extractor(images=image, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)
    logits = outputs.logits

# 上采样到原图大小
upsampled = torch.nn.functional.interpolate(
    logits, size=image.size[::-1], mode="bilinear", align_corners=False
)
pred = upsampled.argmax(dim=1)[0].numpy()

# 可视化（使用ADE20K颜色映射）
unique_classes = np.unique(pred)
print(f"检测到的类别数: {len(unique_classes)}")
for cls_id in unique_classes:
    pixel_count = (pred == cls_id).sum()
    label = model.config.id2label[cls_id]
    print(f"  {label}: {pixel_count} 像素")

5. OCR文字识别

5.1 PaddleOCR

PaddleOCR是百度开源的OCR工具库，支持中英文识别，精度高且部署灵活。

# pip install paddlepaddle paddleocr
from paddleocr import PaddleOCR

# 初始化OCR引擎
ocr = PaddleOCR(
    use_angle_cls=True,  # 启用文字方向分类
    lang="ch",           # 中文识别
    use_gpu=True
)

# 识别
result = ocr.ocr("document.jpg", cls=True)

for line in result[0]:
    box, (text, confidence) = line
    print(f"文字: {text}")
    print(f"置信度: {confidence:.4f}")
    print(f"位置: {box}\n")

# 批量处理
def batch_ocr(image_paths):
    """批量OCR处理"""
    all_results = []
    for path in image_paths:
        result = ocr.ocr(path, cls=True)
        texts = [line[1][0] for line in result[0]]
        all_results.append({
            "file": path,
            "texts": texts,
            "full_text": "\n".join(texts)
        })
    return all_results

# 版面分析
from paddleocr import PPStructure

engine = PPStructure(show_log=False, lang="ch")
result = engine("complex_document.jpg")

for block in result:
    if block["type"] == "text":
        print(f"[文本区域] {block['res'][0]['text']}")
    elif block["type"] == "table":
        print(f"[表格区域] {block['res']['html']}")
    elif block["type"] == "figure":
        print("[图片区域]")

5.2 Tesseract OCR

import pytesseract
from PIL import Image

# 基础文字识别
image = Image.open("text_image.jpg")
text = pytesseract.image_to_string(image, lang="chi_sim+eng")
print(text)

# 获取详细信息（位置、置信度）
data = pytesseract.image_to_data(image, output_type=pytesseract.Output.DICT)
for i, word in enumerate(data["text"]):
    if word.strip():
        conf = int(data["conf"][i])
        x, y, w, h = data["left"][i], data["top"][i], data["width"][i], data["height"][i]
        print(f"'{word}' 置信度:{conf} 位置:({x},{y},{w},{h})")

# 表格识别
import pandas as pd
tables = pytesseract.image_to_data(image, output_type="data.frame")

6. 视频分析与行为识别

6.1 视频目标跟踪

import cv2
from ultralytics import YOLO

# 使用YOLOv8进行视频跟踪
model = YOLO("yolov8n.pt")

# 处理视频流
cap = cv2.VideoCapture("video.mp4")
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

# 输出视频
out = cv2.VideoWriter("output_tracked.mp4",
                       cv2.VideoWriter_fourcc(*"mp4v"),
                       fps, (width, height))

track_history = {}  # 跟踪轨迹记录

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break

    # 使用ByteTrack进行多目标跟踪
    results = model.track(frame, persist=True, tracker="bytetrack.yaml")

    if results[0].boxes.id is not None:
        boxes = results[0].boxes
        for box in boxes:
            track_id = int(box.id[0])
            cls = int(box.cls[0])
            xyxy = box.xyxy[0].tolist()

            # 记录轨迹
            center = ((xyxy[0]+xyxy[2])/2, (xyxy[1]+xyxy[3])/2)
            if track_id not in track_history:
                track_history[track_id] = []
            track_history[track_id].append(center)

    # 绘制结果
    annotated = results[0].plot()
    out.write(annotated)

cap.release()
out.release()

# 分析轨迹
for track_id, points in track_history.items():
    print(f"目标 {track_id}: 轨迹点数={len(points)}, "
          f"移动距离={sum(((points[i][0]-points[i-1][0])**2 + (points[i][1]-points[i-1][1])**2)**0.5 for i in range(1, len(points))):.1f}px")

6.2 行为识别

import torch
from transformers import VideoMAEForVideoClassification, VideoMAEFeatureExtractor
import cv2
import numpy as np

# 使用VideoMAE进行行为识别
model = VideoMAEForVideoClassification.from_pretrained(
    "MCG-NJU/videomae-base-finetuned-kinetics"
)
feature_extractor = VideoMAEFeatureExtractor.from_pretrained(
    "MCG-NJU/videomae-base-finetuned-kinetics"
)

def extract_frames(video_path, num_frames=16):
    """从视频中均匀采样帧"""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)

    frames = []
    for idx in indices:
        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
        ret, frame = cap.read()
        if ret:
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            frames.append(frame)
    cap.release()
    return frames

# 推理
frames = extract_frames("action_video.mp4", num_frames=16)
inputs = feature_extractor(frames, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)
    probs = torch.softmax(outputs.logits, dim=-1)

top5 = torch.topk(probs[0], 5)
for score, idx in zip(top5.values, top5.indices):
    label = model.config.id2label[idx.item()]
    print(f"{label}: {score.item():.4f}")

7. 3D视觉：点云与深度估计

7.1 单目深度估计

import torch
from transformers import DPTForDepthEstimation, DPTFeatureExtractor
from PIL import Image
import numpy as np

# 使用DPT进行单目深度估计
feature_extractor = DPTFeatureExtractor.from_pretrained("Intel/dpt-large")
model = DPTForDepthEstimation.from_pretrained("Intel/dpt-large")

image = Image.open("indoor_scene.jpg")
inputs = feature_extractor(images=image, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)
    predicted_depth = outputs.predicted_depth

# 插值到原图大小
depth = torch.nn.functional.interpolate(
    predicted_depth.unsqueeze(1),
    size=image.size[::-1],
    mode="bicubic",
    align_corners=False
).squeeze().numpy()

print(f"深度图尺寸: {depth.shape}")
print(f"深度范围: {depth.min():.2f} ~ {depth.max():.2f} 米")

# 归一化用于可视化
depth_normalized = ((depth - depth.min()) / (depth.max() - depth.min()) * 255).astype(np.uint8)
# Image.fromarray(depth_normalized).save("depth_map.png")

7.2 点云处理基础

import numpy as np

# 从深度图生成点云
def depth_to_pointcloud(depth_map, intrinsic, rgb_image=None):
    """将深度图转换为3D点云"""
    h, w = depth_map.shape
    fx, fy = intrinsic[0, 0], intrinsic[1, 1]
    cx, cy = intrinsic[0, 2], intrinsic[1, 2]

    # 生成像素网格
    u, v = np.meshgrid(np.arange(w), np.arange(h))

    # 反投影到3D空间
    z = depth_map
    x = (u - cx) * z / fx
    y = (v - cy) * z / fy

    # 组合点云
    points = np.stack([x, y, z], axis=-1).reshape(-1, 3)

    # 过滤无效点
    valid = (z.reshape(-1) > 0) & (z.reshape(-1) < 100)
    points = points[valid]

    if rgb_image is not None:
        colors = rgb_image.reshape(-1, 3)[valid] / 255.0
        return points, colors

    return points

# 使用Open3D可视化（可选）
# import open3d as o3d
# pcd = o3d.geometry.PointCloud()
# pcd.points = o3d.utility.Vector3dVector(points)
# pcd.colors = o3d.utility.Vector3dVector(colors)
# o3d.visualization.draw_geometries([pcd])

# 简单的相机内参示例
intrinsic = np.array([
    [525.0, 0, 319.5],
    [0, 525.0, 239.5],
    [0, 0, 1]
])

8. 视觉Transformer：ViT与Swin

8.1 Vision Transformer (ViT)

ViT将图像分割为固定大小的patch，将每个patch视为一个token，直接应用标准Transformer编码器。

import torch
from transformers import ViTForImageClassification, ViTFeatureExtractor
from PIL import Image

# 加载预训练ViT
model = ViTForImageClassification.from_pretrained(
    "google/vit-base-patch16-224"
)
feature_extractor = ViTFeatureExtractor.from_pretrained(
    "google/vit-base-patch16-224"
)

# 推理
image = Image.open("dog.jpg")
inputs = feature_extractor(images=image, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)
    probs = torch.softmax(outputs.logits, dim=-1)

top5 = torch.topk(probs[0], 5)
for score, idx in zip(top5.values, top5.indices):
    print(f"{model.config.id2label[idx.item()]}: {score.item():.4f}")

# 查看ViT内部结构
print(f"隐藏维度: {model.config.hidden_size}")      # 768
print(f"注意力头数: {model.config.num_attention_heads}")  # 12
print(f"Transformer层数: {model.config.num_hidden_layers}")  # 12
print(f"Patch大小: {model.config.patch_size}")        # 16

8.2 Swin Transformer

Swin Transformer通过层级结构和滑动窗口注意力机制，在多种视觉任务上取得了优异表现。

from transformers import AutoImageProcessor, SwinForImageClassification
from PIL import Image
import torch

# 加载Swin Transformer
processor = AutoImageProcessor.from_pretrained("microsoft/swin-tiny-patch4-window7-224")
model = SwinForImageClassification.from_pretrained("microsoft/swin-tiny-patch4-window7-224")

# 推理
image = Image.open("cat.jpg")
inputs = processor(images=image, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)
    probs = torch.softmax(outputs.logits, dim=-1)

pred_class = probs.argmax(-1).item()
print(f"预测类别: {model.config.id2label[pred_class]}")
print(f"置信度: {probs[0][pred_class].item():.4f}")

# Swin的层级特性：不同stage的特征图分辨率
# Stage 1: H/4 x W/4,   dim=96
# Stage 2: H/8 x W/8,   dim=192
# Stage 3: H/16 x W/16, dim=384
# Stage 4: H/32 x W/32, dim=768

8.3 CNN vs Transformer 对比

import time
import torch
from torchvision import models
from transformers import ViTForImageClassification

def benchmark_model(model, input_shape, device="cuda", num_runs=100):
    """模型性能基准测试"""
    model = model.to(device).eval()
    x = torch.randn(*input_shape).to(device)

    # 预热
    for _ in range(10):
        with torch.no_grad():
            model(x)

    # 计时
    start = time.time()
    for _ in range(num_runs):
        with torch.no_grad():
            model(x)
    elapsed = (time.time() - start) / num_runs

    # 参数量
    params = sum(p.numel() for p in model.parameters())

    return {
        "参数量": f"{params/1e6:.1f}M",
        "推理时间": f"{elapsed*1000:.2f}ms",
        "FPS": f"{1/elapsed:.1f}"
    }

# 对比测试
# resnet50 = models.resnet50(pretrained=True)
# vit = ViTForImageClassification.from_pretrained("google/vit-base-patch16-224")
# input_shape = (1, 3, 224, 224)
# print("ResNet50:", benchmark_model(resnet50, input_shape))
# print("ViT-Base:", benchmark_model(vit, input_shape))

9. 边缘端视觉部署

将视觉模型部署到边缘设备（手机、嵌入式、IoT设备）需要考虑模型大小、推理速度和功耗。

9.1 模型量化

import torch
from torchvision import models

model = models.resnet50(pretrained=True)
model.eval()

# 动态量化（推理时量化权重）
quantized_dynamic = torch.quantization.quantize_dynamic(
    model, {torch.nn.Linear}, dtype=torch.qint8
)

# 静态量化（需要校准数据）
model.qconfig = torch.quantization.get_default_qconfig("qnnpack")
torch.quantization.prepare(model, inplace=True)

# 校准（使用少量样本）
# with torch.no_grad():
#     for calibration_sample in calibration_data:
#         model(calibration_sample)

torch.quantization.convert(model, inplace=True)

# 模型大小对比
import os
torch.save(model.state_dict(), "model_original.pth")
torch.save(quantized_dynamic.state_dict(), "model_quantized.pth")
# 量化后模型通常缩小3-4倍

9.2 ONNX导出与部署

import torch
from torchvision import models

model = models.resnet50(pretrained=True)
model.eval()

# 导出ONNX
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(
    model,
    dummy_input,
    "resnet50.onnx",
    export_params=True,
    opset_version=11,
    do_constant_folding=True,
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={
        "input": {0: "batch_size"},
        "output": {0: "batch_size"}
    }
)

# 使用ONNX Runtime推理
import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("resnet50.onnx",
    providers=["CUDAExecutionProvider", "CPUExecutionProvider"])

input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)
outputs = session.run(None, {"input": input_data})
print(f"输出形状: {outputs[0].shape}")

9.3 TensorRT加速

# TensorRT部署示例（需要NVIDIA GPU）
import tensorrt as trt

def build_engine(onnx_path, engine_path, fp16=True):
    """将ONNX模型转换为TensorRT引擎"""
    logger = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(logger)
    network = builder.create_network(
        1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
    )
    parser = trt.OnnxParser(network, logger)

    # 解析ONNX
    with open(onnx_path, "rb") as f:
        if not parser.parse(f.read()):
            for error in range(parser.num_errors):
                print(parser.get_error(error))
            return None

    # 配置构建参数
    config = builder.create_builder_config()
    config.max_workspace_size = 1 << 30  # 1GB

    if fp16 and builder.platform_has_fast_fp16:
        config.set_flag(trt.BuilderFlag.FP16)

    # 构建引擎
    engine = builder.build_serialized_network(network, config)

    with open(engine_path, "wb") as f:
        f.write(engine)

    print(f"TensorRT引擎已保存: {engine_path}")
    return engine

10. 实战案例：智能安防视觉系统

下面展示一个完整的智能安防视觉系统，整合了目标检测、行为识别和告警功能：

import cv2
import torch
import time
from datetime import datetime
from collections import defaultdict
from ultralytics import YOLO

class SmartSurveillanceSystem:
    """智能安防视觉系统"""

    def __init__(self, config=None):
        self.config = config or {
            "detection_model": "yolov8m.pt",
            "confidence_threshold": 0.5,
            "alert_classes": ["person"],
            "intrusion_zones": [],  # 侵入区域坐标
            "loitering_threshold": 30,  # 停留告警阈值(秒)
            "output_path": "surveillance_output.mp4"
        }

        # 加载模型
        self.model = YOLO(self.config["detection_model"])
        self.track_history = defaultdict(list)
        self.alerts = []

    def define_intrusion_zone(self, points):
        """定义入侵检测区域（多边形）"""
        self.config["intrusion_zones"].append(points)

    def check_intrusion(self, point):
        """检查目标是否在入侵区域内"""
        for zone in self.config["intrusion_zones"]:
            if cv2.pointPolygonTest(zone, point, False) >= 0:
                return True
        return False

    def process_frame(self, frame, timestamp):
        """处理单帧图像"""
        results = self.model.track(
            frame, persist=True,
            conf=self.config["confidence_threshold"],
            tracker="bytetrack.yaml"
        )

        detections = []
        frame_alerts = []

        if results[0].boxes.id is not None:
            for box in results[0].boxes:
                track_id = int(box.id[0])
                cls = int(box.cls[0])
                conf = float(box.conf[0])
                class_name = self.model.names[cls]
                xyxy = box.xyxy[0].tolist()
                center = ((xyxy[0]+xyxy[2])/2, (xyxy[1]+xyxy[3])/2)

                detection = {
                    "track_id": track_id,
                    "class": class_name,
                    "confidence": conf,
                    "bbox": xyxy,
                    "center": center,
                    "timestamp": timestamp
                }
                detections.append(detection)

                # 记录轨迹
                self.track_history[track_id].append({
                    "center": center,
                    "time": timestamp
                })

                # 检查入侵
                if class_name in self.config["alert_classes"]:
                    if self.check_intrusion(center):
                        alert = {
                            "type": "intrusion",
                            "track_id": track_id,
                            "class": class_name,
                            "time": timestamp,
                            "position": center
                        }
                        frame_alerts.append(alert)

                    # 检查停留
                    history = self.track_history[track_id]
                    if len(history) > 1:
                        duration = (timestamp - history[0]["time"]).total_seconds()
                        if duration > self.config["loitering_threshold"]:
                            # 检查是否移动很小
                            movement = sum(
                                ((h["center"][0]-history[0]["center"][0])**2 +
                                 (h["center"][1]-history[0]["center"][1])**2)**0.5
                                for h in history[-10:]
                            ) / min(10, len(history))
                            if movement < 20:  # 移动距离小于20像素
                                alert = {
                                    "type": "loitering",
                                    "track_id": track_id,
                                    "duration": duration,
                                    "time": timestamp
                                }
                                frame_alerts.append(alert)

        self.alerts.extend(frame_alerts)
        return detections, frame_alerts, results[0].plot()

    def run(self, video_source=0):
        """运行安防系统"""
        cap = cv2.VideoCapture(video_source)
        fps = cap.get(cv2.CAP_PROP_FPS) or 30
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

        out = cv2.VideoWriter(
            self.config["output_path"],
            cv2.VideoWriter_fourcc(*"mp4v"),
            fps, (width, height)
        )

        print(f"安防系统启动 - 源: {video_source}")
        print(f"监控告警类别: {self.config['alert_classes']}")
        print(f"入侵区域数量: {len(self.config['intrusion_zones'])}")

        frame_count = 0
        start_time = time.time()

        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break

            timestamp = datetime.now()
            detections, alerts, annotated = self.process_frame(frame, timestamp)

            # 显示告警信息
            for alert in alerts:
                print(f"[{alert['time']}] ⚠️ {alert['type'].upper()}: "
                      f"目标 {alert['track_id']}")

            # 在画面上显示统计
            info = f"Frame: {frame_count} | Detections: {len(detections)}"
            cv2.putText(annotated, info, (10, 30),
                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

            if alerts:
                cv2.putText(annotated, "ALERT!", (width-200, 30),
                           cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 3)

            out.write(annotated)
            frame_count += 1

        elapsed = time.time() - start_time
        cap.release()
        out.release()

        print(f"\n处理完成:")
        print(f"  总帧数: {frame_count}")
        print(f"  处理时间: {elapsed:.1f}s")
        print(f"  平均FPS: {frame_count/elapsed:.1f}")
        print(f"  告警数量: {len(self.alerts)}")

        return self.alerts


# 使用示例
if __name__ == "__main__":
    system = SmartSurveillanceSystem()

    # 定义入侵检测区域（矩形）
    zone = [(100, 100), (500, 100), (500, 400), (100, 400)]
    import numpy as np
    system.define_intrusion_zone(np.array(zone, dtype=np.int32))

    # 运行
    # alerts = system.run("surveillance_video.mp4")

11. 评估指标与优化策略

11.1 常用评估指标

import numpy as np

def compute_iou(box1, box2):
    """计算两个边界框的IoU"""
    x1 = max(box1[0], box2[0])
    y1 = max(box1[1], box2[1])
    x2 = min(box1[2], box2[2])
    y2 = min(box1[3], box2[3])

    intersection = max(0, x2 - x1) * max(0, y2 - y1)
    area1 = (box1[2] - box1[0]) * (box1[3] - box1[1])
    area2 = (box2[2] - box2[0]) * (box2[3] - box2[1])
    union = area1 + area2 - intersection

    return intersection / union if union > 0 else 0

def compute_ap(predictions, ground_truths, iou_threshold=0.5):
    """计算平均精度（AP）"""
    # 按置信度排序
    predictions = sorted(predictions, key=lambda x: x["score"], reverse=True)

    tp = []
    fp = []
    matched_gt = set()

    for pred in predictions:
        best_iou = 0
        best_gt_idx = -1

        for i, gt in enumerate(ground_truths):
            if i in matched_gt:
                continue
            iou = compute_iou(pred["bbox"], gt["bbox"])
            if iou > best_iou:
                best_iou = iou
                best_gt_idx = i

        if best_iou >= iou_threshold and best_gt_idx >= 0:
            tp.append(1)
            fp.append(0)
            matched_gt.add(best_gt_idx)
        else:
            tp.append(0)
            fp.append(1)

    # 计算累积precision和recall
    tp_cumsum = np.cumsum(tp)
    fp_cumsum = np.cumsum(fp)
    recalls = tp_cumsum / len(ground_truths)
    precisions = tp_cumsum / (tp_cumsum + fp_cumsum)

    # 计算AP（11点插值法）
    ap = 0
    for t in np.arange(0, 1.1, 0.1):
        precisions_at_recall = precisions[recalls >= t]
        if len(precisions_at_recall) > 0:
            ap += np.max(precisions_at_recall) / 11

    return ap

def compute_map(all_predictions, all_ground_truths, num_classes, iou_threshold=0.5):
    """计算mAP（各类别AP的平均值）"""
    aps = []
    for cls_id in range(num_classes):
        cls_preds = [p for p in all_predictions if p["class"] == cls_id]
        cls_gts = [g for g in all_ground_truths if g["class"] == cls_id]
        if len(cls_gts) > 0:
            ap = compute_ap(cls_preds, cls_gts, iou_threshold)
            aps.append(ap)

    return np.mean(aps) if aps else 0

# 分割任务指标：mIoU
def compute_miou(pred_mask, gt_mask, num_classes):
    """计算平均交并比（mIoU）"""
    ious = []
    for cls in range(num_classes):
        pred_cls = (pred_mask == cls)
        gt_cls = (gt_mask == cls)
        intersection = (pred_cls & gt_cls).sum()
        union = (pred_cls | gt_cls).sum()
        if union > 0:
            ious.append(intersection / union)
    return np.mean(ious) if ious else 0

11.2 优化策略

# 数据增强策略
from torchvision import transforms

train_transform = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.8, 1.0)),
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomVerticalFlip(p=0.1),
    transforms.RandomRotation(15),
    transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),
    transforms.RandomAffine(degrees=0, translate=(0.1, 0.1)),
    transforms.RandomGrayscale(p=0.05),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                         std=[0.229, 0.224, 0.225]),
    transforms.RandomErasing(p=0.2),
])

# 混合精度训练
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=0.01)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)

def train_epoch(model, dataloader, criterion, optimizer, scaler):
    model.train()
    total_loss = 0

    for images, labels in dataloader:
        images, labels = images.cuda(), labels.cuda()
        optimizer.zero_grad()

        with autocast():  # 混合精度前向传播
            outputs = model(images)
            loss = criterion(outputs, labels)

        scaler.scale(loss).backward()  # 缩放梯度反向传播
        scaler.step(optimizer)
        scaler.update()

        total_loss += loss.item()

    scheduler.step()
    return total_loss / len(dataloader)

# 知识蒸馏
class DistillationLoss(torch.nn.Module):
    def __init__(self, temperature=4.0, alpha=0.7):
        super().__init__()
        self.temperature = temperature
        self.alpha = alpha
        self.ce_loss = torch.nn.CrossEntropyLoss()
        self.kl_loss = torch.nn.KLDivLoss(reduction="batchmean")

    def forward(self, student_logits, teacher_logits, labels):
        # 硬标签损失
        hard_loss = self.ce_loss(student_logits, labels)

        # 软标签损失
        soft_student = torch.log_softmax(student_logits / self.temperature, dim=-1)
        soft_teacher = torch.softmax(teacher_logits / self.temperature, dim=-1)
        soft_loss = self.kl_loss(soft_student, soft_teacher) * (self.temperature ** 2)

        return self.alpha * soft_loss + (1 - self.alpha) * hard_loss

总结

计算机视觉技术已经从实验室走向了大规模工业应用。从基础的图像分类到复杂的视频理解，从云端推理到边缘部署，整个技术栈日趋成熟。关键要点包括：

• 选择合适的架构：CNN适合特征明确的任务，Transformer在大规模数据上表现更优
• 善用预训练模型：迁移学习能以极少的数据达到不错的性能
• 注重工程实践：模型量化、ONNX导出、TensorRT加速是生产部署的必备技能
• 端到端思维：从数据采集、标注、训练到部署监控，每个环节都需要关注
• 评估指标驱动：mAP、mIoU等指标是优化方向的指南针

在实际项目中，建议从预训练模型+微调起步，逐步优化数据质量和模型架构。视觉系统的效果往往80%取决于数据，20%取决于模型。