AI 工程实践系列

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from ultralytics import YOLO

# ========== YOLOv8 ==========
model_v8 = YOLO("yolov8n.pt")      #  nano
model_v8 = YOLO("yolov8s.pt")      #  small
model_v8 = YOLO("yolov8m.pt")      #  medium
model_v8 = YOLO("yolov8l.pt")      #  large
model_v8 = YOLO("yolov8x.pt")      #  extra large

# ========== YOLO11 ==========
model_11 = YOLO("yolo11n.pt")      #  nano
model_11 = YOLO("yolo11s.pt")      #  small
model_11 = YOLO("yolo11m.pt")      #  medium
model_11 = YOLO("yolo11l.pt")      #  large
model_11 = YOLO("yolo11x.pt")      #  extra large

# ========== YOLO26 (2026最新) ==========
model_26 = YOLO("yolo26n.pt")      #  nano  推荐边缘部署
model_26 = YOLO("yolo26s.pt")      #  small
model_26 = YOLO("yolo26m.pt")      #  medium
model_26 = YOLO("yolo26l.pt")      #  large
model_26 = YOLO("yolo26x.pt")      #  extra large

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from ultralytics import YOLO

# 加载模型（以YOLO26为例）
model = YOLO("yolo26n.pt")

# 单张图片检测
results = model("test.jpg", conf=0.25, iou=0.45)

# 结果处理
for result in results:
    boxes = result.boxes          # 检测框
    masks = result.masks          # 分割掩码
    probs = result.probs          # 分类概率
    
    # 打印检测结果
    for box in boxes:
        print(f"类别: {result.names[int(box.cls)]}, "
              f"置信度: {box.conf.item():.3f}, "
              f"坐标: {box.xyxy.tolist()[0]}")
    
    # 保存可视化结果
    result.save("result.jpg")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from ultralytics import YOLO

model = YOLO("yolo26n.pt")

# 视频文件检测
results = model.predict(
    source="input.mp4",
    save=True,           # 保存结果视频
    conf=0.3,
    show=False,          # 是否实时显示
    stream=True          # 流式处理，节省内存
)

# 逐帧处理
for result in results:
    # 自定义后处理逻辑
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from ultralytics import YOLO
import cv2

model = YOLO("yolo26n.pt")

