企业官网建设流程全解析

1. 为什么服务层是整个部署链路里最容易被低估的一环

模型转换完了，NPU 推理也跑通了，单张图片测下来延迟稳定在 20ms 出头，这时候很多人会觉得"活儿干完了"。我一开始也是这么想的，直到真正要把这套东西接到摄像头上、做成一个能持续跑的服务，才发现前面六篇攒下来的那点顺畅感，在服务层这里几乎全部要重新交学费。

这一篇要聊的是 RK3588 上 YOLOv5s 全链路部署的最后一公里：用 FastAPI 把 NPU 推理封装成一个常驻服务，接上摄像头做实时取流，然后处理那个让我折腾了整整两个晚上才定位到的坑。关键词里出现的 RK3588、YOLOv5s、FastAPI、NPU、摄像头，在这一篇里会全部串起来，不再是各自独立的模块。

先说清楚这套服务层到底解决什么问题。裸的推理脚本是这样的：读一张图、跑一次模型、打印结果、退出。但真实场景里，摄像头是 7x24 在出流的，你不可能每来一帧就重启一次 Python 进程——RK3588 上加载一次 RKNN 模型、初始化 NPU 上下文，冷启动开销在 1.5 到 3 秒之间，每帧都重启的话帧率直接掉到 0.3fps，完全没法用。所以服务层要干的事，本质上是三件：把模型常驻在内存里、把摄像头的视频流持续喂进来、对外暴露一个稳定的接口让别的程序来拿结果。

FastAPI 在这里的角色，就是那个"对外接口"。选它而不是 Flask，理由很实际：FastAPI 原生支持异步，配合uvicorn跑起来之后，处理并发请求时不会像 Flask 那样一个请求堵住整个线程；而且它自带的/docs交互文档，在调试阶段能省掉大量写测试脚本的时间。至于摄像头，我这次用的是 RTSP 网络摄像头，走OpenCV的VideoCapture拉流，这也是目前最通用、最不挑硬件的方案。

适合谁来参考这篇内容？如果你已经跟着前面几篇把 RKNN 模型跑通了，现在卡在"怎么把它变成一个真正能用的服务"，那这篇就是给你写的。如果你还没跑通模型推理，建议先回去把 NPU 那部分搞定，否则服务层的问题会和推理层的问题混在一起，排查起来非常痛苦。

2. 服务层的整体架构设计与选型考量

2.1 三种架构方案的取舍

在动手写代码之前，我先在纸上画了三种可能的架构，最后选了第三种。把这三条路摆出来，是因为我踩过前两条的坑，希望你能直接跳过。

方案一：单进程同步阻塞式。一个 Python 进程，主线程里一个while True循环，读一帧、推理一帧、把结果存到一个全局变量里，FastAPI 的接口去读这个全局变量。这是最直觉的写法，我第一版就是这么干的。问题在于：VideoCapture.read()是阻塞的，NPU 推理也是阻塞的，整个循环的节奏被最慢的那一环拖住。实测下来，1080p 的 RTSP 流读取大概 30ms，推理 25ms，加起来一帧 55ms，理论 18fps。但只要有一个 HTTP 请求进来要读结果，GIL 就会让主循环卡顿一下，帧率立刻掉到 12fps 左右，而且请求响应时间不稳定。

方案二：多进程 + 共享内存。读流一个进程，推理一个进程，FastAPI 一个进程，用multiprocessing.shared_memory传帧。这个方案性能最好，但复杂度陡增——共享内存的生命周期管理、进程崩溃后的清理、帧的同步锁，每一样都是坑。我在 RK3588 上试了一版，跑了两小时之后出现过一次共享内存段没释放，导致后续启动直接报错，排查了半天。对于个人项目来说，这个复杂度不值得。

方案三：单进程 + 多线程 + 异步接口。这是最终选的方案。核心思路是：读流和推理放在一个独立的后台线程里跑，这个线程只管闷头处理，把最新的推理结果写进一个线程安全的容器；FastAPI 跑在主线程的异步事件循环里，接口被调用时只是去读那个容器里的最新值，几乎不耗时。这样读流、推理、接口三者互不阻塞，接口响应时间稳定在 5ms 以内。

提示：这里的"最新值"语义很重要。实时视频场景下，你永远只关心最新一帧的结果，旧帧的结果没有意义。所以容器不需要是队列，只需要是一个能被原子替换的引用，这样天然避免了队列堆积导致的延迟累积。

2.2 为什么用线程而不是 asyncio 协程做推理

