建立您的第一個圖

欄位	值
類別	圖形與管線
難度	初級
預估閱讀時間	5 minutes
標籤	graph, build, run, pipeline

第 001 章使用三行程式碼執行一個模型。這種便利性隱藏了一個兩部分的生命週期，每個非trivial的 Neat 程式都會直接使用：首先，您會描述一個管線，作為一個 Graph，然後將該描述建構成一個可執行的 Run。本章通過建構盡可能小的管線（一個輸入節點連接到一個輸出節點，中間沒有模型），並將單個幀推送到其中，使這個生命週期可見。

其好處在於概念上：一個 Graph 是一個可重複使用的定義，您只需建構一次，然後多次執行，而不是一次性的呼叫。到本章結束時，您將建立一個圖，將其轉換為可執行的管線，並讀取輸出張量的等級——證明該幀已成功通過。

操作指南

描述輸入

在連接節點之前，先宣告幀的外觀。InputOptions 就是這個合約：像素 format、width/height、通道 depth，以及執行階段是否會為每個緩衝區添加時間戳。從這些選項建構的輸入節點會驗證傳入的幀是否符合管線預期的形狀。

C++： C++ 額外設定 is_live = false，以標記這是一個非即時（檔案/張量）來源。

tutorials/004_build_inference_pipeline/build_inference_pipeline.cpp

simaai::neat::InputOptions in;
in.format = "RGB";
in.width = width;
in.height = height;
in.depth = 3;
in.is_live = false;
in.do_timestamp = true;

建構圖

現在建構結構。一個新的 Graph 是一個空的建構表面，而 add() 會按順序附加節點。我們添加了兩個節點——一個輸入節點（如上所述設定）和一個空白的輸出節點。這就是整個拓撲：幀從輸入端進入，從輸出端離開，中間沒有任何內容。這是稍後章節中，模型或預處理階段將插入的位置。

C++： 節點來自 simaai::neat::nodes::Input(...) 和 nodes::Output()。

Python： 節點來自 pyneat.nodes.input(...) 和 pyneat.nodes.output()。

tutorials/004_build_inference_pipeline/build_inference_pipeline.cpp

simaai::neat::Graph graph;
graph.add(simaai::neat::nodes::Input(in));
graph.add(simaai::neat::nodes::Output());

建構管線

build() 是從描述到可執行的轉換。它將添加的節點解析為一個具體的管線，根據實際樣本驗證輸入/輸出合約，並建立一個可重複使用的 Run 句柄。我們傳遞一個代表性的幀，以便 build() 可以鎖定協商後的張量形狀；下一步使用 Run::run(...) 進行確定性的逐一呼叫。

C++： 樣本幀是一個 cv::Mat，並且 run_opt.output_memory = Owned 要求執行階段傳回擁有的輸出緩衝區。

Python： 我們首先將幀轉換為一個 Tensor，來自一個 NumPy 陣列，使用 Tensor.from_numpy(...)，然後使用它進行建構。

tutorials/004_build_inference_pipeline/build_inference_pipeline.cpp

auto run = graph.build(std::vector<cv::Mat>{input}, run_opt);

執行一個幀並讀取結果

手邊有一個 Run，run() 會同步推送一個框架並拉取一個結果。由於沒有模型，因此輸出會反映輸入合約——因此，只需讀取張量的 rank，即可確認一個框架已完成整個流程。在實際的管線中，這個相同的 run()/push/pull 介面就是驅動推論的方式。

C++： run() 會傳回一個 TensorList；讀取 sample.front().shape.size()。

Python： 使用張量輸入的 run() 會傳回一個 TensorList；讀取 len(outputs[0].shape)。

tutorials/004_build_inference_pipeline/build_inference_pipeline.cpp

simaai::neat::TensorList sample = run.run(std::vector<cv::Mat>{input}, /*timeout_ms=*/1000);

執行

執行它，您應該會看到輸出張量的 rank 輸出到 stdout。從 Neat 安裝根目錄（包含 share/ 和 lib/ 的目錄）執行 Python 和 C++（預先建置） 命令；從 儲存庫根目錄執行 從原始碼建置 命令。本章不需要模型封存檔。

C++ (prebuilt):

./lib/sima-neat/tutorials/tutorial_004_build_inference_pipeline \
  --width 320 --height 240

C++ (build from source):

./build.sh --target tutorial_004_build_inference_pipeline
./build/tutorials-standalone/tutorial_004_build_inference_pipeline \
  --width 320 --height 240

預期輸出：

tensor_rank=3
[OK] 004_build_inference_pipeline

（Python 建置會輸出 output_rank=...。）若要將本章的 C++ 原始碼整合到您自己的專案中，並使用自訂的 CMakeLists.txt（不需要額外的資料夾），請參閱登陸頁面上的 如何執行教學。

實務應用

如何將 build/run、執行模式、推/拉介面，以及 RunOptions 整合在一起，一旦您超越單次同步呼叫。

建構與執行

• Graph::build(...) 建構管線，並傳回一個 Run 控點，用於推/拉控制。
• Graph::run(...) 是一個同步的便捷路徑：它會建構（如果需要），推入一個輸入，並拉出一個輸出。

