Django + YOLOv8 鸟类识别平台的“反脆弱”架构：异步事件循环、显存零拷贝与动态NMS阈值调优

# model_loader.py
import torch
from ultralytics import YOLO
from threading import Lock

class YOLOv8Singleton:
    _instance = None
    _lock = Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    # 关键点：map_location 固定至当前CUDA设备，避免多卡争抢
                    cls._instance.model = YOLO("weights/bird_best.pt")
                    # 预热：空跑一次，提前分配CUDA Context
                    cls._instance.model.predict("assets/warmup.jpg", verbose=False)
        return cls._instance

    def predict(self, image_bytes):
        return self.model.predict(source=image_bytes, verbose=False)

# inference_context.py
import gc
import torch
from contextlib import contextmanager

@contextmanager
def safe_inference(image_tensor):
    """推理上下文管理器，确保异常退出时也能回收显存"""
    try:
        yield image_tensor
    finally:
        # 1. 删除当前批次的所有中间变量引用
        del image_tensor
        # 2. 触发 Python 垃圾回收，释放循环引用
        gc.collect()
        # 3. 清空 PyTorch 的显存缓存池（最重要的一步）
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
            # 显存碎片整理（针对长时间运行）
            torch.cuda.synchronize()

@router.post("/detect")
def detect_bird(request, file: UploadFile = File(...)):
    img_bytes = file.read()
    model = YOLOv8Singleton().predict(img_bytes)
    
    with safe_inference(img_bytes):
        # 转为Tensor并移至GPU
        tensor = preprocess(img_bytes).cuda()
        with torch.no_grad():
            results = model.predict(tensor, imgsz=640, half=True)  # 开启半精度 FP16
        # 后处理逻辑...
        return {"bboxes": parse_results(results)}

def dynamic_nms_filter(boxes, scores, img_shape):
    """
    boxes: xyxy 格式 Tensor (N, 4)
    面积公式: (x2-x1) * (y2-y1)
    """
    # 计算归一化面积 (相对于 640x640)
    areas = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
    img_area = img_shape[0] * img_shape[1]
    norm_areas = areas / img_area
    
    keep_indices = []
    # 按置信度排序
    sorted_idx = torch.argsort(scores, descending=True)
    
    while len(sorted_idx) > 0:
        current = sorted_idx[0]
        keep_indices.append(current)
        if len(sorted_idx) == 1:
            break
        
        # 计算剩余框与当前框的 IoU
        rest_boxes = boxes[sorted_idx[1:]]
        ious = compute_iou(boxes[current], rest_boxes)
        
        # 动态判定：面积越小，允许的 IoU 重叠越大
        current_area = norm_areas[current]
        # 当面积 < 10% 时，将 IoU 阈值放宽至 0.65 (默认0.45)
        dynamic_thresh = 0.45 + (0.2 * (1 - min(current_area, 0.2) / 0.2))
        
        # 保留 IoU < dynamic_thresh 的索引
        keep_mask = ious < dynamic_thresh
        sorted_idx = sorted_idx[1:][keep_mask]
    
    return keep_indices

# views.py 异步视图
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 单独开辟线程池，防止与主循环争抢 GIL
inference_executor = ThreadPoolExecutor(max_workers=2) 

async def async_detect(request):
    data = await request.body()
    
    # 使用 asyncio.to_thread 让同步推理代码不阻塞事件循环
    loop = asyncio.get_running_loop()
    # 关键：将模型推理放到线程池，并设置 timeout 防止饿死
    try:
        result = await asyncio.wait_for(
            loop.run_in_executor(inference_executor, sync_inference_function, data),
            timeout=5.0
        )
        return JsonResponse({"status": "success", "data": result})
    except asyncio.TimeoutError:
        # 超时保护：清理线程池中的僵尸任务
        inference_executor.shutdown(wait=False)
        return JsonResponse({"status": "error", "msg": "Inference timeout"}, status=504)

class GPUMemoryGuard:
    def __enter__(self):
        if torch.cuda.is_available():
            # 获取当前显存占用率
            used = torch.cuda.memory_allocated() / 1024**3
            total = torch.cuda.get_device_properties(0).total_memory / 1024**3
            if used / total > 0.85:
                # 触发降级：强制使用 CPU 推理（虽然慢，但不至于 OOM 崩溃）
                self.fallback_to_cpu = True
                log.critical("GPU memory over 85%, downgrading to CPU inference!")
            else:
                self.fallback_to_cpu = False
        return self