有人会问，既然 FastAPI 是异步的，为什么不干脆把推理也写成协程？答案是：NPU 推理是 CPU/硬件阻塞操作，不是 IO 等待。协程的优势在于 IO 等待时能切换出去干别的，但 RKNN 的inference()调用会实打实占住当前线程直到算完，你把它塞进协程里，事件循环照样被堵住，反而失去了异步的意义。

正确的做法是用asyncio.to_thread()或者干脆起一个threading.Thread，把阻塞操作扔到独立线程。我选的是后者，因为读流本身就是一个无限循环，用线程更自然。这里有个细节：Python 的 GIL 会让多线程在纯计算场景下退化成串行，但 RKNN 推理的耗时主要花在 NPU 硬件上，Python 层只是发起调用和等待，GIL 在等待期间是释放的，所以多线程在这里是有效的。

2.3 目录结构怎么组织才不乱

FastAPI 项目最容易写成一坨，尤其是这种"推理 + 接口 + 摄像头"混在一起的项目。我用的结构是这样的，你可以直接抄：

rk3588-yolo-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口，路由注册 │ ├── config.py # 配置：模型路径、RTSP 地址、阈值 │ ├── detector.py # RKNN 推理封装 │ ├── camera.py # 摄像头读流线程 │ ├── state.py # 全局状态容器（线程安全） │ └── routers/ │ ├── infer.py # 单图推理接口 │ └── stream.py # 实时结果接口 ├── models/ │ └── yolov5s.rknn ├── requirements.txt └── run.sh

把detector、camera、state拆成独立模块，好处是每个模块可以单独测试。比如摄像头读流有问题，我可以只跑camera.py看它能不能正常出帧，不用把整个服务拉起来。这个习惯在排查那个"折腾最久的坑"时救了我——后面会讲到。

3. 核心模块的实操实现与关键细节

3.1 RKNN 推理封装：别在每次调用时重新加载模型

detector.py的核心是把 RKNN 的初始化和推理分开。初始化只做一次，推理可以被反复调用。这是性能的关键，也是新手最容易犯的错——把rknn.load_rknn()写在推理函数里，每次调用都重新加载，单次开销直接多出 1 秒以上。

# detector.py from rknnlite.api import RKNNLite import numpy as np import cv2 class YOLOv5Detector: def __init__(self, model_path, conf_thres=0.25, iou_thres=0.45): self.conf_thres = conf_thres self.iou_thres = iou_thres self.rknn = RKNNLite() ret = self.rknn.load_rknn(model_path) if ret != 0: raise RuntimeError(f"load_rknn failed: {ret}") # 关键：指定 NPU 核心。RK3588 有三个 NPU 核心 ret = self.rknn.init_runtime(core_mask=RKNNLite.NPU_CORE_0) if ret != 0: raise RuntimeError(f"init_runtime failed: {ret}") def preprocess(self, img, input_size=(640, 640)): # letterbox 保持长宽比，避免形变影响精度 h, w = img.shape[:2] scale = min(input_size[0] / w, input_size[1] / h) nw, nh = int(w * scale), int(h * scale) resized = cv2.resize(img, (nw, nh)) canvas = np.full((input_size[1], input_size[0], 3), 114, dtype=np.uint8) dw, dh = (input_size[0] - nw) // 2, (input_size[1] - nh) // 2 canvas[dh:dh+nh, dw:dw+nw] = resized return canvas, scale, dw, dh def infer(self, img): blob, scale, dw, dh = self.preprocess(img) outputs = self.rknn.inference(inputs=[blob]) return self.postprocess(outputs, scale, dw, dh, img.shape[:2])

这里有几个参数值得展开说。core_mask指定用哪个 NPU 核心，RK3588 有 0/1/2 三个核心。单模型跑的时候用NPU_CORE_0就够了；如果你要同时跑多个模型（比如检测 + 分类），可以分别绑到不同核心上，避免互相抢资源。我实测过，单模型绑单核和绑三核（NPU_CORE_0_1_2）的推理耗时差别不大，因为 YOLOv5s 这个体量根本喂不饱三个核心，绑多核反而增加了调度开销。

preprocess里的 letterbox 是必须的。YOLOv5 训练时用的就是 letterbox，如果你直接resize到 640x640，长宽比变了，检测框会偏移，尤其是细长物体（比如人）的框会明显不准。填充值 114 是 YOLO 系列的惯例，这个数字来自训练时的数据增强配置，用别的值精度会掉一点点。

3.2 后处理：从 NPU 输出到可用的检测框