# 打开摄像头
cap = cv2.VideoCapture(0)  # 0为默认摄像头

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break
    
    # 推理
    results = model(frame, verbose=False)
    
    # 绘制结果
    annotated_frame = results[0].plot()
    
    # 显示
    cv2.imshow("YOLO Real-time", annotated_frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

1
2
3
4
5

# 安装
pip install labelImg

# 启动
labelImg

1
2

pip install labelme
labelme

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from ultralytics import YOLO

# 加载模型
# model = YOLO("yolov8n.yaml")  # 从头训练
# model = YOLO("yolo11n.pt")    # 基于预训练权重
model = YOLO("yolo26n.pt")      # 2026推荐，边缘部署首选

# 开始训练
results = model.train(
    # 基础配置
    data="data.yaml",        # 数据集配置
    epochs=100,              # 训练轮数
    imgsz=640,               # 输入尺寸
    batch=16,                # 批次大小
    workers=8,               # 数据加载线程数
    
    # 优化器配置
    optimizer="auto",        # YOLO26自动使用MuSGD
    lr0=0.01,                # 初始学习率
    lrf=0.01,                # 最终学习率因子
    momentum=0.937,          # SGD动量
    weight_decay=0.0005,     # 权重衰减
    
    # 数据增强
    mosaic=1.0,
    mixup=0.1,
    copy_paste=0.1,
    
    # 其他配置
    device=0,                # GPU设备，"cpu"为CPU
    project="runs/train",    # 保存路径
    name="yolo26_exp1",      # 实验名称
    exist_ok=False,          # 是否覆盖
    pretrained=True,         # 使用预训练
    verbose=True,            # 详细日志
    seed=42,                 # 随机种子
)

# 验证模型
metrics = model.val()
print(f"mAP50: {metrics.box.map50:.3f}")
print(f"mAP50-95: {metrics.box.map:.3f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 大模型作为教师，小模型作为学生
teacher = YOLO("yolo26x.pt")
student = YOLO("yolo26n.yaml")

# 蒸馏训练（Ultralytics内置支持）
student.train(
    data="data.yaml",
    distill="yolo26x.pt",  # 教师模型
    distill_ratio=0.5,     # 蒸馏损失比例
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import torch
import torch.nn.utils.prune as prune

# 对卷积层进行 L1 非结构化剪枝
model = YOLO("yolo26n.pt")
for name, module in model.model.named_modules():
    if isinstance(module, torch.nn.Conv2d):
        prune.l1_unstructured(module, name="weight", amount=0.3)
        prune.remove(module, "weight")  # 使剪枝永久化

# 通道剪枝使用 torch-pruning 库
# pip install torch-pruning
import torch_pruning as tp

model = YOLO("yolo26n.pt").model
DG = tp.DependencyGraph()
DG.build_dependency(model, example_inputs=torch.randn(1, 3, 640, 640))

# 按 L1 范数剪枝 20% 通道
pruning_plan = DG.get_pruning_plan(
    model.model[4], tp.prune_conv,
    pruning_dim=0,  # 输出通道维度
    idxs=list(range(0, 64, 5))  # 每 5 个通道保留一个
)
pruning_plan.exec()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

model = YOLO("yolo26n.pt")

# INT8量化（需要校准数据）
model.export(
    format="engine",      # TensorRT
    int8=True,
    data="data.yaml",     # 校准数据集
    batch=8,
)

# ONNX动态量化
model.export(
    format="onnx",
    dynamic=True,
    simplify=True,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from ultralytics import YOLO

model = YOLO("yolo26n.pt")

# ========== 各种格式导出 ==========
# 1. ONNX（跨平台通用）
model.export(format="onnx", simplify=True, dynamic=True)

# 2. TensorRT（NVIDIA GPU最佳）
model.export(format="engine", half=True, workspace=4)

# 3. OpenVINO（Intel CPU最佳）
model.export(format="openvino", half=True)

# 4. CoreML（Apple设备）
model.export(format="coreml", int8=True)

# 5. TFLite（Android/iOS移动端）
model.export(format="tflite", int8=True)

# 6. NCNN（移动端）
model.export(format="ncnn")

# 7. PaddlePaddle
model.export(format="paddle")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import onnxruntime as ort
import cv2
import numpy as np

# 加载ONNX模型
session = ort.InferenceSession(
    "yolo26n.onnx",
    providers=["CUDAExecutionProvider", "CPUExecutionProvider"]
)

def preprocess(image, imgsz=640):
    """图片预处理"""
    img = cv2.resize(image, (imgsz, imgsz))
    img = img.transpose(2, 0, 1) / 255.0
    return img[np.newaxis].astype(np.float32)

# 推理
image = cv2.imread("test.jpg")
input_data = preprocess(image)
outputs = session.run(None, {"images": input_data})

# YOLO26特别注意：无需NMS后处理！
# 输出已是最终检测结果

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np
import time

# ========== 1. 引擎加载与上下文创建 ==========
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
runtime = trt.Runtime(TRT_LOGGER)

with open("yolo26n.engine", "rb") as f:
    engine = runtime.deserialize_cuda_engine(f.read())

context = engine.create_execution_context()

# ========== 2. CUDA 内存分配 ==========
stream = cuda.Stream()
bindings = []

for i in range(engine.num_io_tensors):
    name = engine.get_tensor_name(i)
    shape = engine.get_tensor_shape(name)
    dtype = trt.nptype(engine.get_tensor_dtype(name))
    size = trt.volume(shape)
    
    host_mem = cuda.pagelocked_empty(size, dtype)   # 主机锁页内存
    device_mem = cuda.mem_alloc(host_mem.nbytes)    # 设备显存
    bindings.append({"name": name, "host": host_mem, "device": device_mem,
                     "shape": shape, "size": size, "dtype": dtype})

# ========== 3. 异步推理循环 ==========
def async_infer(input_blob):
    # 输入拷贝：主机 → 设备
    np.copyto(bindings[0]["host"], input_blob.ravel())
    cuda.memcpy_htod_async(bindings[0]["device"], bindings[0]["host"], stream)
    
    # 设置张量地址并执行
    context.set_tensor_address(bindings[0]["name"], int(bindings[0]["device"]))
    context.set_tensor_address(bindings[1]["name"], int(bindings[1]["device"]))
    context.execute_async_v3(stream.handle)
    
    # 输出拷贝：设备 → 主机
    cuda.memcpy_dtoh_async(bindings[1]["host"], bindings[1]["device"], stream)
    stream.synchronize()
    
    return bindings[1]["host"].copy()

# ========== 4. 性能基准测试 ==========
def benchmark(warmup=10, runs=100):
    dummy = np.random.randn(1, 3, 640, 640).astype(np.float32)
    for _ in range(warmup):
        async_infer(dummy)
    
    latencies = []
    for _ in range(runs):
        t0 = time.perf_counter()
        async_infer(dummy)
        latencies.append((time.perf_counter() - t0) * 1000)
    
    latencies.sort()
    print(f"TensorRT FP16 | 平均: {np.mean(latencies):.1f}ms | "
          f"P50: {latencies[runs//2]:.1f}ms | "
          f"P99: {latencies[int(runs*0.99)]:.1f}ms | "
          f"吞吐: {1000/np.mean(latencies):.0f} FPS")

benchmark()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

import openvino as ov
import cv2
import numpy as np
import time

# ========== 1. ONNX → OpenVINO 转换 ==========
# Ultralytics 统一导出:
#   model.export(format="openvino", half=True)

core = ov.Core()
model = core.read_model("yolo26n_openvino/yolo26n.xml")

# ========== 2. CPU 推理 ==========
compiled_cpu = core.compile_model(model, device_name="CPU")
infer_request = compiled_cpu.create_infer_request()

def openvino_infer(image):
    img = cv2.resize(image, (640, 640))
    blob = img.transpose(2, 0, 1)[np.newaxis].astype(np.float32) / 255.0
    outputs = infer_request.infer({"images": blob})
    return outputs[next(iter(outputs))]

# ========== 3. 异步推理管线（提升吞吐）==========
def async_pipeline(images, num_requests=4):
    """多请求异步推理管线"""
    requests = [core.compile_model(model, "CPU").create_infer_request()
                for _ in range(num_requests)]
    results = [None] * len(images)
    
    def completion_callback(request, userdata):
        idx = userdata
        results[idx] = request.get_output_tensor().data.copy()
    
    for req in requests:
        req.set_callback(completion_callback)
    
    for i, img in enumerate(images):
        req = requests[i % num_requests]
        req.start_async({"images": preprocess(img)}, userdata=i)
    
    for req in requests:
        req.wait()
    
    return results

# ========== 4. CPU vs NPU 基准对比 ==========
def benchmark_openvino():
    dummy = np.random.randn(1, 3, 640, 640).astype(np.float32)
    
    for device in ["CPU", "AUTO"]:
        compiled = core.compile_model(model, device)
        req = compiled.create_infer_request()
        
        # 预热（避免首次 kernel 编译开销）
        for _ in range(20):
            req.infer({"images": dummy})
        
        times = []
        for _ in range(200):
            t0 = time.perf_counter()
            req.infer({"images": dummy})
            times.append((time.perf_counter() - t0) * 1000)
        
        times.sort()
        print(f"OpenVINO {device}: "
              f"平均 {np.mean(times):.1f}ms | "
              f"P99 {times[int(199*0.99)]:.1f}ms | "
              f"{1000/np.mean(times):.0f} FPS")

benchmark_openvino()

1
2
3
4
5
6

# 检查驱动版本（Driver Version 需 >= 支持CUDA的最低版本）
nvidia-smi
# 检查CUDA工具包版本
nvcc --version
# 重新安装对应CUDA版本的PyTorch
pip3 install torch torchvision --index-url https://download.pytorch.org/whl/cu121