边缘部署实战：让算法在有限算力下稳定运行

在这里插入图片描述

关于作者

我接触视觉整整 10 年。

机器视觉、烟草、煤矿等行业都有深度开发经验。从硬件选型、算法开发、模型训练，到上位机开发及部署，都在一线磨过。

之前是多家公司人工智能团队的技术负责人。现在自己创业了，还在继续做视觉落地这件事。

作者说

在做视觉这件事之前，我以为最难的是算法开发和模型精度。

后来发现，真正让人崩溃的从来不是算法本身，而是现场那些"没想到"的情况——

  • 摄像头突然逆光，画面一片白
  • 皮带头晃动，带动的尘雾让检测全乱
  • 机器停了，但传感器报告还在"运行"
  • 边缘设备算力不够，一运行就卡死

这些问题是实验室里遇不到的。

所以我打算写一个系列，记录我们在真实工业场景下踩过的坑、解决过的问题。不讲多么酷炫的算法，只聊怎么让算法稳定跑在客户现场。

这是第五篇。

踩坑实录：现场实测完美，一到客户那就崩了

又一个项目交付的时候，我已经做好了算法。

实验室测试——完美。
去客户现场部署——

第一天还好。

第二天，客户打电话过来：系统卡死了。

我远程一看，内存占用 98%，进程被 OOM Kill 了。

后来排查发现：现场网络不稳定，视频流时断时续，我的代码没有做好断线重连和内存管理。每次断线重连都重新创建模型，内存一直涨，最后直接爆了。

这是我踩的第四个大坑：边缘部署轻视不得。

从那以后，我们花了整整半个月时间，专门打磨边缘部署的稳定性。

01 边缘设备的选择

1.1 常见选择

设备类型 算力 典型场景 推荐指数
Jetson Nano ~0.5 TOPS 入门 ⭐⭐
Jetson Xavier NX ~21 TOPS 主流工业部署 ⭐⭐⭐⭐
Jetson Orin Nano ~40 TOPS 新项目推荐 ⭐⭐⭐⭐⭐
Jetson Orin NX ~70 TOPS 高性能 ⭐⭐⭐⭐⭐

1.2 我们的选择

以前我们用 TX2 比较多，现在新项目基本都换成 Orin Nano / Orin NX：

  • 算力足够跑 YOLO + 光流
  • 功耗适中，可靠性高
  • 支持 CUDA，模型加速方便
  • 工业级稳定性较好

1.3 选型经验

核心原则：留足算力余量。

我们一般这样算：

  • 理论算力需求 × 2 = 实际选型算力

因为工业现场要考虑：

  • 算力波动（高温降频）
  • 同时运行多个模型
  • 系统开销

02 模型加速

2.1 为什么需要加速？

YOLO 原版模型太大，推理太慢：

模型 Jetson TX2 (TensorRT)
YOLOv5n ~10–15ms
YOLOv5s ~20–30ms

建议用 YOLOv5n 或 YOLOv8n——参数量小，精度损失可接受。

2.2 加速手段

手段一：模型量化（TensorRT）
# PyTorch → ONNX → TensorRT (FP16/int8)

# 1. 导出 ONNX
python export.py --weights yolov5n.pt --img 640 --batch 1 --onnx

# 2. 导出 TensorRT (FP16)
python yolov5_to_trt.py --onnx yolov5n.onnx --engine yolov5n.fp16.engine --fp16

# 3. 导出 TensorRT (int8) - 需要校准数据
python yolov5_to_trt.py --onnx yolov5n.onnx --engine yolov5n.int8.engine --int8 --calib-img /path/to/calib/images

效果对比：

精度 推理加速 精度损失
FP32 1x 基准
FP16 ~2x <1%
INT8 ~3-4x ~2-3%

Jetson 上通常 默认使用 FP16，INT8 需要校准数据集。

手段二：输入尺寸优化
输入尺寸 推理时间 mAP 变化
640×640 基准 基准
512×512 -20% -1~2%
416×416 -35% -3~5%

经验：在工业场景中，适当降低输入尺寸往往是最有效的优化手段。