RKNN 的输出是三个尺度的特征图，需要做解码和 NMS。这部分逻辑比较长，我把它单独放在postprocess里。核心步骤是：把三个尺度的输出 reshape 成[num_anchors, 85]（85 = 4 个框坐标 + 1 个置信度 + 80 个类别），然后按置信度过滤，再做 NMS。

def postprocess(self, outputs, scale, dw, dh, orig_shape): # outputs 是三个尺度的列表，每个 shape 类似 [1, 255, 80, 80] predictions = [] strides = [8, 16, 32] anchors = [[10,13, 16,30, 33,23], [30,61, 62,45, 59,119], [116,90, 156,198, 373,326]] for i, out in enumerate(outputs): out = out[0].transpose(1, 2, 0) # [80, 80, 255] h, w = out.shape[:2] out = out.reshape(h, w, 3, 85) # 解码逻辑略，核心是 sigmoid + anchor 还原 ... # NMS keep = self.nms(predictions, self.iou_thres) # 坐标还原到原图 ... return final_boxes

注意：RKNN 输出的顺序和维度在不同版本的工具链里可能不一样，一定要先用一张已知结果的图打印输出 shape 确认。我第一次跑的时候没确认，直接按 PyTorch 的顺序解码，结果框全错位，排查了一个多小时才发现是输出维度顺序的问题。

3.3 摄像头读流线程：缓冲区和超时是重点

camera.py负责从 RTSP 拉流。这里有两个参数必须调，否则会出大问题。

# camera.py import cv2 import threading import time class CameraReader(threading.Thread): def __init__(self, rtsp_url, state): super().__init__(daemon=True) self.rtsp_url = rtsp_url self.state = state self.running = True self.cap = None def run(self): # 关键参数一：设置缓冲区为 1，避免读到旧帧 self.cap = cv2.VideoCapture(self.rtsp_url, cv2.CAP_FFMPEG) self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) while self.running: ret, frame = self.cap.read() if not ret: # 关键参数二：断流后重连，别死循环空转 time.sleep(1) self.cap.release() self.cap = cv2.VideoCapture(self.rtsp_url, cv2.CAP_FFMPEG) self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) continue self.state.update_frame(frame)

CAP_PROP_BUFFERSIZE设成 1 是重中之重。OpenCV 默认的缓冲区可能有好几帧，意味着你读到的永远是几帧之前的画面，实时性直接崩掉。设成 1 之后，每次read()拿到的都是最新帧。这个参数不是所有后端都支持，但 FFMPEG 后端是支持的，所以显式指定cv2.CAP_FFMPEG很重要。

断流重连的逻辑也不能省。RTSP 流在网络抖动时会断，如果不处理，read()会一直返回False，线程空转，CPU 跑满但什么也不干。加个sleep(1)再重连，既避免了空转，又给了网络恢复的时间。

3.4 线程安全的状态容器

state.py是读流线程和 FastAPI 之间的桥梁。它要保证：读流线程写的时候，接口读到的要么是完整的旧值，要么是完整的新值，不能读到写了一半的数据。

# state.py import threading class SharedState: def __init__(self): self._lock = threading.Lock() self._frame = None self._result = None self._timestamp = 0 def update_frame(self, frame): with self._lock: self._frame = frame def get_frame(self): with self._lock: return self._frame def update_result(self, result): with self._lock: self._result = result self._timestamp = time.time() def get_result(self): with self._lock: return self._result, self._timestamp

用锁而不是队列，是因为前面说的"只关心最新值"语义。锁的粒度很小，只包住赋值操作，不会成为性能瓶颈。

4. FastAPI 接口层：从单图推理到实时结果推送

4.1 两个核心接口的设计

服务对外暴露两个接口就够了：一个用于单图推理（调试和离线用），一个用于获取实时结果。

# routers/infer.py from fastapi import APIRouter, UploadFile, File import numpy as np import cv2 router = APIRouter() @router.post("/infer") async def infer(file: UploadFile = File(...)): contents = await file.read() img = cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR) result = detector.infer(img) return {"boxes": result, "count": len(result)}

单图接口用UploadFile接收，注意await file.read()是异步的，不会阻塞事件循环。解码用cv2.imdecode而不是cv2.imread，因为后者需要文件路径，而上传的文件在内存里。

实时结果接口更简单，直接读状态容器：

# routers/stream.py @router.get("/latest") async def latest(): result, ts = state.get_result() if result is None: return {"status": "warming_up"} return { "status": "ok", "timestamp": ts, "age_ms": (time.time() - ts) * 1000, "boxes": result }

