多模型编排的边缘推理流水线架构设计与实现基于 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/ResizeBGR→RGBbr/归一化 to float] PRE -- MD[运动检测模型br/MobileNetV3-Sbr/输入: 224×224br/延迟: ~8ms] PRE -- OD[目标检测模型br/YOLO-Nanobr/输入: 416×416br/延迟: ~35ms] MD --|运动分数 阈值| OD MD --|运动分数 阈值| SKIP[跳帧处理] SKIP -- INPUT OD --|边界框列表| ROI[ROI裁剪节点br/根据检测框br/裁剪原始图像] ROI -- CLS1[分类模型 #1br/ResNet-18br/对ROI_1分类] ROI -- CLS2[分类模型 #2br/ResNet-18br/对ROI_2分类] ROI -- CLS_N[分类模型 #Nbr/ResNet-18br/对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::functionbool(void* input, void* output) work_fn; size_t input_size; // 输入缓冲区大小字节 size_t output_size; // 输出缓冲区大小字节 // 依赖关系 std::vectorDagNode* predecessors; // 前驱节点列表 std::vectorDagNode* successors; // 后继节点列表 // 调度状态 std::atomicint 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_guardstd::mutex lock(mtx_); stop_ true; } cv_.notify_all(); for (auto w : workers_) { if (w.joinable()) w.join(); } } void enqueue(std::functionvoid() task) { { std::lock_guardstd::mutex lock(mtx_); tasks_.push(std::move(task)); } cv_.notify_one(); } size_t pending_count() const { std::lock_guardstd::mutex lock(mtx_); return tasks_.size(); } private: void worker_loop() { while (true) { std::functionvoid() task; { std::unique_lockstd::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::vectorstd::thread workers_; std::queuestd::functionvoid() 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::functionbool(void*, void*) work_fn, size_t input_size, size_t output_size) { auto node std::make_uniqueDagNode(); 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_predecessors0的节点入队 * 2. 并行执行所有就绪节点 * 3. 每个节点完成后遍历successors并递减其pending_predecessors * 4. 任何节点pending_predecessors归零 → 入队 * 5. 循环直到所有节点处理完毕或任一节点失败 */ bool execute(size_t num_workers, void* initial_input) { SimpleThreadPool pool(num_workers); std::atomicint running_count{0}; std::atomicint completed_count{0}; std::atomicbool has_failure{false}; std::mutex ready_mutex; std::vectorDagNode* ready_nodes; const int total_nodes static_castint(nodes_.size()); // 分配节点的输入/输出缓冲区 // 生产代码中应使用内存池预分配这里简化为new/delete std::unordered_mapDagNode*, std::vectoruint8_t input_bufs; std::unordered_mapDagNode*, std::vectoruint8_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_guardstd::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_guardstd::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_castint(n-state), (unsigned long)n-elapsed_us); } } private: std::vectorstd::unique_ptrDagNode 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 的目标检测与帧 N1 的运动检测并行类似 CPU 指令流水线中的多级流水。但这要求 DAG 支持帧间状态传递和缓冲管理复杂度显著增加。五、总结基于 DAG 的多模型编排框架为边缘推理流水线提供了一种结构化的设计范式DAG 建模将前处理、检测、分类、后处理抽象为计算图节点通过拓扑排序 并行调度最大化推理吞吐。门控节点运动检测作为低计算成本的前置筛选器减少后续高成本模型的无效调用。并行分类多个 ROI 的分类任务天然可并行DAG 的多后继边语义自然支持这种模式。容错设计异常通过任一节点失败则阻塞后续的传播规则避免虚假告警。框架的局限在于当前以帧级处理为主对于需要跨帧状态跟踪的场景如多目标跟踪的 ID 分配需在节点内维护帧间状态。后续改进方向包括引入流水线并行、条件跳过语义和动态节点优先级调整——这些特性在实时视频分析场景中有直接的应用价值。