03 抽帧策略：算力不够，帧数来凑

3.1 核心思想

边缘设备算力有限，不是每一帧都需要做复杂推理。

我们的策略是：

  • 正常状态：低频检测（每秒 1-2 次）
  • 检测到运动：提升频率（每秒 5-10 次）
  • 异常状态：持续高频检测

3.2 代码实现

class AdaptiveFrameStrategy:
    """自适应抽帧策略"""
    
    def __init__(self, fps=30):
        self.fps = fps
        self.current_interval = 2  # 当前检测间隔（帧）
        self.min_interval = 1      # 最小间隔
        self.max_interval = 10      # 最大间隔
        
        self.motion_count = 0
        self.static_count = 0
        self.last_state = "static"
    
    def should_detect(self, frame_idx):
        """判断当前帧是否需要检测"""
        return frame_idx % self.current_interval == 0
    
    def update(self, motion_detected):
        """根据运动状态动态调整检测频率"""
        if motion_detected:
            self.motion_count += 1
            self.static_count = 0
            
            # 连续运动，增加检测频率
            if self.motion_count > 5:
                self.current_interval = max(self.min_interval, self.current_interval - 1)
                self.last_state = "motion"
        else:
            self.static_count += 1
            self.motion_count = 0
            
            # 连续静止，降低检测频率
            if self.static_count > 10:
                self.current_interval = min(self.max_interval, self.current_interval + 1)
                self.last_state = "static"

3.3 在光流计算中的应用

在我原来的代码里，是类似的逻辑：

# 原始代码片段
if self.prev_pts is None or len(self.prev_pts) < 30:
    run_optical_flow = True  # 特征点不足，必须更新
elif recent_motion > 0:
    run_optical_flow = True  # 最近有运动，全帧计算
elif diff_score > 0.005:
    run_optical_flow = True  # 有轻微变化，间隔性计算
elif self.frame_counter % 3 == 0:
    run_optical_flow = True  # 场景静止时，降频计算

核心思想是一样的：根据场景动态调整计算频率。

04 内存管理：防止 OOM

4.1 常见内存问题

  • 模型重复加载：每次推理都创建新模型
  • 缓存积累：历史数据一直累积不释放
  • 内存泄漏：某些对象无法被 GC 回收

4.2 场景优化

真实工业系统一般不会 动态删除模型。

因为重新加载模型：

  • 非常慢
  • GPU重新分配显存
  • 可能导致卡顿

更真实的做法是：

  • 限制缓存
  • 重启进程
  • watchdog

4.3 断线重连策略

class VideoStreamManager:
    """视频流管理器"""
    
    def __init__(self, url, reconnect_interval=5, max_reconnect=10):
        self.url = url
        self.cap = None
        self.reconnect_interval = reconnect_interval
        self.max_reconnect = max_reconnect
        self.reconnect_count = 0
    
    def get_frame(self):
        """获取帧，带自动重连"""
        if self.cap is None or not self.cap.isOpened():
            self._connect()
        
        if self.cap is not None and self.cap.isOpened():
            ret, frame = self.cap.read()
            if ret:
                return frame
        
        # 读取失败，尝试重连
        self._reconnect()
        return None
    
    def _connect(self):
        """连接视频流"""
        self.cap = cv2.VideoCapture(self.url)
        self.reconnect_count = 0
    
    def _reconnect(self):
        """重连"""
        if self.reconnect_count >= self.max_reconnect:
            print("最大重连次数 reached")
            return
        
        self.reconnect_count += 1
        print(f"尝试重连 ({self.reconnect_count}/{self.max_reconnect})...")
        
        # 释放旧资源
        if self.cap is not None:
            self.cap.release()
        
        time.sleep(self.reconnect_interval)
        self._connect()
    
    def release(self):
        """释放资源"""
        if self.cap is not None:
            self.cap.release()
            self.cap = None

05 断线重连与状态恢复

5.1 问题

工业现场网络不稳定，视频流可能中断。

5.2 解决思路

  • 自动重连：检测到断线后自动重连
  • 状态恢复：重连后恢复之前的检测状态
  • 数据补齐：用缓存的数据暂时填充

