多模型编排的边缘推理流水线架构设计与实现：基于 DAG 调度器的前处理、检测、分类与后处理全链路-港品优选

多模型编排的边缘推理流水线架构设计与实现：基于 DAG 调度器的前处理、检测、分类与后处理全链路

一、单模型不够用——边缘智能场景的"模型链"需求

单模型推理已经不能满足边缘智能的工程需求。以智能安防摄像头为例，一条典型的推理链路是：运动检测（轻量模型，决定是否唤醒后续模型）→目标检测（YOLO-Nano，检出人和车的边界框）→目标分类（ResNet-18，对 ROI 区域做精细分类）→后处理（卡尔曼滤波跟踪 + 业务规则判定）。四个模型串行执行，前一个的输出是后一个的输入，形成一个有向无环图（DAG）。

在资源受限的边缘 SoC（如 RK3588 的 NPU 或算能 BM1684）上，同时加载并执行多个模型面临三个核心挑战：

内存碎片化：4 个模型的权重 + 运行时张量可能占用 200MB+ 的 DRAM，在仅有 512MB-1GB 的嵌入式内存中必须做精细的分配策略。
流水线延迟累积：串行推理的总耗时 = 各模型推理耗时之和 + 前/后处理耗时。YOLO 推理 50ms、ResNet 推理 30ms、前后处理 20ms，单帧端到端延迟已达 100ms。4 路视频流 = 400ms/frame，远超过实时要求。
异常传播与容错：当链路中某个模型超时或输出异常（如检测器漏检导致 ROI 为空），下游模型应如何响应？简单的"停止整条链路"会导致漏报，而"跳过出错模型继续"又可能产生虚假结果。

本文提出一种基于 DAG 调度器的多模型编排框架，核心思想是将推理链路建模为计算图，通过拓扑排序确定执行顺序，并利用流水线并行和指针内存池实现低延迟推理。

二、DAG 调度模型：从串行链路到并行流水

将四个模型及其前后处理建模为 DAG 节点，边的方向表示数据依赖。调度器的任务是在满足依赖关系的前提下，最大化并行度。

