1. 从模型跑通到服务可用中间隔着多少东西把 YOLOv5s 在 RK3588 的 NPU 上跑出第一帧检测结果那个瞬间确实爽。但爽完之后你会发现距离一个能用的东西还差得远。模型推理只是整条链路里的一环前面要有摄像头把画面喂进来后面要有服务层把结果吐出去中间还得处理格式转换、线程调度、异常恢复这些脏活累活。我见过太多人卡在模型能跑但不知道怎么用的阶段包括我自己。这篇是系列实践的第七篇专门讲服务层和摄像头接入这两块。选 FastAPI 做服务层不是因为它多时髦而是它在轻量和够用之间平衡得最好——异步支持完善、自带交互式文档、依赖注入清晰对于边缘设备上的推理服务来说这些特性比性能数字更实在。摄像头这块我用的是 MIPI 接口的模组走 V4L2 采集中间踩的坑比模型部署本身还多。整篇内容适合已经在 RK3588 上跑通 YOLOv5s 推理、准备把它包装成实际服务的开发者。如果你还在折腾模型转换和 NPU 推理建议先看前几篇。下面直接进入正题从服务层的设计取舍开始讲。2. FastAPI 服务层的设计取舍与目录组织2.1 为什么是 FastAPI 而不是 Flask边缘设备上的推理服务核心诉求其实就三条能异步处理并发请求、能方便地做输入校验、部署依赖别太重。Flask 是同步框架遇到推理这种耗时操作要么开多线程要么上 Gunicorn 多 worker前者受 GIL 限制后者在 RK3588 这种内存有限的板子上开多个进程会直接把内存吃满。FastAPI 基于 ASGI原生支持 async配合 uvicorn 单进程就能处理并发内存占用可控。另一个实际考量是自动生成的 API 文档。调试阶段用/docs页面直接发请求测试比写 curl 命令快得多。Pydantic 做请求体校验也很省事客户端传错参数会直接返回 422 和具体错误字段不用自己写一堆 if-else。不过 FastAPI 也不是没代价。它的异步模型要求你理解 event loop如果推理代码是同步阻塞的NPU 推理基本都是直接放在 async 函数里会把整个事件循环卡死。这个坑后面会专门讲怎么绕。2.2 一个能落地的目录结构网上 FastAPI 教程给的目录结构大多偏 Web 项目套到推理服务上不太合适。我调整后的结构是这样的rknn_service/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口注册路由和生命周期 │ ├── config.py # 配置项模型路径、摄像头参数等 │ ├── schemas.py # Pydantic 模型定义 │ ├── routers/ │ │ ├── __init__.py │ │ ├── detect.py # 检测相关接口 │ │ └── camera.py # 摄像头控制接口 │ ├── services/ │ │ ├── __init__.py │ │ ├── inference.py # NPU 推理封装 │ │ └── capture.py # 摄像头采集封装 │ └── utils/ │ ├── __init__.py │ └── image.py # 图像格式转换工具 ├── models/ │ └── yolov5s.rknn └── requirements.txt这个结构的关键在于services层和routers层分离。路由只负责接收请求、调用服务、返回响应不掺业务逻辑。推理和采集各自封装成独立的服务类方便单独测试和替换。比如你后面想把 YOLOv5s 换成 YOLOv8只需要改inference.py路由层完全不用动。config.py用 Pydantic 的 BaseSettings 来管理配置支持从环境变量读取部署时不用改代码from pydantic_settings import BaseSettings class Settings(BaseSettings): model_path: str models/yolov5s.rknn camera_device: str /dev/video0 camera_width: int 1920 camera_height: int 1080 conf_threshold: float 0.25 nms_threshold: float 0.45 class Config: env_prefix RKNN_ settings Settings()这样启动时用RKNN_MODEL_PATH/path/to/model.rknn uvicorn app.main:app就能覆盖默认值比硬编码灵活得多。2.3 生命周期管理模型只加载一次FastAPI 的 lifespan 机制是管理模型加载的正确位置。很多人图省事在路由函数里加载模型结果每个请求都重新加载一遍NPU 初始化就要好几秒服务直接不可用。from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型 app.state.engine InferenceEngine(settings.model_path) app.state.capture CameraCapture(settings.camera_device) yield # 关闭时释放资源 app.state.capture.release() app.state.engine.release() app FastAPI(lifespanlifespan)模型实例挂在app.state上路由里通过request.app.state.engine访问。这样整个服务生命周期内模型只初始化一次NPU 上下文也只创建一次。注意RKNN 的推理上下文不是线程安全的。如果你用多线程处理请求每个线程需要独立的推理上下文或者用锁串行化。我实测下来单线程串行推理在 RK3588 上处理 640x640 输入大概 30-40ms 一帧对于大多数场景够用了没必要为了并发引入复杂度。3. 摄像头接入V4L2 采集的完整链路3.1 MIPI 摄像头在 RK3588 上的识别流程RK3588 支持 MIPI CSI 接口但摄像头模组能不能被识别取决于设备树里有没有对应的配置。我用的模组需要先在设备树里使能对应的 I2C 和 MIPI 通道编译后重新烧录。这一步如果没做对/dev/video*设备节点根本不会出现。识别流程可以按这个顺序排查ls /dev/video*看设备节点是否存在v4l2-ctl --list-devices看设备名称和对应的 video 节点v4l2-ctl -d /dev/video0 --list-formats-ext看支持的像素格式和分辨率media-ctl -p看 media pipeline 的拓扑结构第三步特别重要。不同模组支持的格式不一样有的只出 YUYV有的支持 NV12有的能直接出 MJPEG。YOLOv5s 需要 RGB 输入所以采集到的数据要经过格式转换。如果模组支持 NV12转换到 RGB 比 YUYV 转 RGB 快不少因为 NV12 是半平面格式内存布局更紧凑。我踩的第一个坑就在这里模组默认输出 YUYV1920x1080 下一帧就是 192010802 4MB30fps 就是 120MB/s 的带宽。USB 2.0 的摄像头根本扛不住MIPI 接口虽然带宽够但格式转换的 CPU 开销也不小。后来改成 1280x720 分辨率带宽降到 53MB/sCPU 占用明显下降。3.2 OpenCV 采集 vs 直接 V4L2 采集用 OpenCV 的cv2.VideoCapture采集最省事但它在 RK3588 上有两个问题一是默认走 GStreamer 后端配置不对会报各种 pipeline 错误二是它内部会做格式转换你拿到的已经是 BGR 的 numpy 数组但转换过程不透明出了问题不好排查。直接走 V4L2 的 ioctl 接口采集代码量大一些但每一步都可控。核心流程是import fcntl import mmap import struct from v4l2 import * def open_camera(device, width, height): fd open(device, rb, buffering0) # 查询能力 cap v4l2_capability() fcntl.ioctl(fd, VIDIOC_QUERYCAP, cap) # 设置格式 fmt v4l2_format() fmt.type V4L2_BUF_TYPE_VIDEO_CAPTURE fmt.fmt.pix.width width fmt.fmt.pix.height height fmt.fmt.pix.pixelformat V4L2_PIX_FMT_NV12 fmt.fmt.pix.field V4L2_FIELD_NONE fcntl.ioctl(fd, VIDIOC_S_FMT, fmt) # 申请缓冲区 req v4l2_requestbuffers() req.count 4 req.type V4L2_BUF_TYPE_VIDEO_CAPTURE req.memory V4L2_MEMORY_MMAP fcntl.ioctl(fd, VIDIOC_REQBUFS, req) # mmap 映射 buffers [] for i in range(req.count): buf v4l2_buffer() buf.type V4L2_BUF_TYPE_VIDEO_CAPTURE buf.memory V4L2_MEMORY_MMAP buf.index i fcntl.ioctl(fd, VIDIOC_QUERYBUF, buf) mm mmap.mmap(fd, buf.length, mmap.MAP_SHARED, mmap.PROT_READ | mmap.PROT_WRITE, offsetbuf.m.offset) buffers.append((mm, buf.length)) return fd, buffers用 4 个缓冲区做双缓冲队列采集线程和推理线程各拿各的避免互相等待。NV12 转 RGB 用 OpenCV 的cv2.cvtColor配合COLOR_YUV2BGR_NV12就行底层有 NEON 优化1280x720 转换大概 3-5ms。提示V4L2 的缓冲区数量不是越多越好。4 个是经验值太少容易丢帧太多会增加内存占用和延迟。RK3588 上 1280x720 NV12 一帧 1.3MB4 个缓冲区才 5MB 多可以接受。3.3 采集线程与推理线程的协作摄像头采集和 NPU 推理是两个独立的节奏。采集是固定帧率比如 30fps推理是能跑多快跑多快比如 25fps。如果串行执行采集一帧、推理一帧实际帧率会被推理速度拖累。我的做法是采集线程持续往队列里放帧推理线程从队列取帧。队列长度设为 2满了就丢最旧的帧。这样保证推理用的永远是最新的画面不会因为积压导致延迟越来越大。import queue import threading frame_queue queue.Queue(maxsize2) def capture_loop(cap, queue): while running: frame cap.read_frame() if frame is None: continue if queue.full(): try: queue.get_nowait() # 丢弃旧帧 except queue.Empty: pass queue.put(frame) def inference_loop(engine, queue): while running: frame queue.get() results engine.infer(frame) # 处理结果...这个模式在 RK3588 上实测很稳CPU 占用比串行方式低因为采集和推理可以并行NPU 工作时 CPU 在准备下一帧的数据。4. 那个折腾最久的坑uvicorn 日志丢失与异步阻塞4.1 现象日志时有时无推理偶尔卡死服务跑起来之后我发现两个诡异的现象。一是 uvicorn 的访问日志有时候打印有时候不打印二是推理接口偶尔会卡住好几秒才返回。单独测试推理代码没问题单独测试 FastAPI 也没问题合在一起就出问题。排查过程走了不少弯路。先怀疑是 NPU 驱动的问题换了几个版本的 rknn-toolkit2 和 runtime没改善。又怀疑是内存泄漏用free -h监控了半天内存稳定。最后用py-spy dump抓了一下运行中的线程栈才发现问题所在。4.2 根因同步推理阻塞了事件循环FastAPI 的 async 路由函数运行在事件循环里。如果路由函数里直接调用同步的 NPU 推理整个事件循环会被阻塞。uvicorn 的日志输出也是通过事件循环调度的事件循环被阻塞日志自然就丢了。推理卡顿是因为事件循环被阻塞期间其他请求全部排队等待。# 错误写法同步推理直接放在 async 函数里 app.post(/detect) async def detect(image: UploadFile): img await image.read() result engine.infer(img) # 阻塞事件循环 return result正确做法有两种。一是把路由函数改成普通 defFastAPI 会自动把它放到线程池里执行不阻塞事件循环# 正确写法一用 def 让 FastAPI 自动走线程池 app.post(/detect) def detect(image: UploadFile): img image.file.read() result engine.infer(img) return result二是保持 async但用run_in_executor把推理丢到线程池# 正确写法二显式丢到线程池 import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers1) app.post(/detect) async def detect(image: UploadFile): img await image.read() loop asyncio.get_event_loop() result await loop.run_in_executor(executor, engine.infer, img) return result我选了第二种因为线程池大小可控设为 1保证推理串行而且 async 函数里还能做其他异步操作比如写日志、更新状态。注意线程池大小不要超过 1。RKNN 推理上下文不是线程安全的多个线程同时推理会出各种奇怪的问题包括结果错乱和进程崩溃。串行推理虽然吞吐量低但稳定压倒一切。4.3 uvicorn 日志配置的正确姿势解决了阻塞问题日志还是不全。uvicorn 默认的日志配置在--reload模式下会有些奇怪的行为而且它不接管 Python 标准库的 logging。我的做法是在main.py里显式配置 loggingimport logging import sys logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[ logging.StreamHandler(sys.stdout), logging.FileHandler(service.log) ] ) # 让 uvicorn 的 logger 也走这个配置 for name in [uvicorn, uvicorn.access, uvicorn.error]: logger logging.getLogger(name) logger.handlers logging.getLogger().handlers这样日志同时输出到控制台和文件格式统一排查问题时直接看service.log就行。文件日志记得配 rotate不然跑几天就把磁盘写满了。5. 服务层与摄像头的联调实战5.1 实时检测接口的设计服务层对外提供的核心接口有两个一个是上传图片做单次检测一个是启动/停止实时摄像头检测。单次检测接口比较简单接收 multipart 上传的图片返回检测框坐标和类别。实时检测接口复杂一些需要管理摄像头状态和推流。实时检测我用的是 MJPEG 流的方式客户端用浏览器直接打开就能看。实现上是一个 StreamingResponse不断从推理结果里取最新帧画上检测框编码成 JPEG 推出去from fastapi.responses import StreamingResponse import cv2 def generate_frames(): while True: frame, results app.state.latest_result if frame is None: continue for box in results: x1, y1, x2, y2, conf, cls box cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText(frame, f{cls} {conf:.2f}, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) _, jpeg cv2.imencode(.jpg, frame, [cv2.IMWRITE_JPEG_QUALITY, 80]) yield (b--frame\r\n bContent-Type: image/jpeg\r\n\r\n jpeg.tobytes() b\r\n) app.get(/stream) def stream(): return StreamingResponse(generate_frames(), media_typemultipart/x-mixed-replace; boundaryframe)这里有个细节推理线程把最新结果写到app.state.latest_result推流线程只读不写。用简单的变量赋值就行Python 的 GIL 保证单个赋值操作的原子性。不要用锁锁会拖慢推流帧率。5.2 摄像头参数动态调整实际使用中经常需要调整摄像头的曝光、增益、分辨率。V4L2 提供了VIDIOC_S_CTRL接口来设置这些参数。我封装了一个接口接收参数名和值通过 ioctl 下发from v4l2 import v4l2_control, VIDIOC_S_CTRL def set_camera_ctrl(fd, ctrl_id, value): ctrl v4l2_control() ctrl.id ctrl_id ctrl.value value fcntl.ioctl(fd, VIDIOC_S_CTRL, ctrl)曝光对应V4L2_CID_EXPOSURE_ABSOLUTE增益对应V4L2_CID_GAIN。注意有些模组的自动曝光和手动曝光是互斥的设置手动曝光前要先关掉自动曝光V4L2_CID_EXPOSURE_AUTO设为V4L2_EXPOSURE_MANUAL。提示调整摄像头参数后建议丢弃接下来 3-5 帧再开始推理。参数切换需要几帧时间稳定直接拿切换瞬间的帧做推理结果会偏暗或偏亮。5.3 异常恢复摄像头掉线怎么办MIPI 摄像头在长时间运行后偶尔会掉线表现为read_frame返回 None 或者 ioctl 报错。如果不处理服务会一直空转。我的做法是在采集线程里加异常捕获检测到连续多次读取失败就重新初始化摄像头def capture_loop(cap, queue): fail_count 0 while running: try: frame cap.read_frame() if frame is None: fail_count 1 if fail_count 30: # 连续30次失败约1秒 cap.reinit() fail_count 0 continue fail_count 0 # 放入队列... except Exception as e: logging.error(fCapture error: {e}) cap.reinit() time.sleep(1)reinit就是关闭设备重新打开重新申请缓冲区。这个过程大概 200-300ms期间推流会卡一下但总比服务彻底挂掉强。6. 性能实测与几个值得记录的细节6.1 端到端延迟的构成在 RK3588 上跑 1280x720 输入、YOLOv5s 模型端到端延迟大概 60-80ms。拆解下来环节耗时摄像头采集5-10msNV12 转 RGB3-5ms预处理resize 归一化5-8msNPU 推理30-40ms后处理NMS 解码5-10msJPEG 编码推流用5-8msNPU 推理是大头但预处理和后处理加起来也有 15-25ms优化空间不小。预处理可以用 RGA 硬件加速后处理的 NMS 可以用 C 重写但那是另一个话题了。6.2 内存与温度监控长时间运行要关注两个指标内存和温度。RK3588 跑满 NPU 时温度能到 70 度以上不加散热会降频。我用psutil和thermal_zone做了个简单的监控接口import psutil app.get(/health) def health(): mem psutil.virtual_memory() with open(/sys/class/thermal/thermal_zone0/temp) as f: temp int(f.read()) / 1000 return { memory_percent: mem.percent, memory_available_mb: mem.available // 1024 // 1024, cpu_temp: temp, uptime_seconds: time.time() - start_time }内存占用稳定在 400-500MB其中模型和 NPU 上下文占 200MB 左右Python 运行时和缓冲区占剩下的。温度超过 80 度就要考虑加风扇了。6.3 几个容易忽略的配置项第一个是 uvicorn 的--workers参数。在 RK3588 上不要设大于 1多 worker 会各自加载一份模型内存直接翻倍而且 NPU 上下文冲突。单 worker 配合线程池就够了。第二个是--limit-concurrency。默认不限制并发连接数客户端如果开很多连接内存会被连接缓冲区吃光。设成 10-20 比较稳妥。第三个是 OpenCV 的线程数。cv2.setNumThreads(2)限制一下不然 OpenCV 会开一堆线程跟推理线程抢 CPU。RK3588 是 8 核4 大核 4 小核留 2 个核给 OpenCV 做图像处理其他核给系统和其他任务。7. 写在最后的一点个人体会这套服务层加摄像头的方案我在两个项目里实际跑过连续运行一周没出过大问题。最深的体会是边缘设备上的服务稳定性比性能重要得多。与其追求把推理优化到 20ms不如先把异常恢复、日志、监控这些基础设施做扎实。摄像头掉线能自动恢复、推理卡住能超时返回、内存泄漏能及时发现这些才是让服务真正可用的关键。另外FastAPI 的异步模型确实需要花点时间理解但一旦搞明白事件循环和线程池的关系写出来的代码会比同步框架清晰很多。那个日志丢失的坑本质上就是没理解async 函数里不能有阻塞调用这条规则。希望这篇记录能帮你少走点弯路。