5.3 在代码中的实现

# 断线重连逻辑
def _init_after_reconnect(self):
    """断流后重启，变量重置"""
    # 保留关键状态
    self.last_decision_time = time.time()
    
    # 重置检测状态
    self.dust_frame_counter = 0
    self.dust_triggered = False
    self.person_mode = False
    
    # 光流状态需要重新初始化
    self.motion_flag_history.clear()
    self.prev_gray = None
    self.prev_pts = None
    self.frame_counter = 0

06 监控与告警

6.1 核心指标

指标 阈值 动作
内存使用率 >85% 告警 + 清理缓存
CPU 使用率 >90% 告警 + 降低帧率
GPU温度 >80° 告警
推理时间 >100ms 记录日志
帧率 <10fps 告警
断线次数 >5次/小时 告警

6.2 代码实现

class SystemMonitor:
    """系统监控"""
    
    def __init__(self):
        self.metrics = {
            "fps": [],
            "inference_time": [],
            "memory_usage": []
        }
        self.alert_thresholds = {
            "memory_ratio": 0.85,
            "fps_min": 10,
            "inference_time_max": 100
        }
    
    def record(self, fps, inference_time, memory_ratio):
        """记录指标"""
        self.metrics["fps"].append(fps)
        self.metrics["inference_time"].append(inference_time)
        self.metrics["memory_usage"].append(memory_ratio)
    
    def check_alerts(self):
        """检查是否需要告警"""
        alerts = []
        
        avg_fps = np.mean(self.metrics["fps"][-30:])
        avg_inference = np.mean(self.metrics["inference_time"][-30:])
        avg_memory = np.mean(self.metrics["memory_usage"][-30:])
        
        if avg_fps < self.alert_thresholds["fps_min"]:
            alerts.append(f"FPS过低: {avg_fps:.1f}")
        
        if avg_inference > self.alert_thresholds["inference_time_max"]:
            alerts.append(f"推理时间过长: {avg_inference:.1f}ms")
        
        if avg_memory > self.alert_thresholds["memory_ratio"]:
            alerts.append(f"内存使用过高: {avg_memory*100:.1f}%")
        
        return alerts

07 总结

本文分享了边缘部署的实战经验，核心要点：

  1. 设备选型：留足算力余量，推荐 Jetson TX2 / Orin
  2. 模型加速：TensorRT 量化 + 输入尺寸优化
  3. 抽帧策略：根据场景动态调整检测频率
  4. 内存管理：单例模型 + 缓存限制 + 自动清理
  5. 断线重连：自动重连 + 状态恢复
  6. 系统监控：关键指标实时监控 + 告警

核心教训：在现场稳定运行，比算法精度更重要。

08 写在最后

工业视觉落地，拼的不是算法多先进，而是能不能在客户现场稳定跑起来。

边缘部署是一个"细节决定成败"的领域——一个内存泄漏、一个断线没处理，就可能导致整个系统崩掉。

希望这篇文章能帮你避坑。

如果你也有工业视觉落地的困扰，欢迎一起交流。

*本文所有代码均为示意，核心思路可复现，具体参数需根据实际场景调整。

📎 相关阅读

# 目标检测 # opencv # 人工智能
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

MoteDB：重新定义具身智能数据库，AI原生多模态数据管理的破局者

avatar AtomGit开源社区
cover

微表情检测系统，yolov8，支持图像、视频和摄像实时检测【pytorch框架、python源码】

avatar AtomGit开源社区

AI Agent 核心概念与 ReAct 框架完全指南

概念一句话类比Agent会思考和决策的 AI 执行者项目经理Skill某领域的专业能力包部门团队Tool执行单一操作的函数具体工具/员工MCP工具与 AI 之间的标准接口协议USB/HTTP 标准核心关系：Agent 通过 MCP 协议发现和调用 Tool，将相关 Tool 组合成 Skill，最终完成复杂任务。│ ││ ReAct = 让 AI 像人一样： ││ ││ 先想清楚 → 再动手做 →

avatar AtomGit开源社区