graph TD INPUT[输入帧 1920×1080] --> PRE[前处理节点<br/>Resize+BGR→RGB<br/>归一化 to float] PRE --> MD[运动检测模型<br/>MobileNetV3-S<br/>输入: 224×224<br/>延迟: ~8ms] PRE --> OD[目标检测模型<br/>YOLO-Nano<br/>输入: 416×416<br/>延迟: ~35ms] MD -->|运动分数 > 阈值| OD MD -->|运动分数 < 阈值| SKIP[跳帧处理] SKIP --> INPUT OD -->|边界框列表| ROI[ROI裁剪节点<br/>根据检测框<br/>裁剪原始图像] ROI --> CLS1[分类模型 #1<br/>ResNet-18<br/>对ROI_1分类] ROI --> CLS2[分类模型 #2<br/>ResNet-18<br/>对ROI_2分类] ROI --> CLS_N[分类模型 #N<br/>ResNet-18<br/>对ROI_N分类] CLS1 --> MERGE[结果聚合节点<br/>合并检测框+分类标签<br/>+置信度] CLS2 --> MERGE CLS_N --> MERGE MERGE --> POST[后处理节点<br/>卡尔曼滤波跟踪<br/>越界/逗留判定] POST --> OUTPUT[输出: 结构化告警事件] style MD fill:#fff3cd style OD fill:#e8f5e9 style CLS1 fill:#e3f2fd style CLS2 fill:#e3f2fd style CLS_N fill:#e3f2fd

这个 DAG 的拓扑特性：

运动检测和目标检测可并行执行（都只依赖前处理结果，相互独立）
多个 ROI 的分类可并行执行（N 个 ROI 互不依赖）
合并节点是同步点（必须等所有分类完成才聚合）

在优化方案中，运动检测提前执行是为了"门控"：当场景没有运动时，跳过昂贵的目标检测和分类。这一决策可节省约 40% 的推理时间（静态场景占总帧数的比例）。

三、DAG 调度器的 C++ 实现

调度器的核心数据结构是任务节点和边。每个节点封装了一个推理操作（前处理/模型推理/后处理），边的依赖关系通过原子计数器管理就绪状态。

/** * dag_scheduler.h - 有向无环图推理调度器 * * 设计要点: * 1. 每个节点维护一个"前驱完成计数"， * 当计数归零时表示该节点就绪可调度 * 2. 多个就绪节点通过线程池并行执行 * 3. 节点执行完后遍历后继节点，递减其前驱计数 * 4. 内存管理使用预分配内存池，避免运行时malloc * * 线程安全: 前驱计数操作使用原子变量， * 就绪队列使用无锁MPSC队列 */ #pragma once #include <atomic> #include <functional> #include <string> #include <vector> #include <queue> #include <memory> #include <thread> #include <mutex> #include <condition_variable> // ====== 节点状态 ====== enum class NodeState { IDLE, // 未就绪（前驱未全部完成） READY, // 就绪可调度 RUNNING, // 正在执行 COMPLETED, // 执行完成 FAILED // 执行失败 }; // ====== DAG节点定义 ====== struct DagNode { std::string name; // 工作函数：输入为前驱节点的输出数据（JSON-like variant），返回本节点输出 // 在边缘设备上建议使用 flatbuffers/protobuf 代替 variant 以减少序列化开销 std::function<bool(void* input, void* output)> work_fn; size_t input_size; // 输入缓冲区大小（字节） size_t output_size; // 输出缓冲区大小（字节） // 依赖关系 std::vector<DagNode*> predecessors; // 前驱节点列表 std::vector<DagNode*> successors; // 后继节点列表 // 调度状态 std::atomic<int> pending_predecessors{0}; // 未完成的前驱计数 NodeState state{NodeState::IDLE}; // 执行结果 bool success{false}; // 耗时统计（微秒） uint64_t elapsed_us{0}; }; // ====== 线程池（简易实现，仅用于演示） ====== class SimpleThreadPool { public: explicit SimpleThreadPool(size_t num_threads) { for (size_t i = 0; i < num_threads; i++) { workers_.emplace_back([this] { worker_loop(); }); } } ~SimpleThreadPool() { { std::lock_guard<std::mutex> lock(mtx_); stop_ = true; } cv_.notify_all(); for (auto& w : workers_) { if (w.joinable()) w.join(); } } void enqueue(std::function<void()> task) { { std::lock_guard<std::mutex> lock(mtx_); tasks_.push(std::move(task)); } cv_.notify_one(); } size_t pending_count() const { std::lock_guard<std::mutex> lock(mtx_); return tasks_.size(); } private: void worker_loop() { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(mtx_); cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); }); if (stop_ && tasks_.empty()) return; task = std::move(tasks_.front()); tasks_.pop(); } task(); } } std::vector<std::thread> workers_; std::queue<std::function<void()>> tasks_; mutable std::mutex mtx_; std::condition_variable cv_; bool stop_{false}; }; // ====== DAG 调度器 ====== class DagScheduler { public: /** * 添加节点到DAG * @param name 节点名称（用于追踪和日志） * @param work_fn 工作函数，返回false表示执行失败 * @param input_size 输入缓冲区大小 * @param output_size 输出缓冲区大小 * @return 节点指针，用于后续添加边 */ DagNode* add_node(const std::string& name, std::function<bool(void*, void*)> work_fn, size_t input_size, size_t output_size) { auto node = std::make_unique<DagNode>(); node->name = name; node->work_fn = std::move(work_fn); node->input_size = input_size; node->output_size = output_size; node->pending_predecessors.store(0); DagNode* ptr = node.get(); nodes_.push_back(std::move(node)); return ptr; } /** * 添加依赖边: from → to * 调用此函数后，to的pending_predecessors自增1 */ void add_edge(DagNode* from, DagNode* to) { from->successors.push_back(to); to->predecessors.push_back(from); to->pending_predecessors.fetch_add(1); } /** * 执行DAG中所有节点（阻塞直到全部完成或任一失败） * @param num_workers 线程池大小 * @param initial_input 入口节点的初始输入（如原始帧数据指针） * @return true 全部成功, false 有节点失败 * * 算法: 拓扑驱动的并行调度 * 1. 初始化: 将所有pending_predecessors==0的节点入队 * 2. 并行执行所有就绪节点 * 3. 每个节点完成后，遍历successors并递减其pending_predecessors * 4. 任何节点pending_predecessors归零 → 入队 * 5. 循环直到所有节点处理完毕或任一节点失败 */ bool execute(size_t num_workers, void* initial_input) { SimpleThreadPool pool(num_workers); std::atomic<int> running_count{0}; std::atomic<int> completed_count{0}; std::atomic<bool> has_failure{false}; std::mutex ready_mutex; std::vector<DagNode*> ready_nodes; const int total_nodes = static_cast<int>(nodes_.size()); // 分配节点的输入/输出缓冲区 // 生产代码中应使用内存池预分配，这里简化为new/delete std::unordered_map<DagNode*, std::vector<uint8_t>> input_bufs; std::unordered_map<DagNode*, std::vector<uint8_t>> output_bufs; for (auto& n : nodes_) { input_bufs[n.get()].resize(n->input_size); output_bufs[n.get()].resize(n->output_size); } // 阶段1: 找出所有入度为0的节点（源节点） for (auto& n : nodes_) { if (n->pending_predecessors.load() == 0) { ready_nodes.push_back(n.get()); } } // 阶段2: 主循环——持续提交就绪节点直到全部完成 while (completed_count.load() < total_nodes && !has_failure.load()) { // 提交所有就绪节点到线程池 { std::lock_guard<std::mutex> lock(ready_mutex); for (auto* node : ready_nodes) { if (node->state != NodeState::READY) continue; node->state = NodeState::RUNNING; running_count.fetch_add(1); pool.enqueue([this, node, &input_bufs, &output_bufs, &running_count, &completed_count, &has_failure, &ready_mutex, &ready_nodes]() { // 从各前驱收集输入数据（合并/最后前驱的输出） void* input = input_bufs[node].data(); if (!node->predecessors.empty()) { // 取最后一个前驱的输出作为输入 DagNode* last_pred = node->predecessors.back(); input = output_bufs[last_pred].data(); } // 执行节点的工作函数 bool ret = node->work_fn(input, output_bufs[node].data()); node->success = ret; if (!ret) { has_failure.store(true); } node->state = ret ? NodeState::COMPLETED : NodeState::FAILED; // 通知后继节点: 前驱计数减1 for (auto* succ : node->successors) { int remaining = succ->pending_predecessors.fetch_sub(1) - 1; if (remaining == 0) { // 所有前驱已完成，该后继就绪 std::lock_guard<std::mutex> lk(ready_mutex); succ->state = NodeState::READY; ready_nodes.push_back(succ); } } running_count.fetch_sub(1); completed_count.fetch_add(1); }); } ready_nodes.clear(); // 已提交完毕 } // 等待有新的就绪节点或全部完成 // 简易实现: 忙等待 + yield（生产代码应使用条件变量） while (running_count.load() > 0 && ready_nodes.empty() && completed_count.load() < total_nodes && !has_failure.load()) { std::this_thread::yield(); } // 如果有节点失败且还有在运行的节点，等待它们完成 if (has_failure.load()) { // 失败处理: 等待正在运行的节点完成 while (running_count.load() > 0) { std::this_thread::yield(); } return false; // 流水线执行失败 } } return !has_failure.load(); } /** * 打印DAG统计信息（用于性能分析） */ void print_stats() const { for (const auto& n : nodes_) { printf("[DAG] %-20s state=%d elapsed=%lu us\n", n->name.c_str(), static_cast<int>(n->state), (unsigned long)n->elapsed_us); } } private: std::vector<std::unique_ptr<DagNode>> nodes_; };

使用示例：构建四阶段推理流水线

/** * dag_pipeline_usage.cpp - 构建运动检测→目标检测→分类→跟踪的DAG流水线 */ void build_surveillance_pipeline(void) { DagScheduler scheduler; // 定义节点（实际work_fn需要封装各推理引擎的调用，此处简化为函数指针） auto* preprocess = scheduler.add_node("Preprocess", preprocess_fn, 0, 1920*1080*3); auto* motion_det = scheduler.add_node("MotionDet", motion_det_fn, 1920*1080*3, 4); auto* obj_detect = scheduler.add_node("ObjectDet", obj_detect_fn, 1920*1080*3, 1024); auto* classify_1 = scheduler.add_node("Classify#1", classify_fn, 256*256*3, 4); auto* classify_2 = scheduler.add_node("Classify#2", classify_fn, 256*256*3, 4); auto* merge = scheduler.add_node("Merge", merge_fn, 0, 4096); auto* postprocess = scheduler.add_node("Postprocess", postprocess_fn, 4096, 1024); // 构建依赖关系 scheduler.add_edge(preprocess, motion_det); // 前处理 → 运动检测 scheduler.add_edge(preprocess, obj_detect); // 前处理 → 目标检测 scheduler.add_edge(motion_det, obj_detect); // 运动检测 → 目标检测（门控） scheduler.add_edge(obj_detect, classify_1); // 目标检测 → 分类#1 scheduler.add_edge(obj_detect, classify_2); // 目标检测 → 分类#2 scheduler.add_edge(classify_1, merge); // 分类#1 → 合并 scheduler.add_edge(classify_2, merge); // 分类#2 → 合并 scheduler.add_edge(merge, postprocess); // 合并 → 后处理 // 执行: 4 个工作线程（匹配4个推理模型的并行度） bool ok = scheduler.execute(4, nullptr); scheduler.print_stats(); }

四、流水线的边界约束与工程权衡

内存池碎片化与共享策略

多个模型共用 DRAM 时的最大风险是内存碎片。当模型 A（占用 60MB 连续区域）被卸载、模型 B（需要 50MB）加载时，如果 60MB 释放出来的区域是碎片化的（如两块 30MB 的不连续区域），模型 B 的加载就会失败。解决方案是使用共享内存池——在流水线初始化时一次性分配一块大内存（如 256MB），所有模型从这个池中分配，模型卸载后内存归还池而非释放给 OS。

这一策略的代价是"峰值内存占用"可能高于"按需加载"方案（因为预分配了可能用不到的空间），但换来了零碎片的保证和可预测的内存行为——对嵌入式系统来说，可预测性通常价值更高。

异常传播策略的选择

当目标检测模型漏检（没有返回任何边界框）时，下游的分类节点应该如何响应？两种方案：

硬中断：分类节点直接返回空结果，合并节点封装为"无检测"事件。优点是逻辑清晰，缺点是下游节点仍被调度执行（浪费 CPU 周期）。
软降级：分类节点被通知"上游无输出"，自动跳过推理，返回空标记。需要 DAG 节点支持"条件跳过"语义。

本框架选择硬中断策略，因为实现简单。如果目标平台 CPU 极为紧张，建议改进为：在work_fn中检查输入是否有效，无效时直接返回 false（标记失败），由调度器终止整条链路。

流式处理 vs 帧级处理

当前 DAG 是以"帧"为粒度——每帧执行一次完整流水线。但在连续视频流场景中，帧级处理的延迟累积会导致输出帧率远低于输入帧率。改进方向是流水线并行：帧 N 的目标检测与帧 N+1 的运动检测并行，类似 CPU 指令流水线中的多级流水。但这要求 DAG 支持帧间状态传递和缓冲管理，复杂度显著增加。

五、总结

基于 DAG 的多模型编排框架为边缘推理流水线提供了一种结构化的设计范式：

DAG 建模：将前处理、检测、分类、后处理抽象为计算图节点，通过拓扑排序 + 并行调度最大化推理吞吐。
门控节点：运动检测作为低计算成本的前置筛选器，减少后续高成本模型的无效调用。
并行分类：多个 ROI 的分类任务天然可并行，DAG 的多后继边语义自然支持这种模式。
容错设计：异常通过"任一节点失败则阻塞后续"的传播规则，避免虚假告警。

框架的局限在于当前以帧级处理为主，对于需要跨帧状态跟踪的场景（如多目标跟踪的 ID 分配），需在节点内维护帧间状态。后续改进方向包括引入流水线并行、条件跳过语义和动态节点优先级调整——这些特性在实时视频分析场景中有直接的应用价值。

企业官网建设流程全解析