同步與非同步

• 對於簡單的一次性呼叫，請使用 Graph::run(...)。
• 當您想要一個可重複使用的執行器，以及明確的 push(...) / pull(...) 控制時，請使用 Graph::build(...) — 請參閱 非同步執行推論。

推/拉 API

Run 公開：

• push(...) / try_push(...) 用於輸入（cv::Mat、Tensor 或 Sample）。
• pull(...)、pull_tensor(...)、pull_tensor_or_throw(...) 用於輸出。

如果您需要輸出中繼資料（時間戳、串流 ID），請使用 pull() 以取得 Sample。如果您只需要張量有效載荷，請使用 pull_tensor()。

RunOptions（簡單 API）

常見的設定：

• preset：延遲/安全性設定檔（Realtime、Balanced、Reliable）。
• queue_depth：執行階段佇列深度。
• overflow_policy：佇列溢出行為（Block、KeepLatest、DropIncoming）。
• output_memory：輸出擁有權原則（Auto、ZeroCopy、Owned）。
• on_input_drop：用於已捨棄輸入事件的回呼掛鉤。

對於負載下的佇列深度、溢出和測量，請參閱 調整輸送量和佇列深度。

RunAdvancedOptions（專家 API）

進階設定可在 RunOptions::advanced 下選擇性啟用：

• advanced.max_input_bytes：限制輸入緩衝區的增長。
• advanced.copy_input：強制進行防禦性輸入複製。

使用 Run::start_measurement() 在單個測量視窗中檢查延遲、輸送量、輸入計數器和外掛程式/邊緣計時。

若要在同一個視窗內記錄板子電源、溫度、CPU 使用率和記憶體，請在該視窗前後執行 TelemetryMonitor。它會從板子遙測常駐程式 Sentinel 讀取這些資料：

auto run = graph.build(inputs, run_opt);
simaai::neat::TelemetryMonitor telemetry_monitor;
telemetry_monitor.start();
auto scope = run.start_measurement();
run.push(inputs);
(void)run.pull_tensors(5000);
auto report = scope.stop();
simaai::neat::Telemetry telemetry = telemetry_monitor.stop();
std::cout << telemetry; // YAML summary

run = graph.build(tensor, run_opt)
telemetry_monitor = neat.TelemetryMonitor()
telemetry_monitor.start()
scope = run.start_measurement()
run.push(tensor)
_ = run.pull_tensors(5000)
report = scope.stop()
telemetry = telemetry_monitor.stop()
print(telemetry)  # YAML summary

Sentinel 每 2 秒取樣一次，因此短於此長度的視窗可能沒有任何樣本；此時 telemetry.status 會指出這一點。請參閱 記錄板子遙測。

完整原始碼

顯示完整原始碼程式

tutorials/004_build_inference_pipeline/build_inference_pipeline.cpp

// Build a minimal Graph (Input -> Output), run a frame, read the tensor rank.
//
// Usage:
//   tutorial_004_build_inference_pipeline [--width <w>] [--height <h>]

#include "neat.h"

#include <opencv2/core.hpp>

#include <iostream>
#include <stdexcept>
#include <string>

namespace {

bool get_arg(int argc, char** argv, const std::string& key, std::string& out) {
  for (int i = 1; i + 1 < argc; ++i) {
    if (key == argv[i]) {
      out = argv[i + 1];
      return true;
    }
  }
  return false;
}

int parse_int_arg(int argc, char** argv, const std::string& key, int def) {
  std::string value;
  if (!get_arg(argc, argv, key, value))
    return def;
  return std::stoi(value);
}

} // namespace

int main(int argc, char** argv) {
  try {
    const int width = parse_int_arg(argc, argv, "--width", 320);
    const int height = parse_int_arg(argc, argv, "--height", 240);

    cv::Mat input(height, width, CV_8UC3, cv::Scalar(30, 60, 90));
    if (!input.isContinuous())
      input = input.clone();

    simaai::neat::InputOptions in;
    in.format = "RGB";
    in.width = width;
    in.height = height;
    in.depth = 3;
    in.is_live = false;
    in.do_timestamp = true;

    simaai::neat::RunOptions run_opt;
    run_opt.output_memory = simaai::neat::OutputMemory::Owned;

    // CORE LOGIC
    // Compose a Graph from Input and Output nodes, then build+run one frame.
    simaai::neat::Graph graph;
    graph.add(simaai::neat::nodes::Input(in));
    graph.add(simaai::neat::nodes::Output());
    auto run = graph.build(std::vector<cv::Mat>{input}, run_opt);
    simaai::neat::TensorList sample = run.run(std::vector<cv::Mat>{input}, /*timeout_ms=*/1000);

    if (sample.empty())
      throw std::runtime_error("missing tensor output");
    std::cout << "tensor_rank=" << sample.front().shape.size() << "\n";
    std::cout << "[OK] 004_build_inference_pipeline\n";
    return 0;
  } catch (const std::exception& e) {
    std::cerr << "[FAIL] " << e.what() << "\n";
    return 1;
  }
}

來源

• C++
• Python
• README