age_ms这个字段很有用，它告诉你当前结果有多"新鲜"。如果这个值持续大于 500ms，说明推理线程卡住了，需要排查。

4.2 启动时把推理线程拉起来

FastAPI 的lifespan是管理后台线程生命周期的正确位置。别用@app.on_event("startup")，那个在新版本里已经废弃了。

# main.py from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): # 启动：初始化 detector、state、camera、推理线程 global detector, state state = SharedState() detector = YOLOv5Detector("models/yolov5s.rknn") camera = CameraReader(RTSP_URL, state) camera.start() infer_thread = InferenceWorker(detector, state) infer_thread.start() yield # 关闭：停止线程、释放资源 camera.running = False infer_thread.running = False app = FastAPI(lifespan=lifespan)

推理线程InferenceWorker的逻辑是：从 state 拿最新帧，推理，把结果写回 state，循环。这里要注意加一个小的sleep，否则在没有新帧的时候会疯狂空转。

4.3 用 uvicorn 跑起来

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 1

--workers必须是 1。因为我们的模型和摄像头状态是进程内的全局变量，多 worker 会导致每个 worker 各自加载一份模型、各自拉一路流，NPU 和网络资源直接翻倍，而且接口返回的结果可能来自不同的 worker，状态不一致。这是 FastAPI 部署里一个非常常见的误区。

提示：如果你确实需要多 worker 来扛并发，正确的做法是把推理服务拆成独立的进程，通过本地 socket 或消息队列通信，而不是靠 uvicorn 的 workers。但对个人项目来说，单 worker 完全够用。

5. 那个折腾最久的坑：uvicorn 日志丢失与线程静默死亡

5.1 现象描述：服务"活着"，但结果不再更新

服务跑起来之后，前 20 分钟一切正常，/latest接口返回的结果一直在更新。然后某一次请求，age_ms突然变成了一个很大的值，而且之后一直不降。但服务进程还在，接口还能响应，只是结果永远停在某一帧。

第一反应是摄像头断了。我去看摄像头，画面正常。第二反应是推理线程崩了。但如果是线程崩了，Python 应该会打印 traceback 才对——然而日志里什么都没有。

5.2 排查过程：从日志入手

第一步是确认日志到底有没有输出。我在推理线程里加了一行print，发现服务启动后能看到，但过一段时间就再也看不到了。这说明推理线程确实停了，但没有任何异常信息。

这里就遇到了第一个坑：uvicorn 默认会接管 stdout/stderr，后台线程里的 print 和 traceback 可能被吞掉。尤其是当 uvicorn 以某种方式重定向了输出流之后，子线程的输出不一定能到终端。

解决办法是显式配置 logging，把日志写到文件：

import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(threadName)s] %(levelname)s: %(message)s', handlers=[ logging.FileHandler('/var/log/yolo-service.log'), logging.StreamHandler() ] )

加上文件日志之后，重新跑，终于抓到了真相：推理线程抛了一个RuntimeError: RKNN inference failed，然后线程就退出了。因为线程是 daemon 的，它死了主进程不会感知，服务继续跑，只是结果不再更新。

5.3 根因分析：NPU 上下文在长时间运行后失效

为什么 RKNN 推理会突然失败？我查了 RKNN 的文档和社区讨论，结合自己的复现，定位到两个可能的原因：

原因一：内存泄漏导致的资源耗尽。RKNN 的inference()每次调用都会在 NPU 侧分配临时缓冲区，如果工具链版本有 bug，这些缓冲区可能不会被完全释放。跑几个小时之后，NPU 侧内存耗尽，推理就失败了。我用cat /proc/meminfo观察过，长时间运行后可用内存确实在缓慢下降。

原因二：输入数据的边界情况。摄像头偶尔会出一帧尺寸异常的图像（比如网络抖动导致的半帧），如果预处理没处理好，传给 RKNN 的 blob 形状不对，推理直接报错。

5.4 解决方案：三重防护

针对这两个原因，我做了三层防护，实测跑 48 小时没有再出现静默死亡。

第一层：线程异常捕获 + 自动重启。把推理循环包在 try/except 里，捕获到异常就记录日志，然后重建 detector 和线程。

class InferenceWorker(threading.Thread): def run(self): while self.running: try: frame = self.state.get_frame() if frame is None: time.sleep(0.01) continue result = self.detector.infer(frame) self.state.update_result(result) except Exception as e: logging.error(f"Inference failed: {e}", exc_info=True) # 重建 detector self.detector = YOLOv5Detector(MODEL_PATH) time.sleep(1)

