1. 为什么服务层是整个链路里最容易被低估的一环模型转换完了NPU 推理也跑通了单张图片测下来延迟 30ms 以内看起来一切都很美好。但真正要把这套东西用起来你不可能每次都手动敲命令、传图片、等结果。你需要一个服务层把推理能力包装成接口让摄像头采集的画面自动流进去结果自动吐出来。这一步听起来简单实际上是我整个部署过程中折腾最久的部分。模型转换有官方文档NPU 推理有现成的 C demo但服务层涉及的东西太杂了Python 环境、异步框架、摄像头协议、多线程资源竞争、内存管理任何一个环节出问题都会让你卡半天。这篇文章面向的是已经在 RK3588 上跑通了 YOLOv5s 模型推理、准备把整套东西串起来做实际应用的开发者。如果你还在折腾模型转换或者 NPU 驱动建议先把前面几步搞定再来看这篇。我会把服务层的设计思路、FastAPI 的具体实现、摄像头接入的几种方案、以及我踩过的那个折腾最久的坑全部拆开讲清楚。核心关键词贯穿全文RK3588、YOLOv5s、FastAPI、NPU、摄像头。这五个东西串起来就是一条从硬件到应用的完整链路。2. 服务层架构设计与技术选型2.1 为什么选 FastAPI 而不是 Flask服务层的核心任务很明确接收图像数据调用 NPU 推理返回检测结果。听起来用什么框架都行但我最终选了 FastAPI原因有几个。第一是异步支持。摄像头取流和 NPU 推理都是 IO 密集和计算密集混合的场景Flask 的同步模型在高并发下会很快成为瓶颈。FastAPI 基于 Starlette原生支持 async/await可以在等待 NPU 结果的时候处理其他请求。第二是自动生成 API 文档。FastAPI 自带 Swagger UI 和 ReDoc你定义好 Pydantic 模型之后接口文档自动就有了。调试阶段这个太省事了不用手动写文档前端同事直接看/docs就能对接。第三是性能。FastAPI 底层是 UvicornASGI 服务器实测下来比 Flask 加 Gunicorn 的组合吞吐量高不少。在 RK3588 这种边缘设备上每一分性能都很宝贵。当然 Flask 也有它的优势生态更成熟中间件更多。但如果你是从零开始搭一个推理服务FastAPI 的学习曲线并不陡投入产出比更高。2.2 服务层的整体架构我的架构设计是这样的摄像头 → 取流模块 → 帧队列 → 推理引擎NPU→ 结果队列 → API 响应拆开来看分成四个模块取流模块负责从摄像头USB 或 RTSP拉取视频帧做初步的预处理缩放、格式转换帧队列一个线程安全的缓冲队列解决取流速度和推理速度不匹配的问题推理引擎封装 RKNN 的推理接口从队列取帧调用 NPU输出检测结果API 层FastAPI 提供 REST 接口支持单帧推理、批量推理、状态查询这个架构的核心思路是生产者-消费者模式。取流模块是生产者推理引擎是消费者队列是缓冲区。这样做的好处是取流和推理解耦任何一方短暂卡顿不会直接影响另一方。注意队列长度一定要设上限。我一开始用了无界队列结果推理速度跟不上取流速度内存直接爆了。后来改成maxsize10满了就丢最旧的帧问题解决。2.3 目录结构规划FastAPI 项目的目录结构我参考了官方推荐的大项目布局结合推理服务的特殊性做了调整yolov5s_service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置管理 │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py # Pydantic 模型 │ ├── core/ │ │ ├── __init__.py │ │ ├── camera.py # 摄像头取流 │ │ ├── inference.py # NPU 推理封装 │ │ └── queue.py # 帧队列管理 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes.py # 路由定义 │ │ └── dependencies.py # 依赖注入 │ └── utils/ │ ├── __init__.py │ └── image.py # 图像处理工具 ├── models/ # RKNN 模型文件 │ └── yolov5s.rknn ├── tests/ ├── requirements.txt └── run.py # 启动脚本这个结构的好处是职责清晰。core目录放核心逻辑api目录放接口定义models放数据模型。后面要加新功能比如增加一个 RTSP 摄像头只需要在core/camera.py里加一个类就行不用动其他文件。3. FastAPI 服务层核心实现细节3.1 推理引擎的封装推理引擎是整个服务的心脏。我的封装思路是把 RKNN 的初始化、推理、释放三个阶段的逻辑都包在一个类里import numpy as np from rknnlite.api import RKNNLite class YOLOv5sInference: def __init__(self, model_path, target_size(640, 640)): self.model_path model_path self.target_size target_size self.rknn None self._init_rknn() def _init_rknn(self): self.rknn RKNNLite() ret self.rknn.load_rknn(self.model_path) if ret ! 0: raise RuntimeError(f加载 RKNN 模型失败错误码: {ret}) ret self.rknn.init_runtime(core_maskRKNNLite.NPU_CORE_0) if ret ! 0: raise RuntimeError(f初始化 NPU 运行时失败错误码: {ret}) def preprocess(self, img): # 缩放、归一化、通道转换 img cv2.resize(img, self.target_size) img cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img img.astype(np.float32) / 255.0 img np.expand_dims(img, axis0) return img def infer(self, img): input_data self.preprocess(img) outputs self.rknn.inference(inputs[input_data]) return self.postprocess(outputs) def postprocess(self, outputs): # NMS、坐标还原等后处理 ... def release(self): if self.rknn: self.rknn.release()这里有几个关键点值得展开说。core_mask 的选择。RK3588 有三个 NPU 核心NPU_CORE_0、NPU_CORE_1、NPU_CORE_2还支持NPU_CORE_0_1_2自动调度。单模型推理用单核就够了多模型并行的时候可以用多核。我实测下来YOLOv5s 在单核上跑 640x640 输入大概 25-35ms三核自动调度反而因为调度开销没有明显提升。预处理的位置。预处理放在 CPU 上做还是 NPU 上做这个有讲究。RKNN 支持把部分预处理算子融合到模型里但 YOLOv5s 的预处理比较简单缩放归一化放 CPU 上做对整体延迟影响不大。如果你追求极致性能可以把 resize 和 normalize 也转进 RKNN 模型。内存管理。RKNN 的 inference 每次调用会分配输出内存频繁调用要注意释放。我在infer方法里没有显式释放因为 RKNNLite 内部有内存池管理。但如果你用的是 C 接口一定要手动管理。3.2 FastAPI 路由与依赖注入路由层我设计了三个接口from fastapi import APIRouter, UploadFile, File, Depends from app.core.inference import YOLOv5sInference from app.api.dependencies import get_inference_engine router APIRouter() router.post(/detect) async def detect_image( file: UploadFile File(...), engine: YOLOv5sInference Depends(get_inference_engine) ): contents await file.read() img cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR) results engine.infer(img) return {code: 0, data: results} router.get(/health) async def health_check(): return {status: ok} router.get(/camera/stream) async def camera_stream(): # 返回 MJPEG 流 ...依赖注入这里用了一个单例模式。推理引擎的初始化很重加载模型、初始化 NPU不能每个请求都重新创建。我用lru_cache做了一个简单的单例from functools import lru_cache lru_cache(maxsize1) def get_inference_engine(): return YOLOv5sInference(models/yolov5s.rknn)注意lru_cache在多进程模式下会有问题。如果你用uvicorn --workers 4启动每个 worker 会各自创建一个推理引擎实例NPU 资源会冲突。边缘设备上建议单 worker 运行用异步并发来处理请求。3.3 异步推理的坑FastAPI 的 async 很好用但 NPU 推理是同步阻塞的。如果你直接在 async 函数里调用engine.infer()整个事件循环会被阻塞其他请求全部排队。正确的做法是用run_in_executor把推理放到线程池里import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers2) router.post(/detect) async def detect_image(file: UploadFile File(...)): contents await file.read() img cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR) loop asyncio.get_event_loop() results await loop.run_in_executor(executor, engine.infer, img) return {code: 0, data: results}线程池大小设多少我的经验是设成 NPU 核心数。RK3588 有三个 NPU 核心但单模型推理只能用一个核心所以设 2 就够了。设太大反而会因为线程切换和资源竞争降低吞吐。3.4 配置文件管理配置我用了 Pydantic 的 BaseSettings支持从环境变量和.env文件读取from pydantic_settings import BaseSettings class Settings(BaseSettings): model_path: str models/yolov5s.rknn camera_source: str 0 # 0 表示 USB 摄像头也可以是 RTSP URL frame_width: int 640 frame_height: int 640 queue_size: int 10 conf_threshold: float 0.25 iou_threshold: float 0.45 class Config: env_file .env settings Settings()这样做的好处是部署的时候不用改代码改环境变量就行。比如你在开发机上用 USB 摄像头部署到现场用 RTSP 摄像头只需要改CAMERA_SOURCE环境变量。4. 摄像头接入方案与实操4.1 USB 摄像头 vs RTSP 摄像头摄像头接入有两种主流方案各有适用场景对比项USB 摄像头RTSP 网络摄像头延迟低50ms中等100-300ms布线需要 USB 线长度受限网线或 WiFi灵活分辨率通常 1080P 以内支持 2K/4K多路支持受 USB 带宽限制理论上无限成本低中等适用场景原型验证、近距离实际部署、远距离我一开始用 USB 摄像头做验证后来换成了 RTSP 网络摄像头。原因很简单RK3588 开发板通常放在机柜里摄像头装在设备上距离远USB 线拉不过去。4.2 OpenCV 取流的正确姿势用 OpenCV 取 RTSP 流是最简单的方式import cv2 cap cv2.VideoCapture(rtsp://admin:password192.168.1.100:554/stream1) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # 关键减少缓冲 while True: ret, frame cap.read() if not ret: break # 处理帧这里有一个极其重要的设置CAP_PROP_BUFFERSIZE。OpenCV 默认会缓冲多帧导致你读到的画面延迟好几秒。设成 1 之后每次读到的都是最新帧。但即使设了 buffersizeOpenCV 的 RTSP 取流在长时间运行后还是会出问题花屏、卡死、断流。这是 FFmpeg 底层的问题不是 OpenCV 的锅。4.3 更稳定的 RTSP 取流方案经过多次踩坑我总结了一个更稳定的方案用 FFmpeg 子进程取流通过管道传给 Python。import subprocess import numpy as np class RTSPCamera: def __init__(self, url, width640, height640): self.url url self.width width self.height height self.process None def start(self): cmd [ ffmpeg, -rtsp_transport, tcp, # 关键用 TCP 而不是 UDP -i, self.url, -f, rawvideo, -pix_fmt, bgr24, -s, f{self.width}x{self.height}, -r, 25, - ] self.process subprocess.Popen( cmd, stdoutsubprocess.PIPE, stderrsubprocess.DEVNULL ) def read(self): frame_size self.width * self.height * 3 raw self.process.stdout.read(frame_size) if len(raw) ! frame_size: return None return np.frombuffer(raw, dtypenp.uint8).reshape( (self.height, self.width, 3) )这个方案的核心优势-rtsp_transport tcp强制用 TCP 传输避免 UDP 丢包导致的花屏直接输出 rawvideo省去解码环节Python 端直接拿到 numpy 数组子进程隔离FFmpeg 崩了不影响主进程可以自动重启实测下来这个方案连续跑 72 小时没有断流。OpenCV 方案大概 2-3 小时就会出问题。4.4 摄像头参数调优摄像头的参数对推理效果影响很大。几个关键参数分辨率YOLOv5s 输入是 640x640如果摄像头输出 1080P需要缩放。建议摄像头直接输出 640x640 或者接近的分辨率减少 CPU 缩放开销。帧率25fps 足够了。太高会增加取流和推理压力太低会影响实时性。码率RTSP 摄像头的码率建议设 2-4Mbps。太高网络压力大太低画质差影响检测。编码格式H.264 兼容性最好H.265 省带宽但解码开销大。提示海康摄像头的 RTSP 地址格式通常是rtsp://用户名:密码IP:554/Streaming/Channels/101101 表示主码流102 表示子码流。子码流分辨率低、码率小适合做推理输入。5. 那个折腾最久的坑多线程下的 NPU 资源竞争5.1 问题现象服务层写好了摄像头也接上了单请求测试一切正常。但当我同时开两个请求的时候问题出现了第一个请求正常返回第二个请求卡住等第一个完成后才返回有时候直接报错RKNN model inference failed更诡异的是有时候两个请求都返回了但结果是错的——第二个请求的检测框画在了第一个请求的图片上。5.2 排查过程我一开始怀疑是 FastAPI 的异步问题加了日志发现两个请求确实并发进来了。然后怀疑是线程池的问题把max_workers改成 1问题依旧。接着怀疑是 RKNN 的问题。查了 RK3588 的文档发现关键信息RKNNLite 实例不是线程安全的。多个线程同时调用同一个实例的inference方法会导致内部状态混乱。这就解释了为什么第二个请求会卡住——RKNN 内部有锁但锁的实现有问题导致死锁或者结果错乱。5.3 解决方案解决方案有三种我逐一试过方案一加全局锁import threading inference_lock threading.Lock() def infer(self, img): with inference_lock: input_data self.preprocess(img) outputs self.rknn.inference(inputs[input_data]) return self.postprocess(outputs)这个方案能解决问题但完全串行化了吞吐量上不去。两个请求的延迟从 30ms 变成了 60ms。方案二每个线程一个 RKNN 实例import threading class InferencePool: def __init__(self, model_path, pool_size2): self.model_path model_path self.pool [] self.lock threading.Lock() for _ in range(pool_size): self.pool.append(YOLOv5sInference(model_path)) def infer(self, img): with self.lock: engine self.pool.pop() try: result engine.infer(img) finally: with self.lock: self.pool.append(engine) return result这个方案实现了真正的并行两个请求可以同时推理。但内存占用翻倍每个 RKNN 实例大概占 100-200MB。方案三用 RKNN 的多核特性RK3588 有三个 NPU 核心可以给每个实例分配不同的核心# 实例 1 rknn.init_runtime(core_maskRKNNLite.NPU_CORE_0) # 实例 2 rknn.init_runtime(core_maskRKNNLite.NPU_CORE_1)这样两个实例真正并行互不干扰。我最终用了这个方案配合方案二的实例池实现了 2 路并行推理每路延迟 30ms 左右。5.4 最终的服务层架构经过这一轮折腾最终的服务层架构变成了请求 1 → 线程池线程 1 → 推理实例 1NPU Core 0 请求 2 → 线程池线程 2 → 推理实例 2NPU Core 1 请求 3 → 排队等待线程池大小 推理实例数 NPU 核心数实际用了 2 个留一个给系统。注意RKNN 实例的创建和销毁很耗时加载模型大概 1-2 秒一定要在服务启动时创建好不要每次请求都创建。我用 FastAPI 的lifespan事件在启动时初始化实例池关闭时释放。from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动时 app.state.inference_pool InferencePool(models/yolov5s.rknn, pool_size2) yield # 关闭时 for engine in app.state.inference_pool.pool: engine.release()6. 常见问题速查与避坑指南6.1 服务层常见问题问题现象可能原因解决方法启动报错RKNN model load failed模型路径错误或模型损坏检查路径重新转换模型推理结果全为 0预处理格式不对检查输入是否归一化、通道顺序服务运行一段时间后卡死内存泄漏检查队列是否无界RKNN 是否释放并发请求结果错乱RKNN 实例线程不安全用实例池每个线程独立实例Uvicorn 日志丢失多 worker 日志冲突用单 worker或配置日志轮转请求超时推理阻塞事件循环用 run_in_executor 放到线程池6.2 摄像头常见问题问题现象可能原因解决方法RTSP 连接失败地址错误或网络不通用 VLC 先测试地址画面花屏UDP 丢包强制 TCP 传输延迟越来越大缓冲区堆积设 CAP_PROP_BUFFERSIZE1运行几小时后断流FFmpeg 崩溃用子进程方案加自动重启画面过暗/过亮摄像头曝光设置通过 ONVIF 或 Web 界面调整帧率不稳定网络带宽不足降低码率或分辨率6.3 我的独家避坑技巧技巧一用 systemd 管理服务不要用nohup或者screen跑服务用 systemd。好处是开机自启、崩溃自动重启、日志统一管理。[Unit] DescriptionYOLOv5s Inference Service Afternetwork.target [Service] Typesimple Userroot WorkingDirectory/opt/yolov5s_service ExecStart/usr/bin/python3 run.py Restartalways RestartSec5 [Install] WantedBymulti-user.target技巧二加一个看门狗线程服务跑久了总会出各种问题加一个看门狗线程定期检查import threading import time def watchdog(camera, inference_pool): while True: time.sleep(60) if not camera.is_alive(): camera.restart() if inference_pool.is_stuck(): inference_pool.reset()技巧三日志分级和轮转边缘设备的存储有限日志一定要轮转。用 Python 的RotatingFileHandlerfrom logging.handlers import RotatingFileHandler handler RotatingFileHandler( logs/service.log, maxBytes10*1024*1024, backupCount5 )技巧四温度监控RK3588 跑 NPU 推理发热不小加一个温度监控超过阈值就降频或者暂停服务def get_cpu_temp(): with open(/sys/class/thermal/thermal_zone0/temp) as f: return int(f.read()) / 1000 if get_cpu_temp() 80: logger.warning(温度过高暂停推理) time.sleep(10)6.4 性能优化建议如果你觉得当前性能不够可以尝试以下优化模型量化YOLOv5s 用 INT8 量化后推理速度可以提升 30-50%精度损失在 1-2% 以内输入分辨率640x640 是精度和速度的平衡点如果场景简单可以降到 416x416多核并行RK3588 有三个 NPU 核心多模型场景可以充分利用零拷贝摄像头数据直接传给 NPU避免 CPU 内存拷贝批处理如果延迟要求不高可以攒几帧一起推理提高 NPU 利用率7. 实际部署中的一些体会整套服务层跑通之后我在实际场景里连续运行了两周。有几个体会比较深。第一稳定性比性能重要。边缘设备部署在现场你不可能随时去重启。与其追求极致的推理速度不如把异常处理和自动恢复做好。我的服务现在遇到任何异常都会自动重启相关模块两周下来没有人工干预过。第二日志是你的救命稻草。出问题的时候没有日志你只能瞎猜。我在关键路径上都加了日志取流帧率、推理耗时、队列长度、内存占用、温度。这些数据不仅能帮你排查问题还能帮你发现性能瓶颈。第三不要过度设计。我一开始想搞一个很复杂的微服务架构后来发现完全没必要。一个 FastAPI 进程两个推理实例一个摄像头取流线程足够了。边缘设备的资源有限简单可靠才是王道。第四测试要覆盖异常场景。正常流程谁都能跑通真正考验的是异常处理摄像头断流了怎么办NPU 推理失败了怎么办内存不够了怎么办这些场景我在测试阶段都模拟过确保服务能自动恢复。最后再分享一个小技巧如果你用的是海康或宇视的摄像头可以通过 ONVIF 协议获取摄像头的 RTSP 地址不用手动拼 URL。Python 有现成的onvif-zeep库几行代码就能自动发现局域网内的摄像头并获取流地址。这个在批量部署的时候特别有用不用一个个手动配置。整套东西跑下来从模型转换到服务上线大概花了三周时间。其中服务层和摄像头接入占了一半以上。但跑通之后后面再做类似的项目就快多了大部分代码可以直接复用。如果你也在做类似的事情希望这篇能帮你少走一些弯路。