售货柜视觉识别流水线实战:RTSP拉流、抽帧与YOLO部署 做售货柜视觉识别项目时最核心的一条流水线就是 IPC 拉流、抽帧、YOLO 识别。我刚接手这类项目的时候以为难点全在模型识别率上结果真正卡住进度的全是拉流不稳定、抽帧节奏不对、队列把内存打爆这些小问题。这篇文章把我从零搭到稳定运行的完整链路写出来包括 IPC 选型、RTSP 拉流细节、抽帧策略、YOLO 模型部署和最后的业务判定闭环给正在做或者准备做类似项目的人一个可以直接参考的路线。1. 为什么售货柜不能来一帧识别一帧从业务需求倒推技术选型1.1 业务场景对延迟和准确率的硬性约束售货柜和常规监控项目不一样它的核心诉求不是事后查录像而是在顾客开关柜门的十几秒里判断出他到底拿走了什么、拿了几个。这个判断结果直接决定订单结算金额错了就是客诉。我这边实际跑过的柜子开门到关门一般是 8 到 20 秒顾客可能中间犹豫、拿起又放下、遮挡商品。留给识别系统的时间窗口其实不宽裕但也没有苛刻到必须每帧都判。比较合理的预期是关门后 2 到 5 秒内给出结果准确率 95% 以上。延迟卡在 5 秒内是因为顾客还站在柜子前等手机弹结算通知等太久体验极差。另一个硬约束是成本。售货柜本质是零售终端毛利没那么高不可能给每台柜子配一台顶配 GPU 服务器。目前主流方案是柜内放一台工业级边缘盒子或者旧工控机CPU 或者轻量 NPU 去跑模型。算力有限就意味着每一帧都过 YOLO 是不现实的。还有一点容易被忽略相邻帧之间高度冗余。视频 25 帧每秒顾客拿一罐可乐可能只有 0.8 秒但前后画面几乎没变化。如果每帧都推理结果会剧烈抖动——同一瓶饮料有时候检到、有时候没检到反而让判断逻辑很难写。所以从业务规则上就需要抽帧来稳定输入。1.2 三段式流水线的职责划分以及为什么不把识别塞进摄像机我最终搭的流水线逻辑上分成三层拉流层从网络摄像机IPC取 RTSP 视频流负责取到最新的一帧并保证断流后能自动恢复。抽帧层决定哪些帧值得送进模型也就是所谓的采样节奏避免无效计算。识别层跑 YOLO 目标检测输出商品类别、坐标、置信度再结合业务规则算出哪些商品被拿走。很多刚入门的人会问能不能直接在 IPC 里做识别市面上确实有带智能算法的 IPC能识别人员闯入、越界这种固定场景但售货柜里要识别的是几十上百种具体商品和柜内光线、摆放角度、包装反光强相关厂商内置算法根本不可能覆盖。所以老老实实把视频流拉到本地用自训练模型识别是目前最可落地的方式。拆成三段还有个好处每一层可以独立调试。拉流出问题就只查网络和协议识别率低就只调模型和数据集不会互相干扰。后面我会按这个顺序一层层讲。2. IPC选型与RTSP拉流细节稳定取流是整条流水线的前提2.1 柜内 IPC 怎么选重点看哪几个参数售货柜里的摄像头一般不是给你自由发挥的空间柜体结构已经预留了安装位和线束。选型时我重点看四个参数分辨率、焦距、编码格式、供电方式。分辨率不用盲目上 400 万像素200 万像素1080p在柜内 1 到 2 米的距离下完全够用像素太高反而拖慢解码。焦距要选广角柜内纵深有限通常 2.8mm 左右比较合适能覆盖单层货架的大半视野。编码格式优先 H.265码率比 H.264 省一半以上对磁带库和带宽都是好事。供电这块能选 PoE 就选 PoE一根网线把电和图像都解决。不过很多旧柜子内部只有 DC 供电口那就要注意 IPC 的电压规格是不是 12V别现场再变电。还有一个细节买回来一定要先测试 RTSP 协议是否开放。部分消费级摄像头默认开启云平台转发却不开放 RTSP 或者需要临时开启这种就不能用因为后续我们所有取流都走 RTSP。2.2 RTSP 地址里藏的坑主码流、子码流与鉴权拿到一台 IPC第一件事是拼 RTSP 地址。以海康为例常见格式是rtsp://admin:password192.168.1.64:554/Streaming/Channels/101admin:password是摄像头的登录账号密码必须得有访问权限。192.168.1.64是摄像头 IP。554是默认 RTSP 端口。/Streaming/Channels/101表示取通道 1 的主码流102表示通道 1 的子码流。不同品牌路径差异很大。大华一般是/cam/realmonitor?channel1subtype0宇视是/live/ch0。所以选型时要尽量统一品牌或者在代码里做一张设备型号到 URL 模板的映射表否则维护成本很高。这里有个非常容易踩的坑主码流分辨率高、码流大清晰度好但解码耗时也高子码流分辨率低、延迟小适合传输和预览却不适合精细识别。我在实际项目里会拉主码流做识别因为商品小字、包装图案都需要细节。如果只是想看画面通不通才用子码流测试。鉴权信息不要硬编码在代码里。虽然柜子内网相对封闭但摄像机密码有时候要定期更换写成配置文件更便于运维。2.3 用 OpenCV 还是 FFmpeg 拉流以及自动重连策略拉 RTSP 流最简单的是 OpenCV 的VideoCaptureimport cv2 cap cv2.VideoCapture(rtsp://admin:password192.168.1.64:554/Streaming/Channels/101) ret, frame cap.read()这段代码跑通很容易但问题在cap.read()默认会缓存最近一帧如果网络抖动返回的画面可能是几秒前的旧画面延迟忽高忽低。OpenCV 底层用的还是 FFmpeg 的解码能力只是接口简化了所以遇到复杂场景我不太建议直接裸用。我的做法是分两种情况快速验证、原型 demo用 OpenCV 足够。生产环境用 FFmpeg 拉流管道接 Python或者直接用支持缓冲控制的 GStreamer 管道。用 FFmpeg 子进程推流进管道的例子ffmpeg -rtsp_transport tcp -i rtsp://admin:password192.168.1.64:554/Streaming/Channels/101 -f rawvideo -pix_fmt bgr24 -an -sn - pipePython 侧从管道读原始帧再转numpy数组。这种方式延迟可控但代码量比 OpenCV 大不少。无论用哪种方式重连策略必须做。IPC 这东西没人敢保证一年不掉线碰到升级固件、异常断电、网线松动流就断了。我用的策略是连续 3 秒拿不到新帧就触发重连重连失败后间隔指数退避从 2 秒开始最多退到 30 秒避免空转打满 CPU。import time import cv2 def connect_with_retry(url, max_retries10): cap cv2.VideoCapture(url, cv2.CAP_FFMPEG) if not cap.isOpened(): raise RuntimeError(can not open stream) return cap # 使用时持续 read超时则重建连接3. 抽帧模块设计不是简单sleep 一下就能完事3.1 抽帧解决的三个问题冗余、抖动、算力抽帧字面上是从视频流里挑出部分帧。但它背后解决的问题有三个。第一是降低算力消耗。YOLOv5s 在一块普通 CPU 上推理一帧可能 200 到 400 毫秒而 IPC 一秒出 25 帧如果满负荷跑CPU 永远追不上帧会在队列里越积越多。抽帧后推理频率降到每秒 2 到 3 次负载就完全可控。第二是消除检测结果抖动。目标检测对画面变化很敏感视频压缩产生的噪点、光线微变、货架反光都可能导致同一目标在相邻帧里检出又漏掉。抽帧让输入信号稳定后端的消失判定才做得准。第三是减少重复计算。顾客拿一瓶水可能 3 秒内画面几乎没变化重复推理这些帧没有业务价值纯属浪费。不过抽帧不是越少越好抽太少了会漏掉关键动作。我见过有人把抽帧间隔设成 3 秒结果顾客快速连拿两个商品只拍到第二个第一个直接漏判。抽帧节奏必须和业务动作速度匹配。3.2 基于时间戳的等间隔抽帧实现别用固定帧号一开始我犯过一个错用帧计数来做抽帧比如每 10 帧送一次推理。问题是 IPC 实际帧率不稳定繁忙时资源不够会掉帧固定帧号间隔的时间跨度忽长忽短。后来我改成按时间戳抽帧。核心逻辑很简单每收到一帧记录当前系统时间如果距离上一次送入推理的时间间隔大于我们设定的阈值比如 0.5 秒就送这一帧否则直接丢弃或者仅用于显示。import time import threading class FrameSampler: def __init__(self, interval0.5): self.interval interval self.last_infer_time 0.0 self.lock threading.Lock() def should_infer(self, frame_stampNone): if frame_stamp is None: frame_stamp time.time() with self.lock: if frame_stamp - self.last_infer_time self.interval: self.last_infer_time frame_stamp return True return False这里的interval怎么定我一般先看模型单帧推理耗时t_infer然后取max(t_infer * 1.5, 0.3)作为抽帧间隔保证推理不积压同时又不会为追求低负载丢掉动作细节。如果用的是 GPU 或者 NPUt_infer 很短抽帧间隔主要根据业务需要来定。3.3 抽帧后的画面连续性录像回放不受影响判断逻辑要用基准帧有次和客户对需求对方很担心抽帧会把视频弄卡其实这是概念混淆。我们抽的是喂给模型做推理的帧不是存进硬盘的录像帧。如果项目里还需要原始录像完全可以从 IPC 的另一个码流比如子码流持续录像或者在拉流端做副本存储抽帧只影响识别模块的输入不影响录像完整性。但抽帧确实会让业务判断逻辑变复杂。因为不是每一帧都进模型后端如果要判断某个商品消失了就需要在抽帧之间维护一个基准状态。我的做法是抽帧送入推理后把识别结果缓存为last_result下一次新结果出来后和它做对比得出商品变化。这个基准帧不需要每帧更新只有新的有效结果才刷新。4. YOLO 识别模型选择、后处理与业务规则闭环4.1 模型尺寸和硬件平台怎么匹配YOLO 系列到现在已经出了很多版本我这边实际用过的有 YOLOv5、YOLOv8也试过 YOLO11。版本本身不是最重要的关键看部署平台支持什么。如果是 NVIDIA Jetson 或者有 GPU 的工控机优先考虑 TensorRT 加速YOLOv8 可以直接导出为 engine 格式推理速度很快。如果是瑞芯微 RK3588 这类平台通常导出 RKNN 格式YOLOv5 的适配资料最全YOLOv8 也行但编译工具版本得对应。如果是纯 CPU 环境老实选 nano 或者 small 尺寸的模型。模型尺寸选择我给一个参考场景模型尺寸推理设备单帧耗时参考快速原型YOLOv8nCPU80-150ms量产边缘盒子YOLOv8n/sRK3588 NPU20-50ms更高精度场景YOLOv8s/mJetson Orin / 低端 GPU15-40ms纯 CPU 老工控机YOLOv5nuCPU150-300ms有一点非常重要预训练的 COCO 权重不能直接用于售货柜商品识别。COCO 的 80 类里根本没有某种口味的饮料某品牌薯片这种细分类。所以必须自采数据、自标注、重新训练。我建议每类商品至少采集 500 到 1000 个样本覆盖不同光照、角度、遮挡情况只有柜内真实拍摄的照片才有效。4.2 预处理、NMS 和类别过滤这些细节决定上限很多教程演示 YOLO 都是直接调用ultralytics库一行代码出结果。但在生产环境我通常不用它的高层 API而是导出 ONNX 后用 ONNX Runtime 或者 TensorRT 跑这样部署灵活也不会被框架版本绑架。推理流程拆开看预处理把摄像头原始帧缩放成模型输入尺寸比如 640×640这一步不能直接拉伸会破坏宽高比得用 letterbox 方式在边缘补灰边。推理inputs是一张归一化后的张量outputs是检测框、类别概率、置信度。后处理先按置信度阈值过滤再做 NMS非极大值抑制去掉重叠框最后把坐标缩放回原始图像尺寸。置信度阈值我通常设在 0.25 到 0.45 之间。调低会召回更多目标但误检也变多调高则相反。售货柜业务里漏检比误检更难处理因为商品没被识别门店损失所以我一般偏低一点把误检交给业务层去修正。4.3 从检测框到商品被拿走判定逻辑才是业务闭环的核心识别模型只是个眼睛真正决定这笔订单怎么算的是后端的判定逻辑。我第一版犯的错误是直接看两帧的检测框数量差比如前一帧检出 3 瓶可乐后一帧检出 2 瓶就判定拿走 1 瓶。但顾客的手伸进去拿东西时手会遮挡商品中间若干帧可能少检了最后其实什么都没拿。后来我改成多帧确认 形态快照的思路柜门刚刚打开时采集连续 3 到 5 帧有效结果作为基准快照记录每个类别的数量。柜门关闭前再采集连续 3 到 5 帧有效结果计算最终快照。最终快照和基准快照做数量差才生成拿走清单。如果两者数量差异较大需要结合柜门的开关信号确认是否真的完成了一次购买。这里的关键点是不用中间帧做决策中间帧只负责更新当前状态避免手部遮挡造成的误判。这个逻辑帮我大幅压低了拿起又放回却扣了钱的客诉。5. 整条流水线的工程化落地线程队列、掉线重连与异常兜底5.1 生产者-消费者模型拉流、抽帧、识别三个线程怎么分工把流水线做成单线程顺序执行代码最简单但有一个致命问题拉流等待 I/O、模型推理都是阻塞操作如果它们串行执行拉流慢的时候推理空转推理慢的时候拉流缓冲积累。实际部署时我用了三个线程通过队列解耦。大致结构import threading import queue class Pipeline: def __init__(self, rtsp_url, infer_interval0.5): self.url rtsp_url self.frame_queue queue.Queue(maxsize10) # 原始帧缓冲 self.infer_queue queue.Queue(maxsize5) # 待推理帧缓冲 self.sampler FrameSampler(infer_interval) def pull_loop(self): # 循环拉流塞入 frame_queue pass def sample_loop(self): # 从 frame_queue 取帧按时间戳抽帧塞入 infer_queue pass def infer_loop(self): # 从 infer_queue 取帧跑 YOLO 后处理回调结果 pass def start(self): threading.Thread(targetself.pull_loop, daemonTrue).start() threading.Thread(targetself.sample_loop, daemonTrue).start() threading.Thread(targetself.infer_loop, daemonTrue).start()线程消息是 Python但要注意 GIL 问题。如果你的推理用 ONNX Runtime、TensorRT 这类 C 扩展核心计算会释放 GIL三个线程并行度还可以。如果纯用 CPU 跑 PyTorch 推理GIL 会限制并行效果这时候我建议要么把推理放进子进程要么干脆退化为拉流 → 抽帧 → 推理单线程顺序流减少上下文切换。5.2 队列长度、内存释放和延迟的三方平衡队列设置要非常小心。拉流线程往队列塞帧很快如果推理速度跟不上队列会越积越长延迟越来越大最后清理的时候内存一路冲高。我的经验是队列长度宁可短不要长。frame_queue设置 maxsize10缓冲区满了就把最旧的一帧丢掉保证永远拿最新的画面。待推理队列 maxsize5满了就丢弃当前最新帧防止内存膨胀。有人问丢帧会不会影响业务判定不会因为抽帧后的判断本身就不依赖连续帧丢一两帧不会改变最终数量对比。内存释放上Python 的numpy数组复用是个容易被忽略的坑。如果你在循环里反复cap.read()拿到新帧旧帧在没有引用后会被回收但为了减少 GC 压力可以在用完后直接del frame或者用数组池复用内存。5.3 掉线重连、看门狗与异常兜底稳定运行期最大的敌人是网络。IPC 掉线后拉流线程会抛异常或者阻塞我做了几个兜底拉流线程自己维护一个最近收到帧时间last_frame_time如果超过 5 秒没有新帧主动销毁当前VideoCapture然后按 2 秒、4 秒、8 秒……的退避策略重连。识别线程如果连续多次推理失败比如模型加载异常、推理结果解析失败不能死循环卡死要记录日志并继续下一帧。整条流水线加一个外部看门狗定时检查三个线程是否还活着如果某个线程意外退出尝试重启该线程而不是重启整个进程。这些逻辑看起来繁琐但实际部署后很有用。我维护的一批柜子大部分异常都是断网导致的没有自动重连机制的话掉一次线就得跑一次现场。6. 实测性能与调优经验从 demo 到稳定交付6.1 一个典型配置下的实测数据我拿一套典型的量产配置做过压测1080p IPC主码流 4MbpsH.265 编码边缘盒子是 RK3588YOLOv8n 导出 RKNN抽帧间隔 0.5 秒。连续跑了 72 小时结果如下项目实测值说明拉流到拿到帧的平均延迟120-180ms局域网 RTSP over TCP单帧 YOLO 推理耗时25-35msRKNN 加速抽帧间隔0.5s每秒约 2 次识别单次拿取动作的判定耗时关门后 1.5-3s含尾帧确认内存占用400-500MBPython ONNX RuntimeCPU 占用15-25%四核中低频运行这个数据说明什么说明整个链路没有瓶颈留给业务判断的时间非常充裕。如果推理耗时超过 100ms我会怀疑是不是模型太大或者平台驱动没装好。6.2 我踩过的高频坑和对应方案最后把几个高频坑列出来每一个都是真金白银换来的RTSP 拉流地址超时。IPC 默认可能只允许一个 RTSP 会话如果你同时拉主码流和子码流有些廉价摄像头会直接拒绝。解决方法是确认设备最大会话数或者只用一路。柜内反光导致识别率骤降。饮料瓶、薯片包装都容易反光特别是到了晚上柜内补光灯一照反光区域直接盖住商品。我后期的做法是把补光灯改成漫反射灯条同时训练数据里加入傍晚和夜晚样本。YOLO 后处理坐标越界。letterbox 缩放后坐标回原图时需要减去 padding 值再除以缩放比例很多新手在这里直接乘比例导致画框偏移。这属于代码细节但影响很大。误检商品导致扣错款。有顾客手里拿着手机靠近摄像头模型把手机识别成某个商品。这种情况没法只靠模型解决我加了区域限制只在货架区域做目标匹配超出 ROI 的目标直接忽略。时间戳不一致。不同摄像头的时间不同步如果一台柜子有多个摄像头各自抽帧的时间点不同在做跨镜头的同一商品匹配时会出问题。方案是统一以服务器时间为准在收到帧时打本地时间戳。一点收尾经验售货柜识别项目做到后期我的体会是模型精度只占成功的一半另一半是拉流稳定性、抽帧节奏和业务判定逻辑。如果你正准备启动类似项目我建议先把拉流和重连模块调稳定再用最少的数据把模型跑通一个端到端 demo最后才去抠识别率。这个顺序能让你少走非常多弯路。