本项目已开源如果有用帮助的话请麻烦点点star谢谢。本人最近也在找实习/工作有需要的老板可以考虑一下。开源链接引言在嵌入式部署中单纯的模型推理Inference往往不是瓶颈真正的性能杀手通常隐藏在图像前后处理和硬件利用率中。本文将深入解析 RT-DETR 在 RK3588 上的高效实现逻辑并分享一套工业级的多线程异步推理框架。RT-DETR的前后处理前处理对于 RT-DETR 模型前处理Pre-process的目标是将图像转换为模型定义的张量格式如640 × 640 640 \times 640640×640, RGB, Float32。 避坑指南在做数据类型转换时img img.astype(np.float32) / 255.0 这一步必须严格对齐你量化模型时的设定。如果量化时开启了预处理下沉Mean/Std 在模型内这里则不需要除以 255。当前现状目前我们主要使用 OpenCV 在 CPU 上处理。但在高分辨率如 4K场景下CPU 缩放将成为瓶颈。进阶方向后续建议引入 Rockchip 官方的 RGA 硬件加速引擎实现零 CPU 负载的图像缩放与格式转换。使用OpenCV做预处理很简单代码如下orig_img cv2.imread(args.img_path) # 预处理Resize (使用 LetterBox 更严谨这里为了快速验证使用简单 Resize) img cv2.resize(orig_img, IMG_SIZE) img cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img img.astype(np.float32) / 255.0 img np.expand_dims(img, axis0) # [1, 640, 640, 3]需要注意的是输入图像的数据类型img img.astype(np.float32) / 255.0需要参考模型量化/转换时的输入要求请一定注意。推理outputs rknn.inference(inputs[img])很简单。后处理后处理指目标检测后处理是对模型输出的原始框、置信度和类别分数进行筛选与优化去除无效、重复的检测结果。它通过NMS非极大值抑制等操作保留置信度最高、位置最准确的预测框得到最终清晰可用的检测输出。这里需要指出RT-DETR与YOLO系列后处理的区别特性RT-DETRYOLO匹配策略一对多 (One-to-Many)一对一 (One-to-One)核心瓶颈无 NMS推理耗时更稳定NMS 耗时且难以在 NPU 上并行密集目标能够更精准地区分紧挨着的物体容易因为 IoU 阈值误删重叠目标DETR系列后处理使用的是离散的分配算法匈牙利算法简单来说的就是无 NMSYOLOv10系列也是该思路无锚框YOLOV5有锚框v8则分别预测位置与类别无锚框RT-DETR 模型会直接输出 300 个经过匈牙利匹配的最优预测框RT-DETR原始后处理def postprocess(self, preds, img, orig_imgs): Post-processes predictions for an image and returns them. return preds def __call__(self, sourceNone, modelNone, streamFalse, *args, **kwargs): Performs inference on an image or stream. self.stream stream if stream: return self.stream_inference(source, model, *args, **kwargs) else: return list(self.stream_inference(source, model, *args, **kwargs)) # merge list of Result into onemodel/rtdetr/predictor.py中def postprocess(self, preds, img, orig_imgs): # 1. 拆分模型输出bboxes (300个框) scores (类别分数) # RT-DETR 固定输出300个预测框匈牙利匹配已完成无冗余 nd preds[0].shape[-1] bboxes, scores preds[0].split((4, nd - 4), dim-1) results [] # 遍历每张图片的300个预测框 for i, bbox in enumerate(bboxes): # 2. 框格式转换xywh → xyxy标准检测框格式 bbox ops.xywh2xyxy(bbox) # 3. 取每个框的最高置信度 对应类别 score, cls scores[i].max(-1, keepdimTrue) # 4. 过滤低置信度框根据 conf 阈值 idx score.squeeze(-1) self.args.conf # 5. 过滤指定类别可选 if self.args.classes is not None: idx (cls torch.tensor(self.args.classes, devicecls.device)).any(1) idx pred torch.cat([bbox, score, cls], dim-1)[idx] # 6. 坐标映射把模型输出框 → 缩放到原图尺寸 orig_img orig_imgs[i] oh, ow orig_img.shape[:2] pred[..., [0, 2]] * ow pred[..., [1, 3]] * oh # 封装结果返回 results.append(Results(orig_img, pathimg_path, namesself.model.names, boxespred)) return results写后处理的逻辑就是根据源代码改写rknn代码如下def post_process(combined_output, orig_shape, conf_thres0.45): # 1. 挤掉 Batch 维度变成 [300, 14] output np.squeeze(combined_output, axis0) # 2. 严格对齐官方切片前 4 是坐标后 10 是概率得分 boxes output[:, :4] scores output[:, 4:] # ⚠️ 注意这里已经是概率了绝对不能加 sigmoid # 3. 获取每个框的最大类别和分数 class_ids np.argmax(scores, axis-1) max_scores np.max(scores, axis-1) # 4. 阈值过滤 (这里直接用 max_scores 比较) mask max_scores conf_thres valid_boxes boxes[mask] valid_scores max_scores[mask] valid_class_ids class_ids[mask] if len(valid_boxes) 0: return [] # 5. 坐标反归一化[cx, cy, w, h] (0~1) - [xmin, ymin, xmax, ymax] h_orig, w_orig orig_shape # 乘以原图尺寸 valid_boxes[:, 0] * w_orig # cx valid_boxes[:, 1] * h_orig # cy valid_boxes[:, 2] * w_orig # w valid_boxes[:, 3] * h_orig # h # cxcywh 转 xyxy x_min valid_boxes[:, 0] - valid_boxes[:, 2] / 2 y_min valid_boxes[:, 1] - valid_boxes[:, 3] / 2 x_max valid_boxes[:, 0] valid_boxes[:, 2] / 2 y_max valid_boxes[:, 1] valid_boxes[:, 3] / 2 # 裁剪到图像边界内 x_min np.clip(x_min, 0, w_orig) y_min np.clip(y_min, 0, h_orig) x_max np.clip(x_max, 0, w_orig) y_max np.clip(y_max, 0, h_orig) results [] for i in range(len(valid_boxes)): results.append({ class_id: int(valid_class_ids[i]), score: float(valid_scores[i]), box: [int(x_min[i]), int(y_min[i]), int(x_max[i]), int(y_max[i])] }) return results输出结果经测试python单NPU只有2.6帧左右。多线程推理如果采用“串行”模式读一帧 - 推一帧 - 画一帧NPU 绝大多数时间都在等待 CPU 的前后处理导致整体 FPS 只有惨淡的 2.7 帧。为了实现真正的实时性我们引入三段式接力流水线架构。Gemini 说这篇实战笔记干货满满尤其是从 Python 逻辑验证到 C 多线程工程化实现的跨越是边缘侧部署最核心的门槛。为了让你的博客更具吸引力和专业感我针对内容逻辑、排版美化以及深度扩展建议进行了全方位优化。 [RT-DETR 边缘部署实战] RK3588 前后处理优化与多线程流水线架构引言在嵌入式部署中单纯的模型推理Inference往往不是瓶颈真正的性能杀手通常隐藏在图像前后处理和硬件利用率中。本文将深入解析 RT-DETR 在 RK3588 上的高效实现逻辑并分享一套工业级的多线程异步推理框架。一、 RT-DETR 的前后处理从“一对多”到“一对一”1.1 前处理不仅仅是缩放对于 RT-DETR 模型前处理Pre-process的目标是将图像转换为模型定义的张量格式如 640×640, RGB, Float32。 避坑指南在做数据类型转换时img img.astype(np.float32) / 255.0 这一步必须严格对齐你量化模型时的设定。如果量化时开启了预处理下沉Mean/Std 在模型内这里则不需要除以 255。当前现状目前我们主要使用 OpenCV 在 CPU 上处理。但在高分辨率如 4K场景下CPU 缩放将成为瓶颈。进阶方向后续建议引入 Rockchip 官方的 RGA 硬件加速引擎实现零 CPU 负载的图像缩放与格式转换。1.2 后处理再见繁琐的 NMSRT-DETR 与传统 YOLO 系列在后处理逻辑上有着本质的区别特性 RT-DETR YOLO (v5/v8)匹配策略 一对一 (Hungarian Matching) 一对多 (Static/Dynamic Assign)核心优势 NMS-Free推理耗时极度稳定 依赖 NMS密集目标下易误删部署友好度 极高逻辑线性易于并行化 一般NMS 难以在 NPU 上加速逻辑拆解RT-DETR 默认输出 300 个预测框且每一个框都已经经过了“去重”处理。我们的后处理任务非常纯粹置信度过滤 坐标反归一化。二、 C 多线程流水线榨干 RK3588 的每一分算力如果采用“串行”模式读一帧 - 推一帧 - 画一帧NPU 绝大多数时间都在等待 CPU 的前后处理导致整体 FPS 只有惨淡的 2.7 帧。为了实现真正的实时性我们引入三段式接力流水线架构。2.1 厨房里的“三道工序”我们可以把推理流程比喻成一个高效的餐厅厨房洗菜工 (Pre-process)负责从摄像头读图、Resize、转 RGB。主厨 (NPU Infer)负责最核心的计算。RK3588 有三个核心我们可以请 3 位主厨并发炒菜。洗碗工 (Post-process)负责最后的解码和画框。为了保证各功能之间数据传递不冲突使用模板类实现了一个线程安全的队列 SafeQueue通过 std::condition_variable 实现高效的生产者-消费者模型主要多线程定义代码如下#ifndef NPU_PIPELINE_H #define NPU_PIPELINE_H #include vector #include thread #include queue #include mutex #include condition_variable #include string #include opencv2/opencv.hpp #include rknn_detector.h // 引入你的检测器 // // 1. 定义三种“集装箱”流水线各阶段传递的数据 // // 阶段一刚从摄像头/硬盘读出来的原始图像 struct RawTask { int frame_id; cv::Mat orig_img; }; // 阶段二洗好菜前处理完毕准备下锅的图像 struct NpuTask { int frame_id; cv::Mat orig_img; // 保留原图留给最后画框用 cv::Mat preprocessed_img; // 缩放并转换颜色后的图像 (640x640, RGB, Float32) }; // 阶段三炒完菜NPU 推理完毕准备装盘解码画框的数据 struct PostTask { int frame_id; cv::Mat orig_img; std::vectorfloat output_data; // NPU 吐出的裸浮点数特征必须深拷贝出来 int num_boxes; // 特征包含的框数量 }; // // 2. 泛型“传送带” (Templated SafeQueue) // 因为有三种任务我们用 template 避免写三遍重复的队列代码 // template typename T class SafeQueue { private: std::queueT queue_; std::mutex mtx_; std::condition_variable cv_; public: void push(T task) { std::unique_lockstd::mutex lock(mtx_); queue_.push(task); lock.unlock(); cv_.notify_one(); } bool pop(T task, bool is_running) { std::unique_lockstd::mutex lock(mtx_); cv_.wait(lock, [this, is_running]() { return !queue_.empty() || !is_running; }); if (!is_running queue_.empty()) { return false; } task queue_.front(); queue_.pop(); return true; } }; // // 3. 三段式接力工厂 (PipelineManager) // class PipelineManager { private: // 三条传送带 SafeQueueRawTask queue_raw_; // 接收源视频帧 SafeQueueNpuTask queue_npu_; // 接收前处理好的图像 SafeQueuePostTask queue_post_; // 接收 NPU 算完的特征矩阵 // 三组工人花名册 std::vectorstd::thread workers_pre_; std::vectorstd::thread workers_npu_; std::vectorstd::thread workers_post_; bool is_running_; std::string model_path_; // 三种工人的具体工作手册将在 .cc 中实现 void worker_preprocess(); // 洗菜工 void worker_npu_infer(int core_id); // NPU 炒菜大厨 void worker_postprocess(); // 洗碗工解码 public: // 构造函数指定各类工人的数量 PipelineManager(int num_pre, int num_npu, int num_post, const std::string model_path); ~PipelineManager(); // 留给主线程(生产者)调用的入口 void push_image(int frame_id, const cv::Mat img); }; #endif // NPU_PIPELINE_Hmain.cc:#include iostream #include string #include chrono #include opencv2/opencv.hpp #include npu_pipeline.h int main(int argc, char** argv) { if (argc 3) return -1; std::string model_path argv[1]; std::string img_path argv[2]; int loop_count (argc 4) ? std::stoi(argv[3]) : 1; cv::Mat orig_img cv::imread(img_path); if (orig_img.empty()) return -1; auto start_time std::chrono::high_resolution_clock::now(); { // 核心配置2个洗菜工3个NPU大厨1个洗碗工 PipelineManager pipeline(2, 3, 1, model_path); for (int i 0; i loop_count; i) { pipeline.push_image(i, orig_img); } std::cout - 生产者已下发 loop_count 帧任务流水线全速轰鸣中... std::endl; } // 执行到这里会触发 PipelineManager 析构阻塞直到所有帧处理完毕 auto end_time std::chrono::high_resolution_clock::now(); auto total_duration_ms std::chrono::duration_caststd::chrono::milliseconds(end_time - start_time).count(); float fps 1000.0f / (static_castfloat(total_duration_ms) / loop_count); std::cout 解耦后极限端到端系统 FPS: fps std::endl; return 0; }深度优化建议下期再优化RGA 硬件加速实战展示如何使用 /dev/rga 替代 cv2.resize释放 CPU 占用。零拷贝技术 (Zero-Copy)介绍如何通过 rknn_inputs_set 直接将 RGA 分配的物理地址传递给 NPU省去内存拷贝开销。