第二层：输入校验。在预处理之前检查帧的形状，异常帧直接跳过。

if frame is None or frame.shape[0] < 32 or frame.shape[1] < 32: continue

第三层：定期重建 NPU 上下文。这是最粗暴但最有效的办法。每处理 N 帧（我设的是 50000 帧，大概 40 分钟），主动释放并重建 RKNN 上下文，把可能泄漏的资源清掉。

self.frame_count += 1 if self.frame_count % 50000 == 0: logging.info("Rebuilding RKNN context to prevent leak") self.detector.rknn.release() self.detector = YOLOv5Detector(MODEL_PATH)

注意：release()之后一定要重新load_rknn和init_runtime，不能只 release 不重建，否则后续推理会直接段错误。我第一次改的时候漏了重建，服务直接 core dump，又排查了半天。

5.5 这个坑给我的教训

最大的教训是：daemon 线程的死亡是静默的，必须主动监控。后来我在/latest接口里加了一个健康检查字段，如果age_ms超过 2 秒，就返回status: "stalled"，配合一个简单的定时脚本去轮询，一旦发现 stalled 就告警。这样即使线程又出问题，我也能第一时间知道，而不是等用户反馈。

第二个教训是：日志一定要落盘。uvicorn 的默认输出行为在不同版本、不同启动方式下不一致，把关键日志写到文件是最稳妥的。这个习惯在排查所有后台服务问题时都适用。

6. 常见问题速查与实操避坑清单

6.1 问题速查表

6.2 几条用血换来的经验

关于 NPU 核心绑定：如果你只跑一个模型，别绑多核。我实测过NPU_CORE_0和NPU_CORE_0_1_2，单帧推理耗时分别是 24ms 和 27ms，多核反而更慢，因为调度开销大于并行收益。只有当你同时跑多个模型时，才需要把不同模型绑到不同核心。

关于 RTSP 地址：海康、宇视这些品牌的摄像头，RTSP 地址格式不一样，主码流和子码流也不一样。做实时检测建议用子码流，分辨率低、码率小，读流快，检测精度对 1080p 和 720p 的差别其实不大。主码流留给需要录像的场景。

关于 FastAPI 的异步陷阱：async def里如果调用了阻塞函数（比如cv2.imread、time.sleep），整个事件循环会被堵住。要么用await asyncio.to_thread()包一层，要么干脆用普通def定义路由，FastAPI 会自动把它扔到线程池里跑。我单图推理接口用的是async def但内部调用了阻塞的detector.infer，严格来说是有问题的，只是因为调用频率低没暴露出来。如果你要高频调用，记得改成def。

关于模型路径：RKNN 模型加载时用的是相对路径的话，uvicorn 的工作目录不同会导致找不到文件。统一用绝对路径，或者在config.py里用os.path.dirname(__file__)拼出来。

关于温度：RK3588 长时间跑 NPU 推理会发热，如果散热不好会降频，推理耗时从 24ms 涨到 40ms 以上。加个散热片，或者用cat /sys/class/thermal/thermal_zone*/temp监控温度，超过 80 度就要注意了。

6.3 性能实测数据

在 RK3588 开发板（4GB 内存版本）上，用 YOLOv5s 640x640 输入，实测数据如下：

这个帧率对于大多数实时检测场景够用了。如果你需要更高帧率，可以考虑把输入降到 416x416，推理耗时会降到 12ms 左右，帧率能上到 20fps 以上，但小目标检测精度会下降。

7. 后续可以继续深挖的几个方向

服务层跑通之后，这套东西其实还有不少可以继续做的。比如把检测结果推给前端做可视化，用 WebSocket 替代轮询，延迟能再降一截；比如加一个简单的跟踪算法（像 ByteTrack），把逐帧检测变成轨迹输出，对视频分析场景更有价值；再比如把多个模型串起来，检测 + 分类 + 属性识别，这时候 NPU 多核绑定的价值就体现出来了。

我个人在实际操作中的体会是，服务层这块最忌讳的就是"能跑就行"。因为它是整个系统里运行时间最长、最接近真实使用场景的一环，任何一个小问题在 7x24 的运行下都会被放大。多花点时间在异常处理、日志、健康检查上，后面能省下大量的排查时间。那个折腾了我两个晚上的坑，本质上就是"没做异常捕获 + 没落盘日志"这两件小事叠加的结果，但代价是整整两天的调试。